from fastapi import FastAPI import databases DATABASE_URL = "sqlite+aiosqlite:///./test.db" database = databases.Database(DATABASE_URL) app = FastAPI() @app.on_event("startup") async def startup(): await database.connect() @app.on_event("shutdown") async def shutdown(): await database.disconnect() @app.get("/users") async def read_users(): query = "SELECT * FROM users" results = await database.fetch_all(query=query) return results
The fetch_all method returns a list of rows matching the query. Since the query selects all users, the response is a list of dictionaries, each dictionary representing a user.
For async PostgreSQL connections with the databases library, the URL must use an async driver like asyncpg. So the correct prefix is postgresql+asyncpg://.
from fastapi import FastAPI import databases DATABASE_URL = "sqlite+aiosqlite:///./test.db" database = databases.Database(DATABASE_URL) app = FastAPI() @app.on_event("startup") async def startup(): await database.connect() @app.on_event("shutdown") async def shutdown(): await database.disconnect()
The database.connect() method is async and must be awaited. The startup event handler is defined as a normal function, so it cannot await. This causes a runtime error.
query = "SELECT COUNT(*) FROM users" user_count = await database.fetch_val(query)
The fetch_val method returns the value of the first column in the first row of the query result. Since the query counts users, it returns the integer 5.
The databases library supports async transactions using an async context manager. Using async with database.transaction(): ensures all queries inside run atomically and commit or rollback together.