from datetime import datetime from decimal import Decimal from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.domain.entities.sales import ( Customer, SalesDelivery, SalesDeliveryLine, SalesInvoice, SalesInvoiceLine, SalesOrder, SalesOrderLine, SalesPayment, SalesPaymentAllocation, ) from app.domain.enums import ( CustomerBusinessType, PaymentTerms, PreferredPaymentMethod, SalesDeliveryStatus, SalesInvoiceStatus, SalesOrderStatus, SalesPaymentStatus, ) from app.domain.repositories.sales_repository import SalesRepository from app.infrastructure.db.models import ( CustomerModel, ItemModel, SalesDeliveryLineModel, SalesDeliveryModel, SalesInvoiceLineModel, SalesInvoiceModel, SalesOrderLineModel, SalesOrderModel, SalesPaymentAllocationModel, SalesPaymentModel, WarehouseModel, new_id, ) from app.infrastructure.repositories.mysql_utils import to_decimal class MySQLSalesRepository(SalesRepository): def __init__(self, session: AsyncSession) -> None: self._session = session # ---- customers ---- def _to_customer( self, row: CustomerModel, warehouse_code: str | None = None, warehouse_name: str | None = None, ) -> Customer: return Customer( id=row.id, company_id=row.company_id, code=row.code, name=row.name, business_type=CustomerBusinessType(row.business_type) if row.business_type else None, contact_person=row.contact_person or "", designation=row.designation, phone=row.phone or "", alternate_phone=row.alternate_phone, email=row.email, website=row.website, ntn=row.ntn, sales_tax_no=row.sales_tax_no, address_line1=row.address_line1 or row.address or "", address_line2=row.address_line2, city=row.city or "", state_province=row.state_province, country=row.country or "Pakistan", postal_code=row.postal_code, notes=row.notes, payment_terms=PaymentTerms(row.payment_terms) if row.payment_terms else None, payment_terms_days=row.payment_terms_days, credit_limit=to_decimal(row.credit_limit), opening_balance=to_decimal(row.opening_balance), currency=row.currency or "PKR", so_prefix=row.so_prefix, default_warehouse_id=row.default_warehouse_id, default_warehouse_code=warehouse_code, default_warehouse_name=warehouse_name, preferred_payment_method=( PreferredPaymentMethod(row.preferred_payment_method) if row.preferred_payment_method else None ), logo=row.logo, address=row.address, account_id=row.account_id, is_active=row.is_active, created_at=row.created_at, updated_at=row.updated_at, ) def _customer_select_query(self): return ( select( CustomerModel, WarehouseModel.code.label("joined_warehouse_code"), WarehouseModel.name.label("joined_warehouse_name"), ) .outerjoin( WarehouseModel, WarehouseModel.id == CustomerModel.default_warehouse_id, ) ) def _customer_from_row(self, row) -> Customer: return self._to_customer( row.CustomerModel, warehouse_code=row.joined_warehouse_code, warehouse_name=row.joined_warehouse_name, ) async def create_customer(self, customer: Customer) -> Customer: row = CustomerModel( id=customer.id or new_id(), company_id=customer.company_id, code=customer.code, name=customer.name, business_type=customer.business_type.value if customer.business_type else None, contact_person=customer.contact_person, designation=customer.designation, phone=customer.phone, alternate_phone=customer.alternate_phone, email=customer.email, website=customer.website, ntn=customer.ntn, sales_tax_no=customer.sales_tax_no, address_line1=customer.address_line1, address_line2=customer.address_line2, city=customer.city, state_province=customer.state_province, country=customer.country, postal_code=customer.postal_code, notes=customer.notes, payment_terms=customer.payment_terms.value if customer.payment_terms else None, payment_terms_days=customer.payment_terms_days, credit_limit=customer.credit_limit, opening_balance=customer.opening_balance, currency=customer.currency, so_prefix=customer.so_prefix, default_warehouse_id=customer.default_warehouse_id, preferred_payment_method=( customer.preferred_payment_method.value if customer.preferred_payment_method else None ), logo=customer.logo, address=customer.address or customer.address_line1, account_id=customer.account_id, is_active=customer.is_active, created_at=customer.created_at, updated_at=customer.updated_at, ) self._session.add(row) await self._session.commit() return await self.get_customer(row.id, customer.company_id) or self._to_customer(row) async def get_customer(self, customer_id: str, company_id: str) -> Customer | None: result = await self._session.execute( self._customer_select_query().where( CustomerModel.id == customer_id, CustomerModel.company_id == company_id, ) ) row = result.first() return self._customer_from_row(row) if row else None async def get_customer_by_code(self, company_id: str, code: str) -> Customer | None: result = await self._session.execute( self._customer_select_query().where( CustomerModel.company_id == company_id, CustomerModel.code == code, ) ) row = result.first() return self._customer_from_row(row) if row else None async def get_next_customer_code(self, company_id: str) -> str: result = await self._session.execute( select(CustomerModel.code).where( CustomerModel.company_id == company_id, CustomerModel.code.like("CUS-%"), ) ) max_num = 0 for (code,) in result.all(): suffix = code.rsplit("-", 1)[-1] if suffix.isdigit(): max_num = max(max_num, int(suffix)) return f"CUS-{max_num + 1:05d}" async def list_customers( self, company_id: str, is_active: bool | None = None, skip: int = 0, limit: int = 100 ) -> list[Customer]: query = self._customer_select_query().where(CustomerModel.company_id == company_id) if is_active is not None: query = query.where(CustomerModel.is_active == is_active) result = await self._session.execute( query.order_by(CustomerModel.code.asc()).offset(skip).limit(limit) ) return [self._customer_from_row(row) for row in result.all()] async def count_customers(self, company_id: str, is_active: bool | None = None) -> int: query = select(func.count()).select_from(CustomerModel).where( CustomerModel.company_id == company_id ) if is_active is not None: query = query.where(CustomerModel.is_active == is_active) result = await self._session.execute(query) return int(result.scalar_one()) async def update_customer(self, customer_id: str, customer: Customer) -> Customer | None: row = await self._session.get(CustomerModel, customer_id) if not row: return None row.code = customer.code row.name = customer.name row.business_type = customer.business_type.value if customer.business_type else None row.contact_person = customer.contact_person row.designation = customer.designation row.phone = customer.phone row.alternate_phone = customer.alternate_phone row.email = customer.email row.website = customer.website row.ntn = customer.ntn row.sales_tax_no = customer.sales_tax_no row.address_line1 = customer.address_line1 row.address_line2 = customer.address_line2 row.city = customer.city row.state_province = customer.state_province row.country = customer.country row.postal_code = customer.postal_code row.notes = customer.notes row.payment_terms = customer.payment_terms.value if customer.payment_terms else None row.payment_terms_days = customer.payment_terms_days row.credit_limit = customer.credit_limit row.opening_balance = customer.opening_balance row.currency = customer.currency row.so_prefix = customer.so_prefix row.default_warehouse_id = customer.default_warehouse_id row.preferred_payment_method = ( customer.preferred_payment_method.value if customer.preferred_payment_method else None ) row.logo = customer.logo row.address = customer.address or customer.address_line1 row.account_id = customer.account_id row.is_active = customer.is_active row.updated_at = datetime.utcnow() await self._session.commit() return await self.get_customer(customer_id, customer.company_id) async def delete_customer(self, customer_id: str) -> bool: row = await self._session.get(CustomerModel, customer_id) if not row: return False await self._session.delete(row) await self._session.commit() return True # ---- helpers ---- async def _next_doc_number( self, company_id: str, model, company_col, number_col, prefix: str, ) -> str: year = datetime.utcnow().year pattern = f"{prefix}-{year}-%" result = await self._session.execute( select(number_col) .where(company_col == company_id, number_col.like(pattern)) .order_by(number_col.desc()) .limit(1) ) last = result.scalar_one_or_none() seq = 1 if last: try: seq = int(str(last).rsplit("-", 1)[-1]) + 1 except ValueError: seq = 1 return f"{prefix}-{year}-{seq:05d}" def _so_line_model(self, order_id: str, line: SalesOrderLine) -> SalesOrderLineModel: return SalesOrderLineModel( id=line.id or new_id(), sales_order_id=order_id, line_number=line.line_number, item_id=line.item_id, description=line.description, quantity=line.quantity, reserved_quantity=line.reserved_quantity, picked_quantity=line.picked_quantity, shipped_quantity=line.shipped_quantity, invoiced_quantity=line.invoiced_quantity, available_quantity=line.available_quantity, is_available=line.is_available, unit_price=line.unit_price, tax_rate=line.tax_rate, tax_amount=line.tax_amount, line_total=line.line_total, ) def _to_so_line( self, row: SalesOrderLineModel, *, item_sku: str | None = None, item_name: str | None = None, ) -> SalesOrderLine: return SalesOrderLine( id=row.id, sales_order_id=row.sales_order_id, line_number=row.line_number, item_id=row.item_id, item_sku=item_sku, item_name=item_name, description=row.description, quantity=to_decimal(row.quantity), reserved_quantity=to_decimal(row.reserved_quantity), picked_quantity=to_decimal(row.picked_quantity), shipped_quantity=to_decimal(row.shipped_quantity), invoiced_quantity=to_decimal(row.invoiced_quantity), available_quantity=to_decimal(row.available_quantity), is_available=bool(row.is_available), unit_price=to_decimal(row.unit_price), tax_rate=to_decimal(row.tax_rate), tax_amount=to_decimal(row.tax_amount), line_total=to_decimal(row.line_total), ) async def _load_so_line_meta(self, lines: list[SalesOrderLineModel]) -> dict[str, dict]: if not lines: return {} item_ids = {line.item_id for line in lines} items: dict[str, ItemModel] = {} if item_ids: result = await self._session.execute( select(ItemModel).where(ItemModel.id.in_(item_ids)) ) items = {row.id: row for row in result.scalars().all()} meta: dict[str, dict] = {} for line in lines: item = items.get(line.item_id) meta[line.id or ""] = { "item_sku": item.sku if item else None, "item_name": item.name if item else None, } return meta def _to_so( self, row: SalesOrderModel, *, customer_code: str | None = None, customer_name: str | None = None, customer_email: str | None = None, customer_phone: str | None = None, warehouse_code: str | None = None, warehouse_name: str | None = None, line_meta: dict[str, dict] | None = None, ) -> SalesOrder: meta = line_meta or {} lines = [ self._to_so_line( line, item_sku=meta.get(line.id or "", {}).get("item_sku"), item_name=meta.get(line.id or "", {}).get("item_name"), ) for line in (row.lines or []) ] return SalesOrder( id=row.id, company_id=row.company_id, so_number=row.so_number, customer_id=row.customer_id, customer_code=customer_code, customer_name=customer_name, customer_email=customer_email, customer_phone=customer_phone, warehouse_id=row.warehouse_id, warehouse_code=warehouse_code, warehouse_name=warehouse_name, order_date=row.order_date, delivery_date=row.delivery_date, delivery_address=row.delivery_address, delivery_contact=row.delivery_contact, delivery_phone=row.delivery_phone, payment_terms=PaymentTerms(row.payment_terms) if row.payment_terms else None, currency=row.currency, notes=row.notes, status=SalesOrderStatus(row.status), subtotal=to_decimal(row.subtotal), tax_amount=to_decimal(row.tax_amount), total_amount=to_decimal(row.total_amount), created_by=row.created_by, confirmed_at=row.confirmed_at, cancelled_at=row.cancelled_at, lines=lines, created_at=row.created_at, updated_at=row.updated_at, ) async def _hydrate_so(self, row: SalesOrderModel) -> SalesOrder: customer = await self._session.get(CustomerModel, row.customer_id) warehouse = ( await self._session.get(WarehouseModel, row.warehouse_id) if row.warehouse_id else None ) line_meta = await self._load_so_line_meta(list(row.lines or [])) return self._to_so( row, customer_code=customer.code if customer else None, customer_name=customer.name if customer else None, customer_email=customer.email if customer else None, customer_phone=customer.phone if customer else None, warehouse_code=warehouse.code if warehouse else None, warehouse_name=warehouse.name if warehouse else None, line_meta=line_meta, ) # ---- sales orders ---- async def create_sales_order(self, order: SalesOrder) -> SalesOrder: order_id = order.id or new_id() row = SalesOrderModel( id=order_id, company_id=order.company_id, so_number=order.so_number, customer_id=order.customer_id, warehouse_id=order.warehouse_id, order_date=order.order_date, delivery_date=order.delivery_date, delivery_address=order.delivery_address, delivery_contact=order.delivery_contact, delivery_phone=order.delivery_phone, payment_terms=order.payment_terms.value if order.payment_terms else None, currency=order.currency, notes=order.notes, status=order.status.value, subtotal=order.subtotal, tax_amount=order.tax_amount, total_amount=order.total_amount, created_by=order.created_by, confirmed_at=order.confirmed_at, cancelled_at=order.cancelled_at, created_at=order.created_at, updated_at=order.updated_at, lines=[self._so_line_model(order_id, line) for line in order.lines], ) self._session.add(row) await self._session.commit() return await self.get_sales_order(order_id, order.company_id) # type: ignore[return-value] async def get_sales_order(self, order_id: str, company_id: str) -> SalesOrder | None: result = await self._session.execute( select(SalesOrderModel) .options(selectinload(SalesOrderModel.lines)) .where( SalesOrderModel.id == order_id, SalesOrderModel.company_id == company_id, ) ) row = result.scalar_one_or_none() return await self._hydrate_so(row) if row else None async def list_sales_orders( self, company_id: str, status: SalesOrderStatus | None = None, customer_id: str | None = None, skip: int = 0, limit: int = 100, ) -> list[SalesOrder]: query = ( select(SalesOrderModel) .options(selectinload(SalesOrderModel.lines)) .where(SalesOrderModel.company_id == company_id) ) if status: query = query.where(SalesOrderModel.status == status.value) if customer_id: query = query.where(SalesOrderModel.customer_id == customer_id) result = await self._session.execute( query.order_by(SalesOrderModel.order_date.desc()).offset(skip).limit(limit) ) return [await self._hydrate_so(row) for row in result.scalars().all()] async def count_sales_orders( self, company_id: str, status: SalesOrderStatus | None = None, customer_id: str | None = None, ) -> int: query = select(func.count()).select_from(SalesOrderModel).where( SalesOrderModel.company_id == company_id ) if status: query = query.where(SalesOrderModel.status == status.value) if customer_id: query = query.where(SalesOrderModel.customer_id == customer_id) result = await self._session.execute(query) return int(result.scalar_one()) async def update_sales_order( self, order_id: str, order: SalesOrder, *, replace_lines: bool = False ) -> SalesOrder | None: result = await self._session.execute( select(SalesOrderModel) .options(selectinload(SalesOrderModel.lines)) .where(SalesOrderModel.id == order_id) ) row = result.scalar_one_or_none() if not row: return None row.customer_id = order.customer_id row.warehouse_id = order.warehouse_id row.order_date = order.order_date row.delivery_date = order.delivery_date row.delivery_address = order.delivery_address row.delivery_contact = order.delivery_contact row.delivery_phone = order.delivery_phone row.payment_terms = order.payment_terms.value if order.payment_terms else None row.currency = order.currency row.notes = order.notes row.status = order.status.value row.subtotal = order.subtotal row.tax_amount = order.tax_amount row.total_amount = order.total_amount row.created_by = order.created_by row.confirmed_at = order.confirmed_at row.cancelled_at = order.cancelled_at row.updated_at = datetime.utcnow() if replace_lines: row.lines.clear() await self._session.flush() for line in order.lines: row.lines.append(self._so_line_model(order_id, line)) else: # sync mutable qty/flag fields on existing lines when present by_id = {line.id: line for line in order.lines if line.id} for existing in row.lines or []: src = by_id.get(existing.id) if not src: continue existing.reserved_quantity = src.reserved_quantity existing.picked_quantity = src.picked_quantity existing.shipped_quantity = src.shipped_quantity existing.invoiced_quantity = src.invoiced_quantity existing.available_quantity = src.available_quantity existing.is_available = src.is_available await self._session.commit() return await self.get_sales_order(order_id, order.company_id) async def update_so_line_quantities( self, line_id: str, *, reserved_quantity: Decimal | None = None, picked_quantity: Decimal | None = None, shipped_quantity: Decimal | None = None, invoiced_quantity: Decimal | None = None, available_quantity: Decimal | None = None, is_available: bool | None = None, ) -> None: row = await self._session.get(SalesOrderLineModel, line_id) if not row: return if reserved_quantity is not None: row.reserved_quantity = reserved_quantity if picked_quantity is not None: row.picked_quantity = picked_quantity if shipped_quantity is not None: row.shipped_quantity = shipped_quantity if invoiced_quantity is not None: row.invoiced_quantity = invoiced_quantity if available_quantity is not None: row.available_quantity = available_quantity if is_available is not None: row.is_available = is_available await self._session.commit() async def get_next_so_number(self, company_id: str) -> str: return await self._next_doc_number( company_id, SalesOrderModel, SalesOrderModel.company_id, SalesOrderModel.so_number, "SO", ) # ---- deliveries ---- def _to_delivery_line(self, row: SalesDeliveryLineModel) -> SalesDeliveryLine: return SalesDeliveryLine( id=row.id, sales_delivery_id=row.sales_delivery_id, sales_order_line_id=row.sales_order_line_id, line_number=row.line_number, item_id=row.item_id, quantity=to_decimal(row.quantity), notes=row.notes, ) def _to_delivery(self, row: SalesDeliveryModel) -> SalesDelivery: return SalesDelivery( id=row.id, company_id=row.company_id, delivery_number=row.delivery_number, sales_order_id=row.sales_order_id, customer_id=row.customer_id, delivery_date=row.delivery_date, status=SalesDeliveryStatus(row.status), notes=row.notes, confirmed_at=row.confirmed_at, lines=[self._to_delivery_line(line) for line in (row.lines or [])], created_at=row.created_at, updated_at=row.updated_at, ) async def create_delivery(self, delivery: SalesDelivery) -> SalesDelivery: delivery_id = delivery.id or new_id() row = SalesDeliveryModel( id=delivery_id, company_id=delivery.company_id, delivery_number=delivery.delivery_number, sales_order_id=delivery.sales_order_id, customer_id=delivery.customer_id, delivery_date=delivery.delivery_date, status=delivery.status.value, notes=delivery.notes, confirmed_at=delivery.confirmed_at, created_at=delivery.created_at, updated_at=delivery.updated_at, lines=[ SalesDeliveryLineModel( id=line.id or new_id(), sales_delivery_id=delivery_id, sales_order_line_id=line.sales_order_line_id, line_number=line.line_number, item_id=line.item_id, quantity=line.quantity, notes=line.notes, ) for line in delivery.lines ], ) self._session.add(row) await self._session.commit() return await self.get_delivery(delivery_id, delivery.company_id) # type: ignore[return-value] async def get_delivery(self, delivery_id: str, company_id: str) -> SalesDelivery | None: result = await self._session.execute( select(SalesDeliveryModel) .options(selectinload(SalesDeliveryModel.lines)) .where( SalesDeliveryModel.id == delivery_id, SalesDeliveryModel.company_id == company_id, ) ) row = result.scalar_one_or_none() return self._to_delivery(row) if row else None async def list_deliveries( self, company_id: str, status: SalesDeliveryStatus | None = None, sales_order_id: str | None = None, skip: int = 0, limit: int = 100, ) -> list[SalesDelivery]: query = ( select(SalesDeliveryModel) .options(selectinload(SalesDeliveryModel.lines)) .where(SalesDeliveryModel.company_id == company_id) ) if status: query = query.where(SalesDeliveryModel.status == status.value) if sales_order_id: query = query.where(SalesDeliveryModel.sales_order_id == sales_order_id) result = await self._session.execute( query.order_by(SalesDeliveryModel.delivery_date.desc()).offset(skip).limit(limit) ) return [self._to_delivery(row) for row in result.scalars().all()] async def count_deliveries( self, company_id: str, status: SalesDeliveryStatus | None = None, sales_order_id: str | None = None, ) -> int: query = select(func.count()).select_from(SalesDeliveryModel).where( SalesDeliveryModel.company_id == company_id ) if status: query = query.where(SalesDeliveryModel.status == status.value) if sales_order_id: query = query.where(SalesDeliveryModel.sales_order_id == sales_order_id) result = await self._session.execute(query) return int(result.scalar_one()) async def update_delivery( self, delivery_id: str, delivery: SalesDelivery ) -> SalesDelivery | None: result = await self._session.execute( select(SalesDeliveryModel) .options(selectinload(SalesDeliveryModel.lines)) .where(SalesDeliveryModel.id == delivery_id) ) row = result.scalar_one_or_none() if not row: return None row.status = delivery.status.value row.notes = delivery.notes row.confirmed_at = delivery.confirmed_at row.updated_at = datetime.utcnow() await self._session.commit() return await self.get_delivery(delivery_id, delivery.company_id) async def get_next_delivery_number(self, company_id: str) -> str: return await self._next_doc_number( company_id, SalesDeliveryModel, SalesDeliveryModel.company_id, SalesDeliveryModel.delivery_number, "DLV", ) # ---- invoices ---- def _to_invoice_line(self, row: SalesInvoiceLineModel) -> SalesInvoiceLine: return SalesInvoiceLine( id=row.id, sales_invoice_id=row.sales_invoice_id, line_number=row.line_number, item_id=row.item_id, account_id=row.account_id, description=row.description, quantity=to_decimal(row.quantity), unit_price=to_decimal(row.unit_price), tax_rate=to_decimal(row.tax_rate), line_total=to_decimal(row.line_total), sales_order_line_id=row.sales_order_line_id, sales_delivery_line_id=row.sales_delivery_line_id, ) def _to_invoice(self, row: SalesInvoiceModel) -> SalesInvoice: return SalesInvoice( id=row.id, company_id=row.company_id, invoice_number=row.invoice_number, customer_id=row.customer_id, sales_order_id=row.sales_order_id, delivery_id=row.delivery_id, invoice_date=row.invoice_date, due_date=row.due_date, status=SalesInvoiceStatus(row.status), ar_account_id=row.ar_account_id, notes=row.notes, subtotal=to_decimal(row.subtotal), tax_amount=to_decimal(row.tax_amount), total_amount=to_decimal(row.total_amount), amount_paid=to_decimal(row.amount_paid), voucher_id=row.voucher_id, posted_at=row.posted_at, lines=[self._to_invoice_line(line) for line in (row.lines or [])], created_at=row.created_at, updated_at=row.updated_at, ) async def create_invoice(self, invoice: SalesInvoice) -> SalesInvoice: invoice_id = invoice.id or new_id() row = SalesInvoiceModel( id=invoice_id, company_id=invoice.company_id, invoice_number=invoice.invoice_number, customer_id=invoice.customer_id, sales_order_id=invoice.sales_order_id, delivery_id=invoice.delivery_id, invoice_date=invoice.invoice_date, due_date=invoice.due_date, status=invoice.status.value, ar_account_id=invoice.ar_account_id, notes=invoice.notes, subtotal=invoice.subtotal, tax_amount=invoice.tax_amount, total_amount=invoice.total_amount, amount_paid=invoice.amount_paid, voucher_id=invoice.voucher_id, posted_at=invoice.posted_at, created_at=invoice.created_at, updated_at=invoice.updated_at, lines=[ SalesInvoiceLineModel( id=line.id or new_id(), sales_invoice_id=invoice_id, line_number=line.line_number, item_id=line.item_id, account_id=line.account_id, description=line.description, quantity=line.quantity, unit_price=line.unit_price, tax_rate=line.tax_rate, line_total=line.line_total, sales_order_line_id=line.sales_order_line_id, sales_delivery_line_id=line.sales_delivery_line_id, ) for line in invoice.lines ], ) self._session.add(row) await self._session.commit() return await self.get_invoice(invoice_id, invoice.company_id) # type: ignore[return-value] async def get_invoice(self, invoice_id: str, company_id: str) -> SalesInvoice | None: result = await self._session.execute( select(SalesInvoiceModel) .options(selectinload(SalesInvoiceModel.lines)) .where( SalesInvoiceModel.id == invoice_id, SalesInvoiceModel.company_id == company_id, ) ) row = result.scalar_one_or_none() return self._to_invoice(row) if row else None async def list_invoices( self, company_id: str, status: SalesInvoiceStatus | None = None, customer_id: str | None = None, skip: int = 0, limit: int = 100, ) -> list[SalesInvoice]: query = ( select(SalesInvoiceModel) .options(selectinload(SalesInvoiceModel.lines)) .where(SalesInvoiceModel.company_id == company_id) ) if status: query = query.where(SalesInvoiceModel.status == status.value) if customer_id: query = query.where(SalesInvoiceModel.customer_id == customer_id) result = await self._session.execute( query.order_by(SalesInvoiceModel.invoice_date.desc()).offset(skip).limit(limit) ) return [self._to_invoice(row) for row in result.scalars().all()] async def count_invoices( self, company_id: str, status: SalesInvoiceStatus | None = None, customer_id: str | None = None, ) -> int: query = select(func.count()).select_from(SalesInvoiceModel).where( SalesInvoiceModel.company_id == company_id ) if status: query = query.where(SalesInvoiceModel.status == status.value) if customer_id: query = query.where(SalesInvoiceModel.customer_id == customer_id) result = await self._session.execute(query) return int(result.scalar_one()) async def update_invoice(self, invoice_id: str, invoice: SalesInvoice) -> SalesInvoice | None: result = await self._session.execute( select(SalesInvoiceModel) .options(selectinload(SalesInvoiceModel.lines)) .where(SalesInvoiceModel.id == invoice_id) ) row = result.scalar_one_or_none() if not row: return None row.status = invoice.status.value row.ar_account_id = invoice.ar_account_id row.notes = invoice.notes row.amount_paid = invoice.amount_paid row.voucher_id = invoice.voucher_id row.posted_at = invoice.posted_at row.updated_at = datetime.utcnow() await self._session.commit() return await self.get_invoice(invoice_id, invoice.company_id) async def get_next_invoice_number(self, company_id: str) -> str: return await self._next_doc_number( company_id, SalesInvoiceModel, SalesInvoiceModel.company_id, SalesInvoiceModel.invoice_number, "INV", ) # ---- payments ---- def _to_payment_alloc(self, row: SalesPaymentAllocationModel) -> SalesPaymentAllocation: return SalesPaymentAllocation( id=row.id, sales_payment_id=row.sales_payment_id, sales_invoice_id=row.sales_invoice_id, amount=to_decimal(row.amount), ) def _to_payment(self, row: SalesPaymentModel) -> SalesPayment: return SalesPayment( id=row.id, company_id=row.company_id, payment_number=row.payment_number, customer_id=row.customer_id, payment_date=row.payment_date, status=SalesPaymentStatus(row.status), payment_method=row.payment_method, bank_account_id=row.bank_account_id, ar_account_id=row.ar_account_id, reference=row.reference, notes=row.notes, total_amount=to_decimal(row.total_amount), voucher_id=row.voucher_id, posted_at=row.posted_at, allocations=[self._to_payment_alloc(a) for a in (row.allocations or [])], created_at=row.created_at, updated_at=row.updated_at, ) async def create_payment(self, payment: SalesPayment) -> SalesPayment: payment_id = payment.id or new_id() row = SalesPaymentModel( id=payment_id, company_id=payment.company_id, payment_number=payment.payment_number, customer_id=payment.customer_id, payment_date=payment.payment_date, status=payment.status.value, payment_method=payment.payment_method, bank_account_id=payment.bank_account_id, ar_account_id=payment.ar_account_id, reference=payment.reference, notes=payment.notes, total_amount=payment.total_amount, voucher_id=payment.voucher_id, posted_at=payment.posted_at, created_at=payment.created_at, updated_at=payment.updated_at, allocations=[ SalesPaymentAllocationModel( id=alloc.id or new_id(), sales_payment_id=payment_id, sales_invoice_id=alloc.sales_invoice_id, amount=alloc.amount, ) for alloc in payment.allocations ], ) self._session.add(row) await self._session.commit() return await self.get_payment(payment_id, payment.company_id) # type: ignore[return-value] async def get_payment(self, payment_id: str, company_id: str) -> SalesPayment | None: result = await self._session.execute( select(SalesPaymentModel) .options(selectinload(SalesPaymentModel.allocations)) .where( SalesPaymentModel.id == payment_id, SalesPaymentModel.company_id == company_id, ) ) row = result.scalar_one_or_none() return self._to_payment(row) if row else None async def list_payments( self, company_id: str, status: SalesPaymentStatus | None = None, customer_id: str | None = None, skip: int = 0, limit: int = 100, ) -> list[SalesPayment]: query = ( select(SalesPaymentModel) .options(selectinload(SalesPaymentModel.allocations)) .where(SalesPaymentModel.company_id == company_id) ) if status: query = query.where(SalesPaymentModel.status == status.value) if customer_id: query = query.where(SalesPaymentModel.customer_id == customer_id) result = await self._session.execute( query.order_by(SalesPaymentModel.payment_date.desc()).offset(skip).limit(limit) ) return [self._to_payment(row) for row in result.scalars().all()] async def count_payments( self, company_id: str, status: SalesPaymentStatus | None = None, customer_id: str | None = None, ) -> int: query = select(func.count()).select_from(SalesPaymentModel).where( SalesPaymentModel.company_id == company_id ) if status: query = query.where(SalesPaymentModel.status == status.value) if customer_id: query = query.where(SalesPaymentModel.customer_id == customer_id) result = await self._session.execute(query) return int(result.scalar_one()) async def update_payment( self, payment_id: str, payment: SalesPayment ) -> SalesPayment | None: result = await self._session.execute( select(SalesPaymentModel) .options(selectinload(SalesPaymentModel.allocations)) .where(SalesPaymentModel.id == payment_id) ) row = result.scalar_one_or_none() if not row: return None row.status = payment.status.value row.bank_account_id = payment.bank_account_id row.ar_account_id = payment.ar_account_id row.voucher_id = payment.voucher_id row.posted_at = payment.posted_at row.updated_at = datetime.utcnow() await self._session.commit() return await self.get_payment(payment_id, payment.company_id) async def get_next_payment_number(self, company_id: str) -> str: return await self._next_doc_number( company_id, SalesPaymentModel, SalesPaymentModel.company_id, SalesPaymentModel.payment_number, "RCP", )