Skip to content

core.container.repositories

src.core.container.repositories

Repository dependency factories.

Request-scoped repository instances for domain entity persistence. Each request gets fresh repository instances with shared session.

Reference

See docs/architecture/dependency-injection-architecture.md for complete patterns and usage examples.

Classes

Functions

get_user_repository async 

get_user_repository(
    session: AsyncSession = Depends(get_db_session),
) -> UserRepository

Get user repository (request-scoped).

Creates new repository instance per request with database session. Repository provides CRUD operations for User domain entities.

Parameters:

Name Type Description Default
session AsyncSession

Database session for request duration. Injected via Depends(get_db_session).

 Depends(get_db_session)

Returns:

Type Description
UserRepository

UserRepository instance.
Usage
Application Layer (command handlers)

from src.core.container import get_user_repository user_repo = await anext(get_user_repository()) user = await user_repo.find_by_email("user@example.com")

Presentation Layer (FastAPI Depends)

from fastapi import Depends from src.infrastructure.persistence.repositories import UserRepository

@router.post("/users") async def create_user( user_repo: UserRepository = Depends(get_user_repository) ): await user_repo.save(user)

Reference
  • docs/architecture/authentication-architecture.md (Lines 589-593)
Source code in src/core/container/repositories.py 
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
async def get_user_repository(
    session: AsyncSession = Depends(get_db_session),
) -> "UserRepository":
    """Get user repository (request-scoped).

    Creates new repository instance per request with database session.
    Repository provides CRUD operations for User domain entities.

    Args:
        session: Database session for request duration.
            Injected via Depends(get_db_session).

    Returns:
        UserRepository instance.

    Usage:
        # Application Layer (command handlers)
        from src.core.container import get_user_repository
        user_repo = await anext(get_user_repository())
        user = await user_repo.find_by_email("user@example.com")

        # Presentation Layer (FastAPI Depends)
        from fastapi import Depends
        from src.infrastructure.persistence.repositories import UserRepository

        @router.post("/users")
        async def create_user(
            user_repo: UserRepository = Depends(get_user_repository)
        ):
            await user_repo.save(user)

    Reference:
        - docs/architecture/authentication-architecture.md (Lines 589-593)
    """
    from src.infrastructure.persistence.repositories import UserRepository

    return UserRepository(session=session)

get_provider_connection_repository async 

get_provider_connection_repository(
    session: AsyncSession = Depends(get_db_session),
) -> ProviderConnectionRepository

Get provider connection repository (request-scoped).

Creates new repository instance per request with database session. Repository provides CRUD operations for ProviderConnection domain entities.

Parameters:

Name Type Description Default
session AsyncSession

Database session for request duration. Injected via Depends(get_db_session).

 Depends(get_db_session)

Returns:

Type Description
ProviderConnectionRepository

ProviderConnectionRepository instance.
Usage
Application Layer (command handlers)

from src.core.container import get_provider_connection_repository conn_repo = await anext(get_provider_connection_repository()) conn = await conn_repo.find_by_id(connection_id)

Presentation Layer (FastAPI Depends)

from fastapi import Depends from src.infrastructure.persistence.repositories import ProviderConnectionRepository

@router.get("/providers/connections/{id}") async def get_connection( conn_repo: ProviderConnectionRepository = Depends(get_provider_connection_repository) ): return await conn_repo.find_by_id(id)

Reference
  • docs/architecture/repository-pattern.md
Source code in src/core/container/repositories.py 
 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
async def get_provider_connection_repository(
    session: AsyncSession = Depends(get_db_session),
) -> "ProviderConnectionRepository":
    """Get provider connection repository (request-scoped).

    Creates new repository instance per request with database session.
    Repository provides CRUD operations for ProviderConnection domain entities.

    Args:
        session: Database session for request duration.
            Injected via Depends(get_db_session).

    Returns:
        ProviderConnectionRepository instance.

    Usage:
        # Application Layer (command handlers)
        from src.core.container import get_provider_connection_repository
        conn_repo = await anext(get_provider_connection_repository())
        conn = await conn_repo.find_by_id(connection_id)

        # Presentation Layer (FastAPI Depends)
        from fastapi import Depends
        from src.infrastructure.persistence.repositories import ProviderConnectionRepository

        @router.get("/providers/connections/{id}")
        async def get_connection(
            conn_repo: ProviderConnectionRepository = Depends(get_provider_connection_repository)
        ):
            return await conn_repo.find_by_id(id)

    Reference:
        - docs/architecture/repository-pattern.md
    """
    from src.infrastructure.persistence.repositories import (
        ProviderConnectionRepository,
    )

    return ProviderConnectionRepository(session=session)

get_provider_repository async 

get_provider_repository(
    session: AsyncSession = Depends(get_db_session),
) -> ProviderRepository

Get provider repository (request-scoped).

Creates new repository instance per request with database session. Repository provides CRUD operations for Provider domain entities.

Parameters:

Name Type Description Default
session AsyncSession

Database session for request duration. Injected via Depends(get_db_session).

 Depends(get_db_session)

Returns:

Type Description
ProviderRepository

ProviderRepository instance.
Usage
Application Layer

from src.core.container import get_provider_repository provider_repo = await anext(get_provider_repository()) provider = await provider_repo.find_by_slug("schwab")

Presentation Layer (FastAPI Depends)

from fastapi import Depends

