Proyecto del Módulo: blog en SQLAlchemy + Repository pattern

Descripción

Es hora de consolidar todo: vas a estructurar el blog como una aplicación Python real con modelos, repositorios (capa de acceso a datos), y tests. El Repository pattern encapsula las queries de cada entidad detrás de funciones con nombres semánticos — más legible, testeable, y reutilizable.

Al final del módulo tendrás app/ con todo el código Python necesario para operar el blog desde Python, listo para integrarse con FastAPI en el módulo 6.

---

Estructura final del proyecto

postgresql-sqlalchemy-blog/
├── README.md
├── docker-compose.yml
├── pyproject.toml
├── .env
├── app/
│   ├── __init__.py
│   ├── database.py          ← Engine, Session, Base
│   ├── models.py            ← User, Post, Comment, etc.
│   └── repositories/
│       ├── __init__.py
│       ├── users.py
│       ├── posts.py
│       ├── comments.py
│       └── tags.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_users.py
│   └── test_posts.py
└── db/
    └── ... (scripts SQL, sin cambios)

---

Repository Pattern: ¿qué es?

Un repository es una capa que encapsula el acceso a datos de una entidad. Cada operación común (crear, buscar por ID, listar publicados, etc.) es una función con nombre semántico.

Antes (queries inline):

# En tu API endpoint
@app.get("/users/{username}")
async def get_user(username: str, db: Session):
    user = db.execute(
        select(User)
        .options(selectinload(User.posts))
        .where(User.username == username, User.deleted_at.is_(None))
    ).scalar_one_or_none()
    return user

Después (con repository):

# En tu API endpoint
@app.get("/users/{username}")
async def get_user(username: str, db: Session):
    return users_repo.get_by_username_with_posts(db, username)

Ventajas:

• Reutilizable — la misma función desde cualquier endpoint
• Testeable — mockeas el repository entero
• Cambios en una sola capa — agregar un filtro deleted_at se hace una vez
• Separación de concerns — endpoints son delgados, lógica de DB en repos

---

Crear app/repositories/users.py

# app/repositories/users.py
"""Repository de usuarios."""

from __future__ import annotations

from uuid import UUID

from sqlalchemy import select
from sqlalchemy.orm import Session, selectinload

from app.models import Comment, Post, User


def create(
    session: Session,
    *,
    email: str,
    username: str,
    password_hash: str,
    full_name: str | None = None,
    bio: str | None = None,
) -> User:
    """Crea un nuevo usuario."""
    user = User(
        email=email,
        username=username,
        password_hash=password_hash,
        full_name=full_name,
        bio=bio,
    )
    session.add(user)
    session.flush()  # para que tenga id antes de retornar
    return user


def get_by_id(session: Session, user_id: UUID) -> User | None:
    """Obtiene un usuario por ID. None si no existe o está soft-deleted."""
    return session.execute(
        select(User).where(
            User.id == user_id,
            User.deleted_at.is_(None),
        )
    ).scalar_one_or_none()


def get_by_username(session: Session, username: str) -> User | None:
    """Obtiene un usuario por username."""
    return session.execute(
        select(User).where(
            User.username == username,
            User.deleted_at.is_(None),
        )
    ).scalar_one_or_none()


def get_by_email(session: Session, email: str) -> User | None:
    """Login: busca por email (case-insensitive). Usa idx_users_email_lower."""
    from sqlalchemy import func
    
    return session.execute(
        select(User).where(
            func.lower(User.email) == email.lower(),
            User.deleted_at.is_(None),
        )
    ).scalar_one_or_none()


def get_by_username_with_posts(session: Session, username: str) -> User | None:
    """Usuario + sus posts publicados (eager)."""
    return session.execute(
        select(User)
        .options(
            selectinload(User.posts.and_(Post.published == True))
        )
        .where(
            User.username == username,
            User.deleted_at.is_(None),
        )
    ).scalar_one_or_none()


def list_active(
    session: Session,
    *,
    limit: int = 50,
    offset: int = 0,
) -> list[User]:
    """Lista de usuarios activos paginada."""
    return list(session.execute(
        select(User)
        .where(User.deleted_at.is_(None), User.is_active == True)
        .order_by(User.created_at.desc())
        .limit(limit)
        .offset(offset)
    ).scalars().all())


