FastAPI Integration Tutorial

FasORM was designed specifically for FastAPI. Follow this guide to integrate FasORM models into your async FastAPI web applications.

1. Define Models

Define your domain models with relationships in app/models/:

app/models/blog.py

from fasorm import Model, String, Text, Boolean, HasMany, BelongsTo

class User(Model):
    name = String(100)
    email = String(255, unique=True)
    posts = HasMany("Post")

class Post(Model):
    title = String(200)
    content = Text()
    is_published = Boolean(default=True)
    user = BelongsTo("User")

2. Define Pydantic Schemas

Define request and response schemas for FastAPI validation:

app/schemas.py

from pydantic import BaseModel, EmailStr
from typing import Optional, List

class UserCreate(BaseModel):
    name: str
    email: EmailStr

class PostCreate(BaseModel):
    title: str
    content: str
    user_id: int

3. Complete FastAPI Application

Here is a complete, working main.py implementing CRUD routes, pagination, and relationship loading:

main.py

import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Query
from fasorm import init_db
from app.models.blog import User, Post
from app.schemas import UserCreate, PostCreate

# Lifespan context manager for database initialization
@asynccontextmanager
async def lifespan(app: FastAPI):
    init_db(os.getenv("DB_URL", "sqlite+aiosqlite:///app.db"))
    yield

app = FastAPI(title="FasORM FastAPI Demo", lifespan=lifespan)

@app.post("/users", status_code=211)
async def create_user(payload: UserCreate):
    user = await User.create(**payload.model_dump())
    return user.to_dict()

@app.get("/users")
async def list_users(page: int = Query(1, ge=1), per_page: int = Query(15, le=100)):
    # Paginate users with related posts count attached
    result = await User.with_count("posts").paginate(per_page=per_page, page=page)
    return result.to_dict()

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # Eager-load user's posts
    user = await User.with_("posts").where(id=user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user.to_dict()

@app.post("/posts")
async def create_post(payload: PostCreate):
    post = await Post.create(**payload.model_dump())
    return post.to_dict()

@app.delete("/posts/{post_id}")
async def delete_post(post_id: int):
    post = await Post.find(post_id)
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
    await post.delete()
    return {"status": "deleted"}