app

`AppBase`

Base class of all application in Bamboo.

Bamboo has two core concepts called application and endpoint, and this class implements basic behavior of the former, e.g. containing multiple endpoints, routing requests from URIs to endpoints and so on.

Note

This class is an abstract class. Consider using its subclasses.

Attributes:

Name	Type	Description
`TAG_VERSION`	`str`	Tag used when versions of `Endpoint`s are inserted in front of paths of URIs. If you want, you can override the value and set new favorite tag. By default, the tag is 'v'.

`tree: Uri2Endpoints_t` `property` `readonly`

Tree of the application's endpoints.

`init(self, error_404=DefaultNotFoundErrInfo())` `special`

Parameters:

Name	Type	Description	Default
`error_404`	`ErrInfo`	Error sending if a request to not registered URI or HTTP method comes.	`DefaultNotFoundErrInfo()`

Source code in bamboo/app.py

def __init__(
    self,
    error_404: ErrInfo = DEFAULT_NOT_FOUND_ERROR
) -> None:
    """
    Args:
        error_404: Error sending if a request to not registered URI or
            HTTP method comes.
    """
    self._router: Router[Endpoint_t] = Router()
    self._error_404 = error_404

`graft(self, apps, , onto=())`

Graft other applications as branches of the application's tree.

Parameters:

Name	Type	Description	Default
`*apps`	`AppBase`	Branch applications.	`()`
`onto`	`t.Tuple[StaticLocation_t, ...]`	Root path of the branches.	`()`

Source code in bamboo/app.py

def graft(
    self,
    *apps: AppBase,
    onto: t.Tuple[StaticLocation_t, ...] = ()
) -> None:
    """Graft other applications as branches of the application's tree.

    Args:
        *apps: Branch applications.
        onto: Root path of the branches.
    """
    for app in apps:
        for locs, endpoint in app._router._raw_uri2endpoint.items():
            locs = onto + locs

            ver_config = VersionConfig(endpoint)
            version = ver_config.get(app)
            ver_config.set(self, version=version)
            if len(version):
                version = tuple(f"{self.TAG_VERSION}{v}" for v in version)

            self._router.register(locs, endpoint, version=version)

`route(self, locs, , version=None)`

Register combination of URI and Endpoint for routing.

Parameters:

Name	Type	Description	Default
`*locs`	`Location_t`	Locations of path of the URI bound with the `Endpoint`.	`()`
`version`	`t.Union[int, t.Tuple[int], None]`	Version of the `Endpoint`.	`None`

Returns:

Type	Description
`t.Callable[[t.Type[Endpoint_t]], t.Type[Endpoint_t]]`	Decorator to add combination of URI and `Endpoint`.

Examples:

app = App()

# Set path of URI as `test/data/image` and the version as 1
@app.route("test", "data", "image", version=1)
class MockEndpoint(Endpoint):

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

Source code in bamboo/app.py

def route(
    self,
    *locs: Location_t,
    version: t.Union[int, t.Tuple[int], None] = None
) -> t.Callable[[t.Type[Endpoint_t]], t.Type[Endpoint_t]]:
    """Register combination of URI and `Endpoint` for routing.

    Args:
        *locs: Locations of path of the URI bound with the `Endpoint`.
        version : Version of the `Endpoint`.

    Returns:
        Decorator to add combination of URI and `Endpoint`.

    Examples:
        ```python
        app = App()

        # Set path of URI as `test/data/image` and the version as 1
        @app.route("test", "data", "image", version=1)
        class MockEndpoint(Endpoint):

            def do_GET(self) -> None:
                # Do something...
        ```
    """
    def register_endpoint(
        endpoint: t.Type[Endpoint_t],
    ) -> t.Type[Endpoint_t]:
        if not issubclass(endpoint, self.__avalidable_endpoints):
            raise TypeError(
                f"{endpoint.__name__} is not avalidable in "
                f"the {self.__class__.__name__}."
            )

        locs_normalized = tuple([loc for loc in locs if loc])

        # version setting
        ver_config = VersionConfig(endpoint)
        ver_config.set(self, version)

        # router setting
        _version = ver_config.get(self)
        assert _version is not None
        if len(_version):
            _version = tuple(f"{self.TAG_VERSION}{v}" for v in _version)
        self._router.register(locs_normalized, endpoint, version=_version)

        return endpoint
    return register_endpoint

