Python has become a powerhouse for SaaS API development, combining rapid development velocity with modern async capabilities that handle thousands of concurrent connections. This guide walks through building a production-grade SaaS API with Python, covering FastAPI and Django REST Framework approaches, database patterns, authentication, and the architectural decisions that matter at scale.
Every pattern here has been validated in production SaaS systems. The focus is on idiomatic Python that's readable, testable, and performant.
Choosing Your Framework
Python offers two dominant API frameworks, each suited to different contexts:
FastAPI excels when you need high performance, async operations, and automatic OpenAPI documentation. It's the right choice for new greenfield APIs and microservices.
Django REST Framework (DRF) excels when you need a full-featured admin interface, ORM with migrations, and the massive Django ecosystem. It's the right choice for monolithic SaaS applications with complex data models.
This guide primarily uses FastAPI, with DRF patterns highlighted where they offer distinct advantages.
Project Structure
A well-organized FastAPI project separates concerns clearly:
1├── app/
2│ ├── __init__.py
3│ ├── main.py # Application factory
4│ ├── config.py # Settings management
5│ ├── dependencies.py # Dependency injection
6│ ├── domain/
7│ │ ├── orders/
8│ │ │ ├── __init__.py
9│ │ │ ├── models.py # SQLAlchemy models
10│ │ │ ├── schemas.py # Pydantic schemas
11│ │ │ ├── router.py # API routes
12│ │ │ ├── service.py # Business logic
13│ │ │ └── repository.py # Database queries
14│ │ └── users/
15│ │ ├── __init__.py
16│ │ ├── models.py
17│ │ ├── schemas.py
18│ │ ├── router.py
19│ │ └── service.py
20│ ├── infrastructure/
21│ │ ├── database.py # Database setup
22│ │ ├── cache.py # Redis client
23│ │ ├── security.py # JWT handling
24│ │ └── middleware.py # Custom middleware
25│ └── shared/
26│ ├── exceptions.py # Custom exceptions
27│ ├── responses.py # Response models
28│ └── pagination.py # Cursor pagination
29├── migrations/
30│ └── versions/
31├── tests/
32├── alembic.ini
33├── pyproject.toml
34└── Dockerfile
35
Configuration with Pydantic Settings
Type-safe configuration loading with validation:
python
1from pydantic_settings import BaseSettings
2from functools import lru_cache
3
4class Settings(BaseSettings):
5 # Application
6 app_name: str = "SaaS API"
7 debug: bool = False
8 api_prefix: str = "/api/v1"
9
10 # Database
11 database_url: str
12 db_pool_size: int = 20
13 db_max_overflow: int = 10
14
15 # Redis
16 redis_url: str
17
18 # Auth
19 jwt_secret: str
20 jwt_algorithm: str = "HS256"
21 access_token_expire_minutes: int = 15
22 refresh_token_expire_days: int = 30
23
24 # CORS
25 allowed_origins: list[str] = ["http://localhost:3000"]
26
27 model_config = {"env_file": ".env"}
28
29
30def get_settings() -> Settings:
31 return Settings()
32
Database Layer with SQLAlchemy 2.0
Use SQLAlchemy 2.0's modern async API with proper session management:
python
1from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
2from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
3from sqlalchemy import String, DateTime, Numeric, ForeignKey
4from datetime import datetime
5from uuid import uuid4
6
7class Base(DeclarativeBase):
8 pass
9
10class TimestampMixin:
11 created_at: Mapped[datetime] = mapped_column(
12 DateTime, default=datetime.utcnow, nullable=False
13 )
14 updated_at: Mapped[datetime] = mapped_column(
15 DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False
16 )
17
18# Database setup
19engine = create_async_engine(
20 settings.database_url,
21 pool_size=settings.db_pool_size,
22 max_overflow=settings.db_max_overflow,
23 echo=settings.debug,
24)
25
26AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
27
28async def get_db() -> AsyncSession:
29 async with AsyncSessionLocal() as session:
30 try:
31 yield session
32 finally:
33 await session.close()
34
Domain Models
python
1from sqlalchemy import String, Numeric, ForeignKey, Enum as SAEnum
2from sqlalchemy.orm import relationship
3from decimal import Decimal
4import enum
5
6class OrderStatus(str, enum.Enum):
7 PENDING = "pending"
8 CONFIRMED = "confirmed"
9 PROCESSING = "processing"
10 COMPLETED = "completed"
11 CANCELLED = "cancelled"
12
13class Order(Base, TimestampMixin):
14 __tablename__ = "orders"
15
16 id: Mapped[str] = mapped_column(
17 String(36), primary_key=True, default=lambda: str(uuid4())
18 )
19 tenant_id: Mapped[str] = mapped_column(String(36), index=True, nullable=False)
20 customer_id: Mapped[str] = mapped_column(String(36), nullable=False)
21 status: Mapped[OrderStatus] = mapped_column(
22 SAEnum(OrderStatus), default=OrderStatus.PENDING
23 )
24 total_amount: Mapped[Decimal] = mapped_column(Numeric(10, 2), nullable=False)
25 currency: Mapped[str] = mapped_column(String(3), nullable=False)
26
27 items: Mapped[list["OrderItem"]] = relationship(
28 back_populates="order", cascade="all, delete-orphan"
29 )
30
31class OrderItem(Base):
32 __tablename__ = "order_items"
33
34 id: Mapped[str] = mapped_column(
35 String(36), primary_key=True, default=lambda: str(uuid4())
36 )
37 order_id: Mapped[str] = mapped_column(ForeignKey("orders.id"), nullable=False)
38 product_id: Mapped[str] = mapped_column(String(36), nullable=False)
39 quantity: Mapped[int] = mapped_column(nullable=False)
40 unit_price: Mapped[Decimal] = mapped_column(Numeric(10, 2), nullable=False)
41
42 order: Mapped["Order"] = relationship(back_populates="items")
43
Pydantic Schemas for Request/Response
python
1from pydantic import BaseModel, Field, field_validator
2from decimal import Decimal
3from datetime import datetime
4
5class OrderItemCreate(BaseModel):
6 product_id: str = Field(min_length=1)
7 quantity: int = Field(ge=1)
8 unit_price: Decimal = Field(ge=Decimal("0.01"), decimal_places=2)
9
10class OrderCreate(BaseModel):
11 customer_id: str = Field(pattern=r"^[0-9a-f\-]{36}$")
12 items: list[OrderItemCreate] = Field(min_length=1)
13 currency: str = Field(min_length=3, max_length=3)
14 metadata: dict | None = None
15
16
17
18 def validate_currency(cls, v: str) -> str:
19 valid_currencies = {"USD", "EUR", "GBP", "AED"}
20 if v.upper() not in valid_currencies:
21 raise ValueError(f"Currency must be one of: {valid_currencies}")
22 return v.upper()
23
24class OrderItemResponse(BaseModel):
25 id: str
26 product_id: str
27 quantity: int
28 unit_price: Decimal
29
30 model_config = {"from_attributes": True}
31
32class OrderResponse(BaseModel):
33 id: str
34 customer_id: str
35 status: str
36 total_amount: Decimal
37 currency: str
38 items: list[OrderItemResponse]
39 created_at: datetime
40
41 model_config = {"from_attributes": True}
42
Repository Pattern
python
1from sqlalchemy import select, func
2from sqlalchemy.ext.asyncio import AsyncSession
3from sqlalchemy.orm import selectinload
4
5class OrderRepository:
6 def __init__(self, session: AsyncSession):
7 self.session = session
8
9 async def find_by_id(self, tenant_id: str, order_id: str) -> Order | None:
10 query = (
11 select(Order)
12 .options(selectinload(Order.items))
13 .where(Order.id == order_id, Order.tenant_id == tenant_id)
14 )
15 result = await self.session.execute(query)
16 return result.scalar_one_or_none()
17
18 async def list_orders(
19 self,
20 tenant_id: str,
21 cursor: str | None = None,
22 limit: int = 20,
23 status: OrderStatus | None = None,
24 ) -> tuple[list[Order], str | None]:
25 query = (
26 select(Order)
27 .options(selectinload(Order.items))
28 .where(Order.tenant_id == tenant_id)
29 )
30
31 if status:
32 query = query.where(Order.status == status)
33
34 if cursor:
35 query = query.where(Order.id < cursor)
36
37 query = query.order_by(Order.id.desc()).limit(limit + 1)
38
39 result = await self.session.execute(query)
40 orders = list(result.scalars().all())
41
42 next_cursor = None
43 if len(orders) > limit:
44 next_cursor = orders[limit - 1].id
45 orders = orders[:limit]
46
47 return orders, next_cursor
48
49 async def create(self, order: Order) -> Order:
50 self.session.add(order)
51 await self.session.flush()
52 return order
53
54 async def count_by_tenant(self, tenant_id: str) -> int:
55 query = select(func.count()).select_from(Order).where(
56 Order.tenant_id == tenant_id
57 )
58 result = await self.session.execute(query)
59 return result.scalar_one()
60
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 CallService Layer
python
1from decimal import Decimal
2
3class OrderService:
4 def __init__(self, order_repo: OrderRepository, customer_repo: CustomerRepository):
5 self.order_repo = order_repo
6 self.customer_repo = customer_repo
7
8 async def create_order(
9 self, tenant_id: str, data: OrderCreate
10 ) -> Order:
11 customer = await self.customer_repo.find_by_id(tenant_id, data.customer_id)
12 if not customer:
13 raise ResourceNotFoundError("Customer", data.customer_id)
14
15 if customer.status != "active":
16 raise BusinessRuleError("Cannot create order for inactive customer")
17
18 total = sum(
19 item.unit_price * item.quantity for item in data.items
20 )
21
22 order = Order(
23 tenant_id=tenant_id,
24 customer_id=data.customer_id,
25 total_amount=total,
26 currency=data.currency,
27 status=OrderStatus.PENDING,
28 items=[
29 OrderItem(
30 product_id=item.product_id,
31 quantity=item.quantity,
32 unit_price=item.unit_price,
33 )
34 for item in data.items
35 ],
36 )
37
38 return await self.order_repo.create(order)
39
40 async def get_order(self, tenant_id: str, order_id: str) -> Order:
41 order = await self.order_repo.find_by_id(tenant_id, order_id)
42 if not order:
43 raise ResourceNotFoundError("Order", order_id)
44 return order
45
46 async def list_orders(
47 self,
48 tenant_id: str,
49 cursor: str | None = None,
50 limit: int = 20,
51 status: str | None = None,
52 ) -> tuple[list[Order], str | None]:
53 order_status = OrderStatus(status) if status else None
54 return await self.order_repo.list_orders(
55 tenant_id, cursor, limit, order_status
56 )
57
API Routes
python
1from fastapi import APIRouter, Depends, Query, status
2from typing import Annotated
3
4router = APIRouter(prefix="/orders", tags=["orders"])
5
6
7async def create_order(
8 data: OrderCreate,
9 tenant_id: Annotated[str, Depends(get_current_tenant)],
10 service: Annotated[OrderService, Depends(get_order_service)],
11):
12 order = await service.create_order(tenant_id, data)
13 return order
14
15
16async def get_order(
17 order_id: str,
18 tenant_id: Annotated[str, Depends(get_current_tenant)],
19 service: Annotated[OrderService, Depends(get_order_service)],
20):
21 return await service.get_order(tenant_id, order_id)
22
23
24async def list_orders(
25 tenant_id: Annotated[str, Depends(get_current_tenant)],
26 service: Annotated[OrderService, Depends(get_order_service)],
27 cursor: str | None = Query(None),
28 limit: int = Query(20, ge=1, le=100),
29 status: str | None = Query(None),
30):
31 orders, next_cursor = await service.list_orders(
32 tenant_id, cursor, limit, status
33 )
34 return PaginatedResponse(
35 data=orders,
36 next_cursor=next_cursor,
37 has_more=next_cursor is not None,
38 )
39
JWT Authentication
python
1from fastapi import Depends, HTTPException, status
2from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
3import jwt
4from datetime import datetime, timedelta
5
6security = HTTPBearer()
7
8class JWTService:
9 def __init__(self, secret: str, algorithm: str = "HS256"):
10 self.secret = secret
11 self.algorithm = algorithm
12
13 def create_access_token(
14 self, user_id: str, tenant_id: str, expires_minutes: int = 15
15 ) -> str:
16 payload = {
17 "sub": user_id,
18 "tenant_id": tenant_id,
19 "exp": datetime.utcnow() + timedelta(minutes=expires_minutes),
20 "type": "access",
21 }
22 return jwt.encode(payload, self.secret, algorithm=self.algorithm)
23
24 def decode_token(self, token: str) -> dict:
25 try:
26 return jwt.decode(token, self.secret, algorithms=[self.algorithm])
27 except jwt.ExpiredSignatureError:
28 raise HTTPException(
29 status_code=status.HTTP_401_UNAUTHORIZED,
30 detail="Token has expired",
31 )
32 except jwt.InvalidTokenError:
33 raise HTTPException(
34 status_code=status.HTTP_401_UNAUTHORIZED,
35 detail="Invalid token",
36 )
37
38async def get_current_tenant(
39 credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)],
40 jwt_service: Annotated[JWTService, Depends(get_jwt_service)],
41) -> str:
42 payload = jwt_service.decode_token(credentials.credentials)
43 return payload["tenant_id"]
44
45async def get_current_user(
46 credentials: Annotated[HTTPAuthorizationCredentials, Depends(security)],
47 jwt_service: Annotated[JWTService, Depends(get_jwt_service)],
48) -> str:
49 payload = jwt_service.decode_token(credentials.credentials)
50 return payload["sub"]
51
Error Handling
python
1from fastapi import FastAPI, Request
2from fastapi.responses import JSONResponse
3
4class AppError(Exception):
5 def __init__(self, status_code: int, title: str, detail: str):
6 self.status_code = status_code
7 self.title = title
8 self.detail = detail
9
10class ResourceNotFoundError(AppError):
11 def __init__(self, resource: str, resource_id: str):
12 super().__init__(404, "Not Found", f"{resource} '{resource_id}' not found")
13
14class BusinessRuleError(AppError):
15 def __init__(self, detail: str):
16 super().__init__(422, "Business Rule Violation", detail)
17
18def register_exception_handlers(app: FastAPI):
19
20 async def app_error_handler(request: Request, exc: AppError):
21 return JSONResponse(
22 status_code=exc.status_code,
23 content={
24 "type": exc.title.lower().replace(" ", "_"),
25 "title": exc.title,
26 "status": exc.status_code,
27 "detail": exc.detail,
28 },
29 )
30
31
32 async def unhandled_error_handler(request: Request, exc: Exception):
33 logger.exception("Unhandled exception")
34 return JSONResponse(
35 status_code=500,
36 content={
37 "type": "internal_error",
38 "title": "Internal Server Error",
39 "status": 500,
40 "detail": "An unexpected error occurred",
41 },
42 )
43
Application Factory
python
1from fastapi import FastAPI
2from fastapi.middleware.cors import CORSMiddleware
3from contextlib import asynccontextmanager
4
5
6async def lifespan(app: FastAPI):
7 # Startup
8 await database.connect()
9 await cache.connect()
10 yield
11 # Shutdown
12 await database.disconnect()
13 await cache.disconnect()
14
15def create_app() -> FastAPI:
16 settings = get_settings()
17
18 app = FastAPI(
19 title=settings.app_name,
20 lifespan=lifespan,
21 docs_url="/docs" if settings.debug else None,
22 )
23
24 app.add_middleware(
25 CORSMiddleware,
26 allow_origins=settings.allowed_origins,
27 allow_credentials=True,
28 allow_methods=["*"],
29 allow_headers=["*"],
30 )
31
32 app.include_router(order_router, prefix=settings.api_prefix)
33 app.include_router(user_router, prefix=settings.api_prefix)
34 app.include_router(auth_router, prefix=settings.api_prefix)
35
36 register_exception_handlers(app)
37
38
39 async def health():
40 return {"status": "ok"}
41
42 return app
43
44app = create_app()
45
Conclusion
Python's expressiveness and FastAPI's modern design make building SaaS APIs remarkably productive. The combination of Pydantic for validation, SQLAlchemy 2.0 for async database access, and FastAPI's dependency injection creates a clean architecture where each layer has a single responsibility.
The key to a maintainable Python API is discipline in structure. The repository pattern keeps database queries contained, the service layer encapsulates business rules, and Pydantic schemas define clear API contracts. This separation makes testing straightforward—mock the repository to test services, use TestClient to test routes end-to-end.
Python may not match Go or Rust in raw throughput, but for the vast majority of SaaS applications, the development velocity advantage more than compensates. A well-structured FastAPI application handles thousands of concurrent requests per second, and when you need more performance, the async foundation means you can scale horizontally without architectural changes.