Skip to content

domain.protocols.holding_repository

src.domain.protocols.holding_repository

Holding repository protocol.

Defines the interface for holding persistence operations. Holdings are synced from providers and represent current portfolio positions.

Classes

HoldingRepository

Bases: Protocol

Protocol for holding persistence operations.

Defines the contract for storing and retrieving holding data. Infrastructure layer provides concrete implementations (e.g., PostgreSQL).

Design Principles: - Read methods return domain entities (Holding), not database models - Holdings belong to accounts (account_id FK) - Bulk operations for sync efficiency (save_many, delete_by_account) - Active/inactive filtering for closed positions

Implementation Notes: - Holdings are synced frequently, optimize for bulk operations - provider_holding_id + account_id is unique - Support upsert for sync operations (create or update)

Source code in src/domain/protocols/holding_repository.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
class HoldingRepository(Protocol):
    """Protocol for holding persistence operations.

    Defines the contract for storing and retrieving holding data.
    Infrastructure layer provides concrete implementations (e.g., PostgreSQL).

    **Design Principles**:
    - Read methods return domain entities (Holding), not database models
    - Holdings belong to accounts (account_id FK)
    - Bulk operations for sync efficiency (save_many, delete_by_account)
    - Active/inactive filtering for closed positions

    **Implementation Notes**:
    - Holdings are synced frequently, optimize for bulk operations
    - provider_holding_id + account_id is unique
    - Support upsert for sync operations (create or update)
    """

    async def find_by_id(self, holding_id: UUID) -> Holding | None:
        """Find holding by ID.

        Args:
            holding_id: Unique holding identifier.

        Returns:
            Holding entity if found, None otherwise.

        Example:
            >>> holding = await repo.find_by_id(holding_id)
            >>> if holding:
            ...     print(f"Found: {holding.symbol}")
        """
        ...

    async def find_by_account_and_symbol(
        self, account_id: UUID, symbol: str
    ) -> Holding | None:
        """Find holding by account and symbol.

        Args:
            account_id: Account identifier.
            symbol: Security symbol (e.g., "AAPL").

        Returns:
            Holding entity if found, None otherwise.

        Example:
            >>> holding = await repo.find_by_account_and_symbol(account_id, "AAPL")
            >>> if holding:
            ...     print(f"Owns {holding.quantity} shares")
        """
        ...

    async def find_by_provider_holding_id(
        self, account_id: UUID, provider_holding_id: str
    ) -> Holding | None:
        """Find holding by provider's unique identifier.

        Used for deduplication during sync operations.

        Args:
            account_id: Account identifier.
            provider_holding_id: Provider's unique holding identifier.

        Returns:
            Holding entity if found, None otherwise.

        Example:
            >>> holding = await repo.find_by_provider_holding_id(
            ...     account_id, "SCHWAB-AAPL-123"
            ... )
        """
        ...

    async def list_by_account(
        self, account_id: UUID, *, active_only: bool = True
    ) -> list[Holding]:
        """List holdings for an account.

        Args:
            account_id: Account identifier.
            active_only: If True, only return active holdings (quantity > 0).

        Returns:
            List of holdings for the account.

        Example:
            >>> holdings = await repo.list_by_account(account_id)
            >>> print(f"Found {len(holdings)} positions")
        """
        ...

    async def list_by_user(
        self, user_id: UUID, *, active_only: bool = True
    ) -> list[Holding]:
        """List all holdings across all accounts for a user.

        Requires joining through account -> connection -> user.

        Args:
            user_id: User identifier.
            active_only: If True, only return active holdings.

        Returns:
            List of all holdings for the user.

        Example:
            >>> holdings = await repo.list_by_user(user_id)
            >>> total_value = sum(h.market_value.amount for h in holdings)
        """
        ...

    async def save(self, holding: Holding) -> None:
        """Save a holding (create or update).

        Creates new holding if ID doesn't exist, updates if it does.
        Typically used for individual holding updates.

        Args:
            holding: Holding entity to save.

        Example:
            >>> await repo.save(holding)
        """
        ...

    async def save_many(self, holdings: list[Holding]) -> None:
        """Save multiple holdings in batch.

        Optimized for sync operations. Uses upsert logic:
        - Creates new holdings if they don't exist
        - Updates existing holdings if they do

        Args:
            holdings: List of holdings to save.

        Example:
            >>> await repo.save_many(holdings_from_provider)
        """
        ...

    async def delete(self, holding_id: UUID) -> None:
        """Delete a holding.

        Typically not used - holdings are marked inactive instead.
        May be needed for data cleanup operations.

        Args:
            holding_id: Holding ID to delete.

        Example:
            >>> await repo.delete(holding_id)
        """
        ...

    async def delete_by_account(self, account_id: UUID) -> int:
        """Delete all holdings for an account.

        Used when account is disconnected or for cleanup.

        Args:
            account_id: Account identifier.

        Returns:
            Number of holdings deleted.

        Example:
            >>> deleted = await repo.delete_by_account(account_id)
            >>> print(f"Deleted {deleted} holdings")
        """
        ...
