コンテンツにスキップ

endpoint

ASGIEndpointBase

Base class of endpoints compliant with the ASGI.

This class implements abstract methods of EndpointBase with the ASGI. However, this class doesn't implement some methods to structure responses.

Note

DO NOT use this class as the super class of your endpoints. Consider to use subclasses of the class like ASGIHTTPEndpoint.

asgi_version: str property readonly

ASGI version.

headers: t.Dict[str, str] property readonly

Request headers.

http_version: str property readonly

HTTP Version on communication.

path: str property readonly

Path of requested URI.

raw_path: bytes property readonly

The original HTTP path received from client.

root_path: str property readonly

The root path ASGI application is mounted at.

scheme: str property readonly

Scheme of requested URI.

scope: t.Dict[str, t.Any] property readonly

scope variable received from ASGI server.

scope_type: str property readonly

Scope type on ASGI.

spec_version: str property readonly

Spec version on ASGI.

__init__(self, app, scope, flexible_locs) special

Parameters:

Name Type Description Default
scope t.Dict[str, t.Any]

scope variable received from ASGI server.

 required
flexible_locs t.Tuple[str, ...]

Flexible locations requested.

 required
Source code in bamboo/endpoint.py 
def __init__(
    self,
    app: App_t,
    scope: t.Dict[str, t.Any],
    flexible_locs: t.Tuple[str, ...],
) -> None:
    """
    Args:
        scope: scope variable received from ASGI server.
        flexible_locs: Flexible locations requested.
    """
    self._scope = scope

    # TODO
    #   Consider not mapping in this method.
    req_headers = scope.get("headers")
    self._req_headers = {}
    if req_headers:
        req_headers = [map(codecs.decode, h) for h in req_headers]
        self._req_headers.update(dict(req_headers))

    super().__init__(app, flexible_locs)

get_client_addr(self)

Retrieve client address, pair of its IP address and port.

Note

IP address and port may be None if retrieving the address from server application would fail, so it is recommended to confirm your using server application's spec.

Returns:

Type Description
t.Tuple[t.Optional[str], t.Optional[str]]

Pair of IP and port of client.
Source code in bamboo/endpoint.py 
def get_client_addr(self) -> t.Tuple[t.Optional[str], t.Optional[str]]:
    client = self._scope.get("client")
    if client:
        return tuple(client)
    return (None, None)

get_header(self, name)

Retrive header value from requested headers.

Parameters:

Name Type Description Default
name str

Header name.

 required

Returns:

Type Description
t.Optional[str]

Value of header if existing, None otherwise.
Source code in bamboo/endpoint.py 
def get_header(self, name: str) -> t.Optional[str]:
    name = name.lower().replace("_", "-")
    return self._req_headers.get(name)

get_host_addr(self)

Retrive host name and port from requested headers.

Returns:

Type Description
t.Tuple[t.Optional[str], t.Optional[int]]

Pair of host name and port.
Source code in bamboo/endpoint.py 
def get_host_addr(self) -> t.Tuple[t.Optional[str], t.Optional[int]]:
    http_host = self.get_header("host")
    if http_host:
        http_host = http_host.split(":")
        if len(http_host) == 1:
            return (http_host[0], None)
        else:
            host, port = http_host
            port = int(port)
            return (host, port)
    return (None, None)

get_server_addr(self)

Retrive server address, pair of its IP address and port.

Note

IP address and port may be None if retrieving the address from server application would fail, so it is recommended to confirm your using server application's spec.

Returns:

Type Description
t.Tuple[t.Optional[str], t.Optional[str]]

Pair of IP and port of server.
Source code in bamboo/endpoint.py 
def get_server_addr(self) -> t.Tuple[t.Optional[str], t.Optional[str]]:
    server = self._scope.get("server")
    if server:
        return tuple(server)
    return (None, None)

ASGIHTTPEndpoint

HTTP endpoint class compliant with the ASGI.

This class is a complete class of endpoints, communicating on HTTP. This class has all attributes of ASGIEndpointBase and HTTPMixIn, and you can define its subclass and use them in your response methods.

Examples:

app = ASGIHTTPApp()

@app.route("hello")
class HelloEndpoint(ASGIHTTPEndpoint):

    # RECOMMEND to use `data_format` decorator
    async def do_GET(self) -> None:
        response = {"greeting": "Hello, Client!"}
        self.send_json(response)

    async def do_POST(self) -> None:
        req_body = async self.body
        print(req_body)

