final changes and tests

Author: YunchuWang

azure_functions_worker/constants.py

@@ -63,7 +63,7 @@
 METADATA_PROPERTIES_WORKER_INDEXED = "worker_indexed"
 
 # HostNames
-LOCAL_HOST = "localhost"
+LOCAL_HOST = "127.0.0.1"
 
 # Header names
 X_MS_INVOCATION_ID = "x-ms-invocation-id"

azure_functions_worker/dispatcher.py

@@ -91,6 +91,7 @@ def __init__(self, loop: BaseEventLoop, host: str, port: int,
         self._shmem_mgr = SharedMemoryManager()
         self._old_task_factory = None
         self.function_metadata_result = None
+        self._has_http_func = False
 
         # Used to store metadata returns
         self._function_metadata_result = None
@@ -516,6 +517,7 @@ async def _handle__function_load_request(self, request):
     async def _handle__invocation_request(self, request):
         invocation_time = datetime.utcnow()
         invoc_request = request.invocation_request
+        trigger_metadata = invoc_request.trigger_metadata
         invocation_id = invoc_request.invocation_id
         function_id = invoc_request.function_id
 
@@ -560,13 +562,19 @@ async def _handle__invocation_request(self, request):
                     pytype=pb_type_info.pytype,
                     shmem_mgr=self._shmem_mgr)
 
+            http_v2_enabled = False
             if fi.trigger_metadata.get('type') == HTTP_TRIGGER:
                 from azure.functions.extension.base import HttpV2FeatureChecker
                 http_v2_enabled = HttpV2FeatureChecker.http_v2_enabled()
 
             if http_v2_enabled:
                 http_request = await http_coordinator.get_http_request_async(
                     invocation_id)
+                
+                from azure.functions.extension.base import RequestTrackerMeta
+                route_params = {key: item.string for key, item in trigger_metadata.items() if key not in ['Headers', 'Query']}
+
+                RequestTrackerMeta.get_synchronizer().sync_route_params(http_request, route_params)
                 args[fi.trigger_metadata.get('param_name')] = http_request
 
             fi_context = self._get_context(invoc_request, fi.name, fi.directory)
@@ -581,23 +589,26 @@ async def _handle__invocation_request(self, request):
                 for name in fi.output_types:
                     args[name] = bindings.Out()
 
-            if fi.is_async:
-                call_result = await self._run_async_func(
-                    fi_context, fi.func, args
-                )
-
-            else:
-                call_result = await self._loop.run_in_executor(
-                    self._sync_call_tp,
-                    self._run_sync_func,
-                    invocation_id, fi_context, fi.func, args)
-
-            if call_result is not None and not fi.has_return:
-                raise RuntimeError(f'function {fi.name!r} without a $return '
-                                   'binding returned a non-None value')
-            
-            if http_v2_enabled:
-                http_coordinator.set_http_response(invocation_id, call_result)
+            call_result = None
+            call_error = None
+            try:
+                if fi.is_async:
+                    call_result = await self._run_async_func(fi_context, fi.func, args)
+                else:
+                    call_result = await self._loop.run_in_executor(
+                        self._sync_call_tp,
+                        self._run_sync_func,
+                        invocation_id, fi_context, fi.func, args)
+                
+                if call_result is not None and not fi.has_return:
+                    raise RuntimeError(f'function {fi.name!r} without a $return '
+                                    'binding returned a non-None value')
+            except Exception as e:
+                call_error = e
+                raise
+            finally:
+                if http_v2_enabled:
+                    http_coordinator.set_http_response(invocation_id, call_result if call_result is not None else call_error)
 
             output_data = []
             cache_enabled = self._function_data_cache_enabled
@@ -753,16 +764,23 @@ async def catch_all(request: request_type): # type: ignore
             invoc_id = request.headers.get(X_MS_INVOCATION_ID)
             if invoc_id is None:
                 raise ValueError(f"Header {X_MS_INVOCATION_ID} not found")
-
+            logger.info('Received HTTP request for invocation %s', invoc_id)
             http_coordinator.set_http_request(invoc_id, request)
             http_resp = await http_coordinator.await_http_response_async(invoc_id)
+
+            logger.info('Sending HTTP response for invocation %s', invoc_id)
+            # if http_resp is an python exception, raise it
+            if isinstance(http_resp, Exception):
+                raise http_resp
+            
             return http_resp
 
         web_server = web_server_class(LOCAL_HOST, unused_port, app)
         web_server_run_task = web_server.serve()
 
         loop = asyncio.get_event_loop()
         loop.create_task(web_server_run_task)
+        logger.info('HTTP server starting on %s:%s', LOCAL_HOST, unused_port)
 
         return f"http://{LOCAL_HOST}:{unused_port}"
 
