Source code for faust.web.apps.router

"""HTTP endpoint showing partition routing destinations."""
from faust import web
from faust.web.exceptions import ServiceUnavailable

__all__ = ['TableList', 'TableDetail', 'TableKeyDetail', 'blueprint']

blueprint = web.Blueprint('router')


[docs]@blueprint.route('/', name='list') class TableList(web.View): """List routes for all tables."""
[docs] async def get(self, request: web.Request) -> web.Response: """Return JSON response with list of all table routes.""" router = self.app.router return self.json(router.tables_metadata())
[docs]@blueprint.route('/{name}/', name='detail') class TableDetail(web.View): """List route for specific table."""
[docs] async def get(self, request: web.Request, name: str) -> web.Response: """Return JSON response with table metadata.""" router = self.app.router return self.json(router.table_metadata(name))
[docs]@blueprint.route('/{name}/{key}/', name='key-detail') class TableKeyDetail(web.View): """List information about key."""
[docs] async def get(self, request: web.Request, name: str, key: str) -> web.Response: """Return JSON response after looking up the route of a table key. Arguments: name: Name of table. key: Key to look up node for. Raises: faust.web.exceptions.ServiceUnavailable: if broker metadata has not yet been received. """ router = self.app.router try: dest_url = router.key_store(name, key) except KeyError: raise ServiceUnavailable() else: return self.json(str(dest_url))