`search_uris(self, endpoint)`

Retrieve all URI patterns of Endpoint.

Note

This method uses bamboo.Router.search_uris() method inner. For more information, see the API document.

Parameters:

Name	Type	Description	Default
`endpoint`	`t.Type[Endpoint_t]`	`Endpoint` whose URIs to be searched.	required

Returns:

Type	Description
`t.List[Uri_t]`	Result of searching.

Source code in bamboo/app.py

def search_uris(self, endpoint: t.Type[Endpoint_t]) -> t.List[Uri_t]:
    """Retrieve all URI patterns of `Endpoint`.

    Note:
        This method uses `bamboo.Router.search_uris()` method inner.
        For more information, see the API document.

    Args:
        endpoint: `Endpoint` whose URIs to be searched.

    Returns:
        Result of searching.
    """
    return self._router.search_uris(endpoint)

`set_parcel(self, endpoint, *parcel)`

Set parcel to an endpoint.

This method enables to give objects to Endpoint objects dynamically. A parcel is a set of objects delivered via the method, and the Endpoint object of its destination receives it at EndpointBase.setup() method.

Note

For more information about the bamboo.EndpointBase.setup(), see the API document.

Parameters:

Name	Type	Description	Default
`endpoint`	`t.Type[Endpoint_t]`	`Endpoint` the `parcel` to be set.	required
`*parcel`	`t.Any`	Pacel to be given to the `Endpoint`.	`()`

Source code in bamboo/app.py

def set_parcel(
    self,
    endpoint: t.Type[Endpoint_t],
    *parcel: t.Any,
) -> AppBase:
    """Set parcel to an endpoint.

    This method enables to give objects to `Endpoint` objects
    dynamically. A parcel is a set of objects delivered via the method,
    and the `Endpoint` object of its destination receives it at
    `EndpointBase.setup()` method.

    Note:
        For more information about the `bamboo.EndpointBase.setup()`,
        see the API document.

    Args:
        endpoint: `Endpoint` the `parcel` to be set.
        *parcel: Pacel to be given to the `Endpoint`.
    """
    parcel_config = ParcelConfig(endpoint)
    parcel_config.set(self, parcel)
    return self

`validate(self, uri)`

Validate specified uri and retrieve corresponding Endpoint class.

Note

This method uses bamboo.Router.validate() method inner. For more information, see the API document.

Parameters:

Name	Type	Description	Default
`uri`	`str`	Path of URI to be validated.	required

Returns:

Type	Description
`t.Tuple[t.Tuple[str, ...], t.Optional[t.Type[Endpoint_t]]]`	Pair of values of flexible locations and `Endpoint` class if specified `uri` is valid, otherwise, pari of empty tuple and None.

Source code in bamboo/app.py

def validate(
    self,
    uri: str,
) -> t.Tuple[t.Tuple[str, ...], t.Optional[t.Type[Endpoint_t]]]:
    """Validate specified `uri` and retrieve corresponding `Endpoint` class.

    Note:
        This method uses `bamboo.Router.validate()` method inner.
        For more information, see the API document.

    Args:
        uri: Path of URI to be validated.

    Returns:
        Pair of values of flexible locations and `Endpoint` class if specified
        `uri` is valid, otherwise, pari of empty tuple and None.
    """
    return self._router.validate(uri)

`ASGIApp`

Application compliant with the ASGI.

This class is a subclass of AppBase calss and implements the callbable compliant with the ASGI.

Note

This class can be used only for ASGI server. If you want to use any WSGI servers, consider using WSGIApp.

This class can also route only ASGIHTTPEndpoints and ASGIWebSocketEndpoints. If you want to another type of endpoint, consider implementation class of its corresponding application.

`call(self, scope, recv, send)` `async` `special`

Handle requests compliant with the ASGI.

