python 39 lines · 7 steps

Wiring SQLAlchemy sessions into FastAPI

A dependency-injected session powers two endpoints that create and fetch users with proper HTTP status codes.

Explained by highlit
FastAPI Intermediate dependency-injection orm sessions rest-api error-handling
1from fastapi import Depends, FastAPI, HTTPException
2from sqlalchemy import create_engine
3from sqlalchemy.orm import Session, sessionmaker
4 
5from .models import User
6from .schemas import UserCreate, UserRead
7 
8engine = create_engine("postgresql://localhost/app", pool_pre_ping=True)
9SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
10 
11app = FastAPI()
12 
13 
14def get_db():
15 db = SessionLocal()
16 try:
17 yield db
18 finally:
19 db.close()
20 
21 
22@app.post("/users", response_model=UserRead, status_code=201)
23def create_user(payload: UserCreate, db: Session = Depends(get_db)):
24 if db.query(User).filter(User.email == payload.email).first():
25 raise HTTPException(status_code=409, detail="Email already registered")
26 
27 user = User(email=payload.email, full_name=payload.full_name)
28 db.add(user)
29 db.commit()
30 db.refresh(user)
31 return user
32 
33 
34@app.get("/users/{user_id}", response_model=UserRead)
35def read_user(user_id: int, db: Session = Depends(get_db)):
36 user = db.get(User, user_id)
37 if user is None:
38 raise HTTPException(status_code=404, detail="User not found")
39 return user
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1A generator dependency cleanly opens a resource before the request and guarantees cleanup after it, even on errors.
  2. 2response_model separates your ORM objects from the shape you expose, validating and serializing the output.
  3. 3Mapping domain conditions to explicit HTTP status codes makes the API's contract clear to clients.

Related explainers

python 
from collections import ChainMap
from functools import reduce
from typing import Any, Iterable, Mapping
 

Five ways to merge dicts in Python

dictionaries merging recursion
Intermediate 8 steps
java 
@Service
public class ReportGenerationService {
 
    private final ReportRepository reportRepository;

Async report generation with Spring @Async

async dependency-injection completablefuture
Intermediate 7 steps
ruby 
class PurgeStaleExportsJob < ApplicationJob
  queue_as :maintenance
 
  retry_on ActiveRecord::Deadlocked, wait: :polynomially_longer, attempts: 5

Purging stale exports with an Active Job in Rails

background-jobs batching error-handling
Intermediate 7 steps
javascript 
const { validationResult, matchedData } = require('express-validator');
 
function validate(schema) {
  const runners = schema.map((rule) => rule.run.bind(rule));

A reusable validation middleware in Express

middleware validation higher-order-functions
Intermediate 8 steps
python 
import re
 
_UNIT_SECONDS = {
    "w": 604800,

Parsing duration strings like 1h30m in Python

regex parsing validation
Intermediate 8 steps
python 
import csv
from collections import defaultdict
from dataclasses import dataclass, field
from decimal import Decimal, InvalidOperation

Aggregating CSV sales by category in Python

dataclasses csv-parsing defaultdict
Intermediate 8 steps

Share this explainer

Here's the card — post it anywhere.

Wiring SQLAlchemy sessions into FastAPI — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code