Instruction file imported from smartfastlabs/lykke.day (.cursor/rules/infrastructure.mdc). Copyright stays with the author.

Infrastructure Layer Rules

The Infrastructure layer implements the protocols defined in the application layer. It handles all I/O operations including database persistence, external API calls, and data serialization.

Layer Structure

infrastructure/
├── repositories/      # Repository implementations
│   ├── base/          # UserScopedBaseRepository
│   │   ├── repository.py # Base repository implementation
│   │   └── utils.py   # Repository utilities
│   ├── task.py        # TaskRepository
│   ├── day.py         # DayRepository
│   └── ...
├── gateways/          # External service implementations
│   ├── google.py      # GoogleCalendarGateway
│   ├── web_push.py    # WebPushGateway
│   ├── anthropic_llm.py # Anthropic LLM gateway
│   ├── openai_llm.py  # OpenAI LLM gateway
│   └── redis_pubsub/  # Redis pub/sub gateway
├── database/          # SQLAlchemy database layer
│   ├── tables/        # Table definitions
│   ├── transaction.py # Transaction management
│   └── utils.py       # Database utilities
├── unit_of_work.py    # SqlAlchemyUnitOfWork implementation
├── user_scoped/        # Current-user scoped identity access
├── unauthenticated/    # Cross-user identity access for auth flows
├── auth/              # Authentication infrastructure
│   ├── config.py      # Auth configuration
│   └── schemas.py     # Auth schemas
├── data/              # Static data (default task definitions)
└── workers/           # Background worker configuration

Import Rules

Allowed Imports:

• Standard library
• Third-party libraries (SQLAlchemy, Google APIs, etc.)
• lykke.application.* - Application layer protocols
• lykke.domain.* - Domain layer (entities, value objects)
• lykke.core.* - Core layer (config, constants, exceptions, utils)
• Other infrastructure layer modules

Forbidden Imports:

• lykke.presentation.* - Presentation layer

Repositories

Repositories implement the repository protocols defined in the application layer. They handle persistence via SQLAlchemy (ORM table models exposed as Core tables).

Base Repository Classes

Repositories extend UserScopedBaseRepository[Entity, Query] (all repositories are user-scoped):

from lykke.domain.entities import TaskEntity
from lykke.domain.value_objects import TaskQuery
from lykke.infrastructure.database.tables import tasks_tbl
from lykke.infrastructure.repositories.base import UserScopedBaseRepository
from lykke.domain.entities import UserEntity


class TaskRepository(UserScopedBaseRepository[TaskEntity, TaskQuery]):
    Object = TaskEntity  # Entity class
    table = tasks_tbl  # SQLAlchemy table
    QueryClass = TaskQuery  # Query object type
    excluded_row_fields = {"date"}  # DB-only fields to exclude

    def __init__(self, user: UserEntity) -> None:
        super().__init__(user=user)

Key Rules for Repositories

1. Extend base class - Use UserScopedBaseRepository[Entity, Query]
2. Set class variables - Set Object, table, QueryClass
3. User scoping - User-scoped repositories accept user: UserEntity and are always scoped
4. Implement conversion methods - Implement entity_to_row() and optionally row_to_entity()
5. Override build_query() - Override to add custom query logic
6. Handle JSONB fields - Serialize/deserialize JSONB fields using dataclass_to_json_dict()
7. Handle enums - Convert enums to/from string values
8. Handle UUIDs - Convert UUIDs to/from strings for JSONB storage

Entity to Row Conversion

Repositories must implement entity_to_row() to convert entities to database rows:

@staticmethod
def entity_to_row(task: TaskEntity) -> dict[str, Any]:
    """Convert a Task entity to a database row dict."""
    from lykke.core.utils.serialization import dataclass_to_json_dict

    row: dict[str, Any] = {
        "id": task.id,
        "user_id": task.user_id,
        "scheduled_date": task.scheduled_date,
        "name": task.name,
        "status": task.status.value,  # Convert enum to string
        "category": task.category.value,
        "frequency": task.frequency.value,
        "completed_at": task.completed_at,
        "routine_id": task.routine_id,
    }

    # Handle JSONB fields - serialize nested objects
    if task.task_definition:
        row["task_definition"] = dataclass_to_json_dict(task.task_definition)

    if task.schedule:
        row["schedule"] = dataclass_to_json_dict(task.schedule)

    if task.tags:
        row["tags"] = [tag.value for tag in task.tags]

    return row

