faust.web.drivers.aiohttp

Web driver using https://pypi.org/project/aiohttp/.

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

Web server and framework implementation using https://pypi.org/project/aiohttp/.

driver_version: str = 'aiohttp=3.11.7'

handler_shutdown_timeout: float = 60.0

cors_options: Mapping[str, ResourceOptions]

property cors: CorsConfig

Return CORS config object. :rtype: CorsConfig

async on_start() → None

Call when the embedded web server starts.

Only used for faust worker, not when using wsgi().

Return type:

None

async wsgi() → Any

Call WSGI handler.

Used by https://pypi.org/project/gunicorn/ and other WSGI compatible hosts to access the Faust web entry point.

Return type:

Any

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

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

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

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:

Any

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

async read_request_content(request: Request) → bytes

Return the request body as bytes.

Return type:

bytes

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

Add route for web view or handler.

Return type:

None

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

Add route for static assets.

Return type:

None

bytes_to_response(s: bytes) → Response

Deserialize byte string back into a response object.

Return type:

Response

response_to_bytes(response: Response) → bytes

Convert response to serializable byte string.

The result is a byte string that can be deserialized using bytes_to_response().

Return type:

bytes

async start_server() → None

Start the web server.

Return type:

None

async stop_server() → None

Stop the web server.

Return type:

None

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