Functions
find_by_id async 
find_by_id(holding_id: UUID) -> Holding | None

Find holding by ID.

Parameters:

Name Type Description Default
holding_id UUID

Unique holding identifier.

 required

Returns:

Type Description
Holding | None

Holding entity if found, None otherwise.
Example

holding = await repo.find_by_id(holding_id) if holding: ... print(f"Found: {holding.symbol}")

Source code in src/domain/protocols/holding_repository.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
async def find_by_id(self, holding_id: UUID) -> Holding | None:
    """Find holding by ID.

    Args:
        holding_id: Unique holding identifier.

    Returns:
        Holding entity if found, None otherwise.

    Example:
        >>> holding = await repo.find_by_id(holding_id)
        >>> if holding:
        ...     print(f"Found: {holding.symbol}")
    """
    ...
find_by_account_and_symbol async 
find_by_account_and_symbol(
    account_id: UUID, symbol: str
) -> Holding | None

Find holding by account and symbol.

Parameters:

Name Type Description Default
account_id UUID

Account identifier.

 required
symbol str

Security symbol (e.g., "AAPL").

 required

Returns:

Type Description
Holding | None

Holding entity if found, None otherwise.
Example

holding = await repo.find_by_account_and_symbol(account_id, "AAPL") if holding: ... print(f"Owns {holding.quantity} shares")

Source code in src/domain/protocols/holding_repository.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
async def find_by_account_and_symbol(
    self, account_id: UUID, symbol: str
) -> Holding | None:
    """Find holding by account and symbol.

    Args:
        account_id: Account identifier.
        symbol: Security symbol (e.g., "AAPL").

    Returns:
        Holding entity if found, None otherwise.

    Example:
        >>> holding = await repo.find_by_account_and_symbol(account_id, "AAPL")
        >>> if holding:
        ...     print(f"Owns {holding.quantity} shares")
    """
    ...
find_by_provider_holding_id async 
find_by_provider_holding_id(
    account_id: UUID, provider_holding_id: str
) -> Holding | None

Find holding by provider's unique identifier.

Used for deduplication during sync operations.

Parameters:

Name Type Description Default
account_id UUID

Account identifier.

 required
provider_holding_id str

Provider's unique holding identifier.

 required

Returns:

Type Description
Holding | None

Holding entity if found, None otherwise.
Example

holding = await repo.find_by_provider_holding_id( ... account_id, "SCHWAB-AAPL-123" ... )

Source code in src/domain/protocols/holding_repository.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
async def find_by_provider_holding_id(
    self, account_id: UUID, provider_holding_id: str
) -> Holding | None:
    """Find holding by provider's unique identifier.

    Used for deduplication during sync operations.

    Args:
        account_id: Account identifier.
        provider_holding_id: Provider's unique holding identifier.

    Returns:
        Holding entity if found, None otherwise.

    Example:
        >>> holding = await repo.find_by_provider_holding_id(
        ...     account_id, "SCHWAB-AAPL-123"
        ... )
    """
    ...
list_by_account async 
list_by_account(
    account_id: UUID, *, active_only: bool = True
) -> list[Holding]

List holdings for an account.

Parameters:

Name Type Description Default
account_id UUID

Account identifier.

 required
active_only bool

If True, only return active holdings (quantity > 0).

 True

Returns:

Type Description
list[Holding]

List of holdings for the account.
Example

holdings = await repo.list_by_account(account_id) print(f"Found {len(holdings)} positions")