def soft_delete(session: Session, user_id: UUID) -> bool:
    """
    Soft delete del usuario + sus posts (despublica) + sus comments (marcar deleted).
    Retorna True si se aplicó, False si el usuario no existe.
    """
    from sqlalchemy import update
    from sqlalchemy.sql import func as sql_func
    
    user = session.get(User, user_id)
    if not user or user.deleted_at is not None:
        return False
    
    now = sql_func.now()
    user.deleted_at = now
    user.updated_at = now
    
    # Despublicar sus posts
    session.execute(
        update(Post)
        .where(Post.author_id == user_id)
        .values(published=False, updated_at=now)
    )
    
    # Marcar sus comments como deleted
    session.execute(
        update(Comment)
        .where(Comment.author_id == user_id)
        .values(is_deleted=True, updated_at=now)
    )
    
    return True

---

Crear app/repositories/posts.py

# app/repositories/posts.py
"""Repository de posts."""

from __future__ import annotations

from datetime import datetime, timezone
from uuid import UUID

from sqlalchemy import func, select
from sqlalchemy.orm import Session, joinedload, selectinload

from app.models import Category, Post, Tag, User, post_tags_table


def create(
    session: Session,
    *,
    author_id: UUID,
    title: str,
    slug: str,
    content: str,
    excerpt: str | None = None,
    category_id: UUID | None = None,
    tag_slugs: list[str] | None = None,
    publish: bool = False,
) -> Post:
    """
    Crea un post. Si publish=True, marca published_at = NOW().
    Si tag_slugs se provee, asocia los tags existentes.
    """
    post = Post(
        author_id=author_id,
        category_id=category_id,
        title=title,
        slug=slug,
        content=content,
        excerpt=excerpt,
        published=publish,
        published_at=datetime.now(timezone.utc) if publish else None,
    )
    
    if tag_slugs:
        tags = session.execute(
            select(Tag).where(Tag.slug.in_(tag_slugs))
        ).scalars().all()
        post.tags = list(tags)
    
    session.add(post)
    session.flush()
    return post


def get_by_slug(session: Session, slug: str) -> Post | None:
    """Obtiene un post por slug, con autor, category, tags y comments cargados."""
    return session.execute(
        select(Post)
        .options(
            joinedload(Post.author),
            joinedload(Post.category),
            selectinload(Post.tags),
            selectinload(Post.comments).joinedload(Comment.author),
        )
        .where(Post.slug == slug)
    ).scalar_one_or_none()


def list_published(
    session: Session,
    *,
    limit: int = 10,
    offset: int = 0,
    category_slug: str | None = None,
    tag_slug: str | None = None,
) -> list[Post]:
    """
    Feed paginado de posts publicados, opcionalmente filtrado por category o tag.
    Optimizado: máximo 4 queries para el feed.
    """
    stmt = (
        select(Post)
        .options(
            joinedload(Post.author),
            joinedload(Post.category),
            selectinload(Post.tags),
        )
        .where(Post.published == True)
        .order_by(Post.published_at.desc())
        .limit(limit)
        .offset(offset)
    )
    
    if category_slug:
        stmt = stmt.join(Post.category).where(Category.slug == category_slug)
    
    if tag_slug:
        stmt = stmt.join(Post.tags).where(Tag.slug == tag_slug)
    
    return list(session.execute(stmt).scalars().unique().all())


def search(session: Session, query: str, *, limit: int = 20) -> list[Post]:
    """Búsqueda case-insensitive en título y content."""
    pattern = f"%{query}%"
    return list(session.execute(
        select(Post)
        .options(joinedload(Post.author))
        .where(
            Post.published == True,
            Post.title.ilike(pattern) | Post.content.ilike(pattern),
        )
        .order_by(Post.published_at.desc())
        .limit(limit)
    ).scalars().all())


def list_related(session: Session, post_id: UUID, *, limit: int = 5) -> list[Post]:
    """Posts con tags compartidos al post dado."""
    target_tag_ids = (
        select(post_tags_table.c.tag_id)
        .where(post_tags_table.c.post_id == post_id)
    )
    
    stmt = (
        select(Post, func.count(post_tags_table.c.tag_id).label("shared"))
        .join(post_tags_table, Post.id == post_tags_table.c.post_id)
        .where(
            post_tags_table.c.tag_id.in_(target_tag_ids),
            Post.id != post_id,
            Post.published == True,
        )
        .group_by(Post.id)
        .order_by(func.count(post_tags_table.c.tag_id).desc())
        .limit(limit)
    )
    
    return [post for post, _ in session.execute(stmt).all()]


