fastack.app¶
Fastack (FastAPI)
¶
Fastack application aims to support:
- App settings
- Adding a plugin
- Adding a command
- Adding a controller to create a REST APIs
- Access
app,request,websocket,stateobjects globally (like Flask)
Source code in fastack/app.py
class Fastack(FastAPI):
"""
Fastack application aims to support:
* App settings
* Adding a plugin
* Adding a command
* Adding a controller to create a REST APIs
* Access ``app``, ``request``, ``websocket``, ``state`` objects globally (like Flask)
"""
# Storage for all commands and will be added to the "fastack" command, so you can access it.
cli = Typer()
def set_settings(self, settings: ModuleType):
"""
Set settings for the application.
Args:
settings (ModuleType): settings module
"""
self.state.settings = settings
def get_setting(self, name: str, default: Any = None):
"""
Get setting value by name.
Args:
name (str): Setting name
default (Any, optional): Default value if setting is not found. Defaults to None.
Returns:
Any: Setting value
"""
assert self.state.settings is not None
return getattr(self.state.settings, name, default)
def load_plugins(self):
"""
Load plugins from settings.
"""
for plugin_str in self.get_setting("PLUGINS", []):
plugin_str += ".setup"
plugin = import_attr(plugin_str)
plugin(self)
def load_commands(self):
"""
Load commands from settings.
"""
for command_str in self.get_setting("COMMANDS", []):
command: Union[Callable, Typer] = import_attr(command_str)
if isinstance(command, Typer):
self.cli.add_typer(command)
else:
self.cli.command(command.__name__)(command)
def include_controller(
self,
controller: Controller,
*,
prefix: str = "",
tags: Optional[List[str]] = None,
dependencies: Optional[Sequence[params.Depends]] = None,
default_response_class: Type[Response] = Default(JSONResponse),
responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
callbacks: Optional[List[BaseRoute]] = None,
routes: Optional[List[BaseRoute]] = None,
redirect_slashes: bool = True,
default: Optional[ASGIApp] = None,
dependency_overrides_provider: Optional[Any] = None,
route_class: Type[APIRoute] = APIRoute,
on_startup: Optional[Sequence[Callable[[], Any]]] = None,
on_shutdown: Optional[Sequence[Callable[[], Any]]] = None,
deprecated: Optional[bool] = None,
include_in_schema: bool = True,
):
"""
Include controller to the application.
Args:
controller (Controller): Controller instance
For other parameters, please see the documentation of ``fastapi.APIRouter.add_api_route``.
"""
assert isinstance(
controller, Controller
), f"Controller must be an instance of {Controller!r}"
router = controller.build(
prefix=prefix,
tags=tags,
dependencies=dependencies,
default_response_class=default_response_class,
responses=responses,
callbacks=callbacks,
routes=routes,
redirect_slashes=redirect_slashes,
default=default,
dependency_overrides_provider=dependency_overrides_provider,
route_class=route_class,
on_startup=on_startup,
on_shutdown=on_shutdown,
deprecated=deprecated,
include_in_schema=include_in_schema,
)
self.include_router(router)
@property
def middleware(self) -> MiddlewareManager: # type: ignore[override]
return MiddlewareManager(self)
def app_context(self, with_lifespan: bool = True):
return AppContext(self, with_lifespan=with_lifespan)
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
req_token: Optional[Token] = None
ws_token: Optional[Token] = None
try:
async with self.app_context(with_lifespan=False):
scope_type = scope["type"]
# If the scope is http we will create a request instance object and add it to the global stack,
# so that it can be accessed via ``fastack.globals.request``
if scope_type == "http":
request = Request(scope, receive)
req_token = _request_ctx_stack.set(request)
# Same as above, but for websocket
elif scope_type == "websocket":
websocket = WebSocket(scope, receive, send)
ws_token = _websocket_ctx_stack.set(websocket)
await super().__call__(scope, receive, send)
finally:
# Clean global stack, when app finish processing request
if req_token:
_request_ctx_stack.reset(req_token)
if ws_token:
_websocket_ctx_stack.reset(ws_token)
get_setting(self, name, default=None)
¶
Get setting value by name.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name |
str |
Setting name |
required |
default |
Any |
Default value if setting is not found. Defaults to None. |
None |
Returns:
| Type | Description |
|---|---|
Any |
Setting value |
Source code in fastack/app.py
def get_setting(self, name: str, default: Any = None):
"""
Get setting value by name.
Args:
name (str): Setting name
default (Any, optional): Default value if setting is not found. Defaults to None.
Returns:
Any: Setting value
"""
assert self.state.settings is not None
return getattr(self.state.settings, name, default)
include_controller(self, controller, *, prefix='', tags=None, dependencies=None, default_response_class=<fastapi.datastructures.DefaultPlaceholder object at 0x7f0df3659790>, responses=None, callbacks=None, routes=None, redirect_slashes=True, default=None, dependency_overrides_provider=None, route_class=<class 'fastapi.routing.APIRoute'>, on_startup=None, on_shutdown=None, deprecated=None, include_in_schema=True)
¶
Include controller to the application.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
controller |
Controller |
Controller instance |
required |
For other parameters, please see the documentation of fastapi.APIRouter.add_api_route.
Source code in fastack/app.py
def include_controller(
self,
controller: Controller,
*,
prefix: str = "",
tags: Optional[List[str]] = None,
dependencies: Optional[Sequence[params.Depends]] = None,
default_response_class: Type[Response] = Default(JSONResponse),
responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
callbacks: Optional[List[BaseRoute]] = None,
routes: Optional[List[BaseRoute]] = None,
redirect_slashes: bool = True,
default: Optional[ASGIApp] = None,
dependency_overrides_provider: Optional[Any] = None,
route_class: Type[APIRoute] = APIRoute,
on_startup: Optional[Sequence[Callable[[], Any]]] = None,
on_shutdown: Optional[Sequence[Callable[[], Any]]] = None,
deprecated: Optional[bool] = None,
include_in_schema: bool = True,
):
"""
Include controller to the application.
Args:
controller (Controller): Controller instance
For other parameters, please see the documentation of ``fastapi.APIRouter.add_api_route``.
"""
assert isinstance(
controller, Controller
), f"Controller must be an instance of {Controller!r}"
router = controller.build(
prefix=prefix,
tags=tags,
dependencies=dependencies,
default_response_class=default_response_class,
responses=responses,
callbacks=callbacks,
routes=routes,
redirect_slashes=redirect_slashes,
default=default,
dependency_overrides_provider=dependency_overrides_provider,
route_class=route_class,
on_startup=on_startup,
on_shutdown=on_shutdown,
deprecated=deprecated,
include_in_schema=include_in_schema,
)
self.include_router(router)
load_commands(self)
¶
Load commands from settings.
Source code in fastack/app.py
load_plugins(self)
¶
set_settings(self, settings)
¶
Set settings for the application.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
settings |
ModuleType |
settings module |
required |
create_app(settings, routes=None, title='Fastack', description='Fastack Framework', version='0.1.0', openapi_url='/openapi.json', openapi_tags=None, servers=None, dependencies=None, default_response_class=<fastapi.datastructures.DefaultPlaceholder object at 0x7f0df36da6d0>, docs_url='/docs', redoc_url='/redoc', swagger_ui_oauth2_redirect_url='/docs/oauth2-redirect', swagger_ui_init_oauth=None, middleware=None, exception_handlers=None, on_startup=None, on_shutdown=None, terms_of_service=None, contact=None, license_info=None, openapi_prefix='', root_path='', root_path_in_servers=True, responses=None, callbacks=None, deprecated=None, include_in_schema=True, swagger_ui_parameters=None, **extra)
¶
Create a Fastack application.
Source code in fastack/app.py
def create_app(
settings: ModuleType,
routes: Optional[List[BaseRoute]] = None,
title: str = "Fastack",
description: str = "Fastack Framework",
version: str = "0.1.0",
openapi_url: Optional[str] = "/openapi.json",
openapi_tags: Optional[List[Dict[str, Any]]] = None,
servers: Optional[List[Dict[str, Union[str, Any]]]] = None,
dependencies: Optional[Sequence[Depends]] = None,
default_response_class: Type[Response] = Default(JSONResponse),
docs_url: Optional[str] = "/docs",
redoc_url: Optional[str] = "/redoc",
swagger_ui_oauth2_redirect_url: Optional[str] = "/docs/oauth2-redirect",
swagger_ui_init_oauth: Optional[Dict[str, Any]] = None,
middleware: Optional[Sequence[Middleware]] = None,
exception_handlers: Optional[
Dict[
Union[int, Type[Exception]],
Callable[[Request, Any], Coroutine[Any, Any, Response]],
]
] = None,
on_startup: Optional[Sequence[Callable[[], Any]]] = None,
on_shutdown: Optional[Sequence[Callable[[], Any]]] = None,
terms_of_service: Optional[str] = None,
contact: Optional[Dict[str, Union[str, Any]]] = None,
license_info: Optional[Dict[str, Union[str, Any]]] = None,
openapi_prefix: str = "",
root_path: str = "",
root_path_in_servers: bool = True,
responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
callbacks: Optional[List[BaseRoute]] = None,
deprecated: Optional[bool] = None,
include_in_schema: bool = True,
swagger_ui_parameters: Optional[Dict[str, Any]] = None,
**extra: Any,
):
"""
Create a Fastack application.
"""
if not contact:
contact = {
"name": "Fastack",
"url": "https://github.com/fastack-dev/fastack",
}
app = Fastack(
debug=settings.DEBUG,
routes=routes,
title=title,
description=description,
version=version,
openapi_url=openapi_url,
openapi_tags=openapi_tags,
servers=servers,
dependencies=dependencies,
default_response_class=default_response_class,
docs_url=docs_url,
redoc_url=redoc_url,
swagger_ui_oauth2_redirect_url=swagger_ui_oauth2_redirect_url,
swagger_ui_init_oauth=swagger_ui_init_oauth,
middleware=middleware,
exception_handlers=exception_handlers,
on_startup=on_startup,
on_shutdown=on_shutdown,
terms_of_service=terms_of_service,
contact=contact,
license_info=license_info,
openapi_prefix=openapi_prefix,
root_path=root_path,
root_path_in_servers=root_path_in_servers,
responses=responses,
callbacks=callbacks,
deprecated=deprecated,
include_in_schema=include_in_schema,
swagger_ui_parameters=swagger_ui_parameters,
**extra,
)
app.set_settings(settings)
app.load_plugins()
app.load_commands()
return app
Last update:
January 17, 2022 13:26:42