Parameters:

Name	Type	Description	Default
`scope`	`t.Dict[str, t.Any]`	Connection scope in the ASGI.	required
`recv`	`ASGIRecv_t`	Awaitable callable to receive new event data.	required
`send`	`ASGISend_t`	Awaitable callable to send new event data.	required

Source code in bamboo/app.py

async def __call__(
    self,
    scope: t.Dict[str, t.Any],
    recv: ASGIRecv_t,
    send: ASGISend_t,
) -> None:
    """Handle requests compliant with the ASGI.

    Args:
        scope: Connection scope in the ASGI.
        recv: Awaitable callable to receive new event data.
        send: Awaitable callable to send new event data.
    """
    typ = scope.get("type")
    if typ == ASGIProtocols.http:
        await self.handle_http(scope, recv, send)
    elif typ == ASGIProtocols.websocket:
        await self.handle_websocket(scope, recv, send)
    elif typ == ASGIProtocols.lifespan:
        await self.handle_lifespan(scope, recv, send)
    else:
        raise NotImplementedError

`init(self, error_404=DefaultNotFoundErrInfo(), lifespan_handler=<function default_lifespan_handler at 0x7f28f4f14200>)` `special`

Parameters:

Name	Type	Description	Default
`error_404`	`ErrInfo`	Error sending if a request to not registered URI or HTTP method comes.	`DefaultNotFoundErrInfo()`
`lifespan_handler`	`LifespanHandler_t`	A callback with the type `LifespanHandler_t`.	`<function default_lifespan_handler at 0x7f28f4f14200>`

Source code in bamboo/app.py

def __init__(
    self,
    error_404: ErrInfo = DEFAULT_NOT_FOUND_ERROR,
    lifespan_handler: LifespanHandler_t = default_lifespan_handler,
) -> None:
    """
    Args:
        error_404: Error sending if a request to not registered URI or
            HTTP method comes.
        lifespan_handler: A callback with the type `LifespanHandler_t`.
    """
    super().__init__(error_404=error_404)

    self._lifespan_handler = lifespan_handler

`apply_lifespan_handler(self, lifespan_handler)`

Apply a handler for the Lifespan protocol.

This method sets a callback with the type LifespanHandler_t to handle requests with the Lifespan protocol. This method should be called only if one intends to set one's custom callback.

Parameters:

Name	Type	Description	Default
`lifespan_handler`	`LifespanHandler_t`	A callback with the type `LifespanHandler_t`.	required

Source code in bamboo/app.py

def apply_lifespan_handler(
    self,
    lifespan_handler: LifespanHandler_t,
) -> None:
    """Apply a handler for the Lifespan protocol.

    This method sets a callback with the type `LifespanHandler_t` to
    handle requests with the Lifespan protocol. This method should be
    called only if one intends to set one's custom callback.

    Args:
        lifespan_handler: A callback with the type `LifespanHandler_t`.
    """
    self._lifespan_handler = lifespan_handler

`handle_http(self, scope, recv, send)` `async`

Handle requests with the HTTP protocol.

This method only handles requests with the HTTP and is used in the __call__() method for the ASGI application. Therefore one doesn't have to call this method if one just runs the application.

Parameters:

Name	Type	Description	Default
`scope`	`t.Dict[str, t.Any]`	Connection scope in the ASGI.	required
`recv`	`ASGIRecv_t`	Awaitable callable to receive new event data.	required
`send`	`ASGISend_t`	Awaitable callable to send new event data.	required

Source code in bamboo/app.py