content_length: t.Optional[int] property readonly

Content length of request body if existing.

method: str property readonly

HTTP method requested from client.

__init__(self, app, scope, receive, flexible_locs) special

Parameters:

Name Type Description Default
app ASGIHTTPApp

Application object which routes the endpoint.

 required
scope t.Dict[str, t.Any]

Scope variable received from ASGI server.

 required
receive t.Callable[[], t.Awaitable[t.Dict[str, t.Any]]]

receive method given from ASGI server.

 required
flexible_locs t.Tuple[str, ...]

Flexible locations requested.

 required
Source code in bamboo/endpoint.py 
def __init__(
    self,
    app: ASGIHTTPApp,
    scope: t.Dict[str, t.Any],
    receive: t.Callable[[], t.Awaitable[t.Dict[str, t.Any]]],
    flexible_locs: t.Tuple[str, ...],
) -> None:
    """
    Args:
        app: Application object which routes the endpoint.
        scope: Scope variable received from ASGI server.
        receive: `receive` method given from ASGI server.
        flexible_locs: Flexible locations requested.
    """
    self._receive = receive
    self._is_disconnected = False

    ASGIEndpointBase.__init__(self, app, scope, flexible_locs)
    HTTPMixIn.__init__(self)

__init_subclass__() classmethod special

This method is called when a class is subclassed.

The default implementation does nothing. It may be overridden to extend subclasses.

Source code in bamboo/endpoint.py 
def __init_subclass__(cls) -> None:
    super().__init_subclass__()

    # NOTE
    #   All response methods of its subclass must be awaitables.
    for method in HTTPMethods:
        callback = cls._get_response_method(method)
        if callback and not inspect.iscoroutinefunction(callback):
            raise TypeError(
                f"{cls.__name__}.{callback.__name__} must be an awaitable"
                ", not a callable."
            )

get_req_body_iter(self, bufsize=8192, cache=False)

Make an access to the request body as an iterator.

Note

If the flag cache is True, the request body data is to be cached into the property body, i.e. one always can access to the request body even after the iteration. On the other hand, if the cache is False, caching is not conducted and access to the body will be failed.

Parameters:

Name Type Description Default
bufsize int

Chunk size of each item.

 8192
cache bool

If the request body is to be cached or not.

 False

Returns:

Type Description
t.AsyncGenerator[bytes, None]

Async iterator with binary of the request body.
Source code in bamboo/endpoint.py 
async def get_req_body_iter(
    self,
    bufsize: int = 8192,
    cache: bool = False,
) -> t.AsyncGenerator[bytes, None]:
    """Make an access to the request body as an iterator.

    Note:
        If the flag `cache` is `True`, the request body data is to be
        cached into the property `body`, i.e. one always can access
        to the request body even after the iteration. On the other hand,
        if the `cache` is `False`, caching is not conducted and
        access to the `body` will be failed.

    Args:
        bufsize: Chunk size of each item.
        cache: If the request body is to be cached or not.

    Returns:
        Async iterator with binary of the request body.
    """
    buffer = io.BytesIO()
    more_body = True
    cacher = await self.__class__.body

    while more_body:
        chunk = await self._receive()
        type = chunk.get("type")
        if type == ASGIHTTPEvents.disconnect:
            self._is_disconnected = True
            return

        body = chunk.get("body", b"")
        buffer.write(body)
        buffer.flush()
        more_body = chunk.get("more_body", False)

        while buffer.tell() >= bufsize:
            buffer.seek(0)
            item = buffer.read(bufsize)
            yield item

            if cache:
                # TODO
                #   Seek more efficient ways
                if cacher._has_cache(self):
                    item = cacher._get_cache(self) + item

                cacher._set_cache(self, item)

            rest = buffer.read()
            buffer.close()
            buffer = io.BytesIO(rest)

    # TODO
    #   Seek more efficient ways
    buffer.seek(0)
    item = buffer.read()
    yield item
    if cache:
        if cacher._has_cache(self):
            item = cacher._get_cache(self) + item

        cacher._set_cache(self, item)

    buffer.close()

EndpointBase

Base class of Endpoint to define logic to requests.

Endpoint is one of the core concept of Bamboo, and this class defines its basic behavior. All endpoints must inherit this class.

Note

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

app: App_t property readonly

Application object handling the endpoint.

flexible_locs: t.Tuple[str, ...] property readonly

Flexible locations extracted from requested URI.

http_version: str property readonly

HTTP Version on communication.

