fix: propagate context in ce fns.

Author: taeold

src/functions_framework/aio/__init__.py

@@ -39,6 +39,7 @@
 try:
     from starlette.applications import Starlette
     from starlette.exceptions import HTTPException
+    from starlette.middleware import Middleware
     from starlette.requests import Request
     from starlette.responses import JSONResponse, Response
     from starlette.routing import Route
@@ -150,9 +151,9 @@ async def handler(request):
             await function(event)
         else:
             # TODO: Use asyncio.to_thread when we drop Python 3.8 support
-            # Python 3.8 compatible version of asyncio.to_thread
             loop = asyncio.get_event_loop()
-            await loop.run_in_executor(None, function, event)
+            ctx = contextvars.copy_context()
+            await loop.run_in_executor(None, ctx.run, function, event)
         return Response("OK")
 
     return handler
@@ -295,8 +296,6 @@ def create_asgi_app(target=None, source=None, signature_type=None):
             f"Unsupported signature type for ASGI server: {signature_type}"
         )
 
-    from starlette.middleware import Middleware
-
     app = Starlette(
         debug=False,
         routes=routes,

tests/test_execution_id_async.py

@@ -295,23 +295,47 @@ async def test_maintains_execution_id_for_concurrent_requests(monkeypatch, capsy
 
 
 def test_async_decorator_with_sync_function():
-    """Test that the async decorator handles sync functions properly."""
-
-    # Create a sync function
     def sync_func(request):
         return {"status": "ok"}
 
-    # Apply the decorator
     wrapped = execution_id.set_execution_context_async(enable_id_logging=False)(
         sync_func
     )
 
-    # Create mock request
     request = Mock()
     request.headers = Mock()
     request.headers.get = Mock(return_value="")
 
-    # Call the wrapped function - it should be sync since the original was sync
     result = wrapped(request)
 
     assert result == {"status": "ok"}
+
+
+def test_sync_cloudevent_function_has_execution_context(monkeypatch, capsys):
+    """Test that sync CloudEvent functions can access execution context."""
+    monkeypatch.setenv("LOG_EXECUTION_ID", "true")
+
+    source = TEST_FUNCTIONS_DIR / "execution_id" / "async_main.py"
+    target = "sync_cloudevent_with_context"
+    app = create_asgi_app(target, source, signature_type="cloudevent")
+    client = TestClient(app)
+
+    response = client.post(
+        "/",
+        headers={
+            "ce-specversion": "1.0",
+            "ce-type": "com.example.test",
+            "ce-source": "test-source",
+            "ce-id": "test-id",
+            "Function-Execution-Id": TEST_EXECUTION_ID,
+            "Content-Type": "application/json",
+        },
+        json={"message": "test"},
+    )
+
+    assert response.status_code == 200
+    assert response.text == "OK"
+
+    record = capsys.readouterr()
+    assert f"Execution ID in sync CloudEvent: {TEST_EXECUTION_ID}" in record.err
+    assert "No execution context in sync CloudEvent function!" not in record.err

tests/test_functions/execution_id/async_main.py

@@ -36,7 +36,6 @@ async def async_sleep(request):
 
 
 async def async_trace_test(request):
-    # Get current execution context
     context = execution_id._get_current_context()
     return {
         "execution_id": context.execution_id if context else None,
@@ -51,3 +50,12 @@ def sync_function_in_async_context(request):
         "execution_id": request.headers.get("Function-Execution-Id"),
         "type": "sync",
     }
+
+
+def sync_cloudevent_with_context(cloud_event):
+    """A sync CloudEvent function that accesses execution context."""
+    context = execution_id._get_current_context()
+    if context:
+        logger.info(f"Execution ID in sync CloudEvent: {context.execution_id}")
+    else:
+        logger.error("No execution context in sync CloudEvent function!")