def update(
    session: Session,
    post_id: UUID,
    *,
    title: str | None = None,
    content: str | None = None,
    excerpt: str | None = None,
    category_id: UUID | None = None,
    tag_slugs: list[str] | None = None,
    publish: bool | None = None,
) -> Post | None:
    """Actualiza un post. Solo modifica los campos pasados."""
    post = session.get(Post, post_id)
    if not post:
        return None
    
    if title is not None:
        post.title = title
    if content is not None:
        post.content = content
    if excerpt is not None:
        post.excerpt = excerpt
    if category_id is not None:
        post.category_id = category_id
    
    if tag_slugs is not None:
        # Reemplazar todos los tags
        new_tags = session.execute(select(Tag).where(Tag.slug.in_(tag_slugs))).scalars().all()
        post.tags = list(new_tags)
    
    if publish is not None:
        post.published = publish
        if publish and post.published_at is None:
            post.published_at = datetime.now(timezone.utc)
    
    return post


def delete(session: Session, post_id: UUID) -> bool:
    """Borra un post (físico, con cascade a comments y post_tags)."""
    post = session.get(Post, post_id)
    if not post:
        return False
    session.delete(post)
    return True


def stats_by_author(session: Session) -> list[tuple[str, int, int]]:
    """Retorna lista de (username, num_posts_publicados, num_comments_recibidos)."""
    from app.models import Comment
    
    stmt = (
        select(
            User.username,
            func.count(Post.id.distinct()).label("posts"),
            func.count(Comment.id.distinct()).label("comments_received"),
        )
        .join(User.posts)
        .outerjoin(Comment, Comment.post_id == Post.id)
        .where(Post.published == True)
        .group_by(User.id, User.username)
        .order_by(func.count(Post.id.distinct()).desc())
    )
    
    return [(row.username, row.posts, row.comments_received) for row in session.execute(stmt)]

Nota: importa Comment adentro de la función para evitar circular imports en algunos casos.

---

app/repositories/__init__.py: re-exports

# app/repositories/__init__.py
"""Capa de acceso a datos del blog."""

from app.repositories import users, posts, comments, tags

__all__ = ["users", "posts", "comments", "tags"]

Uso desde otras partes:

from app import repositories

with db_session() as session:
    user = repositories.users.get_by_username(session, "maria")
    feed = repositories.posts.list_published(session, limit=10)

---

app/repositories/comments.py y tags.py

(Implementaciones similares siguiendo el mismo patrón. Incluyo solo los más relevantes.)

# app/repositories/comments.py
from __future__ import annotations

from uuid import UUID

from sqlalchemy import select
from sqlalchemy.orm import Session, joinedload

from app.models import Comment, Post


def create(
    session: Session,
    *,
    post_id: UUID,
    author_id: UUID,
    content: str,
    parent_comment_id: UUID | None = None,
) -> Comment:
    """Crea un comment, opcionalmente como reply de otro."""
    comment = Comment(
        post_id=post_id,
        author_id=author_id,
        content=content,
        parent_comment_id=parent_comment_id,
    )
    session.add(comment)
    session.flush()
    return comment


def list_by_post(session: Session, post_id: UUID) -> list[Comment]:
    """Comments de un post (con autor cargado), excluyendo soft-deleted."""
    return list(session.execute(
        select(Comment)
        .options(joinedload(Comment.author))
        .where(
            Comment.post_id == post_id,
            Comment.is_deleted == False,
        )
        .order_by(Comment.created_at)
    ).scalars().all())


def soft_delete(session: Session, comment_id: UUID) -> bool:
    """Marca un comment como is_deleted=True."""
    comment = session.get(Comment, comment_id)
    if not comment:
        return False
    comment.is_deleted = True
    return True

# app/repositories/tags.py
from __future__ import annotations

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.models import Tag, post_tags_table


def list_all_with_usage(session: Session) -> list[tuple[Tag, int]]:
    """Lista de (tag, num_uses) ordenado por uso descendente."""
    stmt = (
        select(Tag, func.count(post_tags_table.c.post_id).label("usage"))
        .outerjoin(post_tags_table, Tag.id == post_tags_table.c.tag_id)
        .group_by(Tag.id)
        .order_by(func.count(post_tags_table.c.post_id).desc())
    )
    return [(row.Tag, row.usage) for row in session.execute(stmt)]


def get_by_slug(session: Session, slug: str) -> Tag | None:
    return session.execute(select(Tag).where(Tag.slug == slug)).scalar_one_or_none()

---

Tests con pytest

Crea tests/conftest.py con fixtures reutilizables:

# tests/conftest.py
"""Fixtures de pytest para tests del blog."""

from collections.abc import Generator

