func_extension_base.py

# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.

importabc
importos
importtyping
fromloggingimportLogger
from .extension_metaimportExtensionMeta
from .extension_scopeimportExtensionScope
from .function_extension_exceptionimportFunctionExtensionException
from .._abcimportContext


classFuncExtensionBase(metaclass=ExtensionMeta):
"""An abstract class defines the life-cycle hooks which to be implemented
 by customer's extension.

 Everytime when a new extension is initialized in customer function scripts,
 the ExtensionManager._func_exts field records the extension to this
 specific function name.
 """

_scope=ExtensionScope.FUNCTION

@abc.abstractmethod
def__init__(self, file_path: str):
"""Constructor for extension. This needs to be implemented and ensure
 super().__init__(file_path) is called.

 The initializer serializes the extension to a tree. This speeds
 up the worker lookup and reduce the overhead on each invocation.
 _func_exts[<trigger_name>].<hook_name>.(ext_name, ext_impl)
 (e.g. _func_exts['HttpTrigger'].pre_invocation.ext_impl)

 Parameters
 ----------
 file_path: str
 The name of trigger the extension attaches to (e.g. __file__).
 """
script_root=os.getenv('AzureWebJobsScriptRoot')
ifscript_rootisNone:
raiseFunctionExtensionException(
'AzureWebJobsScriptRoot environment variable is not defined. '
'Please ensure the extension is running in Azure Functions.'
 )

# Split will always return ('') in if no folder exist in the path
relpath_to_project_root=os.path.relpath(
os.path.normpath(file_path),
os.path.normpath(script_root)
 )

trigger_name= (relpath_to_project_root.split(os.sep) or [''])[0]
ifnottrigger_nameortrigger_name.startswith(('.', '..')):
raiseFunctionExtensionException(
'Failed to parse trigger name from filename. '
'Function extension should bind to a trigger script, '
'not share folder. Please ensure extension is create inside a'
'trigger while __file__ is passed into the argument. '
'The trigger name is resolved from os.path.relpath(file_path,'
'project_root).'
 )

# This is used in ExtensionMeta._register_function_extension
self._trigger_name=trigger_name

# DO NOT decorate this with @abc.abstractmethod
# since implementation by subclass is not mandatory
defpost_function_load(self,
function_name: str,
function_directory: str,
*args, **kwargs) ->None:
"""This hook will be called right after a customer's function loaded.
 In this stage, the customer's logger is not fully initialized, so it
 is not provided. Please use print() to emit message if necessary.

 Parameters
 ----------
 function_name: str
 The name of customer's function (e.g. HttpTrigger)
 function_directory: str
 The path to customer's function directory
 (e.g. /home/site/wwwroot/HttpTrigger)
 """
pass

# DO NOT decorate this with @abc.abstractmethod
# since implementation by subclass is not mandatory
defpre_invocation(
self,
logger: Logger,
context: Context,
func_args: typing.Optional[typing.Dict[str, object]] =None,
*args,
**kwargs) ->None:
"""This hook will be called right before customer's function
 is being executed.

 Parameters
 ----------
 logger: logging.Logger
 A logger provided by Python worker. Extension developer should
 use this logger to emit telemetry to Azure Functions customers.
 context: azure.functions.Context
 This will include the function_name, function_directory and an
 invocation_id of this specific invocation.
 func_args: typing.Dict[str, object]
 Arguments that are passed into the Azure Functions. The name of
 each parameter is defined in function.json. Extension developers
 may also want to do isinstance() check if you want to apply
 operations to specific trigger types or input binding types.
 """
pass

# DO NOT decorate this with @abc.abstractmethod
# since implementation by subclass is not mandatory
defpost_invocation(
self,
logger: Logger,
context: Context,
func_args: typing.Optional[typing.Dict[str, object]] =None,
func_ret: typing.Optional[object] =None,
*args,
**kwargs) ->None:
"""This hook will be called right after a customer's function
 is executed.

 Parameters
 ----------
 logger: logging.Logger
 A logger provided by Python worker. Extension developer should
 use this logger to emit telemetry to Azure Functions customers.
 context: azure.functions.Context
 This will include the function_name, function_directory and an
 invocation_id of this specific invocation.
 func_args: typing.Dict[str, object]
 Arguments that are passed into the Azure Functions. The name of
 each parameter is defined in function.json. Extension developers
 may also want to do isinstance() check if you want to apply
 operations to specific trigger types or input binding types.
 func_ret: typing.Optional[object]
 Return value from Azure Functions. This is usually the value
 defined in function.json $return section. Extension developers
 may also want to do isinstance() check if you want to apply
 operations to specific types or input binding types.
 """
pass