Azure Function HttpトリガーのPython実装に迫る
はじめに
masamoriです。
僕は普段クラウドサービスはAzureを使っています。中でもよく使うサービスはApp service系のPassなのです。Azure Functionもよく使っているのですが、よく考えたら実装を覗いたことないなー、と思いAzure FunctionのPython Libraryのコードを読んでみました。
Httpトリガー
Functionは様々なAzureサービスをトリガーにして実行することができますが、一番オーソドックスなHttpトリガーについてコードを読みたいと思います。
まず、VSCodeのFunctionツールから作成できるHttpトリガーのコードはこちらです。
import azure.functions as func import logging app = func.FunctionApp(http_auth_level=func.AuthLevel.ANONYMOUS) @app.route(route="http_trigger") def http_trigger(req: func.HttpRequest) -> func.HttpResponse: logging.info('Python HTTP trigger function processed a request.') name = req.params.get('name') if not name: try: req_body = req.get_json() except ValueError: pass else: name = req_body.get('name') if name: return func.HttpResponse(f"Hello, {name}. This HTTP triggered function executed successfully.") else: return func.HttpResponse( "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デコレータで関数をトリガー化しているので、FunctionAppを探します。
FunctionAppはazure/functions/init.pyの中に定義されていて、fucntions配下のdecorators/function_app.pyの中に定義されています。
class FunctionApp(FunctionRegister, TriggerApi, BindingApi, SettingsApi): """FunctionApp object used by worker function indexing model captures user defined functions and metadata. Ref: https://aka.ms/azure-function-ref """ def __init__(self, http_auth_level: Union[AuthLevel, str] = AuthLevel.FUNCTION): """Constructor of :class:`FunctionApp` object. :param http_auth_level: Determines what keys, if any, need to be present on the request in order to invoke the function. """ super().__init__(auth_level=http_auth_level)
インスタンス化する際に必要な引数としてhttp_auth_levelがあります。デフォルトの設定ではFUNCTIONとなっており、特定の関数エンドポイントへのアクセスのみを許可するキーの設定が必要となります。今回の記事ではANONYMOUSにしているので、キーの設定は必要ありません。localで試したい場合はANONYMOUSで十分だと思います。
FunctionAppはTriggerApi, BindingApi, SettingsApiを継承しています。では、TriggerApiを見ましょう。
class TriggerApi(DecoratorApi, ABC): """Interface to extend for using existing trigger decorator functions.""" def route(self, route: Optional[str] = None, trigger_arg_name: str = 'req', binding_arg_name: str = '$return', methods: Optional[ Union[Iterable[str], Iterable[HttpMethod]]] = None, auth_level: Optional[Union[AuthLevel, str]] = None, trigger_extra_fields: Optional[Dict[str, Any]] = None, binding_extra_fields: Optional[Dict[str, Any]] = None ) -> Callable[..., Any]: """The route decorator adds :class:`HttpTrigger` and :class:`HttpOutput` binding to the :class:`FunctionBuilder` object for building :class:`Function` object used in worker function indexing model. This is equivalent to defining HttpTrigger and HttpOutput binding in the function.json which enables your function be triggered when http requests hit the specified route. All optional fields will be given default value by function host when they are parsed by function host. Ref: https://aka.ms/azure-function-binding-http :param route: Route for the http endpoint, if None, it will be set to function name if present or user defined python function name. :param trigger_arg_name: Argument name for :class:`HttpRequest`, defaults to 'req'. :param binding_arg_name: Argument name for :class:`HttpResponse`, defaults to '$return'. :param methods: A tuple of the HTTP methods to which the function responds. :param auth_level: Determines what keys, if any, need to be present on the request in order to invoke the function. :return: Decorator function. :param trigger_extra_fields: Additional fields to include in trigger json. For example, >>> data_type='STRING' # 'dataType': 'STRING' in trigger json :param binding_extra_fields: Additional fields to include in binding json. For example, >>> data_type='STRING' # 'dataType': 'STRING' in binding json """ if trigger_extra_fields is None: trigger_extra_fields = {} if binding_extra_fields is None: binding_extra_fields = {} @self._configure_function_builder def wrap(fb): def decorator(): fb.add_trigger(trigger=HttpTrigger( name=trigger_arg_name, methods=parse_iterable_param_to_enums(methods, HttpMethod), auth_level=parse_singular_param_to_enum(auth_level, AuthLevel), route=route, **trigger_extra_fields)) fb.add_binding(binding=HttpOutput( name=binding_arg_name, **binding_extra_fields)) return fb return decorator() return wrap ...以下略
routeメソッドのroute引数はエンドポイントです。また、Httpリクエストとレスポンスを定義できて、methodも定義できます。今回のケースではapp.route"を使用しているので、httpトリガーが呼び出されます。
また、ここではデコレーターが定義されていますが、Functionがどのように機能しているかに焦点を当てると、@self._configure_function_builderが重要になってきます。このデコレータでFunctionBuilderが生成され、自分で定義した関数に@appとしてデコレータを付与すると、その関数はトリガーとして機能するようになります。
では、TriggerAPIには'DecoratorApi'が継承されているので、見ていきましょう。
class DecoratorApi(ABC): """Interface which contains essential decorator function building blocks to extend for creating new function app or blueprint classes. """ def __init__(self, *args, **kwargs): self._function_builders: List[FunctionBuilder] = [] self._app_script_file: str = SCRIPT_FILE_NAME ... 略 def _validate_type(self, func: Union[Callable[..., Any], FunctionBuilder]) \ -> FunctionBuilder: """Validate the type of the function object and return the created :class:`FunctionBuilder` object. :param func: Function object passed to :meth:`_configure_function_builder` :raises ValueError: Raise error when func param is neither :class:`Callable` nor :class:`FunctionBuilder`. :return: :class:`FunctionBuilder` object. """ if isinstance(func, FunctionBuilder): fb = self._function_builders.pop() elif callable(func): fb = FunctionBuilder(func, self._app_script_file) else: raise ValueError( "Unsupported type for function app decorator found.") return fb def _configure_function_builder(self, wrap) -> Callable[..., Any]: """Decorator function on user defined function to create and return :class:`FunctionBuilder` object from :class:`Callable` func. """ def decorator(func): fb = self._validate_type(func) self._function_builders.append(fb) return wrap(fb) return decorator
_configure_function_builderというデコレータを作成し、そのメソッドを通してFunctionBuilderをラップしています。ここでFunctionBuilderのバリデーションをして、FunctionBuilderオブジェクトをリストに追加しています。そしてDecoratorApiを継承したTriggerApiの中で、_configure_function_builderがデコレータとして使用され、FunctionBuilderオブジェクトにトリガーが追加されます。(TriggerApiでHttpTriggerとなっていますね)。これはroute(Httpトリガー)を@app.routeで呼び出しているからです。
例えば、timer_trigerを呼び出したかったら、routeの代わりに.
def timer_trigger(self, arg_name: str, schedule: str, run_on_startup: Optional[bool] = None, use_monitor: Optional[bool] = None, data_type: Optional[Union[DataType, str]] = None, **kwargs: Any) -> Callable[..., Any]: """The schedule or timer decorator adds :class:`TimerTrigger` to the :class:`FunctionBuilder` object for building :class:`Function` object used in worker function indexing model. This is equivalent to defining TimerTrigger in the function.json which enables your function be triggered on the specified schedule. All optional fields will be given default value by function host when they are parsed by function host. Ref: https://aka.ms/azure-function-binding-timer :param arg_name: The name of the variable that represents the :class:`TimerRequest` object in function code. :param schedule: A string representing a CRON expression that will be used to schedule a function to run. :param run_on_startup: If true, the function is invoked when the runtime starts. :param use_monitor: Set to true or false to indicate whether the schedule should be monitored. :param data_type: Defines how Functions runtime should treat the parameter value. :return: Decorator function. """ @self._configure_function_builder def wrap(fb): def decorator(): fb.add_trigger( trigger=TimerTrigger( name=arg_name, schedule=schedule, run_on_startup=run_on_startup, use_monitor=use_monitor, data_type=parse_singular_param_to_enum(data_type, DataType), **kwargs)) return fb return decorator() return wrap schedule = timer_trigger
こいつを@app.timer_triggerデコレーターとして呼び出せばいいです。TriggerApiの中に、他サービスで使えるトリガーの定義が書いてあるので、自分の使いたいトリガーの挙動が意図しないものであった場合、このTriggerApiクラスを見ることをお勧めします。どんなパラメーターが使用できるのか確認もできます。
まとめ
流れとして、
- FunctionApp: Azure Functions アプリケーション全体を管理し、ユーザー定義の関数とトリガーの結びつきを定義。
- TriggerApi: HTTP トリガーや他のトリガーを関数に紐付け、どのリクエストやイベントで関数が実行されるかを定義。
- DecoratorApi: デコレータを通じて、関数に FunctionBuilder を適用し、トリガーやバインディングを設定するための基盤を提供。
- FunctionBuilder: 関数のトリガーやバインディングを設定し、関数を構築します。最終的に、Azure Functions で動作するための関数が完成。
このような感じでしょうか。
似たような実装部分が多く、読みにくさも感じないコードなので、また機会があったらソースコードを読んでみたいと思います。
