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) → None[source]
view_name_separator = ':'
cache(timeout: Union[datetime.timedelta, float, str] = None, key_prefix: str = None, backend: Union[Type[faust.types.web.CacheBackendT], str] = None) → faust.types.web.CacheT[source]
Return type:CacheT
route(uri: str, *, name: Optional[str] = None, base: Type[faust.types.web.View] = <class 'faust.types.web.View'>) → Callable[Union[Type[faust.types.web.View], Callable[[faust.types.web.View, faust.types.web.Request], Union[Coroutine[[Any, Any], faust.types.web.Response], Awaitable[faust.types.web.Response]]], Callable[[faust.types.web.View, faust.types.web.Request, Any, Any], Union[Coroutine[[Any, Any], faust.types.web.Response], Awaitable[faust.types.web.Response]]]], Union[Type[faust.types.web.View], Callable[[faust.types.web.View, faust.types.web.Request], Union[Coroutine[[Any, Any], faust.types.web.Response], Awaitable[faust.types.web.Response]]], Callable[[faust.types.web.View, faust.types.web.Request, Any, Any], Union[Coroutine[[Any, Any], faust.types.web.Response], Awaitable[faust.types.web.Response]]]]][source]
Return type: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]]]]]
static(uri: str, file_or_directory: Union[str, pathlib.Path], *, name: Optional[str] = None) → None[source]
Return type:None
register(app: faust.types.app.AppT, *, url_prefix: Optional[str] = None) → None[source]

Register a virtual subclass of an ABC.

Returns the subclass, to allow usage as a class decorator.

Return type:None
init_webserver(web: faust.types.web.Web) → None[source]
Return type:None
on_webserver_init(web: faust.types.web.Web) → None[source]
Return type:None