bamboo.sticky.http

CacheControlInfo dataclass

CacheControlInfo(public: bool = False, private: bool = False, no_cache: bool = False, no_store: bool = False, max_age: Union[int, NoneType] = None, s_maxage: Union[int, NoneType] = None, max_stale: Union[int, NoneType] = None, min_fresh: Union[int, NoneType] = None, stale_while_revalidate: Union[int, NoneType] = None, stale_if_error: Union[int, NoneType] = None, must_revalidate: bool = False, proxy_revalidate: bool = False, immutable: bool = False, no_transform: bool = False, only_if_cached: bool = False)

ClientInfo dataclass

ClientInfo(ip: Union[ipaddress.IPv4Address, ipaddress.IPv6Address], port: Union[int, NoneType] = None)

CookieInfo dataclass

CookieInfo(cookie_name: str, expires: Union[str, NoneType] = None, max_age: Union[int, NoneType] = None, domain: Union[str, NoneType] = None, path: Union[str, NoneType] = None, secure: bool = True, http_only: bool = True, samesite: Union[str, NoneType] = None)

DataFormatInfo dataclass

dataclass with information of data format at callbacks on Endpoint.

Attributes:

Name	Type	Description
input	Optional[Type[ApiData]]	Input data format.
output	Optional[Type[ApiData]]	Output data format.
is_validate	bool	If input data is to be validate.
err_validate	ErrInfo	Error information sent when validation failes.

MultipleAuthSchemeError

Raised if several authentication schemes of the 'Authorization' header are detected.

PreFlightInfo dataclass

PreFlightInfo(allow_methods: Tuple[str], allow_origins: Tuple[str] = (), allow_headers: Tuple[str] = (), expose_headers: Tuple[str] = (), max_age: Union[int, NoneType] = None, allow_credentials: bool = False, err_not_allowed_origin: bamboo.error.ErrInfo = DefualtCORSOriginNotAllowedErrInfo(), err_not_allowed_method: bamboo.error.ErrInfo = DefualtCORSOriginNotAllowedErrInfo(), add_arg: bool = True)

RequiredHeaderInfo dataclass

dataclass with information of header which should be included in response headers.

Attributes:

Name	Type	Description
header	str	Name of header.
err	Optional[bamboo.error.ErrInfo]	Error information sent when the header is not included.
add_arg	bool	Whether the header is given as a callback's argument.

RequiredQueryInfo dataclass

RequiredQueryInfo(query: str, err_empty: Union[bamboo.error.ErrInfo, NoneType], err_not_unique: Union[bamboo.error.ErrInfo, NoneType], mapf: Union[Callable[[Union[List[str], str, NoneType]], Any], NoneType], add_arg: bool)

SimpleAccessControlInfo dataclass

SimpleAccessControlInfo(origins: Tuple[str] = (), allow_credentials: bool = False, err_not_allowed: bamboo.error.ErrInfo = DefualtCORSOriginNotAllowedErrInfo(), add_arg: bool = True)

basic_auth(err=DefaultAuthHeaderNotFoundErrInfo())

Set callback up to require Authorization header in Basic authentication.

Parameters:

Name	Type	Description	Default
err	ErrInfo	Error sent when Authorization header is not found, received scheme doesn't match, or extracting user ID and password from credentials failes.	DefaultAuthHeaderNotFoundErrInfo()

Returns:

Type	Description
Callable[[Union[Callable[[bamboo.endpoint.WSGIEndpoint, Tuple[Any, ...]], NoneType], Callable[[bamboo.endpoint.ASGIHTTPEndpoint, Tuple[Any, ...]], Awaitable[NoneType]]]], Union[Callable[[bamboo.endpoint.WSGIEndpoint, Tuple[Any, ...]]], Callable[[bamboo.endpoint.ASGIHTTPEndpoint, Tuple[Any, ...]], Awaitable[NoneType]]]]	Decorator to make callback to be set up to require the Authorization header.

