Configure default executor to use --parallel setting.

changelog/pending/20221022--sdk-python--python-runtime-now-respects-parallel-flag.yaml

@@ -0,0 +1,4 @@
+changes:
+- type: fix
+  scope: sdk/python
+  description: Python runtime now respects the --parallel flag.

sdk/python/cmd/pulumi-language-python-exec

@@ -3,11 +3,13 @@
 
 import argparse
 import asyncio
+from typing import Optional
 import logging
 import os
 import sys
 import traceback
 import runpy
+from concurrent.futures import ThreadPoolExecutor
 
 # The user might not have installed Pulumi yet in their environment - provide a high-quality error message in that case.
 try:
@@ -51,6 +53,13 @@ def _get_user_stacktrace(user_program_abspath: str) -> str:
     # we did not detect a __main__ program, return normal traceback
     return traceback.format_exc()
 
+def _set_default_executor(loop, parallelism: Optional[int]):
+    '''configure this event loop to respect the settings provided.'''
+    if parallelism is None:
+        return
+    exec = ThreadPoolExecutor(max_workers=parallelism)
+    loop.set_default_executor(exec)
+
 if __name__ == "__main__":
     # Parse the arguments, program name, and optional arguments.
     ap = argparse.ArgumentParser(description='Execute a Pulumi Python program')
@@ -125,6 +134,9 @@ if __name__ == "__main__":
         except RuntimeError:
             loop = asyncio.new_event_loop()
             asyncio.set_event_loop(loop)
+
+    # Configure the event loop to respect the parallelism value provided as input.
+    _set_default_executor(loop, settings.parallel)
 
     # We are (unfortunately) suppressing the log output of asyncio to avoid showing to users some of the bad things we
     # do in our programming model.

sdk/python/lib/pulumi/runtime/settings.py

@@ -208,10 +208,6 @@ def get_engine() -> Optional[Union[engine_pb2_grpc.EngineStub, Any]]:
     """
     return SETTINGS.engine
 
-
-ROOT: ContextVar[Optional[Resource]] = ContextVar("root_resource", default=None)
-
-
 def get_root_resource() -> Optional["Resource"]:
     """
     Returns the implicit root stack resource for all resources created in this program.
@@ -226,6 +222,9 @@ def set_root_resource(root: "Resource"):
     ROOT.set(root)
 
 
+ROOT: ContextVar[Optional[Resource]] = ContextVar("root_resource", default=None)
+
+
 async def monitor_supports_feature(feature: str) -> bool:
     if feature not in SETTINGS.feature_support:
         monitor = SETTINGS.monitor