@@ -779,7 +797,6 @@ def index_functions(self, function_path: str):
                 indexed_functions)
 
             indexed_function_logs: List[str] = []
-            self._has_http_func = False
             for func in indexed_functions:
                 self._has_http_func = self._has_http_func or func.is_http_function()
                 function_log = "Function Name: {}, Function Binding: {}" \

azure_functions_worker/http_proxy.py

@@ -102,14 +102,12 @@ def set_http_request(self, invoc_id, http_request):
             self._context_references[invoc_id] = AsyncContextReference()
         context_ref = self._context_references.get(invoc_id)
         context_ref.http_request = http_request
-        context_ref.http_request_available_event.set()
 
     def set_http_response(self, invoc_id, http_response):
         if invoc_id not in self._context_references:
             raise Exception("No context reference found for invocation %s", invoc_id)
         context_ref = self._context_references.get(invoc_id)
         context_ref.http_response = http_response
-        context_ref.http_response_available_event.set()
 
     async def get_http_request_async(self, invoc_id):
         if invoc_id not in self._context_references:
@@ -141,8 +139,7 @@ def _pop_http_response(self, invoc_id):
         if response is not None:
             context_ref.http_response = None
             return response
-        
-        raise Exception("No http response found for invocation %s", invoc_id)
+        # If user does not set the response, return nothing and web server will return 200 empty response
 
 
 http_coordinator = HttpCoordinator()

azure_functions_worker/loader.py

@@ -146,7 +146,7 @@ def process_indexed_function(functions_registry: functions.Registry,
                 properties={METADATA_PROPERTIES_WORKER_INDEXED: "True"})
 
             fx_metadata_results.append(function_metadata)
-            return fx_metadata_results
+        return fx_metadata_results
     except Exception as e:
         logger.error(f'Error in process_indexed_function. {e}', exc_info=True)
         raise e

azure_functions_worker/logging.py

@@ -20,8 +20,8 @@
 
 handler: Optional[logging.Handler] = None
 error_handler: Optional[logging.Handler] = None
-local_handler = logging.FileHandler("E:/projects/AzureFunctionsPythonWorker/log.txt")
-logger.addHandler(local_handler)
+# local_handler = logging.FileHandler("E:/projects/AzureFunctionsPythonWorker/log.txt")
+# logger.addHandler(local_handler)
 
 def format_exception(exception: Exception) -> str:
     msg = str(exception) + "\n"

python/test/worker.config.json

@@ -2,7 +2,7 @@
     "description":{
         "language":"python",
         "extensions":[".py"],
-        "defaultExecutablePath":"E:\\projects\\AzureFunctionsPythonWorker\\.venv_3.8\\Scripts\\python.exe",
+        "defaultExecutablePath":"E:\\projects\\AzureFunctionsPythonWorker\\.venv_3.9_debug\\Scripts\\python.exe",
         "defaultWorkerPath":"worker.py",
         "workerIndexing": "true"
     }

setup.py

@@ -110,7 +110,7 @@
         "numpy",
         "pre-commit"
     ],
-    "fastapi": ["azure-functions-extension-fastapi"]
+    "http-v2": ["azure-functions-extension-fastapi", "ujson", "orjson"]
 }
 
 

tests/consumption_tests/test_linux_consumption.py

@@ -336,6 +336,41 @@ def test_reload_variables_after_oom_error(self):
             self.assertNotIn("Failure Exception: ModuleNotFoundError",
                              logs)
 