Examples:

class MockEndpoint(WSGIEndpoint):

    @basic_auth()
    def do_GET(self, user_id: str, password: str) -> None:
        # It is guaranteed that request headers include the
        # `Authorization` header at this point, and user_id and
        # password are the ones extracted from the header.

        # Authenticate with any function working on your system.
        authenticate(user_id, password)

        # Do something...

Source code in bamboo/sticky/http.py

def basic_auth(
    err: ErrInfo = DEFAULT_BASIC_AUTH_HEADER_NOT_FOUND_ERROR,
) -> CallbackDecorator_t:
    """Set callback up to require `Authorization` header in Basic
    authentication.

    Args:
        err: Error sent when `Authorization` header is not found, received
            scheme doesn't match, or extracting user ID and password from
            credentials failes.

    Returns:
        Decorator to make callback to be set  up to require
        the `Authorization` header.

    Examples:
        ```python
        class MockEndpoint(WSGIEndpoint):

            @basic_auth()
            def do_GET(self, user_id: str, password: str) -> None:
                # It is guaranteed that request headers include the
                # `Authorization` header at this point, and user_id and
                # password are the ones extracted from the header.

                # Authenticate with any function working on your system.
                authenticate(user_id, password)

                # Do something...
        ```
    """
    def wrapper(callback: Callback_t) -> Callback_t:
        config = AuthSchemeConfig(callback)
        return config.set(AuthSchemes.basic, err)

    return wrapper

bearer_auth(err=DefaultAuthHeaderNotFoundErrInfo())

Set callback up to require Authorization header in token authentication for OAuth 2.0.

Parameters:

Name	Type	Description	Default
err	ErrInfo	Error sent when Authorization header is not found, or when received scheme doesn't match.	DefaultAuthHeaderNotFoundErrInfo()

Returns:

Type	Description
Callable[[Union[Callable[[bamboo.endpoint.WSGIEndpoint, Tuple[Any, ...]], NoneType], Callable[[bamboo.endpoint.ASGIHTTPEndpoint, Tuple[Any, ...]], Awaitable[NoneType]]]], Union[Callable[[bamboo.endpoint.WSGIEndpoint, Tuple[Any, ...]]], Callable[[bamboo.endpoint.ASGIHTTPEndpoint, Tuple[Any, ...]], Awaitable[NoneType]]]]	Decorator to make callback to be set up to require the Authorization header.

Examples:

class MockEndpoint(WSGIEndpoint):

    @bearer_auth()
    def do_GET(self, token: str) -> None:
        # It is guaranteed that request headers include the
        # `Authorization` header at this point, and token is
        # the one extracted from the header.

        # Authenticate with any function working on your system.
        authenticate(token)

        # Do something...

Source code in bamboo/sticky/http.py

def bearer_auth(
    err: ErrInfo = DEFAULT_BEARER_AUTH_HEADER_NOT_FOUND_ERROR,
) -> CallbackDecorator_t:
    """Set callback up to require `Authorization` header in token
    authentication for OAuth 2.0.

    Args:
        err : Error sent when `Authorization` header is not found, or
            when received scheme doesn't match.

    Returns:
        Decorator to make callback to be set  up to require
        the `Authorization` header.

    Examples:
        ```python
        class MockEndpoint(WSGIEndpoint):

            @bearer_auth()
            def do_GET(self, token: str) -> None:
                # It is guaranteed that request headers include the
                # `Authorization` header at this point, and token is
                # the one extracted from the header.

                # Authenticate with any function working on your system.
                authenticate(token)

                # Do something...
        ```
    """
    def wrapper(callback: Callback_t) -> Callback_t:
        config = AuthSchemeConfig(callback)
        return config.set(AuthSchemes.bearer, err)

    return wrapper