path: str property readonly

Path of requested URI.

scheme: str property readonly

Scheme of requested URI.

__init__(self, app, flexible_locs) special

Note

DO NOT generate its instance. Its object will be initialized by application object.

Parameters:

Name Type Description Default
app App_t

Application object which routes the endpoint.

 required
flexible_locs t.Tuple[str, ...]

Flexible locations requested.

 required
Source code in bamboo/endpoint.py 
def __init__(
    self,
    app: App_t,
    flexible_locs: t.Tuple[str, ...],
) -> None:
    """
    Note:
        DO NOT generate its instance. Its object will be initialized
        by application object.

    Args:
        app: Application object which routes the endpoint.
        flexible_locs: Flexible locations requested.
    """
    self._app = app
    self._flexible_locs = flexible_locs

get_client_addr(self)

Retrieve client address, pair of its IP address and port.

Note

IP address and port may be None if retrieving the address from server application would fail, so it is recommended to confirm your using server application's spec.

Returns:

Type Description
t.Tuple[t.Optional[str], t.Optional[int]]

Pair of IP and port of client.
Source code in bamboo/endpoint.py 
@abstractmethod
def get_client_addr(self) -> t.Tuple[t.Optional[str], t.Optional[int]]:
    """Retrieve client address, pair of its IP address and port.

    Note:
        IP address and port may be None if retrieving the address from
        server application would fail, so it is recommended to confirm
        your using server application's spec.

    Returns:
        Pair of IP and port of client.
    """
    pass

get_header(self, name)

Retrive header value from requested headers.

Parameters:

Name Type Description Default
name str

Header name.

 required

Returns:

Type Description
t.Optional[str]

Value of header if existing, None otherwise.
Source code in bamboo/endpoint.py 
@abstractmethod
def get_header(self, name: str) -> t.Optional[str]:
    """Retrive header value from requested headers.

    Args:
        name: Header name.

    Returns:
        Value of header if existing, None otherwise.
    """
    pass

get_host_addr(self)

Retrive host name and port from requested headers.

Returns:

Type Description
t.Tuple[t.Optional[str], t.Optional[int]]

Pair of host name and port.
Source code in bamboo/endpoint.py 
@abstractmethod
def get_host_addr(self) -> t.Tuple[t.Optional[str], t.Optional[int]]:
    """Retrive host name and port from requested headers.

    Returns:
        Pair of host name and port.
    """
    pass

get_queries(self, name)

Get values of query parameter.

Parameters:

Name Type Description Default
name str

Key name of the parameter

 required

Returns:

Type Description
t.List[str]

Value of the parameter. The value of list may have multiple items if client specifies the parameter in several times.
Source code in bamboo/endpoint.py 
def get_queries(self, name: str) -> t.List[str]:
    """Get values of query parameter.

    Args:
        name: Key name of the parameter

    Returns:
        Value of the parameter. The value of list may have multiple
        items if client specifies the parameter in several times.
    """
    query = self.queries.get(name)
    if query:
        return query
    return []

get_server_addr(self)

Retrive server address, pair of its IP address and port.

Note

IP address and port may be None if retrieving the address from server application would fail, so it is recommended to confirm your using server application's spec.

Returns:

Type Description
t.Tuple[t.Optional[str], t.Optional[int]]

Pair of IP and port of server.
Source code in bamboo/endpoint.py 
@abstractmethod
def get_server_addr(self) -> t.Tuple[t.Optional[str], t.Optional[int]]:
    """Retrive server address, pair of its IP address and port.

    Note:
        IP address and port may be None if retrieving the address from
        server application would fail, so it is recommended to confirm
        your using server application's spec.

    Returns:
        Pair of IP and port of server.
    """
    pass

setup(self, *parcel)

Execute setup of the endpoint object.

This method will execute at initialization of the object by specifying parcel. The parcel is sent with set_parcel() method of the application object which has included the object as one of the its endpoints. This method can be used as a substitute for the __init__ method.

This method is useful in some cases like below:

  • Making an endpoint class a reusable component
  • Injecting environmental dependencies using something like a setting file

Parameters:

Name Type Description Default
*parcel

Parcel sent via application object.

 ()

Examples:

app = WSGIApp()

@app.route("hello")
class HelloEndpoint(WSGIEndpoint):

    def setup(self, server_name: str) -> None:
        self._server_name = server_name

    def do_GET(self) -> None:
        self.send_body(f"Hello from {self._server_name}".encode())

