YouTip

FastAPI can be integrated with various databases, and the most common approach is to use SQLAlchemy as the ORM. This section introduces how to build an API with CRUD (Create, Read, Update, Delete) functionality using FastAPI + SQLAlchemy. * * * ## Install Dependencies pip install sqlalchemy * * * ## Project Structure project/├── main.py # FastAPI application entry point├── database.py # Database connection configuration├── models.py # SQLAlchemy models├── schemas.py # Pydantic data models└── crud.py # Database operation functions * * * ## 1. Database Configuration Configure the database connection in `database.py`: ## Example # database.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, DeclarativeBase # SQLite database (suitable for development) SQLALCHEMY_DATABASE_URL ="sqlite:///./sql_app.db" # Create database engine # connect_args is only needed for SQLite, allows multi-threaded access engine = create_engine( SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False} ) # Create session factory SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) # Declare base class class Base(DeclarativeBase): pass > SQLite is sufficient for the development environment; it is a file-based database that requires no additional database service installation. For production environments, PostgreSQL or MySQL is recommended; simply modify `SQLALCHEMY_DATABASE_URL`. * * * ## 2. SQLAlchemy Models Define the database table structure in `models.py`: ## Example # models.py from sqlalchemy import Column, Integer, String from database import Base class User(Base): __tablename__ ="users"# Table name id= Column(Integer, primary_key=True, index=True)# Primary key, auto-increment email= Column(String, unique=True, index=True)# Unique, indexed hashed_password = Column(String)# Hashed password is_active = Column(Boolean, default=True)# Whether active class Item(Base): __tablename__ ="items" id= Column(Integer, primary_key=True, index=True) title = Column(String, index=True)# Title, indexed description = Column(String)# Description owner_id = Column(Integer, ForeignKey("users.id"))# Foreign key referencing users table * * * ## 3. Pydantic Data Models Define the API input/output data structures in `schemas.py`: ## Example # schemas.py from pydantic import BaseModel # ===== User Models ===== class UserBase(BaseModel): email: str class UserCreate(UserBase): password: str# Password required when creating class UserOut(UserBase): id: int is_active: bool model_config = ConfigDict(from_attributes=True)# Support ORM objects # ===== Item Models ===== class ItemBase(BaseModel): title: str description: str | None=None class ItemCreate(ItemBase): pass class ItemOut(ItemBase): id: int owner_id: int model_config = ConfigDict(from_attributes=True) > Pydantic models and SQLAlchemy models are separate. Pydantic handles data validation at the API layer, while SQLAlchemy handles database table structures. `from_attributes=True` allows Pydantic to read data from SQLAlchemy objects. * * * ## 4. Database Operation Functions Encapsulate database operations in `crud.py`: ## Example # crud.py from sqlalchemy.orm import Session import models, schemas def get_user(db: Session, user_id: int): """Get user by ID""" return db.query(models.User).filter(models.User.id== user_id).first() def get_user_by_email(db: Session,email: str): """Get user by email""" return db.query(models.User).filter(models.User.email==email).first() def get_users(db: Session, skip: int=0, limit: int=100): """Get user list""" return db.query(models.User).offset(skip).limit(limit).all() def create_user(db: Session,user: schemas.UserCreate): """Create user""" fake_hashed_password =user.password + "notreallyhashed"# In practice, use passlib for hashing db_user = models.User( email=user.email, hashed_password=fake_hashed_password ) db.add(db_user) db.commit()# Commit transaction db.refresh(db_user)# Refresh object to get database-generated id return db_user def get_items(db: Session, skip: int=0, limit: int=100): """Get item list""" return db.query(models.Item).offset(skip).limit(limit).all() def create_item(db: Session, item: schemas.ItemCreate, user_id: int): """Create item""" db_item = models.Item(**item.model_dump(), owner_id=user_id) db.add(db_item) db.commit() db.refresh(db_item) return db_item * * * ## 5. FastAPI Application Entry Point Integrate all components in `main.py`: ## Example # main.py from typing import Annotated from fastapi import Depends, FastAPI, HTTPException from sqlalchemy.orm import Session from pydantic import ConfigDict import models, schemas, crud from database import engine, SessionLocal, Base # Create database tables Base.metadata.create_all(bind=engine) app = FastAPI() # Dependency: get database session def get_db(): db = SessionLocal() try: yield db finally: db.close() # ===== User Routes ===== @app.post("/users/", response_model=schemas.UserOut) def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)): # Check if email already exists db_user = crud.get_user_by_email(db,email=user.email) if db_user: raise HTTPException(status_code=400, detail="Email already registered") return crud.create_user(db=db,user=user) @app.get("/users/", response_model=list[schemas.UserOut]) def read_users(skip: int=0, limit: int=100, db: Session = Depends(get_db)): users = crud.get_users(db, skip=skip, limit=limit) return users @app.get("/users/{user_id}", response_model=schemas.UserOut) 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 does not exist") return db_user # ===== Item Routes ===== @app.post("/users/{user_id}/items/", response_model=schemas.ItemOut) def create_item_for_user( user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db), ): return crud.create_item(db=db, item=item, user_id=user_id) @app.get("/items/", response_model=list[schemas.ItemOut]) def read_items(skip: int=0, limit: int=100, db: Session = Depends(get_db)): items = crud.get_items(db, skip=skip, limit=limit) return items * * * ## Database Session Dependency Injection The core pattern is using dependency injection to manage database sessions: def get_db(): db = SessionLocal() # Create session try: yield db # Provide session to route functions finally: db.close() # Close session after request completes Each request gets an independent database session, which is automatically closed after the request completes, preventing resource leaks. * * * ## Connection Strings for Different Databases | Database | Connection String Example | | --- | --- | | SQLite | `sqlite:///./sql_app.db` | | PostgreSQL | `postgresql://user:password@localhost/dbname` | | MySQL | `mysql+pymysql://user:password@localhost/dbname` | > Switching databases only requires modifying the connection string; the rest of the code remains unchanged. This is one of the advantages of using an ORM. * * * ## Summary * Use SQLAlchemy as the ORM, with FastAPI managing database sessions through dependency injection * Pydantic models (`schemas.py`) handle API data validation, while SQLAlchemy models (`models.py`) handle database table structures * CRUD operations are encapsulated in `crud.py`, keeping the code clean * Use SQLite for development; switch to PostgreSQL/MySQL for production by simply changing the connection string * `from_attributes=True` allows Pydantic to read data from ORM objects