data_format(input=None, output=None, is_validate=True, err_validate=DefaultDataFormatErrInfo(), err_noheader=DefaultHeaderNotFoundErrInfo())

Set data format of input/output data as API to callback on Endpoint.

This decorator can be used to add attributes of data format information to a response method, and execute validation if input raw data has expected format defined on input argument.

To represent no data inputs/outputs, specify input/output arguments as None. If input is None, then any data received from client will not be read. If is_validate is False, then validation will not be executed.

Parameters:

Name	Type	Description	Default
input	Optional[Type[bamboo.api.base.ApiData]]	Input data format.	None
output	Optional[Type[bamboo.api.base.ApiData]]	Output data format.	None
is_validate	bool	If input data is to be validated.	True
err_validate	ErrInfo	Error information sent when validation failes.	DefaultDataFormatErrInfo()

Returns:

Type	Description
Callable[[Union[Callable[[bamboo.endpoint.WSGIEndpoint, Tuple[Any, ...]], NoneType], Callable[[bamboo.endpoint.ASGIHTTPEndpoint, Tuple[Any, ...]], Awaitable[NoneType]]]], Union[Callable[[bamboo.endpoint.WSGIEndpoint, Tuple[Any, ...]]], Callable[[bamboo.endpoint.ASGIHTTPEndpoint, Tuple[Any, ...]], Awaitable[NoneType]]]]	Decorator to add attributes of data format information to callback.

Examples:

class UserData(JsonApiData):
    name: str
    email: str
    age: int

class MockEndpoint(WSGIEndpoint):

    @data_format(input=UserData, output=None)
    def do_GET(self, rec_body: UserData) -> None:
        user_name = rec_body.name
        # Do something...

Source code in bamboo/sticky/http.py

def data_format(
    input: t.Optional[t.Type[ApiData]] = None,
    output: t.Optional[t.Type[ApiData]] = None,
    is_validate: bool = True,
    err_validate: ErrInfo = DEFUALT_INCORRECT_DATA_FORMAT_ERROR,
    err_noheader: ErrInfo = DEFAULT_HEADER_NOT_FOUND_ERROR,
) -> CallbackDecorator_t:
    """Set data format of input/output data as API to callback on
    `Endpoint`.

    This decorator can be used to add attributes of data format information
    to a response method, and execute validation if input raw data has
    expected format defined on `input` argument.

    To represent no data inputs/outputs, specify `input`/`output` arguments
    as `None`. If `input` is `None`, then any data received from client will
    not be read. If `is_validate` is `False`, then validation will not be
    executed.

    Args:
        input: Input data format.
        output: Output data format.
        is_validate: If input data is to be validated.
        err_validate: Error information sent when validation failes.

    Returns:
        Decorator to add attributes of data format information to callback.

    Examples:
        ```python
        class UserData(JsonApiData):
            name: str
            email: str
            age: int

        class MockEndpoint(WSGIEndpoint):

            @data_format(input=UserData, output=None)
            def do_GET(self, rec_body: UserData) -> None:
                user_name = rec_body.name
                # Do something...
        ```
    """
    dataformat = DataFormatInfo(
        input,
        output,
        is_validate,
        err_validate,
        err_noheader,
    )

    def wrapper(callback: Callback_t) -> Callback_t:
        config = DataFormatConfig(callback)
        return config.set(dataformat)

    return wrapper

has_header_of(header, err=None, add_arg=True)

Set callback up to receive given header from clients.

If request headers don't include specified header, then response headers and body will be made based on err and sent.

Parameters:

Name	Type	Description	Default
header	str	Name of header.	required
err	Optional[bamboo.error.ErrInfo]	Error information sent when specified header is not found.	None
add_arg	bool	Whether the header is given as a callback's argument.	True

Returns:

