FastAPI has become the go-to framework for building SaaS APIs in Python. Its combination of async support, automatic OpenAPI documentation, and Pydantic validation makes it possible to build production-grade APIs with remarkably little boilerplate. This tutorial walks you through building a complete SaaS API from scratch, covering authentication, multi-tenancy, database design, testing, and deployment.
By the end of this tutorial, you'll have a fully functional multi-tenant SaaS API with JWT authentication, cursor-based pagination, webhook delivery, and comprehensive test coverage.
Prerequisites
- Python 3.12+
- PostgreSQL 15+
- Redis 7+
- Basic familiarity with async Python
Project Setup
Initialize the project with a modern Python toolchain:
bash
1mkdir saas-api && cd saas-api
2python -m venv .venv
3source .venv/bin/activate
4
5pip install fastapi uvicorn[standard] sqlalchemy[asyncio] asyncpg \
6 alembic pydantic-settings python-jose[cryptography] passlib[bcrypt] \
7 redis httpx python-multipart
8
Create the project structure:
bash
1mkdir -p app/{domain/{auth,orders,users,webhooks},infrastructure,shared}
2mkdir -p tests migrations
3touch app/__init__.py app/main.py app/config.py app/dependencies.py
4
Step 1: Configuration
Create type-safe configuration with Pydantic Settings:
python
1# app/config.py
2from pydantic_settings import BaseSettings
3from functools import lru_cache
4
5class Settings(BaseSettings):
6 app_name: str = "SaaS API"
7 debug: bool = False
8
9 database_url: str
10 db_pool_size: int = 20
11 db_max_overflow: int = 10
12
13 redis_url: str
14
15 jwt_secret: str
16 jwt_algorithm: str = "HS256"
17 access_token_expire_minutes: int = 15
18 refresh_token_expire_days: int = 30
19
20 webhook_signing_secret: str
21 webhook_max_retries: int = 5
22
23 allowed_origins: list[str] = ["http://localhost:3000"]
24
25 model_config = {"env_file": ".env"}
26
27
28def get_settings() -> Settings:
29 return Settings()
30
Step 2: Database Setup
Configure SQLAlchemy 2.0 with async support:
python
1# app/infrastructure/database.py
2from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
3from sqlalchemy.orm import DeclarativeBase
4from app.config import get_settings
5
6settings = get_settings()
7
8engine = create_async_engine(
9 settings.database_url,
10 pool_size=settings.db_pool_size,
11 max_overflow=settings.db_max_overflow,
12 echo=settings.debug,
13)
14
15AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
16
17class Base(DeclarativeBase):
18 pass
19
20async def get_db() -> AsyncSession:
21 async with AsyncSessionLocal() as session:
22 try:
23 yield session
24 await session.commit()
25 except Exception:
26 await session.rollback()
27 raise
28 finally:
29 await session.close()
30
Step 3: Domain Models
Define your database models:
python
1# app/domain/orders/models.py
2from sqlalchemy import String, Numeric, ForeignKey, Enum as SAEnum, Index
3from sqlalchemy.orm import Mapped, mapped_column, relationship
4from datetime import datetime
5from decimal import Decimal
6from uuid import uuid4
7import enum
8
9from app.infrastructure.database import Base
10
11class OrderStatus(str, enum.Enum):
12 PENDING = "pending"
13 CONFIRMED = "confirmed"
14 PROCESSING = "processing"
15 COMPLETED = "completed"
16 CANCELLED = "cancelled"
17
18class Order(Base):
19 __tablename__ = "orders"
20
21 id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid4()))
22 tenant_id: Mapped[str] = mapped_column(String(36), nullable=False)
23 customer_id: Mapped[str] = mapped_column(String(36), nullable=False)
24 status: Mapped[OrderStatus] = mapped_column(SAEnum(OrderStatus), default=OrderStatus.PENDING)
25 total_amount: Mapped[Decimal] = mapped_column(Numeric(10, 2), nullable=False)
26 currency: Mapped[str] = mapped_column(String(3), nullable=False)
27 notes: Mapped[str | None] = mapped_column(String(500))
28 created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
29 updated_at: Mapped[datetime] = mapped_column(default=datetime.utcnow, onupdate=datetime.utcnow)
30
31 items: Mapped[list["OrderItem"]] = relationship(
32 back_populates="order", cascade="all, delete-orphan"
33 )
34
35 __table_args__ = (
36 Index("ix_orders_tenant_created", "tenant_id", "created_at"),
37 )
38
39class OrderItem(Base):
40 __tablename__ = "order_items"
41
42 id: Mapped[str] = mapped_column(String(36), primary_key=True, default=lambda: str(uuid4()))
43 order_id: Mapped[str] = mapped_column(ForeignKey("orders.id", ondelete="CASCADE"))
44 product_id: Mapped[str] = mapped_column(String(36), nullable=False)
45 product_name: Mapped[str] = mapped_column(String(200), nullable=False)
46 quantity: Mapped[int] = mapped_column(nullable=False)
47 unit_price: Mapped[Decimal] = mapped_column(Numeric(10, 2), nullable=False)
48
49 order: Mapped["Order"] = relationship(back_populates="items")
50
Step 4: Pydantic Schemas
Define request and response schemas:
python
1# app/domain/orders/schemas.py
2from pydantic import BaseModel, Field, field_validator
3from decimal import Decimal
4from datetime import datetime
5
6class OrderItemCreate(BaseModel):
7 product_id: str = Field(min_length=1)
8 product_name: str = Field(min_length=1, max_length=200)
9 quantity: int = Field(ge=1, le=1000)
10 unit_price: Decimal = Field(ge=Decimal("0.01"), decimal_places=2)
11
12class OrderCreate(BaseModel):
13 customer_id: str = Field(pattern=r"^[0-9a-f\-]{36}$")
14 items: list[OrderItemCreate] = Field(min_length=1, max_length=50)
15 currency: str = Field(min_length=3, max_length=3)
16 notes: str | None = Field(None, max_length=500)
17
18
19
20 def validate_currency(cls, v: str) -> str:
21 return v.upper()
22
23class OrderItemResponse(BaseModel):
24 id: str
25 product_id: str
26 product_name: str
27 quantity: int
28 unit_price: Decimal
29 model_config = {"from_attributes": True}
30
31class OrderResponse(BaseModel):
32 id: str
33 customer_id: str
34 status: str
35 total_amount: Decimal
36 currency: str
37 notes: str | None
38 items: list[OrderItemResponse]
39 created_at: datetime
40 updated_at: datetime
41 model_config = {"from_attributes": True}
42
43class OrderListResponse(BaseModel):
44 data: list[OrderResponse]
45 next_cursor: str | None
46 has_more: bool
47
48class UpdateOrderStatus(BaseModel):
49 status: str = Field(pattern=r"^(confirmed|processing|completed|cancelled)$")
50
Step 5: Repository Layer
python
1# app/domain/orders/repository.py
2from sqlalchemy import select, func
3from sqlalchemy.ext.asyncio import AsyncSession
4from sqlalchemy.orm import selectinload
5from app.domain.orders.models import Order, OrderStatus
6
7class OrderRepository:
8 def __init__(self, session: AsyncSession):
9 self.session = session
10
11 async def create(self, order: Order) -> Order:
12 self.session.add(order)
13 await self.session.flush()
14 await self.session.refresh(order, ["items"])
15 return order
16
17 async def find_by_id(self, tenant_id: str, order_id: str) -> Order | None:
18 query = (
19 select(Order)
20 .options(selectinload(Order.items))
21 .where(Order.id == order_id, Order.tenant_id == tenant_id)
22 )
23 result = await self.session.execute(query)
24 return result.scalar_one_or_none()
25
26 async def list_orders(
27 self,
28 tenant_id: str,
29 cursor: str | None = None,
30 limit: int = 20,
31 status: str | None = None,
32 ) -> tuple[list[Order], str | None]:
33 query = (
34 select(Order)
35 .options(selectinload(Order.items))
36 .where(Order.tenant_id == tenant_id)
37 )
38
39 if status:
40 query = query.where(Order.status == OrderStatus(status))
41
42 if cursor:
43 cursor_order = await self.session.get(Order, cursor)
44 if cursor_order:
45 query = query.where(Order.created_at < cursor_order.created_at)
46
47 query = query.order_by(Order.created_at.desc()).limit(limit + 1)
48
49 result = await self.session.execute(query)
50 orders = list(result.scalars().all())
51
52 next_cursor = None
53 if len(orders) > limit:
54 next_cursor = orders[limit - 1].id
55 orders = orders[:limit]
56
57 return orders, next_cursor
58
Need a second opinion on your saas engineering architecture?
I run free 30-minute strategy calls for engineering teams tackling this exact problem.
Book a Free CallStep 6: Service Layer
python
1# app/domain/orders/service.py
2from decimal import Decimal
3from app.domain.orders.models import Order, OrderItem, OrderStatus
4from app.domain.orders.repository import OrderRepository
5from app.domain.orders.schemas import OrderCreate
6from app.shared.exceptions import NotFoundError, BusinessError
7
8class OrderService:
9 VALID_TRANSITIONS = {
10 OrderStatus.PENDING: [OrderStatus.CONFIRMED, OrderStatus.CANCELLED],
11 OrderStatus.CONFIRMED: [OrderStatus.PROCESSING, OrderStatus.CANCELLED],
12 OrderStatus.PROCESSING: [OrderStatus.COMPLETED, OrderStatus.CANCELLED],
13 }
14
15 def __init__(self, order_repo: OrderRepository):
16 self.order_repo = order_repo
17
18 async def create_order(self, tenant_id: str, data: OrderCreate) -> Order:
19 total = sum(
20 item.unit_price * item.quantity for item in data.items
21 )
22
23 order = Order(
24 tenant_id=tenant_id,
25 customer_id=data.customer_id,
26 total_amount=total,
27 currency=data.currency,
28 notes=data.notes,
29 items=[
30 OrderItem(
31 product_id=item.product_id,
32 product_name=item.product_name,
33 quantity=item.quantity,
34 unit_price=item.unit_price,
35 )
36 for item in data.items
37 ],
38 )
39
40 return await self.order_repo.create(order)
41
42 async def get_order(self, tenant_id: str, order_id: str) -> Order:
43 order = await self.order_repo.find_by_id(tenant_id, order_id)
44 if not order:
45 raise NotFoundError("Order", order_id)
46 return order
47
48 async def list_orders(
49 self,
50 tenant_id: str,
51 cursor: str | None = None,
52 limit: int = 20,
53 status: str | None = None,
54 ) -> tuple[list[Order], str | None]:
55 return await self.order_repo.list_orders(tenant_id, cursor, limit, status)
56
57 async def update_status(
58 self, tenant_id: str, order_id: str, new_status: str
59 ) -> Order:
60 order = await self.get_order(tenant_id, order_id)
61 target = OrderStatus(new_status)
62
63 allowed = self.VALID_TRANSITIONS.get(order.status, [])
64 if target not in allowed:
65 raise BusinessError(
66 f"Cannot transition from {order.status.value} to {target.value}"
67 )
68
69 order.status = target
70 return order
71
Step 7: API Routes
python
1# app/domain/orders/router.py
2from fastapi import APIRouter, Depends, Query, status
3from typing import Annotated
4from sqlalchemy.ext.asyncio import AsyncSession
5
6from app.infrastructure.database import get_db
7from app.dependencies import get_current_tenant
8from app.domain.orders.schemas import (
9 OrderCreate, OrderResponse, OrderListResponse, UpdateOrderStatus,
10)
11from app.domain.orders.service import OrderService
12from app.domain.orders.repository import OrderRepository
13
14router = APIRouter(prefix="/orders", tags=["Orders"])
15
16def get_order_service(db: Annotated[AsyncSession, Depends(get_db)]) -> OrderService:
17 return OrderService(OrderRepository(db))
18
19
20async def create_order(
21 data: OrderCreate,
22 tenant_id: Annotated[str, Depends(get_current_tenant)],
23 service: Annotated[OrderService, Depends(get_order_service)],
24):
25 """Create a new order."""
26 order = await service.create_order(tenant_id, data)
27 return order
28
29
30async def list_orders(
31 tenant_id: Annotated[str, Depends(get_current_tenant)],
32 service: Annotated[OrderService, Depends(get_order_service)],
33 cursor: str | None = Query(None),
34 limit: int = Query(20, ge=1, le=100),
35 status: str | None = Query(None),
36):
37 """List orders with cursor-based pagination."""
38 orders, next_cursor = await service.list_orders(
39 tenant_id, cursor, limit, status
40 )
41 return OrderListResponse(
42 data=orders,
43 next_cursor=next_cursor,
44 has_more=next_cursor is not None,
45 )
46
47
48async def get_order(
49 order_id: str,
50 tenant_id: Annotated[str, Depends(get_current_tenant)],
51 service: Annotated[OrderService, Depends(get_order_service)],
52):
53 """Get a single order by ID."""
54 return await service.get_order(tenant_id, order_id)
55
56
57async def update_order_status(
58 order_id: str,
59 data: UpdateOrderStatus,
60 tenant_id: Annotated[str, Depends(get_current_tenant)],
61 service: Annotated[OrderService, Depends(get_order_service)],
62):
63 """Update order status with transition validation."""
64 return await service.update_status(tenant_id, order_id, data.status)
65
Step 8: Authentication
Implement JWT authentication with refresh tokens:
python
1# app/domain/auth/service.py
2from datetime import datetime, timedelta
3from jose import jwt, JWTError
4from passlib.context import CryptContext
5from app.config import get_settings
6from app.shared.exceptions import UnauthorizedError
7
8settings = get_settings()
9pwd_context = CryptContext(schemes=["bcrypt"])
10
11class AuthService:
12 def create_access_token(self, user_id: str, tenant_id: str) -> str:
13 payload = {
14 "sub": user_id,
15 "tenant_id": tenant_id,
16 "type": "access",
17 "exp": datetime.utcnow() + timedelta(
18 minutes=settings.access_token_expire_minutes
19 ),
20 }
21 return jwt.encode(payload, settings.jwt_secret, settings.jwt_algorithm)
22
23 def create_refresh_token(self, user_id: str, tenant_id: str) -> str:
24 payload = {
25 "sub": user_id,
26 "tenant_id": tenant_id,
27 "type": "refresh",
28 "exp": datetime.utcnow() + timedelta(
29 days=settings.refresh_token_expire_days
30 ),
31 }
32 return jwt.encode(payload, settings.jwt_secret, settings.jwt_algorithm)
33
34 def decode_token(self, token: str) -> dict:
35 try:
36 payload = jwt.decode(
37 token, settings.jwt_secret, algorithms=[settings.jwt_algorithm]
38 )
39 return payload
40 except JWTError:
41 raise UnauthorizedError("Invalid or expired token")
42
43 def verify_password(self, plain: str, hashed: str) -> bool:
44 return pwd_context.verify(plain, hashed)
45
46 def hash_password(self, password: str) -> str:
47 return pwd_context.hash(password)
48
49# app/dependencies.py
50from fastapi import Depends
51from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
52from typing import Annotated
53from app.domain.auth.service import AuthService
54
55security = HTTPBearer()
56
57async def get_current_tenant(
58 credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)],
59) -> str:
60 auth_service = AuthService()
61 payload = auth_service.decode_token(credentials.credentials)
62 if payload.get("type") != "access":
63 raise UnauthorizedError("Invalid token type")
64 return payload["tenant_id"]
65
66async def get_current_user_id(
67 credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)],
68) -> str:
69 auth_service = AuthService()
70 payload = auth_service.decode_token(credentials.credentials)
71 return payload["sub"]
72
Step 9: Error Handling
python
1# app/shared/exceptions.py
2class AppError(Exception):
3 def __init__(self, status_code: int, title: str, detail: str):
4 self.status_code = status_code
5 self.title = title
6 self.detail = detail
7
8class NotFoundError(AppError):
9 def __init__(self, resource: str, resource_id: str):
10 super().__init__(404, "Not Found", f"{resource} '{resource_id}' not found")
11
12class BusinessError(AppError):
13 def __init__(self, detail: str):
14 super().__init__(422, "Business Rule Violation", detail)
15
16class UnauthorizedError(AppError):
17 def __init__(self, detail: str = "Authentication required"):
18 super().__init__(401, "Unauthorized", detail)
19
20# Register handlers in main.py
21from fastapi import FastAPI, Request
22from fastapi.responses import JSONResponse
23
24def register_exception_handlers(app: FastAPI):
25
26 async def app_error_handler(request: Request, exc: AppError):
27 return JSONResponse(
28 status_code=exc.status_code,
29 content={
30 "type": exc.title.lower().replace(" ", "_"),
31 "title": exc.title,
32 "status": exc.status_code,
33 "detail": exc.detail,
34 },
35 )
36
Step 10: Application Factory
python
1# app/main.py
2from fastapi import FastAPI
3from fastapi.middleware.cors import CORSMiddleware
4from contextlib import asynccontextmanager
5from app.config import get_settings
6from app.infrastructure.database import engine
7from app.domain.orders.router import router as order_router
8from app.domain.auth.router import router as auth_router
9from app.shared.exceptions import register_exception_handlers
10
11settings = get_settings()
12
13
14async def lifespan(app: FastAPI):
15 yield
16 await engine.dispose()
17
18app = FastAPI(
19 title=settings.app_name,
20 lifespan=lifespan,
21 docs_url="/docs" if settings.debug else None,
22 redoc_url="/redoc" if settings.debug else None,
23)
24
25app.add_middleware(
26 CORSMiddleware,
27 allow_origins=settings.allowed_origins,
28 allow_credentials=True,
29 allow_methods=["*"],
30 allow_headers=["*"],
31)
32
33app.include_router(auth_router, prefix="/api/v1")
34app.include_router(order_router, prefix="/api/v1")
35
36register_exception_handlers(app)
37
38
39async def health():
40 return {"status": "ok"}
41
Step 11: Testing
python
1# tests/test_orders.py
2import pytest
3from httpx import AsyncClient, ASGITransport
4from app.main import app
5from app.domain.auth.service import AuthService
6
7
8def auth_headers():
9 auth = AuthService()
10 token = auth.create_access_token("user-1", "tenant-1")
11 return {"Authorization": f"Bearer {token}"}
12
13
14async def client():
15 transport = ASGITransport(app=app)
16 async with AsyncClient(transport=transport, base_url="http://test") as c:
17 yield c
18
19
20async def test_create_order(client: AsyncClient, auth_headers: dict):
21 response = await client.post(
22 "/api/v1/orders/",
23 json={
24 "customer_id": "550e8400-e29b-41d4-a716-446655440000",
25 "items": [
26 {
27 "product_id": "prod-1",
28 "product_name": "Widget",
29 "quantity": 2,
30 "unit_price": "29.99",
31 }
32 ],
33 "currency": "USD",
34 },
35 headers=auth_headers,
36 )
37 assert response.status_code == 201
38 data = response.json()
39 assert data["status"] == "pending"
40 assert data["total_amount"] == "59.98"
41 assert len(data["items"]) == 1
42
43
44async def test_create_order_validation_error(client: AsyncClient, auth_headers: dict):
45 response = await client.post(
46 "/api/v1/orders/",
47 json={"customer_id": "invalid", "items": [], "currency": "X"},
48 headers=auth_headers,
49 )
50 assert response.status_code == 422
51
Step 12: Deployment
Create a Dockerfile for production:
dockerfile
1FROM python:3.12-slim as builder
2WORKDIR /app
3COPY requirements.txt .
4RUN pip install --no-cache-dir -r requirements.txt
5
6FROM python:3.12-slim
7WORKDIR /app
8COPY --from=builder /usr/local/lib/python3.12/site-packages /usr/local/lib/python3.12/site-packages
9COPY --from=builder /usr/local/bin /usr/local/bin
10COPY . .
11
12EXPOSE 8000
13CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]
14
Conclusion
FastAPI provides everything you need to build a professional SaaS API without the complexity of heavier frameworks. The combination of Pydantic for validation, SQLAlchemy for database access, and FastAPI's dependency injection creates a clean, testable architecture where each component has a clear responsibility.
The tutorial covered the complete lifecycle: from project structure and database models through authentication, error handling, and testing. Every layer is independently testable, and the dependency injection system makes swapping implementations straightforward. FastAPI automatically generates interactive API documentation from your Pydantic schemas, giving your frontend team a self-serve resource for exploring available endpoints.
As your SaaS grows, this architecture scales naturally. Add new domains by creating new subdirectories with their own models, schemas, repositories, and routes. The service layer prevents business logic from leaking into route handlers, and the repository pattern keeps database queries contained and optimizable.