async def handle_http(
    self,
    scope: t.Dict[str, t.Any],
    recv: ASGIRecv_t,
    send: ASGISend_t,
) -> None:
    """Handle requests with the HTTP protocol.

    This method only handles requests with the HTTP and is used in
    the `__call__()` method for the ASGI application. Therefore one
    doesn't have to call this method if one just runs the application.

    Args:
        scope: Connection scope in the ASGI.
        recv: Awaitable callable to receive new event data.
        send: Awaitable callable to send new event data.
    """
    method = scope.get("method")
    path = scope.get("path")
    sendstart = get_http_sendstart(send)
    sendbody = get_http_sendbody(send)
    send_errinfo = get_http_send_errinfo(send)

    flexible_locs, endpoint_class = self.validate(path)
    if endpoint_class is None:
        await send_errinfo(self._error_404, ())
        return

    parcel_config = ParcelConfig(endpoint_class)
    parcel = parcel_config.get(self)

    pre_callback = endpoint_class._get_pre_response_method(method)
    callback = endpoint_class._get_response_method(method)
    if callback is None:
        await send_errinfo(self._error_404, ())
        return

    endpoint = endpoint_class(self, scope, recv, flexible_locs)
    # NOTE
    #   Subclasses of the ErrInfo must be raised in pre-response
    #   methods or response methods. Otherwise, the errors behave
    #   like ordinary exception objects.
    try:
        endpoint.setup(*parcel)
        if pre_callback:
            await pre_callback(endpoint)
        await callback(endpoint)
    except ErrInfo as e:
        await send_errinfo(e, endpoint._res_headers)
        return

        # NOTE
        #   Other exceptions not inheriting the ErrInfo class
        #   are not to be catched here.
    else:
        status = endpoint._res_status
        headers = endpoint._res_headers
        body = endpoint._res_body

    await sendstart(status, headers)
    await sendbody(body)

`handle_lifespan(self, scope, recv, send)` `async`

Handle requests with the Lifespan protocol.

This method only handles requests with the Lifespan and is used in the __call__() method for the ASGI application. Therefore one doesn't have to call this method if one just runs the application.

Parameters:

Name	Type	Description	Default
`scope`	`t.Dict[str, t.Any]`	Connection scope in the ASGI.	required
`recv`	`ASGIRecv_t`	Awaitable callable to receive new event data.	required
`send`	`ASGISend_t`	Awaitable callable to send new event data.	required

Source code in bamboo/app.py

async def handle_lifespan(
    self,
    scope: t.Dict[str, t.Any],
    recv: ASGIRecv_t,
    send: ASGISend_t,
) -> None:
    """Handle requests with the Lifespan protocol.

    This method only handles requests with the Lifespan and is used in
    the `__call__()` method for the ASGI application. Therefore one
    doesn't have to call this method if one just runs the application.

    Args:
        scope: Connection scope in the ASGI.
        recv: Awaitable callable to receive new event data.
        send: Awaitable callable to send new event data.
    """
    await self._lifespan_handler(scope, recv, send)

`handle_websocket(self, scope, recv, send)` `async`

Handle requests with the WebSocket protocol.

This method only handles requests with the WebSocket and is used in the __call__() method for the ASGI application. Therefore one doesn't have to call this method if one just runs the application.

Parameters:

Name	Type	Description	Default
`scope`	`t.Dict[str, t.Any]`	Connection scope in the ASGI.	required
`recv`	`ASGIRecv_t`	Awaitable callable to receive new event data.	required
`send`	`ASGISend_t`	Awaitable callable to send new event data.	required

Source code in bamboo/app.py

async def handle_websocket(
    self,
    scope: t.Dict[str, t.Any],
    recv: ASGIRecv_t,
    send: ASGISend_t,
) -> None:
    """Handle requests with the WebSocket protocol.

    This method only handles requests with the WebSocket and is used in
    the `__call__()` method for the ASGI application. Therefore one
    doesn't have to call this method if one just runs the application.

    Args:
        scope: Connection scope in the ASGI.
        recv: Awaitable callable to receive new event data.
        send: Awaitable callable to send new event data.
    """
    path = scope.get("path")

    flexible_locs, endpoint_class = self.validate(path)
    if endpoint_class is None:
        await self.send_404(send)
        return

    parcel_config = ParcelConfig(endpoint_class)
    parcel = parcel_config.get(self)
    endpoint = endpoint_class(self, scope, flexible_locs, *parcel)

    # Establish connection
    msg = await recv()
    assert msg["type"] == ASGIWebSocketEvents.connect
    accept = get_websock_accept(send)
    await endpoint.do_ACCEPT(accept)

    # Main communications
    recvmsg = get_websock_recvmsg(recv)
    sendmsg = get_websock_sendmsg(send)
    close = get_websock_close(send)
    await endpoint.do_COMMUNICATE(recvmsg, sendmsg, close)

