[Phase 1] Task 2: Create core/ports.py — Storage interfaces #6

New Issue

shoko · 2026-04-02T18:29:04+02:00

shoko commented

2026-04-02 18:29:04 +02:00

Task 2: Create core/ports.py — Storage interfaces

Labels: phase-1, core
Dependency: Task 1 (issue #5)

Goal

Define abstract storage interfaces (ports) that adapters must implement.

Files to create/modify

core/ports.py

Interfaces to implement

from typing import Protocol, Optional
from core.models import GroupData, UserData, TrackingData, GroupBounty, PersonalBounty

class GroupStoragePort(Protocol):
    """Port for group bounty storage operations."""
    
    def load(self, group_id: int) -> GroupData:
        """Load all data for a group. Returns empty GroupData if not found."""
        ...
    
    def save(self, group_id: int, data: GroupData) -> None:
        """Atomically save group data."""
        ...
    
    def add_bounty(self, group_id: int, bounty: GroupBounty) -> GroupBounty:
        """Add a bounty to the group. Assigns ID from next_id counter."""
        ...
    
    def update_bounty(
        self,
        group_id: int,
        bounty_id: int,
        text: Optional[str] = None,
        link: Optional[str] = None,
        due_date_ts: Optional[int] = None,
        clear_link: bool = False,
        clear_due: bool = False,
    ) -> bool:
        """Update a bounty. Returns True if found and updated."""
        ...
    
    def delete_bounty(self, group_id: int, bounty_id: int) -> bool:
        """Delete a bounty. Returns True if found and deleted."""
        ...
    
    def get_bounty(self, group_id: int, bounty_id: int) -> Optional[GroupBounty]:
        """Get a single bounty by ID. Returns None if not found."""
        ...


class PersonalStoragePort(Protocol):
    """Port for personal bounty storage operations (DM mode)."""
    
    def load(self, user_id: int) -> UserData:
        """Load all personal bounties for a user."""
        ...
    
    def save(self, user_id: int, data: UserData) -> None:
        """Atomically save user data."""
        ...
    
    def add_bounty(self, user_id: int, bounty: PersonalBounty) -> PersonalBounty:
        """Add a personal bounty."""
        ...
    
    def update_bounty(
        self,
        user_id: int,
        bounty_id: int,
        text: Optional[str] = None,
        link: Optional[str] = None,
        due_date_ts: Optional[int] = None,
        clear_link: bool = False,
        clear_due: bool = False,
    ) -> bool:
        """Update a personal bounty."""
        ...
    
    def delete_bounty(self, user_id: int, bounty_id: int) -> bool:
        """Delete a personal bounty."""
        ...
    
    def get_bounty(self, user_id: int, bounty_id: int) -> Optional[PersonalBounty]:
        """Get a single personal bounty."""
        ...


class TrackingStoragePort(Protocol):
    """Port for user tracking storage operations."""
    
    def load(self, group_id: int, user_id: int) -> TrackingData:
        """Load tracking data for a user in a group."""
        ...
    
    def save(self, group_id: int, user_id: int, data: TrackingData) -> None:
        """Atomically save tracking data."""
        ...
    
    def track_bounty(self, group_id: int, user_id: int, bounty_id: int) -> bool:
        """Start tracking a bounty. Returns True if newly tracked, False if already tracking."""
        ...
    
    def untrack_bounty(self, group_id: int, user_id: int, bounty_id: int) -> bool:
        """Stop tracking a bounty. Returns True if was tracking, False if not."""
        ...

Notes

The clear_link and clear_due flags allow explicit clearing (vs None which means "do not change")
Ports use Python Protocols for structural typing (no explicit inheritance needed)

Acceptance Criteria

All three ports defined with proper method signatures
Type hints using domain models from Task 1
No implementation — pure interface only

## Task 2: Create core/ports.py — Storage interfaces **Labels:** phase-1, core **Dependency:** Task 1 (issue #5) ### Goal Define abstract storage interfaces (ports) that adapters must implement. ### Files to create/modify - `core/ports.py` ### Interfaces to implement ```python from typing import Protocol, Optional from core.models import GroupData, UserData, TrackingData, GroupBounty, PersonalBounty class GroupStoragePort(Protocol): """Port for group bounty storage operations.""" def load(self, group_id: int) -> GroupData: """Load all data for a group. Returns empty GroupData if not found.""" ... def save(self, group_id: int, data: GroupData) -> None: """Atomically save group data.""" ... def add_bounty(self, group_id: int, bounty: GroupBounty) -> GroupBounty: """Add a bounty to the group. Assigns ID from next_id counter.""" ... def update_bounty( self, group_id: int, bounty_id: int, text: Optional[str] = None, link: Optional[str] = None, due_date_ts: Optional[int] = None, clear_link: bool = False, clear_due: bool = False, ) -> bool: """Update a bounty. Returns True if found and updated.""" ... def delete_bounty(self, group_id: int, bounty_id: int) -> bool: """Delete a bounty. Returns True if found and deleted.""" ... def get_bounty(self, group_id: int, bounty_id: int) -> Optional[GroupBounty]: """Get a single bounty by ID. Returns None if not found.""" ... class PersonalStoragePort(Protocol): """Port for personal bounty storage operations (DM mode).""" def load(self, user_id: int) -> UserData: """Load all personal bounties for a user.""" ... def save(self, user_id: int, data: UserData) -> None: """Atomically save user data.""" ... def add_bounty(self, user_id: int, bounty: PersonalBounty) -> PersonalBounty: """Add a personal bounty.""" ... def update_bounty( self, user_id: int, bounty_id: int, text: Optional[str] = None, link: Optional[str] = None, due_date_ts: Optional[int] = None, clear_link: bool = False, clear_due: bool = False, ) -> bool: """Update a personal bounty.""" ... def delete_bounty(self, user_id: int, bounty_id: int) -> bool: """Delete a personal bounty.""" ... def get_bounty(self, user_id: int, bounty_id: int) -> Optional[PersonalBounty]: """Get a single personal bounty.""" ... class TrackingStoragePort(Protocol): """Port for user tracking storage operations.""" def load(self, group_id: int, user_id: int) -> TrackingData: """Load tracking data for a user in a group.""" ... def save(self, group_id: int, user_id: int, data: TrackingData) -> None: """Atomically save tracking data.""" ... def track_bounty(self, group_id: int, user_id: int, bounty_id: int) -> bool: """Start tracking a bounty. Returns True if newly tracked, False if already tracking.""" ... def untrack_bounty(self, group_id: int, user_id: int, bounty_id: int) -> bool: """Stop tracking a bounty. Returns True if was tracking, False if not.""" ... ``` ### Notes - The `clear_link` and `clear_due` flags allow explicit clearing (vs `None` which means "do not change") - Ports use Python Protocols for structural typing (no explicit inheritance needed) ### Acceptance Criteria - [ ] All three ports defined with proper method signatures - [ ] Type hints using domain models from Task 1 - [ ] No implementation — pure interface only

shoko referenced this issue

2026-04-02 18:29:54 +02:00

[Phase 1] Task 4: Create core/services.py — Business logic #8

shoko referenced this issue

2026-04-02 18:30:40 +02:00

[Phase 1] Task 5: Create adapters/storage/json_file.py — JSON file adapter #9

shoko referenced this issue

2026-04-02 18:32:56 +02:00

[Phase 1] Architecture Extraction — Full Hexagonal Architecture Refactor #17

shoko added the core phase-1 labels 2026-04-02 18:34:40 +02:00

shoko referenced a pull request that will close this issue

2026-04-03 00:40:26 +02:00