CRUD Operations with FastAPI: A Comprehensive Guide

Last updated on February 1st, 2024

Introduction

FastAPI is a modern Python web framework for building performant and scalable REST APIs. Its declarative style and robust feature set have quickly made it a favorite among Python developers.

A common task when building APIs with FastAPI is implementing CRUD (Create, Read, Update, Delete) functionality for resources. In this comprehensive tutorial, we’ll explore the ins and outs of implementing full CRUD operations for endpoints using FastAPI.

We’ll go through:

• Initial project setup
• Defining models
• Creating, reading, updating and deleting resources
• Input validation and error handling
• Authentication and permissions
• Testing and documentation

By the end, you’ll have all the knowledge needed to build robust CRUD REST APIs with FastAPI. Let’s dive in!

Initial Project Setup with FastAPI

We’ll use Python 3.8+ along with some common packages:

pip install fastapi uvicorn[standard] sqlalchemy psycopg2 python-multipart

This installs FastAPI, the Uvicorn web server, SQLAlchemy for database access, and multipart parsing for file uploads.

For the database, we’ll use PostgreSQL. First, install PostgreSQL then create a database named fastapi:

createdb fastapi

Next, define some initial project structure:

.
├── app
│   ├── main.py  
│   ├── db.py
│   ├── models.py
│   ├── crud.py
├── requirements.txt

The main.py will contain the FastAPI app and routes, db.py handles database connections, models.py defines data models, crud.py contains CRUD logic, and requirements.txt our dependencies.

With the foundation in place, let’s start building out CRUD functionality!

Defining Models

First we need to define Pydantic models for interacting with our data. These models define the schema and validation requirements.

Under models.py:

from pydantic import BaseModel

class UserBase(BaseModel):
    email: str
    
class UserCreate(UserBase):
    password: str
    
class User(UserBase):
    id: int

    class Config: 
        orm_mode = True

This defines a User model with id, email and password. The Config enables some SQLAlchemy integrations.

Setting up the Database

Now let’s connect to PostgreSQL. Under db.py:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
 
engine = create_engine('postgresql://user:pass@localhost/fastapi')
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
Base = declarative_base()

We create an engine connecting to the fastapi PostgreSQL database. It uses SQLAlchemy’s SessionLocal to establish database sessions.

The Base class will be the parent for model classes to integrate with SQLAlchemy.

Creating the User Model

Under models.py, we define the User model inheriting from Base:

from sqlalchemy import Column, Integer, String

from .db import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)

This creates a User table with id, email and hashed_password columns. The table name and columns are inferred from the class and variable names.

Adding CRUD Utilities

It’s good practice to encapsulate CRUD logic in utility functions rather than have it directly in routes.

Let’s create a CrudUser class under crud.py to handle CRUD for User models:

from . import models, schemas

class CrudUser:

    def get(self, db, user_id):
        return db.query(models.User).filter(models.User.id == user_id).first()

    def create(self, db, user):
        db_user = models.User(email=user.email, hashed_password=user.password)
        db.add(db_user)
        db.commit()
        db.refresh(db_user)
        return db_user

    # ... other methods

This encapsulates retrieving and creating users in simple methods. We’ll add the remaining CRUD operations shortly.

Creating Users

With the models and utilities ready, let’s implement creating users.

Under main.py, first load the dependencies:

from fastapi import FastAPI
from . import crud, models, schemas
from .db import SessionLocal

# ...

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

We instantiate the FastAPI app and define a get_db function that yields a new SQLAlchemy session. This handles opening and closing sessions automatically.

Now we can add the user creation endpoint:

@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    return crud.create_user(db=db, user=user)

This maps POST /users/ requests, validates using a UserCreate model, and passes the user data to our CRUD utility to create the user record in PostgreSQL.

The response_model decorates responses so only the defined fields are returned.

Let’s test it:

$ curl -X POST -d '{"email":"john@example.com","password":"secret"}' http://127.0.0.1:8000/users/

{"id":1,"email":"john@example.com"}

It returns the created user’s id and email as expected!

Reading Users

For reading users, we can similarly add a GET endpoint:

@app.get("/users/{user_id}", response_model=schemas.User)  
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found") 
    return db_user

This takes a user_id parameter, gets the user using CRUD utilities, handles 404 errors, and returns the User object.

