Advanced FastAPI: Databases, Auth, and Async

FastAPI does not impose any restrictions on database tools,but SQLAlchemy is one of the most powerful ORMs in the python ecosystem. This section will teach you how to configure a database from scratch.

1.Database Integration with SQLAlchemy

Step 1: Install Dependencies

pip install sqlalchemy

Step 2: Define Data Models

# models.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, index=True)  
    email = Column(String(100), unique=True)                
    hashed_password = Column(String(200))  

declarative_base(): Creates a base class for models, similar to Django’s models.Model
Column: Defines a table field, such as Integer, String, etc.

Step 3: Initialize Database Session

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./app.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Step 4: User Creation and Database Operations with FastAPI

# main.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Connect to SQLite (use PostgreSQL ofrom fastapi import FastAPI, Depends,HTTPException,status
from pydantic import BaseModel

from sqlalchemy.orm import Session
from sqlalchemy import text

from database import engine, SessionLocal
from models import Base, User
import hashlib

app = FastAPI()

class UserCreateRequest(BaseModel):
    username: str
    password: str
    email: str

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

def get_password_hash(password: str) -> str:
    return hashlib.sha256(password.encode()).hexdigest() 

@app.on_event("startup")
async def startup():
    print("Starting up...")
    with engine.begin() as conn:
        table_exists = conn.execute(
            text("SELECT name FROM sqlite_master WHERE type='table' AND name='users'")
        )
        if not table_exists.scalar():
            Base.metadata.create_all(bind=engine)
            print("Database tables created.")
            
@app.get("/users")
async def read_users(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    return user

@app.post("/add_users/", status_code=status.HTTP_201_CREATED)
async def add_user(user_data: UserCreateRequest,db: Session = Depends(get_db)  
):
    
    existing_user = db.query(User).filter(User.user_name == user_data.username).first()
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already registered."
        )

    hashed_password = get_password_hash(user_data.password)
    new_user = User(
        user_name=user_data.username,
        email=user_data.email,
        hashed_password=hashed_password
    )

    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return {"message": "User created successfully", "user_id": new_user.id}

Step 5: Run the application with uvicorn

uvicorn main:app --reload

Step 6: Use Postman for API testing.

• Use Postman to call the add_users endpoint to add a user

• Use Postman to call the users endpoint to view user information from the database

In summary, mastering FastAPI`s advanced features—such as database integration, authentication, and asynchronous programming—empowers you to build robust, secure, and high-performance web APIs with ease.