Generated Code Reference
This page documents all code artifacts generated by the FastAPI ORM Generator.
File Structure
orm/
├── __init__.py # Root exports with duplicate handling
├── database.py # Async session factory and engine
├── CLAUDE.md # AI assistant instructions
└── {app}/
├── __init__.py # App exports
├── enums.py # StrEnum classes from TextChoices (if any)
├── models.py # SQLModel table classes
├── schemas.py # Pydantic schemas
└── crud.py # Async CRUD repositorySQLModel Models (models.py)
Each Django model generates a SQLModel table class:
from datetime import datetime
from typing import Optional
from sqlmodel import SQLModel, Field, Relationship
from sqlalchemy import Column, Index, UniqueConstraint
from sqlalchemy.dialects.postgresql import JSONB, ARRAY, UUID
class User(SQLModel, table=True):
"""User account model."""
__tablename__ = "users_user"
# Primary key
id: int = Field(default=None, primary_key=True)
# Required fields
email: str = Field(max_length=255, unique=True)
username: str = Field(max_length=150)
# Optional fields (nullable)
bio: Optional[str] = Field(default=None)
# Fields with defaults
is_active: bool = Field(default=True, index=True)
# Blank string fields
first_name: Optional[str] = Field(default="", max_length=150)
# Auto timestamps (read-only)
created_at: datetime = Field()
updated_at: datetime = Field()
# PostgreSQL types
metadata: dict = Field(default_factory=dict, sa_column=Column(JSONB))
tags: list[str] = Field(default_factory=list, sa_column=Column(ARRAY(String)))
# Foreign keys
profile_id: Optional[int] = Field(default=None, foreign_key="profiles_profile.id")
# Relationships
profile: Optional["Profile"] = Relationship(back_populates="user")
posts: list["Post"] = Relationship(back_populates="author")
# Table constraints
__table_args__ = (
UniqueConstraint("email", name="users_user_email_key"),
Index("users_user_is_active_idx", "is_active"),
)Field Mapping
| Django Field | Python Type | SQLModel Field |
|---|---|---|
AutoField | int | Field(primary_key=True) |
CharField | str | Field(max_length=N) |
TextField | str | Field() |
IntegerField | int | Field() |
FloatField | float | Field() |
DecimalField | Decimal | Field(max_digits=N, decimal_places=N) |
BooleanField | bool | Field(default=...) |
DateField | date | Field() |
DateTimeField | datetime | Field() |
UUIDField | UUID | Field(sa_column=Column(UUID)) |
JSONField | dict | Field(sa_column=Column(JSONB)) |
ArrayField | list[T] | Field(sa_column=Column(ARRAY(T))) |
ForeignKey | int | Field(foreign_key="table.id") |
Nullable/Optional Handling
# null=True → Optional[T] with None default
bio: Optional[str] = Field(default=None)
# blank=True (strings) → Optional[str] with empty default
first_name: Optional[str] = Field(default="", max_length=150)
# null=False, blank=False → Required
email: str = Field(max_length=255)Pydantic Schemas (schemas.py)
Four schemas are generated per model:
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, ConfigDict
class UserBase(BaseModel):
"""Base schema with shared fields."""
email: str
username: str
bio: Optional[str] = None
first_name: Optional[str] = ""
is_active: bool = True
metadata: dict = {}
class UserCreate(BaseModel):
"""Schema for creating User."""
email: str
username: str
bio: Optional[str] = None
first_name: Optional[str] = ""
is_active: bool = True
metadata: dict = {}
# Note: auto_now_add fields (created_at) excluded
class UserRead(UserBase):
"""Schema for reading User."""
model_config = ConfigDict(from_attributes=True)
id: int
created_at: datetime
updated_at: datetime
class UserUpdate(BaseModel):
"""Schema for updating User (all optional)."""
email: Optional[str] = None
username: Optional[str] = None
bio: Optional[str] = None
first_name: Optional[str] = None
is_active: Optional[bool] = None
metadata: Optional[dict] = NoneSchema Rules
| Schema | Purpose | Fields |
|---|---|---|
{Model}Base | Shared definition | All except PK |
{Model}Create | POST body | Excludes auto_now, auto_now_add, PK |
{Model}Read | GET response | All fields including PK |
{Model}Update | PATCH body | All optional (except PK) |
CRUD Repository (crud.py)
Async repository with standard operations:
from typing import Optional, Sequence
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from .models import User
class UserRepository:
"""Repository for User CRUD operations."""
@staticmethod
async def get_by_id(session: AsyncSession, id: int) -> Optional[User]:
"""Get User by primary key."""
statement = select(User).where(User.id == id)
result = await session.execute(statement)
return result.scalars().first()
@staticmethod
async def get_by_ids(
session: AsyncSession,
ids: list[int],
) -> Sequence[User]:
"""Get multiple Users by IDs."""
statement = select(User).where(User.id.in_(ids))
result = await session.execute(statement)
return result.scalars().all()
@staticmethod
async def get_all(
session: AsyncSession,
skip: int = 0,
limit: int = 100,
) -> Sequence[User]:
"""Get all Users with pagination."""
statement = select(User).offset(skip).limit(limit)
result = await session.execute(statement)
return result.scalars().all()
@staticmethod
async def create(session: AsyncSession, obj: User) -> User:
"""Create a new User."""
session.add(obj)
await session.commit()
await session.refresh(obj)
return obj
@staticmethod
async def create_many(
session: AsyncSession,
objects: list[User],
) -> list[User]:
"""Create multiple Users."""
session.add_all(objects)
await session.commit()
for obj in objects:
await session.refresh(obj)
return objects
@staticmethod
async def update(
session: AsyncSession,
id: int,
data: dict,
) -> Optional[User]:
"""Update User by ID."""
obj = await UserRepository.get_by_id(session, id)
if obj:
for key, value in data.items():
setattr(obj, key, value)
await session.commit()
await session.refresh(obj)
return obj
@staticmethod
async def delete(session: AsyncSession, id: int) -> bool:
"""Delete User by ID."""
obj = await UserRepository.get_by_id(session, id)
if obj:
await session.delete(obj)
await session.commit()
return True
return False
@staticmethod
async def exists(session: AsyncSession, id: int) -> bool:
"""Check if User exists."""
obj = await UserRepository.get_by_id(session, id)
return obj is not None
@staticmethod
async def count(session: AsyncSession) -> int:
"""Count all Users."""
from sqlalchemy import func
statement = select(func.count()).select_from(User)
result = await session.execute(statement)
return result.scalar() or 0Repository Methods
| Method | Description | Returns |
|---|---|---|
get_by_id(id) | Get by primary key | Optional[Model] |
get_by_ids(ids) | Get multiple by IDs | Sequence[Model] |
get_all(skip, limit) | Paginated list | Sequence[Model] |
create(obj) | Insert single | Model |
create_many(objects) | Bulk insert | list[Model] |
update(id, data) | Partial update | Optional[Model] |
delete(id) | Delete by ID | bool |
exists(id) | Check existence | bool |
count() | Total count | int |
Database Configuration (database.py)
Async database setup with connection pooling:
import os
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
DATABASE_URL = os.getenv(
"DATABASE_URL",
"postgresql+asyncpg://postgres:postgres@localhost:5432/mydb"
)
# Convert postgres:// to postgresql+asyncpg://
if DATABASE_URL.startswith("postgres://"):
DATABASE_URL = DATABASE_URL.replace("postgres://", "postgresql+asyncpg://", 1)
engine = create_async_engine(
DATABASE_URL,
echo=False,
pool_pre_ping=True,
pool_size=5,
max_overflow=10,
)
async_session_factory = sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False,
)
async def get_session() -> AsyncGenerator[AsyncSession, None]:
"""FastAPI dependency for database sessions."""
async with async_session_factory() as session:
try:
yield session
except Exception:
await session.rollback()
raise
finally:
await session.close()
async def init_db() -> None:
"""Initialize database tables."""
from sqlmodel import SQLModel
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)Root __init__.py
Exports all models with duplicate name handling:
"""
FastAPI ORM - Auto-generated code.
Generated by django-cfg FastAPI ORM Generator.
"""
# Database
from .database import get_session, init_db
# Models by app
from .users.models import User, Profile
from .products.models import Product, Category
from .cfg.models import Currency as CfgCurrency # Aliased duplicate
__all__ = [
"get_session",
"init_db",
"User",
"Profile",
"Product",
"Category",
"CfgCurrency",
]Enums (enums.py)
For each app that has a CharField or IntegerField with a TextChoices class, the generator creates enums.py with matching StrEnum classes.
Django model:
class LLMProvider(models.TextChoices):
OPENROUTER = "openrouter", "OpenRouter"
ANTHROPIC = "anthropic", "Anthropic"
OPENAI = "openai", "OpenAI"
class LLMConfig(models.Model):
provider = models.CharField(
max_length=50,
choices=LLMProvider.choices,
default=LLMProvider.OPENROUTER,
)Generated llm/enums.py:
"""
StrEnum classes for llm.
Auto-generated by django-cfg FastAPI ORM Generator.
Do not edit manually - changes will be overwritten.
"""
from enum import StrEnum
class LLMProvider(StrEnum):
OPENROUTER = "openrouter"
ANTHROPIC = "anthropic"
OPENAI = "openai"The enum class name is taken directly from the TextChoices class. Member names are derived from values: uppercased, with -, ., and spaces replaced by _. Only string-valued choices are included (integer choices are skipped).
Using Generated Enums
from orm.llm.enums import LLMProvider
from orm.llm.models import LLMConfig
# Filter by enum value
statement = select(LLMConfig).where(LLMConfig.provider == LLMProvider.ANTHROPIC)
# Compare safely
if config.provider == LLMProvider.OPENAI:
...StrEnum is directly comparable to plain strings, so existing code that uses string literals continues to work.
PostgreSQL-Specific Types
JSONB
# Django
metadata = models.JSONField(default=dict)
# Generated SQLModel
metadata: dict = Field(
default_factory=dict,
sa_column=Column(JSONB)
)ARRAY
# Django
tags = ArrayField(models.CharField(max_length=50))
# Generated SQLModel
tags: list[str] = Field(
default_factory=list,
sa_column=Column(ARRAY(String(50)))
)UUID
# Django
uuid = models.UUIDField(default=uuid.uuid4)
# Generated SQLModel
uuid: UUID = Field(
default_factory=uuid4,
sa_column=Column(UUID(as_uuid=True))
)TAGS: models, schemas, crud, repository, postgresql DEPENDS_ON: [overview, configuration]