Source code for inmanta.protocol.decorators

"""
    Copyright 2019 Inmanta

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.

    Contact: code@inmanta.com
"""

import inspect
from typing import Callable, Optional, TypeVar, Union

from inmanta import const
from inmanta.types import Apireturn, HandlerType, MethodType

from . import common

FuncT = TypeVar("FuncT", bound=HandlerType)


[docs] class handle: # noqa: N801 """ Decorator for subclasses of an endpoint to handle protocol methods :param method: A subclass of method that defines the method :param api_version: When specific this handler is only associated with a method of the specific api version. If the version is not defined, the handler is not associated with a rest endpoint. :param kwargs: Map arguments in the message from one name to an other """ def __init__(self, method: Callable[..., Apireturn], api_version: Optional[int] = None, **kwargs: str) -> None: self.method = method self.mapping: dict[str, str] = kwargs self._api_version = api_version def __call__(self, function: FuncT) -> FuncT: """ The wrapping """ if not inspect.iscoroutinefunction(function): raise ValueError(f"{function} is not an async function. Only async def functions may handle requests.") function.__protocol_method__ = self.method function.__protocol_mapping__ = self.mapping function.__api_version__ = self._api_version return function
MethodT = TypeVar("MethodT", bound=MethodType) def method( path: str, operation: str = "POST", reply: bool = True, arg_options: dict[str, common.ArgOption] = {}, timeout: Optional[int] = None, server_agent: bool = False, api: Optional[bool] = None, agent_server: bool = False, validate_sid: Optional[bool] = None, client_types: list[const.ClientType] = [const.ClientType.api], api_version: int = 1, api_prefix: str = "api", envelope: bool = False, envelope_key: str = const.ENVELOPE_KEY, enforce_auth: bool = True, ) -> Callable[..., Callable]: """ Decorator to identify a method as a RPC call. The arguments of the decorator are used by each transport to build and model the protocol. :param path: The url path to use for this call. This path can contain parameter names of the function. These names should be enclosed in < > brackets. :param operation: The type of HTTP operation (verb). :param timeout: nr of seconds before request it terminated. :param api: This is a call from the client to the Server (True if not server_agent and not agent_server). :param server_agent: This is a call from the Server to the Agent (reverse http channel through long poll). :param agent_server: This is a call from the Agent to the Server. :param validate_sid: This call requires a valid session, true by default if agent_server and not api :param client_types: The allowed client types for this call. The valid values are defined by the :const:`inmanta.const.ClientType` enum. :param arg_options: Options related to arguments passed to the method. The key of this dict is the name of the arg to which the options apply. The value is another dict that can contain the following options: header: Map this argument to a header with the following name. reply_header: If the argument is mapped to a header, this header will also be included in the reply getter: Call this method after validation and pass its return value to the method call. This may change the type of the argument. This method can raise an HTTPException to return a 404 for example. :param api_version: The version of the api this method belongs to. :param api_prefix: The prefix of the method: /<prefix>/v<version>/<method_name>. :param envelope: Put the response of the call under an envelope with key envelope_key. :param envelope_key: The envelope key to use. :param enforce_auth: When set to true authentication is enforced on this endpoint. When set to false, authentication is not enforced, even if auth is enabled. """ def wrapper(func: MethodT) -> MethodT: properties = common.MethodProperties( func, path, operation, reply, arg_options, timeout, server_agent, api, agent_server, validate_sid, client_types, api_version, api_prefix, envelope, False, envelope_key, enforce_auth=enforce_auth, ) common.MethodProperties.register_method(properties) return func return wrapper def typedmethod( path: Union[str, list[str]], operation: str = "POST", reply: bool = True, arg_options: dict[str, common.ArgOption] = {}, timeout: Optional[int] = None, server_agent: bool = False, api: Optional[bool] = None, agent_server: bool = False, validate_sid: Optional[bool] = None, client_types: list[const.ClientType] = [const.ClientType.api], api_version: int = 1, api_prefix: str = "api", envelope_key: str = const.ENVELOPE_KEY, strict_typing: bool = True, enforce_auth: bool = True, varkw: bool = False, ) -> Callable[..., Callable]: """ Decorator to identify a method as a RPC call. The arguments of the decorator are used by each transport to build and model the protocol. :param path: The url path(s) to use for this call. This path can contain parameter names of the function. These names should be enclosed in < > brackets. :param operation: The type of HTTP operation (verb) :param timeout: nr of seconds before request it terminated :param api: This is a call from the client to the Server (True if not server_agent and not agent_server) :param server_agent: This is a call from the Server to the Agent (reverse http channel through long poll) :param agent_server: This is a call from the Agent to the Server :param validate_sid: This call requires a valid session, true by default if agent_server and not api :param client_types: The allowed client types for this call. The valid values are defined by the :const:`inmanta.const.ClientType` enum. :param arg_options: Options related to arguments passed to the method. The key of this dict is the name of the arg to which the options apply. The value is another dict that can contain the following options: header: Map this argument to a header with the following name. reply_header: If the argument is mapped to a header, this header will also be included in the reply getter: Call this method after validation and pass its return value to the method call. This may change the type of the argument. This method can raise an HTTPException to return a 404 for example. :param api_version: The version of the api this method belongs to :param api_prefix: The prefix of the method: /<prefix>/v<version>/<method_name> :param envelope_key: The envelope key to use. :param strict_typing: If true, does not allow `Any`. Setting this option to False is heavily discouraged except for some few very specific cases where the type system does not allow the strict type to be specified, for example in case of infinite recursion. :param enforce_auth: When set to true authentication is enforced on this endpoint. When set to false, authentication is not enforced, even if auth is enabled. :param varkw: If true, additional arguments are allowed and will be dispatched to the handler. The handler is responsible for the validation. """ def wrapper(func: MethodT) -> MethodT: paths = path if isinstance(path, list) else [path] for current_path in paths: properties = common.MethodProperties( func, current_path, operation, reply, arg_options, timeout, server_agent, api, agent_server, validate_sid, client_types, api_version, api_prefix, True, True, envelope_key, strict_typing=strict_typing, enforce_auth=enforce_auth, varkw=varkw, ) common.MethodProperties.register_method(properties) return func return wrapper