if __name__ == "__main__":
    SERVER_NAME = "awesome_server"
    app.set_parcel(HelloEndpoint, SERVER_NAME)

    WSGITestExecutor.debug(app, "", 8000)
Source code in bamboo/endpoint.py 
def setup(self, *parcel) -> None:
    """Execute setup of the endpoint object.

    This method will execute at initialization of the object by specifying
    parcel. The parcel is sent with `set_parcel()` method of
    the application object which has included the object as one of
    the its endpoints. This method can be used as a substitute for
    the `__init__` method.

    This method is useful in some cases like below:

    - Making an endpoint class a reusable component
    - Injecting environmental dependencies using something like a
        setting file

    Args:
        *parcel: Parcel sent via application object.

    Examples:
        ```python
        app = WSGIApp()

        @app.route("hello")
        class HelloEndpoint(WSGIEndpoint):

            def setup(self, server_name: str) -> None:
                self._server_name = server_name

            def do_GET(self) -> None:
                self.send_body(f"Hello from {self._server_name}".encode())

        if __name__ == "__main__":
            SERVER_NAME = "awesome_server"
            app.set_parcel(HelloEndpoint, SERVER_NAME)

            WSGITestExecutor.debug(app, "", 8000)
        ```
    """
    pass

HTTPMixIn

Mixin class for HTTP endpoints.

This class assumes that endpoint classes inherit this class for HTTP. So, this class do not work alone.

Note

DO NOT use this class alone. This class work correctly by inheriting it, implementing its abstract methods, and call its __init__() method in the one of the subclass.

content_length: t.Optional[int] property readonly

Content length of request body if existing.

method: str property readonly

HTTP method requested from client.

__init_subclass__() classmethod special

This method is called when a class is subclassed.

The default implementation does nothing. It may be overridden to extend subclasses.

Source code in bamboo/endpoint.py 
def __init_subclass__(cls) -> None:
    super().__init_subclass__()

    cls._pre_methods = {}
    cls._res_methods = {}

    # Check if bufsize is positive
    if not (cls.bufsize > 0 and isinstance(cls.bufsize, int)):
        raise ValueError(
            f"{cls.__name__}.bufsize must be positive integer"
        )

    # Check pre & response methods
    for method in _AVAILABLE_RES_METHODS:
        name_pre_method = _PREFIX_PRE_RESPONSE + method
        name_res_method = _PREFIX_RESPONSE + method

        if hasattr(cls, name_pre_method):
            pre_method = getattr(cls, name_pre_method)
            cls._pre_methods[method] = pre_method

        if hasattr(cls, name_res_method):
            res_method = getattr(cls, name_res_method)
            cls._res_methods[method] = res_method

add_content_length(self, length)

Add Content-Length header of response.

Parameters:

Name Type Description Default
length int

Size of response body.

 required
Source code in bamboo/endpoint.py 
def add_content_length(self, length: int) -> None:
    """Add Content-Length header of response.

    Args:
        length: Size of response body.
    """
    self.add_header("Content-Length", str(length))

add_content_length_body(self, body)

Add Content-Length header of response by response body.

Parameters:

Name Type Description Default
body bytes

Response body.

 required
Source code in bamboo/endpoint.py 
def add_content_length_body(self, body: bytes) -> None:
    """Add Content-Length header of response by response body.

    Args:
        body: Response body.
    """
    self.add_header("Content-Length", str(len(body)))

add_content_type(self, content_type)

Add Content-Type header of response.

Parameters:

Name Type Description Default
content_type ContentType

Information of Content-Type header.

 required
Source code in bamboo/endpoint.py 
def add_content_type(self, content_type: ContentType) -> None:
    """Add Content-Type header of response.

    Args:
        content_type: Information of Content-Type header.
    """
    self.add_header(*content_type.to_header())

add_header(self, name, value, **params)

Add response header with MIME parameters.

Parameters:

Name Type Description Default
name str

Field name of the header.

 required
value str

Value of the field.

 required
**params str

Directives added to the field.

 {}
Source code in bamboo/endpoint.py 
def add_header(
    self,
    name: str,
    value: str,
    **params: str
) -> None:
    """Add response header with MIME parameters.

    Args:
        name: Field name of the header.
        value: Value of the field.
        **params: Directives added to the field.
    """
    self._res_headers.append(make_header(name, value, **params))

add_headers(self, *headers)

Add response headers at once.

Note

This method would be used as a shortcut to register multiple headers. If it requires adding MIME parameters, developers can use the 'add_header' method.