`route(self, locs, , version=None)`

Register combination of URI and Endpoint for routing.

Parameters:

Name	Type	Description	Default
`*locs`	`Location_t`	Locations of path of the URI bound with the `Endpoint`.	`()`
`version`	`t.Union[int, t.Tuple[int], None]`	Version of the `Endpoint`.	`None`

Returns:

Type	Description
`t.Callable[[t.Type[ASGIEndpointBase]], t.Type[ASGIEndpointBase]]`	Decorator to add combination of URI and `Endpoint`.

Examples:

app = App()

# Set path of URI as `test/data/image` and the version as 1
@app.route("test", "data", "image", version=1)
class MockEndpoint(Endpoint):

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

Source code in bamboo/app.py

def route(
    self,
    *locs: Location_t,
    version: t.Union[int, t.Tuple[int], None] = None,
) -> t.Callable[[t.Type[ASGIEndpointBase]], t.Type[ASGIEndpointBase]]:
    return super().route(*locs, version=version)

`search_uris(self, endpoint)`

Retrieve all URI patterns of Endpoint.

Note

This method uses bamboo.Router.search_uris() method inner. For more information, see the API document.

Parameters:

Name	Type	Description	Default
`endpoint`	`t.Type[ASGIHTTPEndpoint]`	`Endpoint` whose URIs to be searched.	required

Returns:

Type	Description
`t.List[Uri_t]`	Result of searching.

Source code in bamboo/app.py

def search_uris(
    self,
    endpoint: t.Type[ASGIHTTPEndpoint],
) -> t.List[Uri_t]:
    return super().search_uris(endpoint)

`set_parcel(self, endpoint, *parcel)`

Set parcel to an endpoint.

This method enables to give objects to Endpoint objects dynamically. A parcel is a set of objects delivered via the method, and the Endpoint object of its destination receives it at EndpointBase.setup() method.

Note

For more information about the bamboo.EndpointBase.setup(), see the API document.

Parameters:

Name	Type	Description	Default
`endpoint`	`t.Type[ASGIEndpointBase]`	`Endpoint` the `parcel` to be set.	required
`*parcel`	`t.Any`	Pacel to be given to the `Endpoint`.	`()`

Source code in bamboo/app.py

def set_parcel(
    self,
    endpoint: t.Type[ASGIEndpointBase],
    *parcel: t.Any,
) -> ASGIApp:
    super().set_parcel(endpoint, *parcel)
    return self

`validate(self, uri)`

Validate specified uri and retrieve corresponding Endpoint class.

Note

This method uses bamboo.Router.validate() method inner. For more information, see the API document.

Parameters:

Name	Type	Description	Default
`uri`	`str`	Path of URI to be validated.	required

Returns:

Type	Description
`t.Tuple[t.Tuple[str, ...], t.Optional[t.Type[ASGIEndpointBase]]]`	Pair of values of flexible locations and `Endpoint` class if specified `uri` is valid, otherwise, pari of empty tuple and None.

Source code in bamboo/app.py

def validate(
    self,
    uri: str,
) -> t.Tuple[t.Tuple[str, ...], t.Optional[t.Type[ASGIEndpointBase]]]:
    return super().validate(uri)

`ParcelConfig`

Operator class for parcel of Endpoint.

This class can be used to get and set parcel of Endpoint safely.

`init(self, endpoint)` `special`

Parameters:

Name	Type	Description	Default
`endpoint`	`t.Type[EndpointBase]`	`Endpoint` whose parcel is to be manipulated.	required

Source code in bamboo/app.py

def __init__(self, endpoint: t.Type[EndpointBase]) -> None:
    """
    Args:
        endpoint : `Endpoint` whose parcel is to be manipulated.
    """
    self._endpoint_class = endpoint

`get(self, app)`

Retrieve parcel belonging to specified app.

Parameters:

