Estoy escribiendo mi primer proyecto en FastAPI y estoy luchando un poco. En particular, no estoy seguro de cómo se supone que debo usar el grupo de conexiones asyncpg en mi aplicación. Actualmente lo que tengo va así
en db.py tengo
pgpool = None async def get_pool(): global pgpool if not pgpool: pgpool = await asyncpg.create_pool(dsn='MYDB_DSN') return pgpooly luego en archivos individuales uso get_pool como dependencia.
@router.post("/user/", response_model=models.User, status_code=201) async def create_user(user: models.UserCreate, pgpool = Depends(get_pool)): # ... do things ...Primero, cada punto final que tengo usa la base de datos, por lo que parece una tontería agregar ese argumento de dependencia para cada función. En segundo lugar, esto parece una forma indirecta de hacer las cosas. Defino un global, luego defino una función que devuelve ese global y luego inyecto la función. Estoy seguro de que hay una forma más natural de hacerlo.
He visto a personas sugerir simplemente agregar lo que necesito como propiedad al objeto de la aplicación
@app.on_event("startup") async def startup(): app.pool = await asyncpg.create_pool(dsn='MYDB_DSN')pero no funciona cuando tengo varios archivos con enrutadores, no sé cómo acceder al objeto de la aplicación desde un objeto de enrutador.
¿Qué me estoy perdiendo?
Puede usar un patrón de fábrica de aplicaciones para configurar su aplicación.
Para evitar usar global o agregar cosas directamente al objeto de la aplicación, puede crear su propia base de datos de clase para mantener su grupo de conexiones.
Para pasar el grupo de conexiones a cada ruta, puede usar un middleware y agregar el grupo a request.state
Aquí está el código de ejemplo:
import asyncio import asyncpg from fastapi import FastAPI, Request class Database(): async def create_pool(self): self.pool = await asyncpg.create_pool(dsn='MYDB_DSN') def create_app(): app = FastAPI() db = Database() @app.middleware("http") async def db_session_middleware(request: Request, call_next): request.state.pgpool = db.pool response = await call_next(request) return response @app.on_event("startup") async def startup(): await db.create_pool() @app.on_event("shutdown") async def shutdown(): # cleanup pass @app.get("/") async def hello(request: Request): print(request.state.pool) return app app = create_app()La forma en que lo hago es en db.py.
class Database: def __init__(self,user,password,host,database,port="5432"): self.user = user self.password = password self.host = host self.port = port self.database = database self._cursor = None self._connection_pool = None async def connect(self): if not self._connection_pool: try: self._connection_pool = await asyncpg.create_pool( min_size=1, max_size=20, command_timeout=60, host=self.host, port=self.port, user=self.user, password=self.password, database=self.database, ssl="require" ) logger.info("Database pool connectionn opened") except Exception as e: logger.exception(e) async def fetch_rows(self, query: str,*args): if not self._connection_pool: await self.connect() else: con = await self._connection_pool.acquire() try: result = await con.fetch(query,*args) return result except Exception as e: logger.exception(e) finally: await self._connection_pool.release(con) async def close(self): if not self._connection_pool: try: await self._connection_pool.close() logger.info("Database pool connection closed") except Exception as e: logger.exception(e)Luego en la aplicación
@app.on_event("startup") async def startup_event(): database_instance = db.Database(**db_arguments) await database_instance.connect() app.state.db = database_instance logger.info("Server Startup") @app.on_event("shutdown") async def shutdown_event(): if not app.state.db: await app.state.db.close() logger.info("Server Shutdown")Luego puede obtener la instancia de db con request.app.state.db pasando un parámetro de solicitud en las rutas.