Parameters:

Name Type Description Default
**headers t.Tuple[str, str]

Header's info whose header is the field name.

 ()
Source code in bamboo/endpoint.py 
def add_headers(self, *headers: t.Tuple[str, str]) -> None:
    """Add response headers at once.

    Note:
        This method would be used as a shortcut to register multiple
        headers. If it requires adding MIME parameters, developers
        can use the 'add_header' method.

    Args:
        **headers: Header's info whose header is the field name.
    """
    for name, val in headers:
        self.add_header(name, val)

send_api(self, api, status=<HTTPStatus.OK: 200>)

Set given api data to the response body.

Parameters:

Name Type Description Default
api ApiData

ApiData object to be sent.

 required
status HTTPStatus

HTTP status of the response.

 <HTTPStatus.OK: 200>

Exceptions:

Type Description
StatusCodeAlreadySetError

Raised if response status code has already been set.
Source code in bamboo/endpoint.py 
def send_api(
    self,
    api: ApiData,
    status: HTTPStatus = HTTPStatus.OK,
) -> None:
    """Set given api data to the response body.

    Args:
        api: ApiData object to be sent.
        status: HTTP status of the response.

    Raises:
        StatusCodeAlreadySetError: Raised if response status code
            has already been set.
    """
    self.send_body(
        api.__extract__(),
        content_type=api.__content_type__,
        status=status,
    )

send_body(self, body, *others, *, content_type=ContentType(media_type='text/plain', charset=None, boundary=None), status=<HTTPStatus.OK: 200>)

Set given binary to the response body.

Note

If the parameter content_type is specified, then the Content-Type header is to be added.

DEFAULT_CONTENT_TYPE_PLAIN has its MIME type of text/plain, and the other attributes are None. If another value of Content-Type is needed, then you should generate new ContentType instance with attributes you want.

Parameters:

Name Type Description Default
body t.Union[bytes, t.Iterable[bytes]]

Binary to be set to the response body.

 required
content_type t.Optional[ContentType]

Content-Type header to be sent.

 ContentType(media_type='text/plain', charset=None, boundary=None)
status HTTPStatus

HTTP status of the response.

 <HTTPStatus.OK: 200>

Exceptions:

Type Description
StatusCodeAlreadySetError

Raised if response status code has already been set.
Source code in bamboo/endpoint.py 
def send_body(
    self,
    body: t.Union[bytes, t.Iterable[bytes]],
    *others: t.Union[bytes, t.Iterable[bytes]],
    content_type: t.Optional[ContentType] = DEFAULT_CONTENT_TYPE_PLAIN,
    status: HTTPStatus = HTTPStatus.OK
) -> None:
    """Set given binary to the response body.

    Note:
        If the parameter `content_type` is specified, then
        the `Content-Type` header is to be added.

        `DEFAULT_CONTENT_TYPE_PLAIN` has its MIME type of `text/plain`,
        and the other attributes are `None`. If another value of
        `Content-Type` is needed, then you should generate new
        `ContentType` instance with attributes you want.

    Args:
        body: Binary to be set to the response body.
        content_type: `Content-Type` header to be sent.
        status: HTTP status of the response.

    Raises:
        StatusCodeAlreadySetError: Raised if response status code has
            already been set.
    """
    self._set_status_safely(status)

    bodies = [body]
    bodies.extend(others)

    is_all_bytes = True
    not_empty = False
    for chunk in bodies:
        is_all_bytes &= isinstance(chunk, bytes)
        if is_all_bytes:
            not_empty |= len(chunk) > 0
        self._res_body.append(chunk)

    if content_type:
        self.add_content_type(content_type)

    # Content-Length if avalidable
    if is_all_bytes and not_empty:
        length = sum(map(len, bodies))
        self.add_content_length(length)

send_file(self, path, fname=None, content_type=ContentType(media_type='text/plain', charset=None, boundary=None), status=<HTTPStatus.OK: 200>)

Set file to be sent as response.

Parameters:

Name Type Description Default
path str

File path.

 required
fname t.Optional[str]

File name to be sent.

 None
content_type str

Content type of the response.

 ContentType(media_type='text/plain', charset=None, boundary=None)
status HTTPStatus

HTTP status of the response.

 <HTTPStatus.OK: 200>
