Documents: - Architecture principles (stateless, DI, repository protocol) - Services overview table - Key patterns (config from request, UNSET sentinel, repository injection) - Service details with usage examples - Testing approach with examples Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
6.2 KiB
6.2 KiB
Services Layer
Business logic layer between API endpoints and data access.
Architecture Principles
- Stateless: Config comes from request, not server state
- DI Pattern: Dependencies injected via constructor
- Repository Protocol: Data access through abstract protocols
- Offline Fork Ready: No hard dependencies on Postgres/Redis
Services Overview
| Service | Responsibility | Dependencies |
|---|---|---|
CardService |
Card definition lookup (singleton) | JSON files |
CollectionService |
User card ownership CRUD | CollectionRepository, CardService |
DeckService |
Deck CRUD + validation | DeckRepository, CollectionRepository, CardService |
DeckValidator |
Pure deck validation functions | CardService (for lookups) |
UserService |
User profile management | Direct DB access |
GameStateManager |
Game state (Redis + Postgres) | Redis, AsyncSession |
JWTService |
Token creation/verification | Settings |
TokenStore |
Refresh token storage | Redis |
Key Patterns
Config from Request
The backend is stateless. Rules come from the request via config objects like DeckConfig:
# Correct - config from caller (frontend provides rules)
await deck_service.create_deck(
user_id=user.id,
cards=request.cards,
deck_config=request.deck_config, # From request body
max_decks=user.max_decks,
)
# Wrong - baked-in config
class DeckService:
def __init__(self, config: DeckConfig): # Don't do this
self._config = config
This enables different game modes (campaign, freeplay, custom) without server-side config.
UNSET Sentinel
For nullable fields that can be explicitly cleared, use UNSET to distinguish between "not provided" and "set to null":
from app.repositories.protocols import UNSET
async def update_deck(
self,
deck_id: UUID,
name: str | None = None, # None = don't change
description: str | None = UNSET, # UNSET = keep, None = clear, str = set
) -> DeckEntry:
...
if description is not UNSET:
# User explicitly provided a value (could be None to clear)
record.description = description
API layer usage with Pydantic:
# Check if field was in the request payload
description = request.description if "description" in request.model_fields_set else UNSET
Repository Injection
Services use constructor injection with repository protocols:
class DeckService:
def __init__(
self,
deck_repository: DeckRepository, # Protocol, not implementation
card_service: CardService,
collection_repository: CollectionRepository | None = None,
) -> None:
self._deck_repo = deck_repository
self._card_service = card_service
self._collection_repo = collection_repository
Benefits:
- Testability: Inject mock repositories
- Offline Fork: Swap PostgresRepository for LocalRepository
- Decoupling: Services don't know about SQLAlchemy
Pure Validation Functions
Validation logic is extracted into pure functions for reuse:
# app/services/deck_validator.py
def validate_deck(
cards: dict[str, int],
energy_cards: dict[str, int],
deck_config: DeckConfig,
card_lookup: Callable[[str], CardDefinition | None],
owned_cards: dict[str, int] | None = None,
) -> ValidationResult:
"""Pure function - all inputs from caller."""
...
Service Details
CardService
Singleton that loads card definitions from JSON files at startup.
from app.services.card_service import get_card_service
card_service = get_card_service()
card = card_service.get_card("a1-001-bulbasaur")
all_cards = card_service.get_all_cards()
CollectionService
Manages user card ownership with source tracking.
# Add cards to collection
await collection_service.add_cards(
user_id=user.id,
card_definition_id="a1-001-bulbasaur",
quantity=4,
source=CardSource.BOOSTER,
)
# Get owned cards for deck validation
owned = await collection_service.get_owned_cards_dict(user.id)
DeckService
Deck CRUD with validation and ownership checks.
# Create deck - validates and stores result
deck = await deck_service.create_deck(
user_id=user.id,
name="My Deck",
cards={"a1-001-bulbasaur": 4, ...},
energy_cards={"grass": 14, "colorless": 6},
deck_config=DeckConfig(),
max_decks=user.max_decks,
validate_ownership=True, # Campaign mode
)
# Deck can be saved even if invalid (for work-in-progress)
if not deck.is_valid:
print(deck.validation_errors)
GameStateManager
Hybrid storage: Redis for fast access, Postgres for durability.
manager = GameStateManager(redis, db_session)
# Save to Redis (fast, every action)
await manager.save_to_redis(game_id, game_state)
# Persist to Postgres (at turn boundaries)
await manager.persist_to_db(game_id, game_state)
Testing
| Directory | Purpose | Database |
|---|---|---|
tests/unit/services/ |
Pure unit tests, mocked deps | No |
tests/services/ |
Integration tests | Yes (testcontainers) |
Unit Test Example
def test_validate_deck_requires_basic_pokemon():
"""Test that decks must contain at least one Basic Pokemon."""
result = validate_deck(
cards={"trainer-card": 40}, # No Pokemon
energy_cards={"colorless": 20},
deck_config=DeckConfig(),
card_lookup=mock_lookup,
)
assert not result.is_valid
assert "Basic Pokemon" in str(result.errors)
Integration Test Example
@pytest.mark.asyncio
async def test_create_deck_enforces_limit(db_session, deck_service):
"""Test that free users are limited to 5 decks."""
user = await UserFactory.create(db_session, max_decks=5)
# Create 5 decks
for i in range(5):
await deck_service.create_deck(user_id=user.id, ...)
# 6th should fail
with pytest.raises(DeckLimitExceededError):
await deck_service.create_deck(user_id=user.id, ...)
See Also
app/repositories/protocols.py- Repository interfaces and DTOsapp/repositories/postgres/- PostgreSQL implementationsapp/core/config.py- DeckConfig and RulesConfigCLAUDE.md- Architecture guidelines