78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
from __future__ import annotations
|
||
|
||
from collections.abc import Sequence
|
||
from datetime import date
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.db.models import ImportBatch, Portfolio
|
||
|
||
|
||
class ImportBatchRepository:
|
||
def __init__(self, session: AsyncSession) -> None:
|
||
self.session = session
|
||
|
||
async def list_by_content(
|
||
self, content_sha256: str
|
||
) -> list[tuple[ImportBatch, str]]:
|
||
"""Загрузки этого файла во все портфели -> (загрузка, код портфеля)."""
|
||
result = await self.session.execute(
|
||
select(ImportBatch, Portfolio.code)
|
||
.join(Portfolio, Portfolio.id == ImportBatch.portfolio_id)
|
||
.where(ImportBatch.content_sha256 == content_sha256)
|
||
.order_by(ImportBatch.created_at, ImportBatch.id)
|
||
)
|
||
return [(batch, code) for batch, code in result.all()]
|
||
|
||
async def list_overlapping(
|
||
self, date_from: date, date_to: date, exclude_sha256: str
|
||
) -> list[tuple[ImportBatch, str]]:
|
||
"""Загрузки других файлов, период которых пересекается с [from, to]."""
|
||
result = await self.session.execute(
|
||
select(ImportBatch, Portfolio.code)
|
||
.join(Portfolio, Portfolio.id == ImportBatch.portfolio_id)
|
||
.where(
|
||
ImportBatch.content_sha256 != exclude_sha256,
|
||
ImportBatch.date_from <= date_to,
|
||
ImportBatch.date_to >= date_from,
|
||
)
|
||
.order_by(ImportBatch.created_at, ImportBatch.id)
|
||
)
|
||
return [(batch, code) for batch, code in result.all()]
|
||
|
||
async def add(
|
||
self,
|
||
*,
|
||
source: str,
|
||
portfolio_id: int,
|
||
file_name: str,
|
||
content_sha256: str,
|
||
date_from: date,
|
||
date_to: date,
|
||
rows_total: int,
|
||
loaded_rows: Sequence[int],
|
||
rows_excluded: int,
|
||
user: str,
|
||
) -> ImportBatch:
|
||
"""Вызывается в одной транзакции с записью операций. Повтор (тот же
|
||
хэш и портфель) отсекает уникальный индекс
|
||
uq_import_batch_content_portfolio — IntegrityError на flush."""
|
||
batch = ImportBatch(
|
||
source=source,
|
||
portfolio_id=portfolio_id,
|
||
file_name=file_name,
|
||
content_sha256=content_sha256,
|
||
date_from=date_from,
|
||
date_to=date_to,
|
||
rows_total=rows_total,
|
||
rows_loaded=len(loaded_rows),
|
||
rows_excluded=rows_excluded,
|
||
loaded_rows=sorted(loaded_rows),
|
||
created_by=user,
|
||
updated_by=user,
|
||
)
|
||
self.session.add(batch)
|
||
await self.session.flush()
|
||
return batch
|