You can connect FastAPI to Postgres asynchronously using asyncpg and its connection pooling feature.
Here’s a FastAPI application that demonstrates connecting to a PostgreSQL database using asyncpg with a connection pool. This approach is efficient because it reuses database connections instead of establishing a new one for every request.
import asyncpg
import uvicorn
from fastapi import FastAPI, HTTPException
# Database connection details
DATABASE_URL = "postgresql://user:password@host:port/database"
# Global variable to hold the connection pool
db_pool = None
app = FastAPI()
@app.on_event("startup")
async def startup_event():
"""
Establishes the connection pool when the FastAPI application starts.
"""
global db_pool
try:
db_pool = await asyncpg.create_pool(DATABASE_URL)
print("Database connection pool created successfully.")
except Exception as e:
print(f"Error creating database connection pool: {e}")
# In a real application, you might want to handle this more robustly,
# e.g., by retrying or exiting.
@app.on_event("shutdown")
async def shutdown_event():
"""
Closes the connection pool when the FastAPI application shuts down.
"""
global db_pool
if db_pool:
await db_pool.close()
print("Database connection pool closed.")
async def get_db():
"""
Dependency function to get a database connection from the pool.
Yields a connection that will be released back to the pool.
"""
global db_pool
if db_pool is None:
raise HTTPException(status_code=500, detail="Database pool not initialized")
async with db_pool.acquire() as connection:
try:
yield connection
finally:
# The connection is automatically released back to the pool
# when exiting the 'async with' block.
pass
@app.get("/items/{item_id}")
async def read_item(item_id: int, connection: asyncpg.Connection = Depends(get_db)):
"""
Retrieves an item from the database using an asynchronous connection.
"""
try:
# Example query: fetch an item by its ID
# Replace 'your_table' and 'id' with your actual table and column names
row = await connection.fetchrow("SELECT * FROM your_table WHERE id = $1", item_id)
if not row:
raise HTTPException(status_code=404, detail="Item not found")
return dict(row) # Convert asyncpg.Record to a dictionary
except Exception as e:
print(f"Database query error: {e}")
raise HTTPException(status_code=500, detail="Database query failed")
@app.post("/items/")
async def create_item(item_data: dict, connection: asyncpg.Connection = Depends(get_db)):
"""
Creates a new item in the database using an asynchronous connection.
"""
try:
# Example query: insert a new item
# Replace 'your_table' and column names with your actual schema
# Assuming item_data is a dictionary like {"name": "New Item", "description": "A test item"}
query = "INSERT INTO your_table (name, description) VALUES ($1, $2) RETURNING id"
new_item_id = await connection.fetchval(query, item_data.get("name"), item_data.get("description"))
return {"id": new_item_id, **item_data}
except Exception as e:
print(f"Database insert error: {e}")
raise HTTPException(status_code=500, detail="Database insert failed")
# To run this application:
# 1. Save it as main.py
# 2. Install required libraries: pip install fastapi uvicorn asyncpg
# 3. Replace DATABASE_URL with your actual PostgreSQL connection string.
# 4. Ensure you have a table named 'your_table' with 'id', 'name', and 'description' columns (or adjust queries).
# 5. Run from your terminal: uvicorn main:app --reload
The core of asynchronous database interaction with FastAPI and asyncpg lies in managing the connection pool. The asyncpg.create_pool() function initializes a pool of connections. When a request needs database access, db_pool.acquire() is used to get a connection from the pool. This connection is then passed to your route handler. Crucially, the async with statement ensures that the connection is automatically returned to the pool once the block is exited, whether due to successful completion or an exception.
The startup_event and shutdown_event functions, decorated with @app.on_event(), are essential for managing the lifecycle of the connection pool. The pool is created when the application starts and closed gracefully when it stops, preventing resource leaks. The get_db function acts as a FastAPI dependency, making it easy to inject database connections into your route handlers.
When you query the database, asyncpg methods like fetchrow, fetchval, or execute are used. These are await-able, meaning they yield control back to the event loop while waiting for the database operation to complete, allowing other requests to be processed concurrently.
The surprising thing about asynchronous database access is how little it changes your application’s structure while dramatically improving its performance. You don’t typically rewrite your entire data access layer; instead, you wrap your existing synchronous database calls (or write new ones) with await and ensure you’re using an asynchronous driver like asyncpg. The real magic is in the event loop and the driver’s ability to perform I/O without blocking, letting you handle hundreds or thousands of concurrent connections efficiently with a small number of actual database connections managed by the pool.
The connection pool itself is a collection of pre-established connections. When a request needs a connection, it takes one from the pool. If no connections are available, the request waits until one is returned. This is far more efficient than creating a new TCP connection to the database for every single request, which involves network latency and database authentication overhead. asyncpg allows you to configure the pool size (min_size, max_size) to match your application’s expected load.
A common pitfall is forgetting to release connections back to the pool. However, asyncpg’s pool.acquire() used with an async with statement handles this automatically. If you were to manually acquire a connection without async with, you would need to explicitly call pool.release(connection).
The next step in scaling this is often implementing robust error handling and connection retry logic within the startup_event or within the get_db dependency if the pool itself is healthy but individual connection acquisitions fail.