Type	Description
Callable[[Union[Callable[[bamboo.endpoint.WSGIEndpoint, Tuple[Any, ...]], NoneType], Callable[[bamboo.endpoint.ASGIHTTPEndpoint, Tuple[Any, ...]], Awaitable[NoneType]]]], Union[Callable[[bamboo.endpoint.WSGIEndpoint, Tuple[Any, ...]]], Callable[[bamboo.endpoint.ASGIHTTPEndpoint, Tuple[Any, ...]], Awaitable[NoneType]]]]	Decorator to make callback to be set up to receive the header.

Examples:

class BasicAuthHeaderNotFoundErrInfo(ErrInfo):
    http_status = HTTPStatus.UNAUTHORIZED

    def get_headers(self) -> List[Tuple[str, str]]:
        return [("WWW-Authenticate", 'Basic realm="SECRET AREA"')]

class MockEndpoint(WSGIEndpoint):

    @has_header_of("Authorization", BasicAuthHeaderNotFoundErrInfo())
    def do_GET(self) -> None:
        # It is guaranteed that request headers include the
        # `Authorization` header at this point.
        header_auth = self.get_header("Authorization")

        # Do something...

Source code in bamboo/sticky/http.py

def has_header_of(
    header: str,
    err: t.Optional[ErrInfo] = None,
    add_arg: bool = True,
) -> CallbackDecorator_t:
    """Set callback up to receive given header from clients.

    If request headers don't include specified `header`, then response
    headers and body will be made based on `err` and sent.

    Args:
        header: Name of header.
        err: Error information sent when specified `header` is not found.
        add_arg: Whether the header is given as a callback's argument.

    Returns:
        Decorator to make callback to be set up to receive the header.

    Examples:
        ```python
        class BasicAuthHeaderNotFoundErrInfo(ErrInfo):
            http_status = HTTPStatus.UNAUTHORIZED

            def get_headers(self) -> List[Tuple[str, str]]:
                return [("WWW-Authenticate", 'Basic realm="SECRET AREA"')]

        class MockEndpoint(WSGIEndpoint):

            @has_header_of("Authorization", BasicAuthHeaderNotFoundErrInfo())
            def do_GET(self) -> None:
                # It is guaranteed that request headers include the
                # `Authorization` header at this point.
                header_auth = self.get_header("Authorization")

                # Do something...
        ```
    """
    info = RequiredHeaderInfo(header, err, add_arg)

    def wrapper(callback: Callback_t) -> Callback_t:
        config = RequiredHeaderConfig(callback)
        return config.set(info)

    return wrapper

has_query_of(query, err_empty=None, err_not_unique=None, mapf=None, add_arg=True)

Set callback up to receive given query parameter from clients.

Source code in bamboo/sticky/http.py

def has_query_of(
    query: str,
    err_empty: t.Optional[ErrInfo] = None,
    err_not_unique: t.Optional[ErrInfo] = None,
    mapf: t.Optional[t.Callable[[t.Union[t.List[str], str, None]], t.Any]] = None,
    add_arg: bool = True,
) -> CallbackDecorator_t:
    """Set callback up to receive given query parameter from clients.
    """
    info = RequiredQueryInfo(query, err_empty, err_not_unique, mapf, add_arg)

    def wrapper(callback: Callback_t) -> Callback_t:
        config = RequiredQueryConfig(callback)
        return config.set(info)

    return wrapper

may_occur(*errors)

Register error classes to callback on Endpoint.

Parameters:

Name	Type	Description	Default
*errors	Type[bamboo.error.ErrInfo]	Error classes which may occuur.	()

Returns:

Type	Description
Callable[[Union[Callable[[bamboo.endpoint.WSGIEndpoint, Tuple[Any, ...]], NoneType], Callable[[bamboo.endpoint.ASGIHTTPEndpoint, Tuple[Any, ...]], Awaitable[NoneType]]]], Union[Callable[[bamboo.endpoint.WSGIEndpoint, Tuple[Any, ...]]], Callable[[bamboo.endpoint.ASGIHTTPEndpoint, Tuple[Any, ...]], Awaitable[NoneType]]]]	CallbackDecorator_t: Decorator to register error classes to callback.