Name	Type	Description	Default
`app`	`AppBase`	Application including the internal `Endpoint`	required

Returns:

Type	Description
`Parcel_t`	Parcel set to `Endpoint`, if not set yet, then `None`.

Source code in bamboo/app.py

def get(self, app: AppBase) -> Parcel_t:
    """Retrieve parcel belonging to specified `app`.

    Args:
        app: Application including the internal `Endpoint`

    Returns:
        Parcel set to `Endpoint`, if not set yet, then `None`.
    """
    if hasattr(self._endpoint_class, ATTR_PARCEL):
        registered = getattr(self._endpoint_class, ATTR_PARCEL)
        return registered.get(app)
    return ()

`get_all(self)`

Retrieve parcels belonging to all AppBase objects.

Returns:

Type	Description
`t.List[t.Tuple[AppBase, Parcel_t]]`	List of tuples of `AppBase` objects and their parcels.

Source code in bamboo/app.py

def get_all(self) -> t.List[t.Tuple[AppBase, Parcel_t]]:
    """Retrieve parcels belonging to all `AppBase` objects.

    Returns:
        List of tuples of `AppBase` objects and their parcels.
    """
    if hasattr(self._endpoint_class, ATTR_PARCEL):
        registered = getattr(self._endpoint_class, ATTR_PARCEL)
        return [(app, parcel) for app, parcel in registered.items()]
    return []

`set(self, app, parcel)`

Ser parcel of Endpoint.

Parameters:

Name	Type	Description	Default
`app`	`AppBase`	Application including the internal `Endpoint`.	required
`parcel`	`Parcel_t`	Parcel to be set.	required

Source code in bamboo/app.py

def set(self, app: AppBase, parcel: Parcel_t) -> None:
    """Ser parcel of `Endpoint`.

    Args:
        app: Application including the internal `Endpoint`.
        parcel: Parcel to be set.
    """
    if not hasattr(self._endpoint_class, ATTR_PARCEL):
        setattr(self._endpoint_class, ATTR_PARCEL, {})

    registered = getattr(self._endpoint_class, ATTR_PARCEL)
    registered[app] = parcel

`VersionConfig`

Operator class for version of Endpoint.

This class can be used to get and set version of Endpoint safely.

`init(self, endpoint)` `special`

Parameters:

Name	Type	Description	Default
`endpoint`	`t.Type[EndpointBase]`	`Endpoint` whose version is to be manipulated	required

Source code in bamboo/app.py

def __init__(self, endpoint: t.Type[EndpointBase]) -> None:
    """
    Args:
        endpoint: `Endpoint` whose version is to be manipulated
    """
    self._endpoint_class = endpoint

`get(self, app)`

Retrieve version belonging to specified app.

Parameters:

Name	Type	Description	Default
`app`	`AppBase`	Application including the internal `Endpoint`.	required

Returns:

Type	Description
`t.Optional[Version_t]`	Version set to `Endpoint`, if not set yet, then None.

Source code in bamboo/app.py

def get(self, app: AppBase) -> t.Optional[Version_t]:
    """Retrieve version belonging to specified `app`.

    Args:
        app: Application including the internal `Endpoint`.

    Returns:
        Version set to `Endpoint`, if not set yet, then None.
    """
    if hasattr(self._endpoint_class, ATTR_VERSION):
        registered = getattr(self._endpoint_class, ATTR_VERSION)
        return registered.get(app)
    return None

`get_all(self)`

Retrieve versions belonging to all AppBase objects.

Returns:

Type	Description
`t.List[t.Tuple[AppBase, Version_t]]`	List of tuples of `AppBase` objects and their versions.

Source code in bamboo/app.py

def get_all(self) -> t.List[t.Tuple[AppBase, Version_t]]:
    """Retrieve versions belonging to all `AppBase` objects.

    Returns:
        List of tuples of `AppBase` objects and their versions.
    """
    if hasattr(self._endpoint_class, ATTR_VERSION):
        registered = getattr(self._endpoint_class, ATTR_VERSION)
        return [(app, version) for app, version in registered.items()]
    return []