Source code in bamboo/endpoint.py 
def send_file(
    self,
    path: str,
    fname: t.Optional[str] = None,
    content_type: str = DEFAULT_CONTENT_TYPE_PLAIN,
    status: HTTPStatus = HTTPStatus.OK
) -> None:
    """Set file to be sent as response.

    Args:
        path: File path.
        fname: File name to be sent.
        content_type: Content type of the response.
        status: HTTP status of the response.
    """
    file_iter = BufferedFileIterator(path)
    self.send_body(file_iter, content_type=content_type, status=status)

    length = os.path.getsize(path)
    self.add_header("Content-Length", str(length))
    if fname:
        self.add_header(
            "Content-Disposition",
            "attachment",
            filename=fname
        )

send_json(self, body, status=<HTTPStatus.OK: 200>, encoding='UTF-8')

Set given json data to the response body.

Parameters:

Name Type Description Default
body t.Dict[str, t.Any]

Json data to be set to the response body.

 required
status HTTPStatus

HTTP status of the response.

 <HTTPStatus.OK: 200>
encoding str

Encoding of the Json data.

 'UTF-8'

Exceptions:

Type Description
StatusCodeAlreadySetError

Raised if response status code has already been set.
Source code in bamboo/endpoint.py 
def send_json(
    self,
    body: t.Dict[str, t.Any],
    status: HTTPStatus = HTTPStatus.OK,
    encoding: str = "UTF-8"
) -> None:
    """Set given json data to the response body.

    Args:
        body: Json data to be set to the response body.
        status: HTTP status of the response.
        encoding: Encoding of the Json data.

    Raises:
        StatusCodeAlreadySetError: Raised if response status code
            has already been set.
    """
    body = json.dumps(body).encode(encoding=encoding)
    content_type = ContentType(MediaTypes.json, encoding)
    self.send_body(body, content_type=content_type, status=status)

send_only_status(self, status=<HTTPStatus.OK: 200>)

Set specified status code to one of response.

This method can be used if a callback doesn't need to send response body.

Parameters:

Name Type Description Default
status HTTPStatus

HTTP status of the response.

 <HTTPStatus.OK: 200>
Source code in bamboo/endpoint.py 
def send_only_status(self, status: HTTPStatus = HTTPStatus.OK) -> None:
    """Set specified status code to one of response.

    This method can be used if a callback doesn't need to send response
    body.

    Args:
        status: HTTP status of the response.
    """
    self._set_status_safely(status)

StaticEndpoint

content_type: ContentType property readonly

Content type of request body.

Returns:

Type Description
ContentType

Content type if existing, None otherwise.

setup(self, doc_root)

Execute setup of the endpoint object.

This method will execute at initialization of the object by specifying parcel. The parcel is sent with set_parcel() method of the application object which has included the object as one of the its endpoints. This method can be used as a substitute for the __init__ method.

This method is useful in some cases like below:

  • Making an endpoint class a reusable component
  • Injecting environmental dependencies using something like a setting file

Parameters:

Name Type Description Default
*parcel

Parcel sent via application object.

 required

Examples:

app = WSGIApp()

@app.route("hello")
class HelloEndpoint(WSGIEndpoint):

    def setup(self, server_name: str) -> None:
        self._server_name = server_name

    def do_GET(self) -> None:
        self.send_body(f"Hello from {self._server_name}".encode())

if __name__ == "__main__":
    SERVER_NAME = "awesome_server"
    app.set_parcel(HelloEndpoint, SERVER_NAME)

    WSGITestExecutor.debug(app, "", 8000)
Source code in bamboo/endpoint.py 
def setup(self, doc_root: str) -> None:
    self._filepath = os.path.join(doc_root, *self.path[1:].split("/"))
    self._content_type = ContentType(file2mime(self._filepath))

    if not os.path.isfile(self.filepath) and not os.path.isdir(self.filepath):
        raise DEFAULT_NOT_FOUND_ERROR

StaticRedirectASGIEndpoint

setup(self, doc_root, suffix)

Execute setup of the endpoint object.

This method will execute at initialization of the object by specifying parcel. The parcel is sent with set_parcel() method of the application object which has included the object as one of the its endpoints. This method can be used as a substitute for the __init__ method.

This method is useful in some cases like below:

  • Making an endpoint class a reusable component
  • Injecting environmental dependencies using something like a setting file

Parameters:

Name Type Description Default
*parcel

Parcel sent via application object.

 required

Examples:

app = WSGIApp()

@app.route("hello")
class HelloEndpoint(WSGIEndpoint):

    def setup(self, server_name: str) -> None:
        self._server_name = server_name

    def do_GET(self) -> None:
        self.send_body(f"Hello from {self._server_name}".encode())

