Connection pooling in FastAPI - Mini Project: Build & Apply

Connection Pooling with FastAPI and Databases

📖 Scenario: You are building a simple FastAPI app that connects to a database. To make your app faster and handle many users, you want to use connection pooling. Connection pooling means reusing database connections instead of opening a new one every time.

🎯 Goal: Create a FastAPI app that uses a connection pool to connect to a SQLite database. You will set up the database URL, configure the connection pool size, write a function to get a connection from the pool, and add a route that uses the connection to fetch data.

📋 What You'll Learn

Create a variable DATABASE_URL with the exact value sqlite+aiosqlite:///./test.db

Create a variable POOL_SIZE and set it to 5

Create an AsyncEngine using create_async_engine with DATABASE_URL and pool_size=POOL_SIZE

Create an async function get_connection that returns a connection from the engine

Create a FastAPI app instance called app

Add a GET route /items that uses get_connection to fetch all rows from a table items and returns them as a list

💡 Why This Matters

🌍 Real World

Connection pooling is used in real web apps to improve speed and handle many users by reusing database connections instead of opening new ones each time.

💼 Career

Understanding connection pooling is important for backend developers working with databases and web frameworks like FastAPI to build scalable and efficient applications.

1

Set up the database URL

Create a variable called DATABASE_URL and set it to the string "sqlite+aiosqlite:///./test.db".

FastAPI

# Create the database URL
# Your code here

Need a hint?

Use a string exactly like "sqlite+aiosqlite:///./test.db" for the database URL.

2

Configure the connection pool size

Create a variable called POOL_SIZE and set it to the number 5.

FastAPI

DATABASE_URL = "sqlite+aiosqlite:///./test.db"
# Set the pool size
# Your code here

Need a hint?

Use a simple integer assignment like POOL_SIZE = 5.

3

Create the async engine with connection pooling

Import create_async_engine from sqlalchemy.ext.asyncio. Then create a variable called engine by calling create_async_engine with DATABASE_URL and pool_size=POOL_SIZE.

FastAPI

DATABASE_URL = "sqlite+aiosqlite:///./test.db"
POOL_SIZE = 5

# Import create_async_engine and create engine with pool size
# Your code here

Need a hint?

Remember to import create_async_engine before using it.

4

Create FastAPI app and route using connection pool

Import FastAPI from fastapi. Create a FastAPI app instance called app. Then create an async function called get_connection that uses async with engine.connect() as conn and returns conn. Finally, add a GET route /items to app that uses get_connection to execute SELECT * FROM items and returns the fetched rows as a list.

FastAPI

from sqlalchemy.ext.asyncio import create_async_engine

DATABASE_URL = "sqlite+aiosqlite:///./test.db"
POOL_SIZE = 5

engine = create_async_engine(DATABASE_URL, pool_size=POOL_SIZE)

# Import FastAPI and create app
# Create async get_connection function
# Add GET /items route
# Your code here

Need a hint?

Use async with engine.connect() as conn to get a connection. Use await conn.execute() to run the query. Return the rows as a list of dictionaries.