-
-
Notifications
You must be signed in to change notification settings - Fork 337
-
I've followed the documentation about wiring injection on Fast API. Everything is working fine except for retrieving the database instance.
The API is a multi-tenant database so I need to change the database name by getting a header key. So I've made this function:
async def get_db_name(request: Request): if 'HTTP_X_DATABASE' not in request.headers: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="HTTP_X_DATABASE is missing" ) db_name = request.headers['HTTP_X_DATABASE'] return db_name
This is the original factory for the database:
class Container(containers.DeclarativeContainer): db = providers.Factory(Db, config=config.db, logger=db_logger)
And I tried to override the database factory like this:
app = FastAPI() app.container = Container() config = ConfigParser().parse() app.container.config.override(config) app.container.core.db.override(Callable(get_db, config=app.container.config.db, logger=app.container.core.db_logger)) @inject def get_db(config, logger, db_name: str = Depends(Provide(get_db_name))): config['name'] = db_name return Db(config, logger)
But I'm getting an error because the db_name injected to the Db class is an instance of Depends.
How can I retrieve the db name from request and the inject it into my app?
Beta Was this translation helpful? Give feedback.
All reactions
Replies: 2 comments
-
I could figure out by using a middleware:
async def get_db_name(request: Request): # Get db from subdomain # db_name = request.base_url.hostname.split('.')[0] # Get db from header if 'HTTP_X_DATABASE' not in request.headers: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="HTTP_X_DATABASE is missing" ) db_name = request.headers['HTTP_X_DATABASE'].replace('-', '_').upper() # Check that database provided exists clients = request.app.container.services.client().get_clients() client = next((c for c in clients if c.db_name == db_name), None) if client is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate database", ) return db_name app = FastAPI() app.container = Container() config = ConfigParser().parse() app.container.config.override(config) @app.middleware('http') async def set_db_context(request: Request, call_next): try: request.app.container.config()['db']['name'] = await get_db_name(request) except HTTPException as e: return JSONResponse(status_code=status.HTTP_401_UNAUTHORIZED, content={'detail': e.detail}) response = await call_next(request) return response
Beta Was this translation helpful? Give feedback.
All reactions
-
Finally, I decided to use another method instead of middleware because I'm getting conflicts in async requests.
I didn't understood how wiring works and how I can use it for my cases so I decided to just create a dependency that return the application on demand:
from functools import lru_cache from fastapi import HTTPException, Request from starlette import status from api import HTTP_X_DATABASE from shared.config_parser import ConfigParser from shared.containers import Application def get_db_name(request: Request): # Get db name from header # On ne lève pas une exception si la valeur est absente, # car certaines routes ne nécessitent pas d'accès à la base. if HTTP_X_DATABASE not in request.headers: return None db_name = request.headers[HTTP_X_DATABASE] return db_name def get_support_app(request: Request) -> Application: return _get_client_app(get_db_name(request)) @lru_cache def _get_client_app(db_name: str | None): application = Application() config = ConfigParser().parse() if db_name: # Vérifie que le client existe if not application.services.client().get_by_db_name(db_name): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate database") config['db']['name'] = db_name application.config.override(config) return application
Then inside a router, i ask for app instead of service
@app.router.get('/testdb', response_model=dict) async def get_test_db(support_app: Application = Depends(get_support_app)): return {'db': support_app.core.db().config['name']}
Beta Was this translation helpful? Give feedback.