Using Asyncpg with FastAPI and Air

"Asyncpg is the connector for PostgreSQL and asyncio-flavored Python. Here's how to use it without other libraries on FastAPI and Air projects. Recently I've been on a few projects using PostgreSQL where SQLAlchemy and SQLModel felt like overkill. Instead of using those libraries I leaned on writing SQL queries and running those directly in [asyncpg](https://pypi.org/project/asyncpg/) instead of using an ORM powered by asyncpg. Here's how I got it to work"

""""A lifespan for maintaining the connection to the PostgreSQL DB Without this, the connection will timeout and queries will fail. """ # app.state is where the connection pool is created, which can # be accessed later inside of views. The is only run once during # app startup. app.state.pool = await asyncpg.create_pool( dsn=DATABASE_URL, min_size=1, max_size=10, ) try: # This is where the app runs all the URL route functons. yield finally: # This is run once when the app is shut down. await app.state.pool.close()"

Define an async lifespan function for ASGI apps to create and maintain an asyncpg connection pool during app startup and close it on shutdown. Use asynccontextmanager to yield control while the app runs and store the pool on app.state for route handlers to access. Create the pool with asyncpg.create_pool(dsn=DATABASE_URL, min_size, max_size) to tune concurrency and prevent timeouts. Pass the lifespan callable to FastAPI or Air app instantiation so the pool is available in request handlers. Close the pool in the finally block with await app.state.pool.close() to release connections.

Read at https://daniel.feldroy.com