`set(self, app, version=None)`

Set version of Endpoint.

Parameters:

Name	Type	Description	Default
`app`	`AppBase`	Application including the internal `Endpoint`.	required
`version`	`t.Union[int, t.Tuple[int], None]`	Version to be set.	`None`
`force`		If forcing to set the `version`.	required

Exceptions:

Type	Description
`ValueError`	Raised if version of the `Endpoint` has already been set.

Source code in bamboo/app.py

def set(
    self,
    app: AppBase,
    version: t.Union[int, t.Tuple[int], None] = None,
) -> None:
    """Set version of `Endpoint`.

    Args:
        app: Application including the internal `Endpoint`.
        version: Version to be set.
        force: If forcing to set the `version`.

    Raises:
        ValueError: Raised if version of the `Endpoint` has already
            been set.
    """
    if not hasattr(self._endpoint_class, ATTR_VERSION):
        setattr(self._endpoint_class, ATTR_VERSION, {})

    registered = getattr(self._endpoint_class, ATTR_VERSION)

    # Format to fit the type Version_t
    if version is None:
        version = ()
    if isinstance(version, int):
        version = (version,)

    registered[app] = version

`WSGIApp`

Application compliant with the WSGI.

This class is a subclass of AppBase calss and implements the callbable compliant with the WSGI.

Note

This class can be used only for WSGI server. If you want to use any ASGI servers, consider using ASGIHTTPApp.

This class can also route only WSGIEndpoints. If you want to another type of endpoint, consider implementation class of its corresponding application.

`route(self, locs, , version=None)`

Register combination of URI and Endpoint for routing.

Parameters:

Name	Type	Description	Default
`*locs`	`Location_t`	Locations of path of the URI bound with the `Endpoint`.	`()`
`version`	`t.Union[int, t.Tuple[int], None]`	Version of the `Endpoint`.	`None`

Returns:

Type	Description
`t.Callable[[t.Type[WSGIEndpoint]], t.Type[WSGIEndpoint]]`	Decorator to add combination of URI and `Endpoint`.

Examples:

app = App()

# Set path of URI as `test/data/image` and the version as 1
@app.route("test", "data", "image", version=1)
class MockEndpoint(Endpoint):

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

Source code in bamboo/app.py

def route(
    self,
    *locs: Location_t,
    version: t.Union[int, t.Tuple[int], None] = None
) -> t.Callable[[t.Type[WSGIEndpoint]], t.Type[WSGIEndpoint]]:
    return super().route(*locs, version=version)

`search_uris(self, endpoint)`

Retrieve all URI patterns of Endpoint.

Note

This method uses bamboo.Router.search_uris() method inner. For more information, see the API document.

Parameters:

Name	Type	Description	Default
`endpoint`	`t.Type[WSGIEndpoint]`	`Endpoint` whose URIs to be searched.	required

Returns:

Type	Description
`t.List[Uri_t]`	Result of searching.

Source code in bamboo/app.py

def search_uris(self, endpoint: t.Type[WSGIEndpoint]) -> t.List[Uri_t]:
    return super().search_uris(endpoint)

`send_404(self, start_response)`

Send 404 error code, i.e. Resource Not Found error.

Parameters:

Name	Type	Description	Default
`start_response`	`WSGIStartRespoint_t`	`start_response` callable given from the WSGI application.	required

Returns:

Type	Description
`BufferedConcatIterator`	Response body of the error.

Source code in bamboo/app.py

def send_404(
    self,
    start_response: WSGIStartRespoint_t,
) -> BufferedConcatIterator:
    """Send `404` error code, i.e. `Resource Not Found` error.

    Args:
        start_response: `start_response` callable given from
            the WSGI application.

    Returns:
        Response body of the error.
    """
    status, headers, res_body = self._error_404.get_all_form()
    start_response(status.wsgi, headers)
    return res_body

`set_parcel(self, endpoint, *parcel)`

Set parcel to an endpoint.

This method enables to give objects to Endpoint objects dynamically. A parcel is a set of objects delivered via the method, and the Endpoint object of its destination receives it at EndpointBase.setup() method.