We can test it:

$ curl http://127.0.0.1:8000/users/1  

{"id":1,"email":"john@example.com"}

Great, it returns the expected user!

Updating Users

To enable updating users, we can add a PUT endpoint:

@app.put("/users/{user_id}", response_model=schemas.User)
def update_user(user_id: int, user: schemas.UserUpdate, db: Session = Depends(get_db)):

    db_user = crud.get_user(db, user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")

    update_data = user.dict(exclude_unset=True)

    if "password" in update_data:
        update_data["hashed_password"] = get_password_hash(update_data["password"])
        del update_data["password"]

    crud.update_user(db, db_user=db_user, obj_in=update_data)  
    return db_user

This takes a partial UserUpdate model containing fields to update, gets the existing user, and applies the update using the CRUD util. Excluding unset fields allows making partial updates.

For example, updating just the email:

$ curl -X PUT -d '{"email":"john@newdomain.com"}' http://127.0.0.1:8000/users/1

{"id":1,"email":"john@newdomain.com"}

The user is updated as expected!

Deleting Users

Finally, enabling deleting users with a DELETE endpoint:

@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    
    db_user = crud.get_user(db=db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")

    crud.remove_user(db=db, user_id=user_id)
    return {"message": "User deleted successfully"}

This uses the CRUD util to delete the user if found, else returns 404.

Testing:

$ curl -X DELETE http://127.0.0.1:8000/users/1

{"message":"User deleted successfully"}

With that, we’ve built out full CRUD REST endpoints for the User resource!

Input Validation

A key part of robust CRUD APIs is validating input data. Pydantic, FastAPI’s data validation system, makes this easy.

For example, we can define stricter validation requirements:

from pydantic import EmailStr, BaseModel

class UserBase(BaseModel):
    email: EmailStr
    
class UserCreate(UserBase):
    password: str
    
    @validator('password')
    def password_strong(cls, v):
        if len(v) < 8:
            raise ValueError('Password must be >= 8 chars')
        return v

This adds an email validator and password length check. FastAPI will automatically ensure inputs match before creating objects.

We can also parameterize validation using configs. For example, defining a min_password_strength setting:

class AppConfig(BaseSettings):
    min_password_strength: int = 8

@validator('password')  
def password_strong(cls, v, values, config: AppConfig):
    if len(v) < config.min_password_strength:
        raise ValueError(f'Password must be >= {config.min_password_strength} chars')
    return v

This allows customizing validation rules via configuration rather than hardcoding.

Handling Errors FastAPI

It’s also essential to handle errors and exceptions well:

from fastapi import HTTPException

@app.exception_handler(HTTPException)
async def http_error_handler(request, exc):
    return JSONResponse({"errors": [exc.detail]}, status_code=exc.status_code)

@app.exception_handler(Exception)
async def general_error_handler(request, exc):
    return JSONResponse({"errors": ["Internal server error"]}, status_code=500)

This overrides the exception handlers to return JSON error responses for both HTTPErrors and general exceptions.

With robust validation and error handling, our API is hardened against bad data.

Adding Authentication in FastAPI

For any real application, we need authentication and authorization.

FastAPI integrates nicely with OAuth 2.0 and JWT tokens for this. Let’s add JWT auth using the fastapi-jwt-auth library:

pip install fastapi-jwt-auth

Then under main.py:

import jwt

from fastapi_jwt_auth import AuthJWT

@app.on_event("startup") 
async def startup():
    await AuthJWT.load_config(settings)


@app.post("/login")
async def login(user: UserAuth, Authorize: AuthJWT = Depends()):
    if user.username != "test" or user.password != "test":
        raise HTTPException(status_code=401, detail="Invalid username or password")

    # Generate access and refresh tokens
    access_token = Authorize.create_access_token(subject=user.username)
    refresh_token = Authorize.create_refresh_token(subject=user.username)

    # Return tokens  
    return {"access_token": access_token, "refresh_token": refresh_token}

# Protected endpoint  
@app.get("/items/", dependencies=[Depends(Authorize)])
async def read_items(user: dict = Depends(get_current_user)):
    return [{"item": "Foo"}, {"item": "Bar"}]

This implements JWT based login flow and authentication. Protected routes can now require a valid access token.

There are many options for implementing sophisticated auth with FastAPI.

Adding Tests

Writing tests ensures code works as expected through refactors and changes.

For testing FastAPI, we can use TestClient to make requests against routes and assert responses:

from fastapi.testclient import TestClient

# Import app normally 
from .main import app

client = TestClient(app)

def test_create_user():
    response = client.post("/users/", json={"email": "foo@example.com"})  
    assert response.status_code == 201
    assert response.json()["email"] == "foo@example.com"

This makes a request to /users/ and asserts the response looks correct. We can add many tests for our critical endpoints this way.

Automated testing is key to preventing regressions as code evolves.

Adding API Documentation

FastAPI automatically generates interactive API documentation using OpenAPI standard.

Simply add:

app.include_router(users.router, prefix="/users", tags=["users"])

This tags the /users routes together under “users” in the docs.

Now browse to /docs while app is running to view the auto-generated docs!

The interactive docs make understanding and using the API easy.

Deployment

Once our API is complete, we can easily deploy it using various Platform-as-a-Service providers like Heroku, AWS Elastic Beanstalk etc.

For example, to deploy on Heroku:

1. Add Procfile:

web: uvicorn main:app --host=0.0.0.0 --port=${PORT:-5000}

2. Install gunicorn
3. Commit to Git
4. Create Heroku app
5. Connect GitHub repo
6. Enable auto-deploy

And our FastAPI app will be live on a public URL!

Conclusion

That wraps up our guide to implementing full CRUD functionality with FastAPI. Here’s a quick summary of what we covered:

• Initial project setup and structure
• Defining models with Pydantic
• SQLAlchemy integration for PostgreSQL
• Encapsulating CRUD operations in utilities
• Creating, reading, updating and deleting resources
• Input validation and error handling
• JWT authentication and permissions
• Testing routes correctly
• Auto-generating interactive API documentation
• Deploying the finished API

FastAPI makes it delightfully easy to build robust and scalable REST APIs. With these techniques, you can develop complete CRUD backends for mobile, web or third-party applications.

The simple but powerful FastAPI framework combined with Python’s flexibility makes building reliable APIs quick and enjoyable. Give FastAPI a try on your next web project!

Frequently Asked Questions

How does FastAPI compare to alternatives like Flask and Django?

Some key differences between FastAPI, Flask and Django:

• FastAPI uses Pydantic for data validation, Flask and Django have their own validation approaches.
• FastAPI uses Starlette as a base and supports async natively.
• FastAPI has automatic interactive docs via OpenAPI.
• Flask is more minimal and unopinionated compared to FastAPI’s batteries included approach.
• Django provides full-stack components like admin, ORM etc. out of box. FastAPI is focused on the API.
• FastAPI aims to be high performance – leverages type hints, code generation etc.

So in summary, FastAPI fits in nicely between Flask and Django – more robust than Flask, more lightweight than Django.

How can I properly handle authentication and permissions for my FastAPI application?

Some best practices for auth in FastAPI:

• Use OAuth 2.0 with Bearer JWT tokens rather than custom token schemes.
• Secure access tokens in httpOnly cookies rather than localStorage.
• Use scopes to implement different permission levels for users.
• Revoke tokens on logout using token blacklisting.
• Use HTTPS to protect against MITM attacks.
• Hash/encrypt sensitive fields of tokens like usernames.
• Rate limit auth attempts to prevent brute force.
• Make access tokens short-lived, use refresh tokens to get new access tokens.

Following OAuth standards and security good practices avoids common pitfalls when implementing authentication.

What are some recommended ways to deploy FastAPI apps in production?

Some good options for deploying FastAPI to production:

• Platforms like Heroku, AWS Elastic Beanstalk, Azure App Service for managed deployment.
• Docker containers on Kubernetes for scalability and portability.
• Serverless platforms like AWS Lambda, Azure Functions for event-driven apps.
• IaaS providers like DigitalOcean, Linode for flexible VPS hosting.
• Reverse proxy with Nginx for handling TLS, authentication, routing etc.
• CI/CD pipelines to automate testing and promote builds.

So in summary, leverage managed platforms where possible, or use Kubernetes and infrastructure tools like Nginx to build robust hosting environments. Automate testing and deployment.