65 lines
2.2 KiB
Python
65 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db.models import Currency, Security
|
|
from app.integrations.moex import SecurityInfo
|
|
|
|
|
|
class SecurityRepository:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self.session = session
|
|
|
|
async def get_by_isin(self, isin: str) -> Security | None:
|
|
result = await self.session.execute(
|
|
select(Security).where(Security.isin == isin)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def list_active(self) -> Sequence[Security]:
|
|
result = await self.session.execute(
|
|
select(Security).where(Security.is_active.is_(True)).order_by(Security.isin)
|
|
)
|
|
return result.scalars().all()
|
|
|
|
async def list_by_names(self, names: Sequence[str]) -> Sequence[Security]:
|
|
"""Бумаги по краткому имени (security.name = SHORTNAME MOEX, D17)."""
|
|
if not names:
|
|
return []
|
|
result = await self.session.execute(
|
|
select(Security).where(Security.name.in_(names)).order_by(Security.isin)
|
|
)
|
|
return result.scalars().all()
|
|
|
|
async def list_with_currency(self) -> list[tuple[Security, str]]:
|
|
result = await self.session.execute(
|
|
select(Security, Currency.code)
|
|
.join(Currency, Currency.id == Security.currency_id)
|
|
.order_by(Security.isin)
|
|
)
|
|
return [(security, code) for security, code in result.all()]
|
|
|
|
async def add(
|
|
self, isin: str, info: SecurityInfo, currency_id: int, user: str
|
|
) -> Security:
|
|
security = Security(
|
|
isin=isin,
|
|
secid=info.secid,
|
|
name=info.short_name or info.secid,
|
|
initial_face_value=info.initial_face_value,
|
|
currency_id=currency_id,
|
|
placement_date=info.issue_date,
|
|
maturity_date=info.mat_date,
|
|
coupon_rate=info.coupon_percent,
|
|
coupon_frequency=info.coupon_frequency,
|
|
is_active=True,
|
|
created_by=user,
|
|
updated_by=user,
|
|
)
|
|
self.session.add(security)
|
|
await self.session.flush()
|
|
return security
|