Created
April 21, 2022 19:41
-
-
Save tafaust/ee9e09b7d68f702ad8f7fb1177e20c93 to your computer and use it in GitHub Desktop.
FastAPI singleton service dependency injection across multiple requests
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import random | |
import click | |
import uvicorn | |
from fastapi import FastAPI, APIRouter, Depends, Request | |
# Service | |
class MySingletonService: | |
def __init__(self): | |
self.random_number = random.randint(1, 100) | |
def __str__(self): | |
return 'my_singleton_service' | |
def say_hello(self) -> str: | |
return f'Hello Python Community! You will see "{self.random_number}" for every request.' | |
# Helper to grab dependencies that live in the app.state | |
def get_my_singleton_service(request: Request) -> MySingletonService: | |
# See application for the key name in `app`. | |
return request.app.state.my_singleton_service | |
# Router | |
router = APIRouter(prefix='/linkedin', tags=['LinkedIn']) | |
@router.get('/hello_python', response_model=str) | |
async def edge_detection(my_singleton_service: MySingletonService = Depends(get_my_singleton_service)) -> str: | |
return my_singleton_service.say_hello() | |
# FastAPI Application Factory | |
def create_app() -> FastAPI: | |
"""Creates the FastAPI application. | |
""" | |
application = FastAPI(title='This is single-instance DI across requests!') | |
application.include_router(router=router) | |
# Register our singleton service for DI once when the app has finished startup | |
@application.on_event('startup') | |
def load_single_instance_service(): | |
application.state.my_singleton_service = MySingletonService() | |
return application | |
@click.command() | |
@click.option('--host', default='127.0.0.1', help='host') | |
@click.option('--port', default='31313', help='port') | |
@click.option('--hot-reload', is_flag=True, default=False, help='Hot reload') | |
@click.option('--log-level', default='info', help='Set the log level. Default: "info".') | |
def main(host, port, hot_reload, log_level): | |
uvicorn.run('application:create_app', host=host, port=int(port), reload=hot_reload, log_level=log_level, factory=True) | |
if __name__ == '__main__': | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment