Develop Python worker extensions for Azure Functions

Article
04/25/2023

Azure Functions lets you integrate custom behaviors as part of Python function execution. This feature enables you to create business logic that customers can easily use in their own function apps. To learn more, see the Python developer reference. Worker extensions are supported in both the v1 and v2 Python programming models.

In this tutorial, you'll learn how to:

Create an application-level Python worker extension for Azure Functions.
Consume your extension in an app the way your customers do.
Package and publish an extension for consumption.

Prerequisites

Before you start, you must meet these requirements:

Python 3.7 or above. To check the full list of supported Python versions in Azure Functions, see the Python developer guide.
The Azure Functions Core Tools, version 4.0.5095 or later, which supports using the extension with the v2 Python programming model. Check your version with func --version.
Visual Studio Code installed on one of the supported platforms.

Create the Python Worker extension

The extension you create reports the elapsed time of an HTTP trigger invocation in the console logs and in the HTTP response body.

Folder structure

The folder for your extension project should be like the following structure:

<python_worker_extension_root>/
 | - .venv/
 | - python_worker_extension_timer/
 | | - __init__.py
 | - setup.py
 | - readme.md

Folder/file	Description
.venv/	(Optional) Contains a Python virtual environment used for local development.
python_worker_extension/	Contains the source code of the Python worker extension. This folder contains the main Python module to be published into PyPI.
setup.py	Contains the metadata of the Python worker extension package.
readme.md	Contains the instruction and usage of your extension. This content is displayed as the description in the home page in your PyPI project.

Configure project metadata

First you create setup.py, which provides essential information about your package. To make sure that your extension is distributed and integrated into your customer's function apps properly, confirm that 'azure-functions >= 1.7.0, < 2.0.0' is in the install_requires section.

In the following template, you should change author, author_email, install_requires, license, packages, and url fields as needed.

from setuptools import find_packages, setup
setup(
    name='python-worker-extension-timer',
    version='1.0.0',
    author='Your Name Here',
    author_email='your@email.here',
    classifiers=[
        'Intended Audience :: End Users/Desktop',
        'Development Status :: 5 - Production/Stable',
        'Intended Audience :: End Users/Desktop',
        'License :: OSI Approved :: Apache Software License',
        'Programming Language :: Python',
        'Programming Language :: Python :: 3.7',
        'Programming Language :: Python :: 3.8',
        'Programming Language :: Python :: 3.9',
        'Programming Language :: Python :: 3.10',
    ],
    description='Python Worker Extension Demo',
    include_package_data=True,
    long_description=open('readme.md').read(),
    install_requires=[
        'azure-functions >= 1.7.0, < 2.0.0',
        # Any additional packages that will be used in your extension
    ],
    extras_require={},
    license='MIT',
    packages=find_packages(where='.'),
    url='https://your-github-or-pypi-link',
    zip_safe=False,
)

Next, you'll implement your extension code in the application-level scope.

Implement the timer extension

Add the following code in python_worker_extension_timer/__init__.py to implement the application-level extension:

import typing
from logging import Logger
from time import time
from azure.functions import AppExtensionBase, Context, HttpResponse
class TimerExtension(AppExtensionBase):
    """A Python worker extension to record elapsed time in a function invocation
    """

    @classmethod
    def init(cls):
        # This records the starttime of each function
        cls.start_timestamps: typing.Dict[str, float] = {}

    @classmethod
    def configure(cls, *args, append_to_http_response:bool=False, **kwargs):
        # Customer can use TimerExtension.configure(append_to_http_response=)
        # to decide whether the elapsed time should be shown in HTTP response
        cls.append_to_http_response = append_to_http_response

    @classmethod
    def pre_invocation_app_level(
        cls, logger: Logger, context: Context,
        func_args: typing.Dict[str, object],
        *args, **kwargs
    ) -> None:
        logger.info(f'Recording start time of {context.function_name}')
        cls.start_timestamps[context.invocation_id] = time()

    @classmethod
    def post_invocation_app_level(
        cls, logger: Logger, context: Context,
        func_args: typing.Dict[str, object],
        func_ret: typing.Optional[object],
        *args, **kwargs
    ) -> None:
        if context.invocation_id in cls.start_timestamps:
            # Get the start_time of the invocation
            start_time: float = cls.start_timestamps.pop(context.invocation_id)
            end_time: float = time()
            # Calculate the elapsed time
            elapsed_time = end_time - start_time
            logger.info(f'Time taken to execute {context.function_name} is {elapsed_time} sec')
            # Append the elapsed time to the end of HTTP response
            # if the append_to_http_response is set to True
            if cls.append_to_http_response and isinstance(func_ret, HttpResponse):
                func_ret._HttpResponse__body += f' (TimeElapsed: {elapsed_time} sec)'.encode()