if __name__ == "__main__":
    SERVER_NAME = "awesome_server"
    app.set_parcel(HelloEndpoint, SERVER_NAME)

    WSGITestExecutor.debug(app, "", 8000)
Source code in bamboo/endpoint.py 
def setup(self, doc_root: str, suffix: str) -> None:
    super().setup(doc_root)

    self._suffix = suffix

StaticRedirectWSGIEndpoint

setup(self, doc_root)

Execute setup of the endpoint object.

This method will execute at initialization of the object by specifying parcel. The parcel is sent with set_parcel() method of the application object which has included the object as one of the its endpoints. This method can be used as a substitute for the __init__ method.

This method is useful in some cases like below:

  • Making an endpoint class a reusable component
  • Injecting environmental dependencies using something like a setting file

Parameters:

Name Type Description Default
*parcel

Parcel sent via application object.

 required

Examples:

app = WSGIApp()

@app.route("hello")
class HelloEndpoint(WSGIEndpoint):

    def setup(self, server_name: str) -> None:
        self._server_name = server_name

    def do_GET(self) -> None:
        self.send_body(f"Hello from {self._server_name}".encode())

if __name__ == "__main__":
    SERVER_NAME = "awesome_server"
    app.set_parcel(HelloEndpoint, SERVER_NAME)

    WSGITestExecutor.debug(app, "", 8000)
Source code in bamboo/endpoint.py 
def setup(self, doc_root: str) -> None:
    super().setup(doc_root)

StatusCodeAlreadySetError

Raised if response status code has already been set.

WSGIEndpoint

HTTP endpoint class compliant with the WSGI.

This class is a complete class of endpoints, communicating on HTTP. This class has all attributes of WSGIEndpointBase and HTTPMixIn, and you can define its subclass and use them in your response methods.

Examples:

app = WSGIApp()

@app.route("hello")
class HelloEndpoint(WSGIEndpoint):

    # RECOMMEND to use `data_format` decorator
    def do_GET(self) -> None:
        response = {"greeting": "Hello, Client!"}
        self.send_json(response)

    def do_POST(self) -> None:
        req_body = self.body
        print(req_body)

content_length: t.Optional[int] property readonly

Content length of request body if existing.

method: str property readonly

HTTP method requested from client.

__init__(self, app, environ, flexible_locs) special

Parameters:

Name Type Description Default
app WSGIApp

Application object which routes the endpoint.

 required
environ t.Dict[str, t.Any]

Environ variable received from WSGI server.

 required
flexible_locs t.Tuple[str, ...]

Flexible locations requested.

 required
Source code in bamboo/endpoint.py 
def __init__(
    self,
    app: WSGIApp,
    environ: t.Dict[str, t.Any],
    flexible_locs: t.Tuple[str, ...],
) -> None:
    """
    Args:
        app: Application object which routes the endpoint.
        environ: Environ variable received from WSGI server.
        flexible_locs: Flexible locations requested.
    """
    WSGIEndpointBase.__init__(self, app, environ, flexible_locs)
    HTTPMixIn.__init__(self)

get_req_body_iter(self, bufsize=8192, cache=False)

Make an access to the request body as an iterator.

Note

If the flag cache is True, the request body data is to be cached into the property body, i.e. one always can access to the request body even after the iteration. On the other hand, if the cache is False, caching is not conducted and access to the body will be failed.

Parameters:

Name Type Description Default
bufsize int

Chunk size of each item.

 8192
cache bool

If the request body is to be cached or not.

 False

Returns:

Type Description
t.Generator[bytes, None, None]

Iterator with binary of the request body.
Source code in bamboo/endpoint.py 
def get_req_body_iter(
    self,
    bufsize: int = 8192,
    cache: bool = False,
) -> t.Generator[bytes, None, None]:
    """Make an access to the request body as an iterator.

    Note:
        If the flag `cache` is `True`, the request body data is to be
        cached into the property `body`, i.e. one always can access
        to the request body even after the iteration. On the other hand,
        if the `cache` is `False`, caching is not conducted and
        access to the `body` will be failed.

    Args:
        bufsize: Chunk size of each item.
        cache: If the request body is to be cached or not.

    Returns:
        Iterator with binary of the request body.
    """
    stream = self.get_req_body_stream()
    cacher = self.__class__.body
    length = self.content_length
    remain = length

    while True:
        if length is None:
            chunk = stream.read(bufsize)
        elif remain <= 0:
            break
        else:
            counts = remain if remain < bufsize else bufsize
            chunk = stream.read(counts)
            remain -= len(chunk)

        if not chunk:
            break
        yield chunk

        # TODO
        #   Seek more efficient ways
        if cache:
            if cacher._has_cache(self):
                chunk = cacher._get_cache(self) + chunk

            cacher._set_cache(self, chunk)

