cardosofelipe/pragma-stack
Template
1
0
Fork 1
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

refactor(docs): update architecture to reflect repository migration

Browse Source 
- Rename CRUD layer to Repository layer throughout architecture documentation.
- Update dependency injection examples to use repository classes.
- Add async SQLAlchemy pattern for Repository methods (`select()` and transactions).
- Replace CRUD references in FEATURE_EXAMPLE.md with Repository-focused implementation details.
- Highlight repository class responsibilities and remove outdated CRUD patterns.
This commit is contained in:
Felipe Cardoso
2026-03-01 11:13:51 +01:00
parent 80d2dc0cb2
commit 68275b1dd3
4 changed files with 349 additions and 773 deletions
Download Patch File Download Diff File Expand all files Collapse all files

178
backend/docs/ARCHITECTURE.md
View File

@@ -117,7 +117,8 @@ backend/
│ ├── api/ # API layer
│ │ ├── dependencies/ # Dependency injection
│ │ │ ├── auth.py # Authentication dependencies
│ │ │ └── permissions.py # Authorization dependencies
│ │ │ ├── permissions.py # Authorization dependencies
│ │ │ └── services.py # Service singleton injection
│ │ ├── routes/ # API endpoints
│ │ │ ├── auth.py # Authentication routes
│ │ │ ├── users.py # User management routes
@@ -131,13 +132,14 @@ backend/
│ │ ├── config.py # Application configuration
│ │ ├── database.py # Database connection
│ │ ├── exceptions.py # Custom exception classes
│ │ ├── repository_exceptions.py # Repository-level exception hierarchy
│ │ └── middleware.py # Custom middleware
│ │
│ ├── crud/ # Database operations
│ │ ├── base.py # Generic CRUD base class
│ │ ├── user.py # User CRUD operations
│ │ ├── session.py # Session CRUD operations
│ │ └── organization.py # Organization CRUD
│ ├── repositories/ # Data access layer
│ │ ├── base.py # Generic repository base class
│ │ ├── user.py # User repository
│ │ ├── session.py # Session repository
│ │ └── organization.py # Organization repository
│ │
│ ├── models/ # SQLAlchemy models
│ │ ├── base.py # Base model with mixins
@@ -153,8 +155,11 @@ backend/
│ │ ├── sessions.py # Session schemas
│ │ └── organizations.py # Organization schemas
│ │
│ ├── services/ # Business logic
│ ├── services/ # Business logic layer
│ │ ├── auth_service.py # Authentication service
│ │ ├── user_service.py # User management service
│ │ ├── session_service.py # Session management service
│ │ ├── organization_service.py # Organization service
│ │ ├── email_service.py # Email service
│ │ └── session_cleanup.py # Background cleanup
│ │
@@ -168,9 +173,9 @@ backend/
│
├── tests/ # Test suite
│ ├── api/ # Integration tests
│ ├── crud/ # CRUD tests
│ ├── repositories/ # Repository unit tests
│ ├── services/ # Service unit tests
│ ├── models/ # Model tests
│ ├── services/ # Service tests
│ └── conftest.py # Test configuration
│
├── docs/ # Documentation
@@ -214,11 +219,11 @@ The application follows a strict 5-layer architecture:
└──────────────────────────┬──────────────────────────────────┘
│ calls
┌──────────────────────────▼──────────────────────────────────┐
│ CRUD Layer (crud/) │
│ Repository Layer (repositories/) │
│ - Database operations │
│ - Query building │
│ - Transaction management │
│ - Error handling │
│ - Custom repository exceptions │
│ - No business logic │
└──────────────────────────┬──────────────────────────────────┘
│ uses
┌──────────────────────────▼──────────────────────────────────┐
@@ -262,7 +267,7 @@ async def get_current_user_info(
**Rules**:
- Should NOT contain business logic
- Should NOT directly perform database operations (use CRUD or services)
- Should NOT directly call repositories (use services injected via `dependencies/services.py`)
- Must validate all input via Pydantic schemas
- Must specify response models
- Should apply appropriate rate limits
@@ -279,9 +284,9 @@ async def get_current_user_info(
**Example**:
```python
def get_current_user(
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db)
db: AsyncSession = Depends(get_db)
) -> User:
"""
Extract and validate user from JWT token.
@@ -295,7 +300,7 @@ def get_current_user(
except Exception:
raise AuthenticationError("Invalid authentication credentials")
user = user_crud.get(db, id=user_id)
user = await user_repo.get(db, id=user_id)
if not user:
raise AuthenticationError("User not found")
@@ -313,7 +318,7 @@ def get_current_user(
**Responsibility**: Implement complex business logic
**Key Functions**:
- Orchestrate multiple CRUD operations
- Orchestrate multiple repository operations
- Implement business rules
- Handle external service integration
- Coordinate transactions
@@ -323,9 +328,9 @@ def get_current_user(
class AuthService:
"""Authentication service with business logic."""
def login(
async def login(
self,
db: Session,
db: AsyncSession,
email: str,
password: str,
request: Request
@@ -339,8 +344,8 @@ class AuthService:
3. Generate tokens
4. Return tokens and user info
"""
# Validate credentials
user = user_crud.get_by_email(db, email=email)
# Validate credentials via repository
user = await user_repo.get_by_email(db, email=email)
if not user or not verify_password(password, user.hashed_password):
raise AuthenticationError("Invalid credentials")
@@ -350,11 +355,10 @@ class AuthService:
# Extract device info
device_info = extract_device_info(request)
# Create session
session = session_crud.create_session(
# Create session via repository
session = await session_repo.create(
db,
user_id=user.id,
device_info=device_info
obj_in=SessionCreate(user_id=user.id, **device_info)
)
# Generate tokens
@@ -373,75 +377,60 @@ class AuthService:
**Rules**:
- Contains business logic, not just data operations
- Can call multiple CRUD operations
- Can call multiple repository operations
- Should handle complex workflows
- Must maintain data consistency
- Should use transactions when needed
#### 4. CRUD Layer (`app/crud/`)
#### 4. Repository Layer (`app/repositories/`)
**Responsibility**: Database operations and queries
**Responsibility**: Database operations and queries — no business logic
**Key Functions**:
- Create, read, update, delete operations
- Build database queries
- Handle database errors
- Raise custom repository exceptions (`DuplicateEntryError`, `IntegrityConstraintError`)
- Manage soft deletes
- Implement pagination and filtering
**Example**:
```python
class CRUDSession(CRUDBase[UserSession, SessionCreate, SessionUpdate]):
"""CRUD operations for user sessions."""
class SessionRepository(RepositoryBase[UserSession, SessionCreate, SessionUpdate]):
"""Repository for user sessions — database operations only."""
def get_by_jti(self, db: Session, jti: UUID) -> Optional[UserSession]:
async def get_by_jti(self, db: AsyncSession, *, jti: str) -> UserSession | None:
"""Get session by refresh token JTI."""
try:
return (
db.query(UserSession)
.filter(UserSession.refresh_token_jti == jti)
.first()
)
except Exception as e:
logger.error(f"Error getting session by JTI: {str(e)}")
return None
result = await db.execute(
select(UserSession).where(UserSession.refresh_token_jti == jti)
)
return result.scalar_one_or_none()
def get_active_by_jti(
self,
db: Session,
jti: UUID
) -> Optional[UserSession]:
"""Get active session by refresh token JTI."""
session = self.get_by_jti(db, jti=jti)
if session and session.is_active and not session.is_expired:
return session
return None
def deactivate(self, db: Session, session_id: UUID) -> bool:
async def deactivate(self, db: AsyncSession, *, session_id: UUID) -> bool:
"""Deactivate a session (logout)."""
try:
session = self.get(db, id=session_id)
session = await self.get(db, id=session_id)
if not session:
return False
session.is_active = False
db.commit()
await db.commit()
logger.info(f"Session {session_id} deactivated")
return True
except Exception as e:
db.rollback()
await db.rollback()
logger.error(f"Error deactivating session: {str(e)}")
return False
```
**Rules**:
- Should NOT contain business logic
- Must handle database exceptions
- Must use parameterized queries (SQLAlchemy does this)
- Must raise custom repository exceptions (not raw `ValueError`/`IntegrityError`)
- Must use async SQLAlchemy 2.0 `select()` API (never `db.query()`)
- Should log all database errors
- Must rollback on errors
- Should use soft deletes when possible
- **Never imported directly by routes** — always called through services
#### 5. Data Layer (`app/models/` + `app/schemas/`)
@@ -546,51 +535,23 @@ SessionLocal = sessionmaker(
#### Dependency Injection Pattern
```python
def get_db() -> Generator[Session, None, None]:
async def get_db() -> AsyncGenerator[AsyncSession, None]:
"""
Database session dependency for FastAPI routes.
Async database session dependency for FastAPI routes.
Automatically commits on success, rolls back on error.
The session is passed to service methods; commit/rollback is
managed inside service or repository methods.
"""
db = SessionLocal()
try:
async with AsyncSessionLocal() as db:
yield db
finally:
db.close()
# Usage in routes
# Usage in routes — always through a service, never direct repository
@router.get("/users")
def list_users(db: Session = Depends(get_db)):
return user_crud.get_multi(db)
```
#### Context Manager Pattern
```python
@contextmanager
def transaction_scope() -> Generator[Session, None, None]:
"""
Context manager for database transactions.
Use for complex operations requiring multiple steps.
Automatically commits on success, rolls back on error.
"""
db = SessionLocal()
try:
yield db
db.commit()
except Exception:
db.rollback()
raise
finally:
db.close()
# Usage in services
def complex_operation():
with transaction_scope() as db:
user = user_crud.create(db, obj_in=user_data)
session = session_crud.create(db, session_data)
return user, session
async def list_users(
user_service: UserService = Depends(get_user_service),
db: AsyncSession = Depends(get_db),
):
return await user_service.get_users(db)
```
### Model Mixins
@@ -782,22 +743,15 @@ def get_profile(
```python
@router.delete("/sessions/{session_id}")
def revoke_session(
async def revoke_session(
session_id: UUID,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
session_service: SessionService = Depends(get_session_service),
db: AsyncSession = Depends(get_db),
):
"""Users can only revoke their own sessions."""
session = session_crud.get(db, id=session_id)
if not session:
raise NotFoundError("Session not found")
# Check ownership
if session.user_id != current_user.id:
raise AuthorizationError("You can only revoke your own sessions")
session_crud.deactivate(db, session_id=session_id)
# SessionService verifies ownership and raises NotFoundError / AuthorizationError
await session_service.revoke_session(db, session_id=session_id, user_id=current_user.id)
return MessageResponse(success=True, message="Session revoked")
```
@@ -1092,8 +1046,8 @@ async def cleanup_expired_sessions():
Runs daily at 2 AM. Removes sessions expired for more than 30 days.
"""
try:
with transaction_scope() as db:
count = session_crud.cleanup_expired(db, keep_days=30)
async with AsyncSessionLocal() as db:
count = await session_repo.cleanup_expired(db, keep_days=30)
logger.info(f"Cleaned up {count} expired sessions")
except Exception as e:
logger.error(f"Error cleaning up sessions: {str(e)}", exc_info=True)
@@ -1110,7 +1064,7 @@ async def cleanup_expired_sessions():
│Integration │ ← API endpoint tests
│ Tests │
├─────────────┤
│ Unit │ ← CRUD, services, utilities
│ Unit │ ← repositories, services, utilities
│ Tests │
└─────────────┘
```

90
backend/docs/CODING_STANDARDS.md
View File

@@ -75,15 +75,14 @@ def create_user(db: Session, user_in: UserCreate) -> User:
### 4. Code Formatting
Use automated formatters:
- **Black**: Code formatting
- **isort**: Import sorting
- **flake8**: Linting
- **Ruff**: Code formatting and linting (replaces Black, isort, flake8)
- **pyright**: Static type checking
Run before committing:
Run before committing (or use `make validate`):
```bash
black app tests
isort app tests
flake8 app tests
uv run ruff format app tests
uv run ruff check app tests
uv run pyright app
```
## Code Organization
@@ -94,19 +93,17 @@ Follow the 5-layer architecture strictly:
```
API Layer (routes/)
↓ calls
Dependencies (dependencies/)
↓ injects
↓ calls (via service injected from dependencies/services.py)
Service Layer (services/)
↓ calls
CRUD Layer (crud/)
Repository Layer (repositories/)
↓ uses
Models & Schemas (models/, schemas/)
```
**Rules:**
- Routes should NOT directly call CRUD operations (use services when business logic is needed)
- CRUD operations should NOT contain business logic
- Routes must NEVER import repositories directly — always use a service
- Services call repositories; repositories contain only database operations
- Models should NOT import from higher layers
- Each layer should only depend on the layer directly below it
@@ -125,7 +122,7 @@ from sqlalchemy.orm import Session
# 3. Local application imports
from app.api.dependencies.auth import get_current_user
from app.crud import user_crud
from app.api.dependencies.services import get_user_service
from app.models.user import User
from app.schemas.users import UserResponse, UserCreate
```
@@ -442,19 +439,19 @@ backend/app/alembic/versions/
4. **Testability**: Easy to mock and test
5. **Consistent Ordering**: Always order queries for pagination
### Use the Async CRUD Base Class
### Use the Async Repository Base Class
Always inherit from `CRUDBase` for database operations:
Always inherit from `RepositoryBase` for database operations:
```python
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.crud.base import CRUDBase
from app.repositories.base import RepositoryBase
from app.models.user import User
from app.schemas.users import UserCreate, UserUpdate
class CRUDUser(CRUDBase[User, UserCreate, UserUpdate]):
"""CRUD operations for User model."""
class UserRepository(RepositoryBase[User, UserCreate, UserUpdate]):
"""Repository for User model — database operations only."""
async def get_by_email(
self,
@@ -467,7 +464,7 @@ class CRUDUser(CRUDBase[User, UserCreate, UserUpdate]):
)
return result.scalar_one_or_none()
user_crud = CRUDUser(User)
user_repo = UserRepository(User)
```
**Key Points:**
@@ -476,6 +473,7 @@ user_crud = CRUDUser(User)
- Use `await db.execute()` for queries
- Use `.scalar_one_or_none()` instead of `.first()`
- Use `T | None` instead of `Optional[T]`
- Repository instances are used internally by services — never import them in routes
### Modern SQLAlchemy Patterns
@@ -563,7 +561,7 @@ async def create_user(
The database session is automatically managed by FastAPI.
Commit on success, rollback on error.
"""
return await user_crud.create(db, obj_in=user_in)
return await user_service.create_user(db, obj_in=user_in)
```
**Key Points:**
@@ -582,12 +580,11 @@ async def complex_operation(
"""
Perform multiple database operations atomically.
The session automatically commits on success or rolls back on error.
Services call repositories; commit/rollback is handled inside
each repository method.
"""
user = await user_crud.create(db, obj_in=user_data)
session = await session_crud.create(db, obj_in=session_data)
# Commit is handled by the route's dependency
user = await user_repo.create(db, obj_in=user_data)
session = await session_repo.create(db, obj_in=session_data)
return user, session
```
@@ -597,10 +594,10 @@ Prefer soft deletes over hard deletes for audit trails:
```python
# Good - Soft delete (sets deleted_at)
await user_crud.soft_delete(db, id=user_id)
await user_repo.soft_delete(db, id=user_id)
# Acceptable only when required - Hard delete
user_crud.remove(db, id=user_id)
await user_repo.remove(db, id=user_id)
```
### Query Patterns
@@ -740,9 +737,10 @@ Always implement pagination for list endpoints:
from app.schemas.common import PaginationParams, PaginatedResponse
@router.get("/users", response_model=PaginatedResponse[UserResponse])
def list_users(
async def list_users(
pagination: PaginationParams = Depends(),
db: Session = Depends(get_db)
user_service: UserService = Depends(get_user_service),
db: AsyncSession = Depends(get_db),
):
"""
List all users with pagination.
@@ -750,10 +748,8 @@ def list_users(
Default page size: 20
Maximum page size: 100
"""
users, total = user_crud.get_multi_with_total(
db,
skip=pagination.offset,
limit=pagination.limit
users, total = await user_service.get_users(
db, skip=pagination.offset, limit=pagination.limit
)
return PaginatedResponse(data=users, pagination=pagination.create_meta(total))
```
@@ -816,19 +812,17 @@ def admin_route(
pass
# Check ownership
def delete_resource(
async def delete_resource(
resource_id: UUID,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db)
resource_service: ResourceService = Depends(get_resource_service),
db: AsyncSession = Depends(get_db),
):
resource = resource_crud.get(db, id=resource_id)
if not resource:
raise NotFoundError("Resource not found")
if resource.user_id != current_user.id and not current_user.is_superuser:
raise AuthorizationError("You can only delete your own resources")
resource_crud.remove(db, id=resource_id)
# Service handles ownership check and raises appropriate errors
await resource_service.delete_resource(
db, resource_id=resource_id, user_id=current_user.id,
is_superuser=current_user.is_superuser,
)
```
### Input Validation
@@ -862,9 +856,9 @@ tests/
├── api/ # Integration tests
│ ├── test_users.py
│ └── test_auth.py
├── crud/ # Unit tests for CRUD
├── models/ # Model tests
└── services/ # Service tests
├── repositories/ # Unit tests for repositories
├── services/ # Unit tests for services
└── models/ # Model tests
```
### Async Testing with pytest-asyncio
@@ -927,7 +921,7 @@ async def test_user(db_session: AsyncSession) -> User:
@pytest.mark.asyncio
async def test_get_user(db_session: AsyncSession, test_user: User):
"""Test retrieving a user by ID."""
user = await user_crud.get(db_session, id=test_user.id)
user = await user_repo.get(db_session, id=test_user.id)
assert user is not None
assert user.email == test_user.email
```

62
backend/docs/COMMON_PITFALLS.md
View File

@@ -616,7 +616,43 @@ def create_user(
return user
```
**Rule**: Add type hints to ALL functions. Use `mypy` to enforce type checking.
**Rule**: Add type hints to ALL functions. Use `pyright` to enforce type checking (`make type-check`).
---
---
### ❌ PITFALL #19: Importing Repositories Directly in Routes
**Issue**: Routes should never call repositories directly. The layered architecture requires all business operations to go through the service layer.
```python
# ❌ WRONG - Route bypasses service layer
from app.repositories.session import session_repo
@router.get("/sessions/me")
async def list_sessions(
current_user: User = Depends(get_current_active_user),
db: AsyncSession = Depends(get_db),
):
return await session_repo.get_user_sessions(db, user_id=current_user.id)
```
```python
# ✅ CORRECT - Route calls service injected via dependency
from app.api.dependencies.services import get_session_service
from app.services.session_service import SessionService
@router.get("/sessions/me")
async def list_sessions(
current_user: User = Depends(get_current_active_user),
session_service: SessionService = Depends(get_session_service),
db: AsyncSession = Depends(get_db),
):
return await session_service.get_user_sessions(db, user_id=current_user.id)
```
**Rule**: Routes import from `app.api.dependencies.services`, never from `app.repositories.*`. Services are the only callers of repositories.
---
@@ -649,6 +685,11 @@ Use this checklist to catch issues before code review:
- [ ] Resource ownership verification
- [ ] CORS configured (no wildcards in production)
### Architecture
- [ ] Routes never import repositories directly (only services)
- [ ] Services call repositories; repositories call database only
- [ ] New service registered in `app/api/dependencies/services.py`
### Python
- [ ] Use `==` not `is` for value comparison
- [ ] No mutable default arguments
@@ -661,21 +702,18 @@ Use this checklist to catch issues before code review:
### Pre-commit Checks
Add these to your development workflow:
Add these to your development workflow (or use `make validate`):
```bash
# Format code
black app tests
isort app tests
# Format + lint (Ruff replaces Black, isort, flake8)
uv run ruff format app tests
uv run ruff check app tests
# Type checking
mypy app --strict
# Linting
flake8 app tests
uv run pyright app
# Run tests
pytest --cov=app --cov-report=term-missing
IS_TEST=True uv run pytest --cov=app --cov-report=term-missing
# Check coverage (should be 80%+)
coverage report --fail-under=80
@@ -693,6 +731,6 @@ Add new entries when:
---
**Last Updated**: 2025-10-31
**Issues Cataloged**: 18 common pitfalls
**Last Updated**: 2026-02-28
**Issues Cataloged**: 19 common pitfalls
**Remember**: This document exists because these issues HAVE occurred. Don't skip it.

792
backend/docs/FEATURE_EXAMPLE.md
View File

File diff suppressed because it is too large Load Diff