Row to Entity Conversion

Repositories can override row_to_entity() for custom deserialization:

@classmethod
def row_to_entity(cls, row: dict[str, Any]) -> TaskEntity:
    """Convert a database row dict to a Task entity."""
    from lykke.infrastructure.repositories.base.utils import (
        filter_init_false_fields,
        normalize_list_fields,
    )

    # Filter out excluded fields and init=False fields
    data = {k: v for k, v in row.items() if k not in cls.excluded_row_fields}
    data = normalize_list_fields(data, TaskEntity)
    data = filter_init_false_fields(data, TaskEntity)

    # Convert enum strings to enums
    if "status" in data and isinstance(data["status"], str):
        data["status"] = TaskStatus(data["status"])

    # Convert JSONB fields - deserialize nested objects
    if data.get("task_definition"):
        data["task_definition"] = TaskDefinitionEntity(**data["task_definition"])

    if data.get("time_window"):
        data["time_window"] = TimeWindow(**data["time_window"])

    return TaskEntity(**data)

Query Building

Repositories can override build_query() to add custom query logic:

def build_query(self, query: TaskQuery) -> Select[tuple]:
    """Build a SQLAlchemy Core select statement from a query object."""
    stmt = super().build_query(query)  # Base query with user scoping

    # Add date filtering
    if query.date is not None:
        stmt = stmt.where(self.table.c.scheduled_date == query.date)

    # Add status filtering
    if query.status is not None:
        stmt = stmt.where(self.table.c.status == query.status.value)

    return stmt

Gateways

Gateways implement the gateway protocols defined in the application layer. They handle external API calls.

Gateway Implementation

from lykke.application.gateways.google_protocol import GoogleCalendarGatewayProtocol
from lykke.domain.entities import CalendarEntity, CalendarEntryEntity


class GoogleCalendarGateway:
    """Implementation of GoogleCalendarGatewayProtocol."""

    async def load_calendar_events(
        self,
        calendar: CalendarEntity,
        lookback: datetime,
        token: AuthToken,
    ) -> list[CalendarEntryEntity]:
        """Load calendar entries from Google Calendar."""
        google_cal = get_google_calendar(calendar, token)
        events = google_cal.get_events(time_min=lookback)

        # Convert Google events to domain entities
        return [
            self._google_event_to_entity(event, calendar) for event in events
        ]

Key Rules for Gateways

1. Implement protocols - Gateways implement gateway protocols from application layer
2. Return domain objects - Methods return domain entities or value objects
3. Accept domain objects - Methods accept domain entities or value objects
4. Handle external APIs - Use third-party libraries for external API calls
5. Error handling - Convert external API errors to domain exceptions
6. Data conversion - Convert external API data to domain objects

Unit of Work

The Unit of Work implementation manages database transactions and coordinates repository operations.

SqlAlchemyUnitOfWork

The Unit of Work implementation is in infrastructure/unit_of_work.py:

class SqlAlchemyUnitOfWork:
    """SQLAlchemy implementation of UnitOfWorkProtocol."""

    # Read-only repository properties (public API)
    task_ro_repo: TaskRepositoryReadOnlyProtocol
    day_ro_repo: DayRepositoryReadOnlyProtocol
    # ... other repos

    def __init__(self, user: UserEntity) -> None:
        self.user = user
        self._connection: AsyncConnection | None = None
        self._added_entities: list[BaseEntityObject] = []

    async def __aenter__(self) -> Self:
        """Enter transaction context."""
        # Create connection and transaction
        # Initialize repositories with connection
        return self

    async def __aexit__(self, ...) -> None:
        """Exit transaction context."""
        # Auto-commit on success, rollback on exception

    def add(self, entity: BaseEntityObject) -> BaseEntityObject:
        """Track entity for persistence."""
        self._added_entities.append(entity)
        return entity

    async def create(self, entity: T) -> T:
        """Create new entity (marks as created + adds to tracking)."""
        entity.create()
        return self.add(entity)

    async def delete(self, entity: BaseEntityObject) -> None:
        """Delete entity (marks as deleted + adds to tracking)."""
        entity.delete()
        self.add(entity)

    async def commit(self) -> None:
        """Commit transaction and dispatch events."""
        # Process added entities (create/update/delete based on events)
        # Dispatch domain events to handlers (BEFORE commit)
        # Commit database transaction
        # Publish events to Redis (AFTER commit)

