⚡ Section 13 — Mid-Level Masterclass

Database Architecture & SQLAlchemy 2.0

Master PostgreSQL SQL queries, SQLAlchemy 2.0 ORM mapping, Alembic schema migrations, connection pooling, and strict layer separation (Model vs Schema vs Repository vs Service).

👁️ Views: 393 Page Views
💡 Data Layer Responsibility

Fat ORM "God Model" vs 4-Layer Responsibility Architecture

Never mix database table mapping with password hashing, HTTP validation, or email sending. Separate responsibilities cleanly.

❌ Fat ORM God-Object (Mixed Concerns)
# ❌ Model does hashing, DB queries & email sending!
class User(Base):
    def register(self):
        self.password = hash_pwd(self.password)
        db.session.add(self)
        send_welcome_email(self.email)
✅ 4-Layer Clean Separation (Decoupled Rules)
# ✅ Model = DB Entity | Schema = DTO | Repo = Queries | Service = Rules
class UserModel(Base): ...      # Entity
class UserCreate(BaseModel): ... # DTO
class UserRepository: ...       # SQL
class UserService: ...          # Logic

🧱 Data Architecture 4-Layer Matrix

1. Model (ORM Entity)

Maps directly to relational SQL table schema, primary keys, and indexes.

2. Schema (DTO)

Validates incoming HTTP payload fields and serializes response JSON.

3. Repository (Data Access)

Encapsulates raw SQL queries, filtering, and database transaction commits.

4. Service (Business Logic)

Orchestrates domain validation rules, password hashing, and external APIs.

💻 Production Blueprint: SQLAlchemy 2.0 & Repository Pattern

Notice how `UserModel` uses SQLAlchemy 2.0 `Mapped` type annotations and `UserRepository` encapsulates SQL execution.

# ─── 1. SQLALCHEMY 2.0 ORM ENTITY (models/user.py) ─────────────────────────
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String

class UserModel(DeclarativeBase):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)

# ─── 2. DATA ACCESS REPOSITORY (repositories/user_repo.py) ────────────────
from sqlalchemy import select

class UserRepository:
    def __init__(self, session): self.session = session

    def get_by_email(self, email: str) -> UserModel | None:
        stmt = select(UserModel).where(UserModel.email == email)
        return self.session.execute(stmt).scalars().first()

⭐ Rate Section 13: Database Architecture & SQLAlchemy 2.0 (1 to 5 Stars)

Your rating is stored directly in the database. Average: 5.0 / 5 (1 rating)

★ ★ ★ ★ ★
← Section 12: FastAPI Arch 🏠 Masterclass Hub Section 14: REST API Design →