Examples:

class MockErrInfo(ErrInfo):
    http_status = HTTPStatus.INTERNAL_SERVER_ERROR

    def get_body(self) -> bytes:
        return b"Intrernal server error occured"

class MockEndpoint(WSGIEndpoint):

    @may_occur(MockErrInfo)
    def do_GET(self) -> None:
        # Do something...

        # It is possible to send error response.
        if is_some_flag():
            self.send_err(MockErrInfo())

        self.send_only_status()

Source code in bamboo/sticky/http.py

def may_occur(*errors: t.Type[ErrInfo]) -> CallbackDecorator_t:
    """Register error classes to callback on `Endpoint`.

    Args:
        *errors: Error classes which may occuur.

    Returns:
        CallbackDecorator_t: Decorator to register
            error classes to callback.

    Examples:
        ```python
        class MockErrInfo(ErrInfo):
            http_status = HTTPStatus.INTERNAL_SERVER_ERROR

            def get_body(self) -> bytes:
                return b"Intrernal server error occured"

        class MockEndpoint(WSGIEndpoint):

            @may_occur(MockErrInfo)
            def do_GET(self) -> None:
                # Do something...

                # It is possible to send error response.
                if is_some_flag():
                    self.send_err(MockErrInfo())

                self.send_only_status()
        ```
    """

    def wrapper(callback: Callback_t) -> Callback_t:
        config = HTTPErrorConfig(callback)
        return config.set(*errors)

    return wrapper

restricts_client(*clients, *, err=DefaultNotApplicableIpErrInfo())

Restrict IP addresses at callback.

Parameters:

Name	Type	Description	Default
*client_ips		IP addresses to be allowed to request	required
err	ErrInfo	Error information sent when request from IP not included specified IPs comes.	DefaultNotApplicableIpErrInfo()

Returns:

Type	Description
Callable[[Union[Callable[[bamboo.endpoint.WSGIEndpoint, Tuple[Any, ...]], NoneType], Callable[[bamboo.endpoint.ASGIHTTPEndpoint, Tuple[Any, ...]], Awaitable[NoneType]]]], Union[Callable[[bamboo.endpoint.WSGIEndpoint, Tuple[Any, ...]]], Callable[[bamboo.endpoint.ASGIHTTPEndpoint, Tuple[Any, ...]], Awaitable[NoneType]]]]	Decorator to make callback to be set up to restrict IP addresses.

Exceptions:

Type	Description
ValueError	Raised if invalid IP address is detected

Examples:

class MockEndpoint(WSGIEndpoint):

    # Restrict to allow only localhost to request
    # to this callback
    @restricts_client(ClientInfo("localhost"))
    def do_GET(self) -> None:
        # Only localhost can access to the callback.

        # Do something...

Source code in bamboo/sticky/http.py

def restricts_client(
    *clients: ClientInfo,
    err: ErrInfo = DEFAULT_NOT_APPLICABLE_IP_ERROR
) -> CallbackDecorator_t:
    """Restrict IP addresses at callback.

    Args:
        *client_ips: IP addresses to be allowed to request
        err: Error information sent when request from IP not included
            specified IPs comes.

    Returns:
        Decorator to make callback to be set up to restrict IP addresses.

    Raises:
        ValueError: Raised if invalid IP address is detected

    Examples:
        ```python
        class MockEndpoint(WSGIEndpoint):

            # Restrict to allow only localhost to request
            # to this callback
            @restricts_client(ClientInfo("localhost"))
            def do_GET(self) -> None:
                # Only localhost can access to the callback.

                # Do something...
        ```
    """
    def wrapper(callback: Callback_t) -> Callback_t:
        config = RestrictedClientsConfig(callback)
        return config.set(*clients, err=err)

    return wrapper