Key Rules for Unit of Work

1. Implement protocol - Must implement UnitOfWorkProtocol
2. Transaction management - Manage database transactions via async context manager
3. Repository coordination - Provide repositories that share the same connection
4. Entity tracking - Track entities via add() method
5. Event collection - Collect domain events from tracked entities
6. Event dispatching - Dispatch events before commit, publish to Redis after commit

Database Tables

Database tables are defined as SQLAlchemy ORM models in infrastructure/database/tables/, and each module exports a *_tbl = Model.__table__ alias for repository query building.

Table Definition

class Task(Base):
    __tablename__ = "tasks"
    id: Mapped[UUID] = mapped_column(Uuid, primary_key=True)
    user_id: Mapped[UUID] = mapped_column(Uuid, nullable=False, index=True)
    scheduled_date: Mapped[date] = mapped_column(Date, nullable=False, index=True)
    name: Mapped[str] = mapped_column(String, nullable=False)
    # ... other columns ...

tasks_tbl = Task.__table__

Key Rules for Tables

1. Use SQLAlchemy table modules - Prefer existing *_tbl exports from table modules
2. Use ORM model + table alias pattern - Model defines schema, repositories query against *_tbl
3. Index foreign keys - Index foreign keys and commonly queried fields
4. Use JSONB for complex types - Store nested objects as JSONB
5. Use UUID for IDs - Use UUID type for primary keys

Gotchas

1. JSONB Field Serialization

Issue: JSONB fields must be serialized/deserialized correctly. Use dataclass_to_json_dict() for serialization.

# Serialization
row["task_definition"] = dataclass_to_json_dict(task.task_definition)

# Deserialization
if data.get("task_definition"):
    data["task_definition"] = TaskDefinitionEntity(**data["task_definition"])

2. Enum Conversion

Issue: Enums must be converted to/from string values for database storage.

# Entity to row
row["status"] = task.status.value  # Convert enum to string

# Row to entity
if "status" in data and isinstance(data["status"], str):
    data["status"] = TaskStatus(data["status"])  # Convert string to enum

3. UUID Conversion in JSONB

Issue: UUIDs in JSONB are stored as strings and must be converted back.

# JSONB stores UUIDs as strings
if "id" in template_data and isinstance(template_data["id"], str):
    template_data["id"] = UUID(template_data["id"])

4. None to Empty List Normalization

Issue: Database may return None for JSONB array fields, but entities expect empty lists.

from lykke.infrastructure.repositories.base.utils import normalize_list_fields

# Normalize None to [] for list fields
data = normalize_list_fields(dict(row), Entity)

5. Init=False Fields

Issue: Entities have fields like _domain_events that shouldn't be passed to constructors.

from lykke.infrastructure.repositories.base.utils import filter_init_false_fields

# Filter out init=False fields
data = filter_init_false_fields(dict(row), Entity)

6. User Scoping in Queries

Issue: All user-scoped repositories automatically filter by user_id. Don't add it manually.

# Base repository automatically adds user_id filter
def build_query(self, query: TaskQuery) -> Select[tuple]:
    stmt = super().build_query(query)  # Already filtered by user_id
    # Add additional filters only
    if query.date is not None:
        stmt = stmt.where(self.table.c.scheduled_date == query.date)
    return stmt

Testing Guidance

Integration Testing Repositories

Test repository implementations with a real database:

@pytest.mark.asyncio
async def test_task_repository_get(task_repo, test_user, test_date):
    """Test getting a task by ID."""
    task = TaskEntity(
        id=uuid4(),
        user_id=test_user.id,
        name="Test Task",
        status=TaskStatus.READY,
        ...
    )
    await task_repo.put(task)

    result = await task_repo.get(task.id)

    assert result.id == task.id
    assert result.name == "Test Task"

Test Location

Infrastructure layer tests go in:

• tests/integration/repositories/ - Repository integration tests
• tests/unit/gateways/ - Gateway unit tests (with mocks/VCR)

Test Patterns

1. Use real database - Integration tests use real database
2. Test serialization - Verify entity-to-row and row-to-entity conversion
3. Test query building - Verify custom query logic
4. Test error handling - Verify error cases (not found, validation errors)