Source code in src/domain/protocols/holding_repository.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
async def list_by_account(
    self, account_id: UUID, *, active_only: bool = True
) -> list[Holding]:
    """List holdings for an account.

    Args:
        account_id: Account identifier.
        active_only: If True, only return active holdings (quantity > 0).

    Returns:
        List of holdings for the account.

    Example:
        >>> holdings = await repo.list_by_account(account_id)
        >>> print(f"Found {len(holdings)} positions")
    """
    ...
list_by_user async 
list_by_user(
    user_id: UUID, *, active_only: bool = True
) -> list[Holding]

List all holdings across all accounts for a user.

Requires joining through account -> connection -> user.

Parameters:

Name Type Description Default
user_id UUID

User identifier.

 required
active_only bool

If True, only return active holdings.

 True

Returns:

Type Description
list[Holding]

List of all holdings for the user.
Example

holdings = await repo.list_by_user(user_id) total_value = sum(h.market_value.amount for h in holdings)

Source code in src/domain/protocols/holding_repository.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
async def list_by_user(
    self, user_id: UUID, *, active_only: bool = True
) -> list[Holding]:
    """List all holdings across all accounts for a user.

    Requires joining through account -> connection -> user.

    Args:
        user_id: User identifier.
        active_only: If True, only return active holdings.

    Returns:
        List of all holdings for the user.

    Example:
        >>> holdings = await repo.list_by_user(user_id)
        >>> total_value = sum(h.market_value.amount for h in holdings)
    """
    ...
save async 
save(holding: Holding) -> None

Save a holding (create or update).

Creates new holding if ID doesn't exist, updates if it does. Typically used for individual holding updates.

Parameters:

Name Type Description Default
holding Holding

Holding entity to save.

 required
Example

await repo.save(holding)

Source code in src/domain/protocols/holding_repository.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
async def save(self, holding: Holding) -> None:
    """Save a holding (create or update).

    Creates new holding if ID doesn't exist, updates if it does.
    Typically used for individual holding updates.

    Args:
        holding: Holding entity to save.

    Example:
        >>> await repo.save(holding)
    """
    ...
save_many async 
save_many(holdings: list[Holding]) -> None

Save multiple holdings in batch.

Optimized for sync operations. Uses upsert logic: - Creates new holdings if they don't exist - Updates existing holdings if they do

Parameters:

Name Type Description Default
holdings list[Holding]

List of holdings to save.

 required
Example

await repo.save_many(holdings_from_provider)

Source code in src/domain/protocols/holding_repository.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
async def save_many(self, holdings: list[Holding]) -> None:
    """Save multiple holdings in batch.

    Optimized for sync operations. Uses upsert logic:
    - Creates new holdings if they don't exist
    - Updates existing holdings if they do

    Args:
        holdings: List of holdings to save.

    Example:
        >>> await repo.save_many(holdings_from_provider)
    """
    ...
delete async 
delete(holding_id: UUID) -> None

Delete a holding.

Typically not used - holdings are marked inactive instead. May be needed for data cleanup operations.

Parameters:

Name Type Description Default
holding_id UUID

Holding ID to delete.

 required
Example

await repo.delete(holding_id)

Source code in src/domain/protocols/holding_repository.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
async def delete(self, holding_id: UUID) -> None:
    """Delete a holding.

    Typically not used - holdings are marked inactive instead.
    May be needed for data cleanup operations.

    Args:
        holding_id: Holding ID to delete.

    Example:
        >>> await repo.delete(holding_id)
    """
    ...
delete_by_account async 
delete_by_account(account_id: UUID) -> int

Delete all holdings for an account.

Used when account is disconnected or for cleanup.

Parameters:

Name Type Description Default
account_id UUID

Account identifier.

 required

Returns:

Type Description
int

Number of holdings deleted.
Example

deleted = await repo.delete_by_account(account_id) print(f"Deleted {deleted} holdings")

Source code in src/domain/protocols/holding_repository.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
async def delete_by_account(self, account_id: UUID) -> int:
    """Delete all holdings for an account.

    Used when account is disconnected or for cleanup.

    Args:
        account_id: Account identifier.

    Returns:
        Number of holdings deleted.

    Example:
        >>> deleted = await repo.delete_by_account(account_id)
        >>> print(f"Deleted {deleted} holdings")
    """
    ...