+    @skipIf(sys.version_info.minor != 10,
+            "This is testing only for python310")
+    def test_http_v2_fastapi_streaming_upload_download(self):
+        """
+        A function app with init indexing enabled
+        """
+        import random as rand
+        with LinuxConsumptionWebHostController(_DEFAULT_HOST_VERSION,
+                                               self._py_version) as ctrl:
+            ctrl.assign_container(env={
+                "AzureWebJobsStorage": self._storage,
+                "SCM_RUN_FROM_PACKAGE": self._get_blob_url("HttpV2FastApiStreaming"),
+                PYTHON_ENABLE_INIT_INDEXING: "true",
+                PYTHON_ISOLATE_WORKER_DEPENDENCIES: "1"
+            })
+
+            def generate_random_bytes_stream():
+                """Generate a stream of random bytes."""
+                yield b'streaming'
+                yield b'testing'
+                yield b'response'
+                yield b'is'
+                yield b'returned'
+
+            req = Request('POST', f'{ctrl.url}/api/http_v2_fastapi_streaming', data=generate_random_bytes_stream())
+            resp = ctrl.send_request(req)
+            self.assertEqual(resp.status_code, 200)
+
+            streamed_data = b''
+            for chunk in resp.iter_content(chunk_size=1024):
+                if chunk:
+                    streamed_data+= chunk
+
+            self.assertEqual(streamed_data, b'streamingtestingresponseisreturned')
+
     def _get_blob_url(self, scenario_name: str) -> str:
         return (
             f'https://pythonworker{self._py_shortform}sa.blob.core.windows.net/'

tests/endtoend/http_functions/http_functions_v2/fastapi/file_name/main.py

@@ -0,0 +1,48 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from datetime import datetime
+import logging
+import time
+
+import azure.functions as func
+
+from azure.functions.extension.fastapi import Request, Response, StreamingResponse, \
+    HTMLResponse, PlainTextResponse, HTMLResponse, JSONResponse, \
+    UJSONResponse, ORJSONResponse, RedirectResponse, FileResponse
+
+app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)
+
+
+@app.route(route="default_template")
+async def default_template(req: Request) -> Response:
+    logging.info('Python HTTP trigger function processed a request.')
+
+    name = req.query_params.get('name')
+    if not name:
+        try:
+            req_body = await req.json()
+        except ValueError:
+            pass
+        else:
+            name = req_body.get('name')
+
+    if name:
+        return Response(
+            f"Hello, {name}. This HTTP triggered function "
+            f"executed successfully.")
+    else:
+        return Response(
+            "This HTTP triggered function executed successfully. "
+            "Pass a name in the query string or in the request body for a"
+            " personalized response.",
+            status_code=200
+        )
+
+
+@app.route(route="http_func")
+def http_func(req: Request) -> Response:
+    time.sleep(1)
+
+    current_time = datetime.now().strftime("%H:%M:%S")
+    return Response(f"{current_time}")

tests/endtoend/http_functions/http_functions_v2/fastapi/function_app.py

@@ -0,0 +1,91 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+import asyncio
+from datetime import datetime
+import logging
+import time
+
+import azure.functions as func
+from azure.functions.extension.fastapi import Request, Response, StreamingResponse, \
+    HTMLResponse, PlainTextResponse, HTMLResponse, JSONResponse, \
+    UJSONResponse, ORJSONResponse, RedirectResponse, FileResponse
+
+app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)
+
+
+@app.route(route="default_template")
+async def default_template(req: Request) -> Response:
+    logging.info('Python HTTP trigger function processed a request.')
+
+    name = req.query_params.get('name')
+    if not name:
+        try:
+            req_body = await req.json()
+        except ValueError:
+            pass
+        else:
+            name = req_body.get('name')
+
+    if name:
+        return Response(
+            f"Hello, {name}. This HTTP triggered function "
+            f"executed successfully.")
+    else:
+        return Response(
+            "This HTTP triggered function executed successfully. "
+            "Pass a name in the query string or in the request body for a"
+            " personalized response.",
+            status_code=200
+        )
+
+
+@app.route(route="http_func")
+def http_func(req: Request) -> Response:
+    time.sleep(1)
+
+    current_time = datetime.now().strftime("%H:%M:%S")
+    return Response(f"{current_time}")
+
+
+@app.route(route="upload_data_stream")
+async def upload_data_stream(req: Request) -> Response:
+    # Define a list to accumulate the streaming data
+    data_chunks = []
+
+    async def process_stream():
+        async for chunk in req.stream():
+            # Append each chunk of streaming data to the list
+            data_chunks.append(chunk)
+
+    await process_stream()
+
+    # Concatenate the data chunks to form the complete data
+    complete_data = b"".join(data_chunks)
+
+    # Return the complete data as the response
+    return Response(content=complete_data, status_code=200)
+
+
+@app.route(route="return_streaming")
+async def return_streaming(req: Request) -> StreamingResponse:
+    async def content():
+        yield b"First chunk\n"
+        yield b"Second chunk\n"
+    return StreamingResponse(content())
+
+@app.route(route="return_html")
+def return_html(req: Request) -> HTMLResponse:
+    html_content = "<html><body><h1>Hello, World!</h1></body></html>"
+    return HTMLResponse(content=html_content, status_code=200)
+
+@app.route(route="return_ujson")
+def return_ujson(req: Request) -> UJSONResponse:
+    return UJSONResponse(content={"message": "Hello, World!"}, status_code=200)
+
+@app.route(route="return_orjson")
+def return_orjson(req: Request) -> ORJSONResponse:
+    return ORJSONResponse(content={"message": "Hello, World!"}, status_code=200)
+
+@app.route(route="return_file")
+def return_file(req: Request) -> FileResponse:
+    return FileResponse("function_app.py")