32 lines
1.1 KiB
Python
32 lines
1.1 KiB
Python
from __future__ import annotations
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db.models import Currency, OperationType, Portfolio
|
|
|
|
|
|
class ReferenceRepository:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self.session = session
|
|
|
|
async def get_currency_id(self, code: str) -> int | None:
|
|
result = await self.session.execute(
|
|
select(Currency.id).where(Currency.code == code)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def currency_ids(self) -> dict[str, int]:
|
|
result = await self.session.execute(select(Currency.code, Currency.id))
|
|
return {code: id_ for code, id_ in result.all()}
|
|
|
|
async def operation_type_ids(self) -> dict[str, int]:
|
|
result = await self.session.execute(
|
|
select(OperationType.code, OperationType.id)
|
|
)
|
|
return {code: id_ for code, id_ in result.all()}
|
|
|
|
async def list_portfolios(self) -> list[Portfolio]:
|
|
result = await self.session.execute(select(Portfolio).order_by(Portfolio.id))
|
|
return list(result.scalars().all())
|