faust.web.base

Base interface for Web server and views.

class faust.web.base.BlueprintManager(initial: Optional[Iterable[Tuple[str, Union[BlueprintT, str]]]] = None)

Manager of all blueprints.

applied: bool

add(prefix: str, blueprint: Union[BlueprintT, str]) → None

Register blueprint with this app.

Return type:

None

apply(web: Web) → None

Apply all blueprints.

Return type:

None

class faust.web.base.Request

HTTP Request.

method: str

headers: Mapping[str, str]

url: URL

rel_url: URL

query_string: str

keep_alive: bool

body_exists: bool

user: Any

if_modified_since: Optional[datetime]

if_unmodified_since: Optional[datetime]

if_range: Optional[datetime]

abstract can_read_body() → bool

Return True if the request has a body.

Return type:

bool

abstract async read() → bytes

Read post data as bytes.

Return type:

bytes

abstract async text() → str

Read post data as text.

Return type:

str

abstract async json() → Any

Read post data and deserialize as JSON.

Return type:

Any

abstract async post() → Mapping[str, str]

Read post data.

Return type:

_GenericAlias[str, str]

abstract property match_info: Mapping[str, str]

Return match info from URL route as a mapping. :rtype: _GenericAlias[str, str]

abstract property query: Mapping[str, str]

Return HTTP query parameters as a mapping. :rtype: _GenericAlias[str, str]

abstract property cookies: Mapping[str, Any]

Return cookies as a mapping. :rtype: _GenericAlias[str, Any]

class faust.web.base.Response

Web server response and status.

abstract property status: int

Return the response status code. :rtype: int

abstract property body: bytes

Return the response body as bytes. :rtype: bytes

abstract property headers: MutableMapping

Return mapping of response HTTP headers. :rtype: _SpecialGenericAlias

abstract property content_length: Optional[int]

Return the size of the response body. :rtype: _UnionGenericAlias[int, None]

abstract property content_type: str

Return the response content type. :rtype: str

abstract property charset: Optional[str]

Return the response character set. :rtype: _UnionGenericAlias[str, None]

abstract property chunked: bool

Return True if response is chunked. :rtype: bool

abstract property compression: bool

Return True if the response body is compressed. :rtype: bool

abstract property keep_alive: Optional[bool]

Return True if HTTP keep-alive enabled. :rtype: _UnionGenericAlias[bool, None]

abstract property body_length: int

Size of HTTP response body. :rtype: int

class faust.web.base.Web(app: AppT, **kwargs: Any)

Web server and HTTP interface.

default_blueprints: ClassVar[Iterable[Tuple[str, Union[BlueprintT, str]]]] = [('/router', 'faust.web.apps.router:blueprint'), ('/table', 'faust.web.apps.tables.blueprint')]

production_blueprints: ClassVar[Iterable[Tuple[str, Union[BlueprintT, str]]]] = [('', 'faust.web.apps.production_index:blueprint')]

debug_blueprints: ClassVar[Iterable[Tuple[str, Union[BlueprintT, str]]]] = [('/graph', 'faust.web.apps.graph:blueprint'), ('', 'faust.web.apps.stats:blueprint')]

driver_version: str

content_separator: ClassVar[bytes] = b'\r\n\r\n'

header_separator: ClassVar[bytes] = b'\r\n'

header_key_value_separator: ClassVar[bytes] = b': '

app: AppT

views: MutableMapping[str, View]

reverse_names: MutableMapping[str, str]

blueprints: BlueprintManager

abstract text(value: str, *, content_type: Optional[str] = None, status: int = 200, reason: Optional[str] = None, headers: Optional[MutableMapping] = None) → Response

Create text response, using “text/plain” content-type.

Return type:

Response

abstract html(value: str, *, content_type: Optional[str] = None, status: int = 200, reason: Optional[str] = None, headers: Optional[MutableMapping] = None) → Response

Create HTML response from string, text/html content-type.

Return type:

Response

abstract json(value: Any, *, content_type: Optional[str] = None, status: int = 200, reason: Optional[str] = None, headers: Optional[MutableMapping] = None) → Response

Create new JSON response.

Accepts any JSON-serializable value and will automatically serialize it for you.

The content-type is set to “application/json”.

Return type:

Response

abstract bytes(value: bytes, *, content_type: Optional[str] = None, status: int = 200, reason: Optional[str] = None, headers: Optional[MutableMapping] = None) → Response

Create new bytes response - for binary data.

Return type:

Response

abstract bytes_to_response(s: bytes) → Response

Deserialize HTTP response from byte string.

Return type:

Response

abstract response_to_bytes(response: Response) → bytes

Serialize HTTP response into byte string.

Return type:

bytes

abstract route(pattern: str, handler: Callable, cors_options: Optional[Mapping[str, ResourceOptions]] = None) → None

Add route for handler.

Return type:

None

abstract add_static(prefix: str, path: Union[Path, str], **kwargs: Any) → None

Add static route.

Return type:

None

abstract async read_request_content(request: Request) → bytes

Read HTTP body as bytes.

Return type:

bytes

abstract async wsgi() → Any

WSGI entry point.

Return type:

Any

add_view(view_cls: Type[View], *, prefix: str = '', cors_options: Optional[Mapping[str, ResourceOptions]] = None) → View

Add route for view.

Return type:

View

url_for(view_name: str, **kwargs: Any) → str

Get URL by view name.

If the provided view name has associated URL parameters, those need to be passed in as kwargs, or a TypeError will be raised.

Return type:

str

init_server() → None

Initialize and setup web server.

Return type:

None

property url: URL

Return the canonical URL to this worker (including port). :rtype: URL

logger: logging.Logger = <Logger faust.web.base (WARNING)>