feat: Adding support for python 3.13 by gavin-aguiar · Pull Request #1666 · Azure/azure-functions-python-worker

gavin-aguiar · 2025-04-11T16:49:53Z

Description

Proxy Worker Changes (merge to dev)

Adds 3.13 to the build & test pipelines
- For python 3.12, the builds include azure_functions_worker ONLY
- For python 3.13+, the builds include proxy_worker ONLY
- We reuse the existing build / testing structure from the current worker, but we make some logical changes to test afw or proxy accordingly based on python version
Proxy_worker: receives & returns gRPC messages from the host
- Majority of the worker functionality is defined in the library worker. The proxy worker will solely receive messages from the host, forward them to the library worker, and respond back.
- Both the V1 and V2 library worker are included in the image. The proxy worker decides what to import based on the function path.

Fixes #

PR information

The title of the PR is clear and informative.
There are a small number of commits, each of which has an informative message. This means that previously merged commits do not appear in the history of the PR. For information on cleaning up the commits in your pull request, see this page.
If applicable, the PR references the bug/issue that it fixes in the description.
New Unit tests were added for the changes made and CI is passing.

Quality of Code and Contribution Guidelines

I have read the contribution guidelines.

…into gaaguiar/proxy_worker

* build: recognize collection_model_binding_data for batch inputs (#1655) * add cmbd * Add * Add * Rm newline * Add tests * Fix cmbd * Fix test * Lint * Rm * Rm * Add back newline * rm ws * Rm list * Rm cmbd from cache * Avoid caching * Keep cmbd check * Add comment * Lint --------- Co-authored-by: Evan Roman <evanroman@microsoft.com> Co-authored-by: hallvictoria <59299039+hallvictoria@users.noreply.github.com> * build: update Python Worker Version to 4.36.1 (#1660) Co-authored-by: AzureFunctionsPython <funcdisc@microsoft.com> * initial changes * Update Python SDK Version to 1.23.0 (#1663) Co-authored-by: AzureFunctionsPython <azfunc@microsoft.com> * merges from ADO * merge fixes * merge fixes * merge fixes * merge fixes * don't run 313 unit tests yet * changes for builds --------- Co-authored-by: Evan <66287338+EvanR-Dev@users.noreply.github.com> Co-authored-by: Evan Roman <evanroman@microsoft.com> Co-authored-by: AzureFunctionsPython <funcdisc@microsoft.com> Co-authored-by: AzureFunctionsPython <azfunc@microsoft.com>

swapnil-nagar · 2025-04-23T03:38:44Z

-@unittest.skipIf(sys.version_info.minor < 13, "For python 3.13+,"
-                                              "this logic is in the"
-                                              "library worker.")
+@unittest.skipIf(sys.version_info.minor >= 13, "For python 3.13+,"


Use a constant to defined the supported python version to minimize the changes in the future.

swapnil-nagar · 2025-04-23T03:41:35Z

+          # Extract minor version as integers
+          PY_MINOR="${PY_VER#*.}"
+
+          if [ "$PY_MINOR" -ge 13 ]  then


NIT: Consider having a file that contains the supported version of python and used at both ci-unit-tests and ci-e2e-tests.yml file.

swapnil-nagar · 2025-04-23T03:43:31Z

-@unittest.skipIf(sys.version_info.minor < 13, "For python 3.13+,"
-                                              "this logic is in the"
-                                              "library worker.")
+@unittest.skipIf(sys.version_info.minor >= 13, "For python 3.13+,"


[Checking] You are skipping tests for 3.13 is that intended?

vrdmr · 2025-04-14T17:15:14Z

Why any change in this?

vrdmr · 2025-04-14T17:16:39Z

        workerPath: 'python/prodV4/worker.py'
+      Python313V4:
+        pythonVersion: '3.13'
+        workerPath: 'python/proxyV4/worker.py'


Can the worker Path be a variable at the top?

vrdmr · 2025-04-23T14:28:00Z

          EVENTGRID_URI: $(LinuxEventGridTopicUriString312)
          EVENTGRID_CONNECTION: $(LinuxEventGridConnectionKeyString312)
+        Python313:
+          PYTHON_VERSION: '3.13'


E2E tests are re-using 312 resources for now until the appropriate 313 resources are created (cosmosdb, sql, eventgrid).

For this - is there a helper script to smoothen the process of creation? If not, please add it in the 3.13 backlog.

vrdmr · 2025-04-23T14:28:37Z

@@ -100,6 +106,9 @@ jobs:
      Python312V4:


In the long run - We should simplify these blocks - the yaml is unnecessarily long.

vrdmr · 2025-04-23T14:29:30Z

            Write-Host "##vso[task.setvariable variable=skipTest;]false"
          }
        displayName: 'Set skipTest variable'
-        condition: or(eq(variables.isSdkRelease, true), eq(variables['USETESTPYTHONSDK'], true))


What did this do?

vrdmr · 2025-04-23T14:30:39Z

+if [[ $version_minor -lt 13 ]]; then
+    cp -r azure_functions_worker/protos "$BUILD_SOURCESDIRECTORY/deps/azure_functions_worker"
+else
+    cp -r proxy_worker/protos "$BUILD_SOURCESDIRECTORY/deps/proxy_worker"


in the long run - the name shouldn't change - can it be done in this iteration itself?

As this seems unnecessary.

vrdmr · 2025-04-23T18:06:17Z

-    UNIX_SHARED_MEMORY_DIRECTORIES,
-)
-from azure_functions_worker.utils.common import get_app_setting, is_envvar_true
+if sys.version_info.minor < 13:


I really don't like this :( Better to create a 3.13 specific util for now.

vrdmr · 2025-04-23T18:08:00Z

            subprocess.run(
-                [sys.executable, '-m', 'flake8', '--config', str(config_path)],
+                [sys.executable, '-m', 'flake8', '--config', str(config_path),
+                 'azure_functions_worker',],


Would it not "flake" other folders?

vrdmr · 2025-04-23T18:09:04Z

 if __name__ == '__main__':
    add_script_root_to_sys_path()
-    main.main()
+    minor_version = sys.version_info[1]


We wanted to get out of the whole minor version thing - Please create a different test folder if needed.

vrdmr · 2025-04-23T18:10:23Z

Why create a separate worker.py?

vrdmr · 2025-04-23T18:11:05Z

+    # third-party user packages over worker packages in PYTHONPATH
+    user_pkg_paths = determine_user_pkg_paths()
+    joined_pkg_paths = os.pathsep.join(user_pkg_paths)
+    env['PYTHONPATH'] = f'{joined_pkg_paths}:{func_worker_dir}'


This needs a documentation update

vrdmr · 2025-04-23T18:15:56Z

+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+VERSION = "4.36.1"


Any specific reason it should be in a python file or would it be in the pyproject.toml itself

vrdmr · 2025-04-23T18:17:22Z


-from azure_functions_worker import protos
-from azure_functions_worker.bindings import datumdef, meta
+if sys.version_info.minor < 13:


Still sad :(

vrdmr · 2025-04-23T18:49:14Z

+    logger.info("Args: %s", args)
+    logger.info('Starting proxy worker.')
+    logger.info('Worker ID: %s, Request ID: %s, Host Address: %s:%s',
+                args.worker_id, args.request_id, args.host, args.port)


Change to 1 or debug logs

vrdmr · 2025-04-23T18:53:18Z

+    return getattr(_invocation_id_local, 'invocation_id', None)
+
+
+class AsyncLoggingHandler(logging.Handler):


Exploration for later - to move to logging

vrdmr · 2025-04-23T18:58:29Z

+        return int(max_workers) if max_workers else None
+
+    async def _handle__worker_init_request(self, request):
+        logger.info('Received WorkerInitRequest, '


the new logs added are intended to help confirm that the library worker is receiving requests appropriately.

No new info logs - make them debug. During cold start, adding a log should be fine but post Evan's validation.

vrdmr · 2025-04-23T19:02:03Z

+
+        global _library_worker
+        directory = request.worker_init_request.function_app_directory
+        v2_directory = os.path.join(directory, get_script_file_name())


Rename - make it explicit that its a function_app entry file for v2 model

vrdmr · 2025-04-23T19:19:02Z

+        v2_directory = os.path.join(directory, get_script_file_name())
+        if os.path.exists(v2_directory):
+            try:
+                import azure_functions_worker_v2  # NoQA


Explore to import this in a separate variable above.

x = importlib.import (azure_functions_worker_v2)

x.handle_dispatch_init in your code.

Decide if you want to run import lib again based on a return DependencyManager.prioritize_customer_dependencies(directory) if it tells you that Cx did bring in a runtime in reqs.txt

swapnil-nagar · 2025-04-23T19:11:06Z

+
+    def on_logging(self, record: logging.LogRecord,
+                   formatted_msg: str) -> None:
+        if record.levelno >= logging.CRITICAL:


Thats a lot uses an if-elif chain to map log levels, consider having a map defined to clean the code.

Can this be switch case itself?

swapnil-nagar · 2025-04-23T19:14:06Z

+    async def _dispatch_grpc_request(self, request):
+        content_type = request.WhichOneof("content")
+
+        match content_type:


Instead of if-else, use a dictionary

So something like

handlers = handlers["content_type"]

And handlers is a dict [<content_type:str>, <func_point:addr>]

swapnil-nagar · 2025-04-23T19:16:42Z

+            self._grpc_resp_queue.put_nowait(self._GRPC_STOP_RESPONSE)
+            self._grpc_thread.join()
+            self._grpc_thread = None
+


[Blocker] You are joining thread and cleanup, there could be exception from these which will left resources in open(or undefined state). I will encourage use to exception and finally to ensure thread safety.

which will left resources in open(or undefined state)

This is already at the end of the workers lifetime - and we can discard any state.

swapnil-nagar · 2025-04-23T19:22:21Z

+        try:
+            forever = self._loop.create_future()
+
+            self._grpc_resp_queue.put_nowait(


[Checking, unaware of python data structures] If multiple thread attempt to put in the queue is that thread safe. We use something called BlockingQueue or ConcurrentQueue in Java

swapnil-nagar · 2025-04-23T19:23:48Z

+
+    def stop(self) -> None:
+        if self._grpc_thread is not None:
+            self._grpc_resp_queue.put_nowait(self._GRPC_STOP_RESPONSE)


Also you should consider having lock, if two thread access this code simultaneously, it will lead to race cndition.

swapnil-nagar · 2025-04-23T19:25:30Z

+        try:
+            for req in grpc_req_stream:
+                self._loop.call_soon_threadsafe(
+                    self._loop.create_task, self._dispatch_grpc_request(req))


[Checking] If _dispatch_grpc_request itself is thread safe? You might need to synchronize the _dispatch_grpc_request

swapnil-nagar · 2025-04-23T19:52:14Z

+    if logger and handler:
+        handler.flush()
+        logger.removeHandler(handler)
+


[Checking] Should you need close handler?

…ons-python-worker into gaaguiar/proxy_worker

vrdmr · 2025-04-25T17:09:07Z

+    logger.info("Args: %s", args)
+    logger.info(
+        'Starting proxy worker. Worker ID: %s, Request ID: %s, Host Address: %s:%s',
+        args.worker_id, args.request_id, args.host, args.port)


Merging these logs

vrdmr · 2025-04-25T17:14:33Z

+        cls._add_to_sys_path(cls.cx_deps_path, True)
+        cls._add_to_sys_path(working_directory, False)
+
+        logger.info(f'Finished prioritize_customer_dependencies: {sys.path}')


Please convert to debug (when GA)

vrdmr · 2025-04-25T17:15:46Z

+        cls._add_to_sys_path(cls.worker_deps_path, True)
+        cls._add_to_sys_path(cls.cx_deps_path, True)
+        cls._add_to_sys_path(working_directory, False)


Can adding to first or last be more explicit - readable

vrdmr · 2025-04-25T17:17:02Z

+        logger.info(f'Finished prioritize_customer_dependencies: {sys.path}')
+
+    @classmethod
+    def _add_to_sys_path(cls, path: str, add_to_first: bool):


move private methods to be earlier than public methods.

vrdmr · 2025-04-25T17:17:37Z

+    @staticmethod
+    def _get_cx_deps_path() -> str:
+        """Get the directory storing the customer's third-party libraries.
+
+        Returns
+        -------
+        str
+            Core Tools: path to customer's site packages
+            Linux Dedicated/Premium: path to customer's site packages
+            Linux Consumption: empty string
+        """
+        prefix: Optional[str] = os.getenv(AZURE_WEBJOBS_SCRIPT_ROOT)
+        cx_paths: List[str] = [
+            p for p in sys.path
+            if prefix and p.startswith(prefix) and ('site-packages' in p)
+        ]
+        # Return first or default of customer path
+        return (cx_paths or [''])[0]


Why is "AZURE_WEBJOBS_SCRIPT_ROOT" showing up here?

First - explore using the dir from grpc request itself. or use cls.Cx_working_directory

vrdmr · 2025-04-25T17:21:53Z

+            The worker packages path
+        """
+        # 1. Try to parse the absolute path python/3.13/LINUX/X64 in sys.path
+        r = re.compile(r'.*python(\/|\\)\d+\.\d+(\/|\\)(WINDOWS|LINUX|OSX).*')


This Regex should be on the top as Constant

vrdmr · 2025-04-25T17:22:21Z

+        # Don't reload proxy_worker
+        to_be_cleared_from_cache = set([
+            module_name for module_name in not_builtin
+            if not module_name.startswith('proxy_worker')


Name - shouldn't be hardcoded. Should be moved to Constant.

vrdmr · 2025-04-25T17:34:50Z

+
+# App Setting constants
+PYTHON_ENABLE_DEBUG_LOGGING = "PYTHON_ENABLE_DEBUG_LOGGING"
+PYTHON_THREADPOOL_THREAD_COUNT = "PYTHON_THREADPOOL_THREAD_COUNT"


Proxy worker shouldn't create the TP. Thus this constant shouldn't be here.

vrdmr · 2025-04-25T17:36:00Z

+        self._sync_call_tp: Optional[concurrent.futures.Executor] = (
+            self._create_sync_call_tp(self._get_sync_tp_max_workers()))
+


We should explore - moving this logic into lib_worker and not to put in proxy worker.

vrdmr · 2025-04-25T17:38:17Z

+    @classmethod
+    async def connect(cls, host: str, port: int, worker_id: str,
+                      request_id: str, connect_timeout: float):
+        loop = asyncio.events.get_event_loop()


Add a task - app setting for enabling "uvloop" - add this feature.

uvloop.new_event_loop vs asyncio.events.get_event_loop()

vrdmr · 2025-04-25T17:43:30Z

+    async def _dispatch_grpc_request(self, request):
+        content_type = request.WhichOneof("content")
+
+        match content_type:


So something like

handlers = handlers["content_type"]

And handlers is a dict [<content_type:str>, <func_point:addr>]

vrdmr · 2025-04-25T17:48:36Z

+            self._grpc_resp_queue.put_nowait(self._GRPC_STOP_RESPONSE)
+            self._grpc_thread.join()
+            self._grpc_thread = None
+


which will left resources in open(or undefined state)

This is already at the end of the workers lifetime - and we can discard any state.

vrdmr

Looks good to go in;

follow up tasks to be created and minor tests.

gavin-aguiar and others added 25 commits February 5, 2025 15:41

Proxy Worker: Initial Commit

a1fb445

Updated worker config to include 3.13

4d3b892

Updated test_setup

a6c80f1

Updated worker.py

5343e0e

Updated dispatcher

d1390d5

Updated syspath in worker.py

bcad815

Updated path in worker.py

f720195

Updated worker.py

f7f4795

Removed reload in dispatcher

43d29a3

Updating v1 library worker name

bcdf819

Added dispatcher logs

bc4a520

Added dispatcher try/catch logs

7c7cc36

Updated sys path

4764ce7

Dispatcher and dependency manager updates

53a3d4e

Updated dispatcher and pyproject

5caa922

Merge branch 'dev' of github.com:Azure/azure-functions-python-worker …

357ed59

…into gaaguiar/proxy_worker

Testing updates and refactoring

dabaec6

Bug fixes and refactoring

7f038bb

Added more unit tests

3ff686d

Added tests and fixed test setup

5ade8e3

Updated test_setup

c465406

Updated test setup to add grpc dir copy

6083d7f

Merging changes

2992d42

linting fixes

dc94eb1