Database Integration with FastAPI
Posted inFastAPI Programming

Database Integration with FastAPI

No Comments

Adding a database integration to your web application is like giving it superpowers; all of a sudden, it can remember, change, and communicate with a plethora of data. FastAPI, renowned for its excellent performance and user-friendliness, further streamlines this procedure. This book will help you integrate databases with FastAPI without any problems, regardless of your level of experience. It’s perfect for beginners who want to get their hands dirty or seasoned developers who want to brush up on the latest techniques.

This article will guide you through every step of setting up your database integration FastAPI project, including selecting the appropriate database, configuring your database models, and carrying out CRUD activities. Upon completion, you will possess a firm grasp on how to integrate your FastAPI application with a database, along with code samples that will expedite your start-up process.

Step 1: Set Up Your FastAPI Project

First, make sure that Uvicorn, an ASGI server, and FastAPI are installed. If not, pip can be used to install them:

python -m venv venv
source venv/bin/activate # On Windows use `venv\Scripts\activate`
pip install fastapi uvicorn

We’re integrating databases with SQLAlchemy in this lesson. Install it together with the database driver for your database (asyncpg is suggested for PostgreSQL, for example):

pip install sqlalchemy asyncpg alembic

Enhanced Directory Structure

Here’s a recommended structure for a FastAPI project with database integration:

fastapi_project/
├── venv/
└── app/
    ├── __init__.py
    ├── main.py
    ├── models.py
    └── database.py

Step 2: Define Your Database Models

Let’s create a simple model for a User. Define your SQLAlchemy model in a file named models.py. Since FastAPI 0.79.0 to 0.108.0 introduced support for SQLAlchemy 2.0’s new style, we’ll use that:

In app/models.py, define your SQLAlchemy models:

from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base, Mapped

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id: Mapped[int] = Column(Integer, primary_key=True, index=True)
    name: Mapped[str] = Column(String, index=True)
    email: Mapped[str] = Column(String, unique=True, index=True)
  • Column, Integer, String: These are imported from sqlalchemy. Column is used to define a column in a database table. Integer and String are types that specify what kind of data each column will store (integer numbers and strings of characters, respectively).
  • declarative_base: This function returns a base class for declarative class definitions. The class we define will inherit from this base class, and SQLAlchemy will recognize it as a model that should be mapped to a database table.
  • Mapped: This is a typing hint introduced in SQLAlchemy 1.4 for type checking and editor support. It informs type checkers that these attributes are SQLAlchemy column objects which are mapped to database columns.
  • Base = declarative_base(): This line creates a base class for the declarative class definition. All model classes should inherit from this Base class. It contains a MetaData object where newly defined Table objects are collected.
  • class User(Base): This defines a new class User that inherits from Base. This class represents a table in the database.
  • __tablename__ = 'users': This attribute is required by SQLAlchemy to know the table name in the database that this class will be mapped to.
  • Attributes with Mapped[Type]: Each attribute of the class represents a column in the database table. The type of each attribute (e.g., int for id, str for name and email) is specified using Mapped[Type], which also enables better type hinting and editor support.
  • Column(...): This function is used to define a column. The first argument specifies the column’s data type.
  • primary_key=True: This indicates that id is the primary key of the table.
  • index=True: This means an index will be created for this column, improving query performance on this column.
  • unique=True: For the email column, this ensures that all values in this column are unique across the table.

Step 3: Establish Database Connection

In a new file, database.py, set up your database connection using SQLAlchemy’s create_async_engine and sessionmaker for asynchronous communication

Create app/database.py to set up database connectivity:

# database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
# For other databases like PostgreSQL, use the appropriate URL

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()
  • create_engine: This function from SQLAlchemy is used to create an engine that manages connections to the database. The engine is the starting point for any SQLAlchemy application.
  • declarative_base: This function returns a base class for class-based model definitions. Each model you create will inherit from this base class.
  • sessionmaker: This is a factory for session objects. A session in SQLAlchemy represents a “work area” for your objects, where you can query and manipulate data before committing those changes to the database
  • SQLALCHEMY_DATABASE_URL: This string represents the connection URL to your database. In this example, it’s configured to connect to a SQLite database named test.db located in the same directory as your project. For other databases like PostgreSQL, you would change this URL accordingly.
  • create_engine(...): This line creates an engine instance that knows how to connect to your database using the URL provided. The connect_args parameter is specific to SQLite and is used to allow the same thread to be used by multiple requests in development. For other databases, this parameter might not be needed or different parameters may be required.
  • sessionmaker(...): This creates a factory (named SessionLocal here) that will generate new Session objects when called, bound to the engine we previously created. autocommit=False means that changes made through the session won’t be committed automatically, giving you control over when to commit transactions. autoflush=False means SQLAlchemy won’t automatically flush database changes (sending pending changes to the database) before each query.
  • declarative_base(): This creates a base class for your declarative class definitions. Every model you create will inherit from this Base class, which associates it with the engine, allowing SQLAlchemy to discover and map the database schema.

Creating Database Tables

Before running the app, ensure the database tables are created. You can use main.py to create tables based on your models.

# main.py

from fastapi import FastAPI
from sqlalchemy.orm import Session
from database import SessionLocal, engine
import models

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