@router.get("/providers/{slug}") async def get_provider( provider_repo: ProviderRepository = Depends(get_provider_repository) ): return await provider_repo.find_by_slug(slug)

Source code in src/core/container/repositories.py 
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
async def get_provider_repository(
    session: AsyncSession = Depends(get_db_session),
) -> "ProviderRepository":
    """Get provider repository (request-scoped).

    Creates new repository instance per request with database session.
    Repository provides CRUD operations for Provider domain entities.

    Args:
        session: Database session for request duration.
            Injected via Depends(get_db_session).

    Returns:
        ProviderRepository instance.

    Usage:
        # Application Layer
        from src.core.container import get_provider_repository
        provider_repo = await anext(get_provider_repository())
        provider = await provider_repo.find_by_slug("schwab")

        # Presentation Layer (FastAPI Depends)
        from fastapi import Depends

        @router.get("/providers/{slug}")
        async def get_provider(
            provider_repo: ProviderRepository = Depends(get_provider_repository)
        ):
            return await provider_repo.find_by_slug(slug)
    """
    from src.infrastructure.persistence.repositories.provider_repository import (
        ProviderRepository,
    )

    return ProviderRepository(session=session)

get_account_repository async 

get_account_repository(
    session: AsyncSession = Depends(get_db_session),
) -> AccountRepository

Get account repository (request-scoped).

Creates new repository instance per request with database session. Repository provides CRUD operations for Account domain entities.

Parameters:

Name Type Description Default
session AsyncSession

Database session for request duration. Injected via Depends(get_db_session).

 Depends(get_db_session)

Returns:

Type Description
AccountRepository

AccountRepository instance.
Usage
Application Layer (command handlers)

from src.core.container import get_account_repository account_repo = await anext(get_account_repository()) account = await repo.find_by_id(account_id)

Presentation Layer (FastAPI Depends)

from fastapi import Depends from src.infrastructure.persistence.repositories import AccountRepository

@router.get("/accounts/{id}") async def get_account( account_repo: AccountRepository = Depends(get_account_repository) ): return await account_repo.find_by_id(id)

Reference
  • docs/architecture/repository-pattern.md
Source code in src/core/container/repositories.py 
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
184
185
186
187
188
async def get_account_repository(
    session: AsyncSession = Depends(get_db_session),
) -> "AccountRepository":
    """Get account repository (request-scoped).

    Creates new repository instance per request with database session.
    Repository provides CRUD operations for Account domain entities.

    Args:
        session: Database session for request duration.
            Injected via Depends(get_db_session).

    Returns:
        AccountRepository instance.

    Usage:
        # Application Layer (command handlers)
        from src.core.container import get_account_repository
        account_repo = await anext(get_account_repository())
        account = await repo.find_by_id(account_id)

        # Presentation Layer (FastAPI Depends)
        from fastapi import Depends
        from src.infrastructure.persistence.repositories import AccountRepository

        @router.get("/accounts/{id}")
        async def get_account(
            account_repo: AccountRepository = Depends(get_account_repository)
        ):
            return await account_repo.find_by_id(id)

    Reference:
        - docs/architecture/repository-pattern.md
    """
    from src.infrastructure.persistence.repositories import AccountRepository

    return AccountRepository(session=session)

get_transaction_repository async 

get_transaction_repository(
    session: AsyncSession = Depends(get_db_session),
) -> TransactionRepository

Get transaction repository (request-scoped).

Creates new repository instance per request with database session. Repository provides CRUD operations for Transaction domain entities.

Parameters:

Name Type Description Default
session AsyncSession

Database session for request duration. Injected via Depends(get_db_session).

 Depends(get_db_session)

Returns:

Type Description
TransactionRepository

TransactionRepository instance.
Usage
Application Layer (command handlers)

from src.core.container import get_transaction_repository tx_repo = await anext(get_transaction_repository()) transactions = await tx_repo.find_by_account_id(account_id)

Presentation Layer (FastAPI Depends)

from fastapi import Depends from src.infrastructure.persistence.repositories import TransactionRepository

@router.get("/transactions") async def list_transactions( tx_repo: TransactionRepository = Depends(get_transaction_repository) ): return await tx_repo.find_by_account_id(account_id, limit=50)

Reference
  • docs/architecture/repository-pattern.md
Source code in src/core/container/repositories.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
async def get_transaction_repository(
    session: AsyncSession = Depends(get_db_session),
) -> "TransactionRepository":
    """Get transaction repository (request-scoped).

    Creates new repository instance per request with database session.
    Repository provides CRUD operations for Transaction domain entities.

    Args:
        session: Database session for request duration.
            Injected via Depends(get_db_session).

    Returns:
        TransactionRepository instance.

    Usage:
        # Application Layer (command handlers)
        from src.core.container import get_transaction_repository
        tx_repo = await anext(get_transaction_repository())
        transactions = await tx_repo.find_by_account_id(account_id)

        # Presentation Layer (FastAPI Depends)
        from fastapi import Depends
        from src.infrastructure.persistence.repositories import TransactionRepository

        @router.get("/transactions")
        async def list_transactions(
            tx_repo: TransactionRepository = Depends(get_transaction_repository)
        ):
            return await tx_repo.find_by_account_id(account_id, limit=50)

    Reference:
        - docs/architecture/repository-pattern.md
    """
    from src.infrastructure.persistence.repositories import TransactionRepository

    return TransactionRepository(session=session)