12c8615615
Adds accounting-service PATCH/archive, fiscal helpers, COA templates and setup status, plus SuperApp Accounting UI (DS, scoreboard, masters, vouchers, ledger, ops modules) with session refresh and HTTPS public API URLs. Co-authored-by: Cursor <cursoragent@cursor.com>
109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
"""Posting repositories."""
|
|
from __future__ import annotations
|
|
|
|
from uuid import UUID
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.models.posting import (
|
|
AccountingAuditLog,
|
|
JournalEntry,
|
|
PostingError,
|
|
PostingLog,
|
|
PostingReference,
|
|
Voucher,
|
|
VoucherLine,
|
|
)
|
|
from app.models.types import VoucherStatus
|
|
from app.repositories.base import TenantBaseRepository
|
|
|
|
|
|
class VoucherRepository(TenantBaseRepository[Voucher]):
|
|
model = Voucher
|
|
|
|
async def get_with_lines(self, tenant_id: UUID, voucher_id: UUID) -> Voucher | None:
|
|
stmt = (
|
|
select(Voucher)
|
|
.options(selectinload(Voucher.lines))
|
|
.where(Voucher.tenant_id == tenant_id, Voucher.id == voucher_id)
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_by_number(self, tenant_id: UUID, voucher_number: str) -> Voucher | None:
|
|
stmt = select(Voucher).where(
|
|
Voucher.tenant_id == tenant_id, Voucher.voucher_number == voucher_number
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def list_by_status(self, tenant_id: UUID, status: VoucherStatus, *, limit: int = 50):
|
|
stmt = (
|
|
select(Voucher)
|
|
.where(Voucher.tenant_id == tenant_id, Voucher.status == status)
|
|
.limit(limit)
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
|
|
class VoucherLineRepository(TenantBaseRepository[VoucherLine]):
|
|
model = VoucherLine
|
|
|
|
|
|
class JournalEntryRepository(TenantBaseRepository[JournalEntry]):
|
|
model = JournalEntry
|
|
|
|
async def list_by_voucher(self, tenant_id: UUID, voucher_id: UUID):
|
|
stmt = select(JournalEntry).where(
|
|
JournalEntry.tenant_id == tenant_id,
|
|
JournalEntry.voucher_id == voucher_id,
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
async def list_by_account(self, tenant_id: UUID, account_id: UUID, *, limit: int = 100):
|
|
stmt = (
|
|
select(JournalEntry)
|
|
.where(
|
|
JournalEntry.tenant_id == tenant_id,
|
|
JournalEntry.account_id == account_id,
|
|
)
|
|
.limit(limit)
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
|
|
class PostingLogRepository(TenantBaseRepository[PostingLog]):
|
|
model = PostingLog
|
|
|
|
|
|
class PostingErrorRepository(TenantBaseRepository[PostingError]):
|
|
model = PostingError
|
|
|
|
|
|
class PostingReferenceRepository(TenantBaseRepository[PostingReference]):
|
|
model = PostingReference
|
|
|
|
async def get_by_source(
|
|
self,
|
|
tenant_id: UUID,
|
|
source_module: str,
|
|
source_document_type: str,
|
|
source_document_id: str,
|
|
) -> PostingReference | None:
|
|
stmt = select(PostingReference).where(
|
|
PostingReference.tenant_id == tenant_id,
|
|
PostingReference.source_module == source_module,
|
|
PostingReference.source_document_type == source_document_type,
|
|
PostingReference.source_document_id == source_document_id,
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
class AuditLogRepository(TenantBaseRepository[AccountingAuditLog]):
|
|
model = AccountingAuditLog
|