The main.py script in your FastAPI application is where the web application instance is created and configured. It also plays a critical role in initializing your database models and setting up the application’s endpoints. Here’s a breakdown of the code snippet provided:

  • from fastapi import FastAPI: Imports the FastAPI class, which is used to create your web application instance.
  • from sqlalchemy.orm import Session: Imports the Session class from SQLAlchemy, which will be used for database sessions to execute queries.
  • from database import SessionLocal, engine: Imports the SessionLocal (a factory for creating new session objects) and engine (your database engine) from your database.py file.
  • import models: Imports the models module, which contains your SQLAlchemy models (tables).
  • models.Base.metadata.create_all(bind=engine): This line is crucial. It tells SQLAlchemy to create all tables defined in your models (inherited from Base) that do not already exist in the database. The bind=engine argument specifies which engine (database connection) to use for creating the tables. This is how your database schema is applied to the actual database, allowing you to work with the defined tables in your application.
  • app = FastAPI(): Creates an instance of the FastAPI application. This app object is used to create routes and handle requests. After defining app, you would typically go on to define various endpoints (API routes) that allow clients to interact with your application by performing actions like reading from or writing to your database.

CRUD Examples

Create a New User

To create a new user, we’ll use a POST request.

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from .models import User
from .database import SessionLocal
from pydantic import BaseModel

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

class UserCreate(BaseModel):
    """
      UserCreate defines the expected request body for creating a new user.
      It uses Pydantic models for data validation.
      
      Attributes:
          name (str): The name of the user.
          email (str): The email address of the user. It must be unique.
    """
    
    name: str
    email: str

@app.post("/users/", response_model=UserCreate)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    """
    Creates a new user in the database.

    Parameters:
        user (UserCreate): The user data sent by the client.
        db (Session, optional): The database session used to execute database operations. 
                                It's injected by FastAPI's Depends mechanism.

    Returns:
        User instance that was added to the database.
    """
    
    db_user = db.query(User).filter(User.email == user.email).first()
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    db_user = User(name=user.name, email=user.email)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

Read Operations

  • Get a Single User by ID
@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
    """
      Retrieves a user by their user_id from the database.
      
      Parameters:
          user_id (int): The unique identifier of the user to retrieve.
          db (Session): The database session used to execute database operations. 
                        This session is automatically provided by FastAPI's dependency injection.
      
      Returns:
          The user instance identified by user_id. If no user is found with the provided ID,
          a 404 error is raised indicating that the user is not found.
      
      Raises:
          HTTPException: A 404 status code is raised if no user with the given user_id exists in the database.
    """
    
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

Get All Users

@app.get("/users/")
def read_users(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    """
    Retrieves a list of users from the database, with pagination support.
    
    This endpoint supports pagination through `skip` and `limit` parameters,
    allowing for a scalable way to fetch users. By default, it returns the first
    10 users. Adjust `skip` and `limit` to retrieve other users or to change the
    number of users returned in a single request.
    
    Parameters:
        skip (int): The number of records to skip before starting to collect the
                    result set. Defaults to 0.
        limit (int): The maximum number of users to return. Defaults to 10.
        db (Session): The database session used to execute database operations.
                      Provided by FastAPI's dependency injection system.
    
    Returns:
        A list of user instances up to the specified `limit`, starting after
        `skip` number of users. If no users are found, returns an empty list.
    """
    users = db.query(User).offset(skip).limit(limit).all()
    return users

Update a User

To update a user, we use a PUT request. We’ll update the user’s name and email for simplicity.

class UserUpdate(BaseModel):
    """
    UserUpdate is used for updating existing users in the database.
    It defines the structure of the request body with fields that can be updated.
    
    Attributes:
        name (str): The new name of the user.
        email (str): The new email address of the user. It must remain unique.
    """
    
    name: str
    email: str


@app.put("/users/{user_id}")
def update_user(user_id: int, user: UserUpdate, db: Session = Depends(get_db)):
    """
    Updates an existing user's information in the database.
    
    Parameters:
        user_id (int): The unique identifier of the user to update.
        user (UserUpdate): An object containing the updated user data.
        db (Session): The database session used to execute database operations. 
                      This session is automatically provided by FastAPI's dependency injection.
    
    Returns:
        The updated user instance if the update is successful.
    
    Raises:
        HTTPException: A 404 status code is raised if no user with the given user_id exists.
    """
    
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    db_user.name = user.name
    db_user.email = user.email
    db.commit()
    db.refresh(db_user)
    return db_user

Delete a User

To delete a user, we’ll use a DELETE request.

@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    """
    Deletes a user from the database by their unique user ID.
    
    Parameters:
        user_id (int): The unique identifier of the user to be deleted.
        db (Session): The database session used to execute database operations. 
                      This session is automatically provided by FastAPI's dependency injection.
    
    Returns:
        A JSON object with a detail key indicating that the user has been successfully deleted.
    
    Raises:
        HTTPException: A 404 status code is raised if no user with the given user_id exists in the database.
    """
    
    db_user = db.query(User).filter(User.id == user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    db.delete(db_user)
    db.commit()
    return {"detail": "User deleted"}

Running Your Application

Ensure all dependencies are installed and run your application with:

uvicorn main:app --reload

Navigate to http://127.0.0.1:8000/docs in your browser to see the interactive API documentation provided by FastAPI, where you can test all these CRUD operations.

Conclusion

This guide provided step-by-step examples for implementing CRUD operations in a FastAPI application with SQLAlchemy. By defining models, creating database sessions, and utilizing FastAPI’s dependency injection system, we demonstrated how to create, read, update, and delete resources in a database. These operations form the backbone of many web applications, providing a solid foundation for building robust APIs.

Leave a response to this article by providing your insights, comments, or requests for future articles. Share the articles with your friends and colleagues on social media.

Post Views: 65
Tags:
Leave a Comment

Comments

No comments yet. Why don’t you start the discussion?

Leave a Reply

Your email address will not be published. Required fields are marked *