get_req_body_stream(self)

Fetch the stream with which request body can be received.

Returns:

Type Description
io.BufferedIOBase

The stream with request body.
Source code in bamboo/endpoint.py 
def get_req_body_stream(self) -> io.BufferedIOBase:
    """Fetch the stream with which request body can be received.

    Returns:
        The stream with request body.
    """
    return self._environ.get("wsgi.input")

WSGIEndpointBase

Base class of endpoints compliant with the WSGI.

This class implements abstract methods of EndpointBase with the WSGI. However, this class doesn't implement some methods to structure responses.

Note

DO NOT use this class as the super class of your endpoints. Consider to use subclasses of the class like WSGIEndpoint.

environ: t.Dict[str, t.Any] property readonly

environ variable received from WSGI server.

http_version: str property readonly

HTTP Version on communication.

path: str property readonly

Path of requested URI.

scheme: str property readonly

Scheme of requested URI.

server_software: str property readonly

Software name of WSGI server.

wsgi_version: str property readonly

WSGI version number.

__init__(self, app, environ, flexible_locs) special

Parameters:

Name Type Description Default
environ t.Dict[str, t.Any]

environ variable received from WSGI server.

 required
flexible_locs t.Tuple[str, ...]

flexible locations requested.

 required
Source code in bamboo/endpoint.py 
def __init__(
    self,
    app: WSGIApp,
    environ: t.Dict[str, t.Any],
    flexible_locs: t.Tuple[str, ...],
) -> None:
    """
    Args:
        environ: environ variable received from WSGI server.
        flexible_locs: flexible locations requested.
    """
    self._environ = environ
    super().__init__(app, flexible_locs)

get_client_addr(self)

Retrieve client address, pair of its IP address and port.

Note

IP address and port may be None if retrieving the address from server application would fail, so it is recommended to confirm your using server application's spec.

Returns:

Type Description
t.Tuple[t.Optional[str], t.Optional[int]]

Pair of IP and port of client.
Source code in bamboo/endpoint.py 
def get_client_addr(self) -> t.Tuple[t.Optional[str], t.Optional[int]]:
    client = self._environ.get("REMOTE_ADDR")
    port = self._environ.get("REMOTE_PORT")
    if port:
        port = int(port)
    return (client, port)

get_header(self, name)

Retrive header value from requested headers.

Parameters:

Name Type Description Default
name str

Header name.

 required

Returns:

Type Description
t.Optional[str]

Value of header if existing, None otherwise.
Source code in bamboo/endpoint.py 
def get_header(self, name: str) -> t.Optional[str]:
    name = name.upper().replace("-", "_")
    if name == "CONTENT_TYPE":
        return self.content_type
    if name == "CONTENT_LENGTH":
        return self._environ.get("CONTENT_LENGTH")

    name = "HTTP_" + name
    return self._environ.get(name)

get_host_addr(self)

Retrive host name and port from requested headers.

Returns:

Type Description
t.Tuple[t.Optional[str], t.Optional[int]]

Pair of host name and port.
Source code in bamboo/endpoint.py 
def get_host_addr(self) -> t.Tuple[t.Optional[str], t.Optional[int]]:
    http_host = self._environ.get("HTTP_HOST")
    if http_host:
        http_host = http_host.split(":")
        if len(http_host) == 1:
            return (http_host[0], None)
        else:
            host, port = http_host
            port = int(port)
            return (host, port)
    return (None, None)

get_server_addr(self)

Retrive server address, pair of its IP address and port.

Note

IP address and port may be None if retrieving the address from server application would fail, so it is recommended to confirm your using server application's spec.

Returns:

Type Description
t.Tuple[t.Optional[str], t.Optional[int]]

Pair of IP and port of server.
Source code in bamboo/endpoint.py 
def get_server_addr(self) -> t.Tuple[t.Optional[str], t.Optional[int]]:
    server = self._environ.get("SERVER_NAME")
    port = self._environ.get("SERVER_PORT")
    if port:
        port = int(port)
    return (server, port)
Back to top