faust.web.blueprints

Blueprints define reusable web apps.

They are lazy and need to be registered to an app to be activated:

from faust import web

blueprint = web.Blueprint('users')
cache = blueprint.cache(timeout=300.0)

@blueprint.route('/', name='list')
class UserListView(web.View):

    @cache.view()
    async def get(self, request: web.Request) -> web.Response:
        return web.json(...)

@blueprint.route('/{user_id}/', name='detail')
class UserDetailView(web.View):

    @cache.view(timeout=10.0)
    async def get(self,
                  request: web.Request,
                  user_id: str) -> web.Response:
        return web.json(...)

At this point the views are realized and can be used from Python code, but the cached get method handlers cannot be called yet.

To actually use the view from a web server, we need to register the blueprint to an app:

app = faust.App(
    'name',
    broker='kafka://',
    cache='redis://',
)

user_blueprint.register(app, url_prefix='/user/')

At this point the web server will have fully-realized views with actually cached method handlers.

The blueprint is registered with a prefix, so the URL for the UserListView is now /user/, and the URL for the UserDetailView is /user/{user_id}/.

Blueprints can be registered to multiple apps at the same time.

class faust.web.blueprints.Blueprint(name: str, *, url_prefix: Optional[str] = None)

Define reusable web application.

view_name_separator: str = ':'

routes: List[FutureRoute]

static_routes: List[FutureStaticRoute]

cache(timeout: Optional[Union[timedelta, float, str]] = None, include_headers: bool = False, key_prefix: Optional[str] = None, backend: Optional[Union[Type[CacheBackendT], str]] = None) → CacheT

Cache API.

Return type:

CacheT

route(uri: str, *, name: ~typing.Optional[str] = None, cors_options: ~typing.Optional[~typing.Mapping[str, ~faust.types.web.ResourceOptions]] = None, base: ~typing.Type[~faust.types.web.View] = <class 'faust.types.web.View'>) → Callable[[Union[Type[View], Callable[[View, Request], Union[Coroutine[Any, Any, Response], Awaitable[Response]]], Callable[[View, Request, Any, Any], Union[Coroutine[Any, Any, Response], Awaitable[Response]]]]], Union[Type[View], Callable[[View, Request], Union[Coroutine[Any, Any, Response], Awaitable[Response]]], Callable[[View, Request, Any, Any], Union[Coroutine[Any, Any, Response], Awaitable[Response]]]]]

Create route by decorating handler or view class.

Return type:

_CallableGenericAlias[_UnionGenericAlias[_GenericAlias[View], _CallableGenericAlias[View, Request, _UnionGenericAlias[_GenericAlias[Any, Any, Response], _GenericAlias[Response]]], _CallableGenericAlias[View, Request, Any, Any, _UnionGenericAlias[_GenericAlias[Any, Any, Response], _GenericAlias[Response]]]], _UnionGenericAlias[_GenericAlias[View], _CallableGenericAlias[View, Request, _UnionGenericAlias[_GenericAlias[Any, Any, Response], _GenericAlias[Response]]], _CallableGenericAlias[View, Request, Any, Any, _UnionGenericAlias[_GenericAlias[Any, Any, Response], _GenericAlias[Response]]]]]

static(uri: str, file_or_directory: Union[str, Path], *, name: Optional[str] = None) → None

Add static route.

Return type:

None

register(app: AppT, *, url_prefix: Optional[str] = None) → None

Register blueprint with app.

Return type:

None

init_webserver(web: Web) → None

Init blueprint for web server start.

Return type:

None

on_webserver_init(web: Web) → None

Call when web server starts.

Return type:

None