import pytest
from sqlalchemy.orm import Session

from app.database import SessionLocal


@pytest.fixture
def db() -> Generator[Session, None, None]:
    """Session aislada en transacción que rollbackea al final."""
    session = SessionLocal()
    try:
        yield session
    finally:
        session.rollback()
        session.close()

Test de un repository:

# tests/test_users.py
"""Tests de users repository."""

from app import repositories
from app.repositories import users as users_repo


def test_get_by_username_existing(db):
    user = users_repo.get_by_username(db, "maria")
    assert user is not None
    assert user.username == "maria"


def test_get_by_username_nonexistent(db):
    user = users_repo.get_by_username(db, "no_existe")
    assert user is None


def test_create_user(db):
    user = users_repo.create(
        db,
        email="test_create@blog.local",
        username="test_create",
        password_hash="$2b$12$x",
    )
    assert user.id is not None
    assert user.username == "test_create"
    # rollback en fixture: la fila no persiste


def test_soft_delete_user(db):
    user = users_repo.get_by_username(db, "luis")
    assert user is not None
    
    success = users_repo.soft_delete(db, user.id)
    assert success
    
    db.refresh(user)
    assert user.deleted_at is not None

Ejecuta:

pytest tests/ -v

---

Verificación final del módulo

✅ Criterio 1: Modelos completos

python -c "from app.models import User, Post, Comment, Category, Tag; print('OK')"

✅ Criterio 2: Smoke test funciona

python -m app

Imprime versión PostgreSQL y modelos registrados.

✅ Criterio 3: Repositories funcionan

from app.database import db_session
from app import repositories

with db_session() as session:
    feed = repositories.posts.list_published(session, limit=5)
    for p in feed:
        print(f"{p.title} — {p.author.username}")

Debe imprimir 5 posts con autores sin queries N+1.

✅ Criterio 4: Tests pasan

pytest tests/ -v

✅ Criterio 5: mypy limpio

mypy app/

Sin errores.

✅ Criterio 6: No N+1 en feed

Activa echo=True, ejecuta posts.list_published(session), verifica que ves máximo 4 queries (post + user + category + tags).

✅ Criterio 7: Composable

# Combinar repos en una operación de mayor nivel
with db_session() as session:
    post = repositories.posts.create(
        session,
        author_id=repositories.users.get_by_username(session, "maria").id,
        title="Test final M4",
        slug="test-final-m4",
        content="C",
        tag_slugs=["postgresql", "tutorial"],
        publish=True,
    )
    print(f"Creado: {post.title} con tags {[t.slug for t in post.tags]}")

---

Si todo pasa: estás listo para el Módulo 5

Has completado el Módulo 4. Tienes:

• ✅ Engine + Session + Base configurados (app/database.py)
• ✅ 5 modelos con relaciones (app/models.py)
• ✅ 4 repositorios con funciones semánticas (app/repositories/)
• ✅ Tests con pytest
• ✅ Type hints completos (mypy passes)
• ✅ Eager loading aplicado para evitar N+1
• ✅ Patrón Repository implementado

En el Módulo 5 (Alembic Migrations) vas a:

• Configurar Alembic apuntando a Base.metadata
• Generar migrations automáticas con --autogenerate
• Aplicar con upgrade head, revertir con downgrade
• Manejar migrations en equipo (branching, merging)
• Data migrations vs schema migrations

---

Resumen del módulo

1. Cápsula 01: contexto y mapa de SQLAlchemy 2.0
2. Cápsula 02: Engine, Session, DeclarativeBase
3. Cápsula 03: Modelos básicos con Mapped[] + mapped_column()
4. Cápsula 04: Relaciones 1:N, N:M, self-reference
5. Cápsula 05: CRUD con Session (Unit of Work, identity map)
6. Cápsula 06: Queries con select() (filter, join, aggregations, CTEs)
7. Cápsula 07: Eager vs lazy loading, N+1 problem
8. Cápsula 08: Repository pattern + tests

---

Recursos Adicionales

1. "Repository Pattern" — Martin Fowler — Patrón original
2. FastAPI + SQLAlchemy 2.0 example — Estructura típica
3. "Cosmic Python — Architecture Patterns with Python" — Libro gratuito sobre Repository, Unit of Work, etc.
4. pytest-postgresql — Tests con DB real
5. "SQLAlchemy 2.0 in Production" — Best practices

---

🎉 Has completado el Módulo 4: SQLAlchemy ORM.

Siguiente módulo: Módulo 5 — Alembic Migrations.