Note

For more information about the bamboo.EndpointBase.setup(), see the API document.

Parameters:

Name	Type	Description	Default
`endpoint`	`t.Type[WSGIEndpoint]`	`Endpoint` the `parcel` to be set.	required
`*parcel`	`t.Any`	Pacel to be given to the `Endpoint`.	`()`

Source code in bamboo/app.py

def set_parcel(
    self,
    endpoint: t.Type[WSGIEndpoint],
    *parcel: t.Any
) -> WSGIApp:
    super().set_parcel(endpoint, *parcel)
    return self

`validate(self, uri)`

Validate specified uri and retrieve corresponding Endpoint class.

Note

This method uses bamboo.Router.validate() method inner. For more information, see the API document.

Parameters:

Name	Type	Description	Default
`uri`	`str`	Path of URI to be validated.	required

Returns:

Type	Description
`t.Tuple[t.Tuple[str, ...], t.Optional[t.Type[WSGIEndpoint]]]`	Pair of values of flexible locations and `Endpoint` class if specified `uri` is valid, otherwise, pari of empty tuple and None.

Source code in bamboo/app.py

def validate(
    self,
    uri: str
) -> t.Tuple[t.Tuple[str, ...], t.Optional[t.Type[WSGIEndpoint]]]:
    return super().validate(uri)

app

AppBase

tree: Uri2Endpoints_t property readonly

__init__(self, error_404=DefaultNotFoundErrInfo()) special

graft(self, *apps, *, onto=())

route(self, *locs, *, version=None)

search_uris(self, endpoint)

set_parcel(self, endpoint, *parcel)

validate(self, uri)

ASGIApp

__call__(self, scope, recv, send) async special

__init__(self, error_404=DefaultNotFoundErrInfo(), lifespan_handler=<function default_lifespan_handler at 0x7f28f4f14200>) special

apply_lifespan_handler(self, lifespan_handler)

handle_http(self, scope, recv, send) async

handle_lifespan(self, scope, recv, send) async

handle_websocket(self, scope, recv, send) async

route(self, *locs, *, version=None)

search_uris(self, endpoint)

set_parcel(self, endpoint, *parcel)

validate(self, uri)

ParcelConfig

__init__(self, endpoint) special

get(self, app)

get_all(self)

set(self, app, parcel)

VersionConfig

__init__(self, endpoint) special

get(self, app)

get_all(self)

set(self, app, version=None)

WSGIApp

route(self, *locs, *, version=None)

search_uris(self, endpoint)

send_404(self, start_response)

set_parcel(self, endpoint, *parcel)

validate(self, uri)

`AppBase`

`tree: Uri2Endpoints_t` `property` `readonly`

`init(self, error_404=DefaultNotFoundErrInfo())` `special`

`graft(self, apps, , onto=())`

`route(self, locs, , version=None)`

`search_uris(self, endpoint)`

`set_parcel(self, endpoint, *parcel)`

`validate(self, uri)`

`ASGIApp`

`call(self, scope, recv, send)` `async` `special`

`init(self, error_404=DefaultNotFoundErrInfo(), lifespan_handler=<function default_lifespan_handler at 0x7f28f4f14200>)` `special`

`apply_lifespan_handler(self, lifespan_handler)`

`handle_http(self, scope, recv, send)` `async`

`handle_lifespan(self, scope, recv, send)` `async`

`handle_websocket(self, scope, recv, send)` `async`

`route(self, locs, , version=None)`

`search_uris(self, endpoint)`

`set_parcel(self, endpoint, *parcel)`

`validate(self, uri)`

`ParcelConfig`

`init(self, endpoint)` `special`

`get(self, app)`

`get_all(self)`

`set(self, app, parcel)`

`VersionConfig`

`init(self, endpoint)` `special`

`get(self, app)`

`get_all(self)`

`set(self, app, version=None)`

`WSGIApp`

`route(self, locs, , version=None)`

`search_uris(self, endpoint)`

`send_404(self, start_response)`

`set_parcel(self, endpoint, *parcel)`

`validate(self, uri)`