worker extensions disabled for python 3.13

[ArvidMartensRenson]

From the docs and the code here I can see that the usage of custom worker extensions is not longer supported in python 3.13. All of our function apps are heavily dependent on these extensions, making a transition to python 3.13+ not as straightforward.

Is this feature going to be supported again in the future (in this form or another like middleware or hooks) or should we look for other solutions for our extensions? If the latter is true, is there any migration path that can be followed?

[hallvictoria]

Hi ArvidMartensRenson, thanks for reporting.

As you mentioned, custom worker extensions are not supported in Python 3.13+. We will be introducing an alternative to worker extensions later and a migration path. I don't have an ETA for this feature as of right now, but I'll update here when we have one.

[ArvidMartensRenson]

Hi hallvictoria ,

thanks for the answer, just out of curiosity is there already an idea how this alternative will be implemented? We are planning to try 3.13 and will have to adjust the extension we developed, would be nice for us that we can already somewhat implement it in the format that is going to be released so when it is released we can easily correct it.

If not known yet, we will implement something custom and migrate later. We are thinking of creating a blueprint class on which middlewares can be registered. Whenever a function needs to be build the python method gets wrapped by the consecutive middlewares and passed to the Function class. In our opinion it will mimick the worker extension behavior and improve it as it allows for some error handling as well. It also changes the current azure-functions (not azure-functions-worker-runtime) code base minimally.

[hallvictoria]

We don't have any finalized design right now - this is still very much so a work-in-progress :)

Our main thinking was changing the extension structure from being an external class / package that users have to create and install to being decorators that you can just add to your function.

Taking the timer extension described in the Develop Python Worker Extensions Doc
Generally speaking, the pre_invocation_app_level function would look something like this with the new design:

// function_app.py

start_timestamps: typing.Dict[str, float] = {}

app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS)

# Code to execute before the invocation
def my_pre_invocation_code(logger: Logger, context: Context, func_args: typing.Dict[str, object]):
    logger.info(f'Recording start time of {context.function_name}')
    start_timestamps[context.invocation_id] = time()


@app.route(route="http_trigger")
# A pre_invocation decorator is added to the function you want to the pre-invoc code to apply to
# Inside the decorator you specify the method / code to execute
@app.pre_invocation(logger, context, func_args, code_to_execute=my_pre_invocation_code)
def http_trigger(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')
    return func.HttpResponse("hello world!")

This was based on feedback from an internal user of worker extensions. Your custom solution seems viable in the meantime while we work on this feature. Would you be willing to share more about your current scenario and how you use worker extensions? We can take that into consideration to make sure we support your use-cases with the new design

[ArvidMartensRenson]

Hi hallvictoria,

So in our team we mainly use two extensions:

• one is used to clean stale database connection before an invocation such that errors are less common. However with the current extension implementation we lack some retry options in case a db error occurs. We could solve this by migrating the extension to a decorator with a try and except clause in it (essentially creating somewhat of a middleware).
• The second extension is for our otel logging setup. So this extension basically configures the otel logging, tracing, and metrics for our application, and starts the necessary spans before every function invocation. That span is stored within the extension for that invocation id. At the end a post invocation method is run to properly close the span and delete the object from the extension. This implementation also has some limitations due to the lack of an error handling option. If an error surfaces in the actual function the span should be set to error (such that it is flagged in the monitoring visualization) but that can't be achieved as the post invocation method is not invoked in case of raised exceptions. To achieve it I had to wrap parts of the azure worker such that the post invocation method is also invoked. The middleware solution we implemented should also solve this.

I like the proposal of the decorators to set pre_invocation and post_invocation methods to a function, but I would certainly include a post_exception decorator as well. Also some option to set the decorators on a blueprint level would also be nice, if you want every function to have the same pre_invocation method to be run (like in our logging case), you don't want to add the decorator to every function.

If you want our take on how we implemented the middleware blueprint, I am happy to share it.

hopefully this clarifies a bit our usage of extensions and what shortcomings we experience today.

[hallvictoria]

Thanks, this is really helpful!

Based on your feedback, I think there are 3 extra parts we can look into including for this feature, but please feel free to add on if you think there's something else missing:

• Retry options for pre / post invocation code
• Post-exception decorator - code that executes if an exception occurs in the function
• A way to apply pre or post invocation decorators at a file level (eg. all functions in blueprint.py) and/or app level (eg. all functions in the app)

I don't have an ETA for when this work will be completed, but once we finalize a design and timeline I'll post an update on here.

[ArvidMartensRenson]

In our team we use this solution to get hooks (mainly the pre_function_invocation and post_function_invocation) working in 3.13. We tried to keep it in line with how middlewares are built for WSGI or ASGI applications, so you can have try-except-finally clauses, have internal variables that are accessible before and after request without storing them in contextvars. We know that it is not complete replacement for the current hooks but maybe it can serve as an inspiration for the new hooks that are being implemented.

class FunctionAppMiddlewareInitializer:
    def __init__(self, middleware_cls: type["FunctionAppMiddleware"], **kwargs):
        self.middleware_cls = middleware_cls
        self.kwargs = kwargs

    def create(self, func: Callable):
        mw_instance = self.middleware_cls(func, **self.kwargs)
        if inspect.iscoroutinefunction(func):
            @functools.wraps(func)
            async def wrapped_method(context: Context, **func_kwargs):
                return await mw_instance.async_call(context, **func_kwargs)
        else:
            @functools.wraps(func)
            def wrapped_method(context: Context, **func_kwargs):
                return mw_instance.sync_call(context, **func_kwargs)
        return wrapped_method

class FunctionAppMiddleware:
    def __init__(self, func: Callable):
        self.func = func

    async def async_call(self, context: Context, **kwargs):
        raise NotImplementedError("Async call not implemented in middleware.")

    def sync_call(self, context: Context, **kwargs):
        raise NotImplementedError("Sync call not implemented in middleware.")


def function_invoker(func: Callable):
    sig = inspect.signature(func)
    has_context = "context" in sig.parameters

    if not has_context:
        params = [inspect.Parameter("context", inspect.Parameter.POSITIONAL_OR_KEYWORD, annotation=Context)]
        params += list(sig.parameters.values())
        new_sig = sig.replace(parameters=params)
    else:
        new_sig = sig

    if inspect.iscoroutinefunction(func):

        @functools.wraps(func)
        async def wrapper(context, **kwargs):
            return await func(context=context, **kwargs) if has_context else await func(**kwargs)
    else:

        @functools.wraps(func)
        def wrapper(context, **kwargs):
            return func(context=context, **kwargs) if has_context else func(**kwargs)

    wrapper.__signature__ = new_sig
    return wrapper


class MiddlewareFunctionAppBlueprint(Blueprint):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.middlewares: list[FunctionAppMiddlewareInitializer] = []

    def add_middleware(self, middleware_cls, *args, **kwargs):
        self.middlewares.append(FunctionAppMiddlewareInitializer(middleware_cls, **kwargs))

    def _validate_type(self, func):
        if isinstance(func, FunctionBuilder):
            fb = self._function_builders.pop()
        elif callable(func):
            if len(self.middlewares) > 0:
                func = function_invoker(func)
            for middleware in self.middlewares:
                func = middleware.create(func=func)
            fb = FunctionBuilder(func, self._app_script_file)
        else:
            raise ValueError("Unsupported type for function app decorator found.")
        return fb

[ArvidMartensRenson]

hallvictoria Varad Meru (@vrdmr)
What is your take on this structure, is it something that could be introduced in the azure-functions code stack or be used as an inspiration?