This code inherits from AppExtensionBase so that the extension applies to every function in the app. You could have also implemented the extension on a function-level scope by inheriting from FuncExtensionBase.

The init method is a class method that's called by the worker when the extension class is imported. You can do initialization actions here for the extension. In this case, a hash map is initialized for recording the invocation start time for each function.

The configure method is customer-facing. In your readme file, you can tell your customers when they need to call Extension.configure(). The readme should also document the extension capabilities, possible configuration, and usage of your extension. In this example, customers can choose whether the elapsed time is reported in the HttpResponse.

The pre_invocation_app_level method is called by the Python worker before the function runs. It provides the information from the function, such as function context and arguments. In this example, the extension logs a message and records the start time of an invocation based on its invocation_id.

Similarly, the post_invocation_app_level is called after function execution. This example calculates the elapsed time based on the start time and current time. It also overwrites the return value of the HTTP response.

Create a readme.md

Create a readme.md file in the root of your extension project. This file contains the instructions and usage of your extension. The readme.md content is displayed as the description in the home page in your PyPI project.

# Python Worker Extension Timer

In this file, tell your customers when they need to call `Extension.configure()`.

The readme should also document the extension capabilities, possible configuration,
and usage of your extension.

Consume your extension locally

Now that you've created an extension, you can use it in an app project to verify it works as intended.

Create an HTTP trigger function

Create a new folder for your app project and navigate to it.
From the appropriate shell, such as Bash, run the following command to initialize the project:
```
func init --python
```
Use the following command to create a new HTTP trigger function that allows anonymous access:
```
func new -t HttpTrigger -n HttpTrigger -a anonymous
```

Activate a virtual environment

Create a Python virtual environment, based on OS as follows:
- Linux
- Windows
```
python3 -m venv .venv
```
```
py -m venv .venv
```
Activate the Python virtual environment, based on OS as follows:
- Linux
- Windows
```
source .venv/bin/activate
```
```
.venv\Scripts\Activate.ps1
```

Configure the extension

Install remote packages for your function app project using the following command:
```
pip install -r requirements.txt
```
Install the extension from your local file path, in editable mode as follows:
```
pip install -e <PYTHON_WORKER_EXTENSION_ROOT>
```
In this example, replace <PYTHON_WORKER_EXTENSION_ROOT> with the root file location of your extension project.

When a customer uses your extension, they'll instead add your extension package location to the requirements.txt file, as in the following examples:
- PyPI
- GitHub
```
# requirements.txt
python_worker_extension_timer==1.0.0
```
```
# requirements.txt
git+https://github.com/Azure-Samples/python-worker-extension-timer@main
```
Open the local.settings.json project file and add the following field to Values:
```
"PYTHON_ENABLE_WORKER_EXTENSIONS": "1" 
```
When running in Azure, you instead add PYTHON_ENABLE_WORKER_EXTENSIONS=1 to the app settings in the function app.
Add following two lines before the main function in __init.py__ file for the v1 programming model, or in the function_app.py file for the v2 programming model:
```
from python_worker_extension_timer import TimerExtension
TimerExtension.configure(append_to_http_response=True)
```
This code imports the TimerExtension module and sets the append_to_http_response configuration value.

Verify the extension

From your app project root folder, start the function host using func host start --verbose. You should see the local endpoint of your function in the output as https://localhost:7071/api/HttpTrigger.

In the browser, send a GET request to https://localhost:7071/api/HttpTrigger. You should see a response like the following, with the TimeElapsed data for the request appended.

This HTTP triggered function executed successfully. Pass a name in the query string or in the request body for a personalized response. (TimeElapsed: 0.0009996891021728516 sec)

Publish your extension

After you've created and verified your extension, you still need to complete these remaining publishing tasks:

Choose a license.
Create a readme.md and other documentation.
Publish the extension library to a Python package registry or a version control system (VCS).

PyPI
GitHub

To publish your extension to PyPI:

Run the following command to install twine and wheel in your default Python environment or a virtual environment:
```
pip install twine wheel
```
Remove the old dist/ folder from your extension repository.
Run the following command to generate a new package inside dist/:
```
python setup.py sdist bdist_wheel
```
Run the following command to upload the package to PyPI:
```
twine upload dist/*
```
You may need to provide your PyPI account credentials during upload. You can also test your package upload with twine upload -r testpypi dist/*. For more information, see the Twine documentation.

After these steps, customers can use your extension by including your package name in their requirements.txt.

For more information, see the official Python packaging tutorial.

Examples

You can view completed sample extension project from this article in the python_worker_extension_timer sample repository.
OpenCensus integration is an open-source project that uses the extension interface to integrate telemetry tracing in Azure Functions Python apps. See the opencensus-python-extensions-azure repository to review the implementation of this Python worker extension.

Next steps

For more information about Azure Functions Python development, see the following resources:

Share via