FastAPI is a modern, fast (high-performance) web framework for building APIs with Python 3.7+ based on standard Python type hints. One of its best features is its support for asynchronous programming, which makes handling database operations more efficient. In this guide, we'll create a simple FastAPI app that performs asynchronous database access using SQLAlchemy.

Prerequisites:

Before we start, ensure you have Python 3.7+ installed, as well as the following libraries. You can install them using pip:

pip install fastapi[all] sqlalchemy asyncpg databases

• FastAPI: The web framework we'll use.
• SQLAlchemy: The ORM (Object Relational Manager) that we'll use for database operations.
• asyncpg: An async PostgreSQL driver.
• databases: An async database support library.

Step 1: Set Up the FastAPI Application

Start by creating a new file called main.py. This is where we will set up our FastAPI app.

from fastapi import FastAPI
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.ext.asyncio import async_sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:password@localhost/testdb"

# Create the database engine
engine = create_async_engine(DATABASE_URL, echo=True)

# Create a base class for declarative models
Base = declarative_base()

# Define a sample model
class Item(Base):
    __tablename__ = 'items'

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)

# Create FastAPI app
app = FastAPI()

Step 2: Create the Database Session

We need a database session to interact with our PostgreSQL database. Let's create a sessionmaker function that can provide a session to our endpoint functions.

# Create an async sessionmaker
async_session = async_sessionmaker(bind=engine, expire_on_commit=False)

# Dependency to get the database session
async def get_db():
    async with async_session() as session:
        yield session

Step 3: Creating CRUD Operations

At this point, we will implement the CRUD operations. Let's create endpoints to create, read, update, and delete Items.

Create an Item

from fastapi import Depends, HTTPException
from sqlalchemy.future import select

@app.post("/items/", response_model=Item)
async def create_item(name: str, db: AsyncSession = Depends(get_db)):
    new_item = Item(name=name)
    db.add(new_item)
    await db.commit()
    await db.refresh(new_item)
    return new_item

Read Items

@app.get("/items/", response_model=list[Item])
async def read_items(skip: int = 0, limit: int = 10, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Item).offset(skip).limit(limit))
    items = result.scalars().all()
    return items

Update an Item

@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: int, name: str, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Item).where(Item.id == item_id))
    item = result.scalar_one_or_none()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    
    item.name = name
    await db.commit()
    await db.refresh(item)
    return item

Delete an Item

@app.delete("/items/{item_id}")
async def delete_item(item_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Item).where(Item.id == item_id))
    item = result.scalar_one_or_none()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    
    await db.delete(item)
    await db.commit()
    return {"detail": "Item deleted"}

Step 4: Running Your FastAPI App

Add the following code to run the FastAPI app using Uvicorn:

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Now you can run your FastAPI application:

python main.py

Conclusion

To test your API, you can use tools like Postman or your web browser. FastAPI automatically generates interactive documentation with Swagger UI at http://localhost:8000/docs. From there you can test your CRUD operations effortlessly.

This guide should give you a solid foundation for building more complex FastAPI applications leveraging async database access!