from datetime import date, datetime from decimal import Decimal from math import ceil from sqlalchemy import and_, case, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.domain.entities.inventory import ( BaseUnit, Brand, DashboardBreakdownSlice, Department, DepartmentIssue, DepartmentIssueLine, DepartmentIssueSummary, InventoryBalance, InventoryTransaction, Item, ItemCategory, ItemGroup, ItemTransaction, ItemTransactionLine, ItemType, Location, StockReport, StockReportRow, StockStatusSlice, StockTopItem, StockTransfer, StockTransferLine, UnitType, Warehouse, ) from app.domain.enums import ( AmountType, DepartmentIssuePriority, DepartmentIssueReason, DepartmentIssueStatus, DepartmentIssueType, DepartmentStatus, InventoryTxnDirection, InventoryTxnType, ItemTransactionReferenceType, ItemTransactionStatus, LowStockAlertTarget, PricingModel, StockReportStatus, StockTransferPriority, StockTransferReason, StockTransferStatus, UnitKind, WarehousePriority, WarehouseStatus, WarehouseType, ) from app.domain.repositories.inventory_repository import InventoryRepository from app.infrastructure.db.models import ( BaseUnitModel, BrandModel, CompanyUserModel, DepartmentIssueLineModel, DepartmentIssueModel, DepartmentModel, InventoryBalanceModel, InventoryTransactionModel, ItemCategoryModel, ItemGroupModel, ItemModel, ItemTransactionLineModel, ItemTransactionModel, ItemTypeModel, LocationModel, StockTransferLineModel, StockTransferModel, UnitTypeModel, UserRegistrationModel, VendorModel, WarehouseModel, new_id, ) from app.infrastructure.repositories.mysql_utils import to_decimal class MySQLInventoryRepository(InventoryRepository): def __init__(self, session: AsyncSession) -> None: self._session = session # ---- mappers ---- def _to_category(self, row: ItemCategoryModel) -> ItemCategory: return ItemCategory( id=row.id, company_id=row.company_id, code=row.code, name=row.name, description=row.description, is_active=row.is_active, created_at=row.created_at, updated_at=row.updated_at, ) def _to_base_unit(self, row: BaseUnitModel) -> BaseUnit: return BaseUnit( id=row.id, company_id=row.company_id, code=row.code, name=row.name, description=row.description, is_active=row.is_active, sort_order=row.sort_order, created_at=row.created_at, updated_at=row.updated_at, ) def _to_warehouse( self, row: WarehouseModel, manager_name: str | None = None, parent_warehouse_code: str | None = None, parent_warehouse_name: str | None = None, ) -> Warehouse: return Warehouse( id=row.id, company_id=row.company_id, code=row.code, name=row.name, warehouse_type=WarehouseType(row.warehouse_type), status=WarehouseStatus(row.status or WarehouseStatus.ACTIVE.value), priority=WarehousePriority(row.priority or WarehousePriority.NORMAL.value), parent_warehouse_id=row.parent_warehouse_id, parent_warehouse_code=parent_warehouse_code, parent_warehouse_name=parent_warehouse_name, manager_id=row.manager_id, manager_name=manager_name, contact_person=row.contact_person, phone=row.phone, email=row.email, address_line1=row.address_line1, address_line2=row.address_line2, country=row.country, state_province=row.state_province, city=row.city, postal_code=row.postal_code, latitude=row.latitude, longitude=row.longitude, address=row.address or "", location=row.location or "", is_active=row.is_active, description=row.description, capacity=row.capacity, capacity_uom=row.capacity_uom, operating_hours=row.operating_hours, notes=row.notes, remarks=row.remarks, allow_stock_in=bool(row.allow_stock_in), allow_stock_out=bool(row.allow_stock_out), allow_stock_transfer=bool(row.allow_stock_transfer), allow_returns=bool(row.allow_returns), attachments=self._parse_json_list(row.attachments), created_at=row.created_at, updated_at=row.updated_at, ) async def _attach_warehouse_relations(self, row: WarehouseModel) -> Warehouse: manager_name = None if row.manager_id: manager = await self._session.get(UserRegistrationModel, row.manager_id) if manager: manager_name = manager.full_name parent_code = None parent_name = None if row.parent_warehouse_id: parent = await self._session.get(WarehouseModel, row.parent_warehouse_id) if parent: parent_code = parent.code parent_name = parent.name return self._to_warehouse( row, manager_name=manager_name, parent_warehouse_code=parent_code, parent_warehouse_name=parent_name, ) def _to_brand(self, row: BrandModel) -> Brand: return Brand( id=row.id, company_id=row.company_id, code=row.code, name=row.name, description=row.description, logo=row.logo, website=row.website, is_active=row.is_active, contact_person=row.contact_person, email=row.email, phone=row.phone, address=row.address, created_at=row.created_at, updated_at=row.updated_at, ) def _to_item_type( self, row: ItemTypeModel, category_code: str | None = None, category_name: str | None = None, ) -> ItemType: return ItemType( id=row.id, company_id=row.company_id, category_id=row.category_id, category_code=category_code, category_name=category_name, code=row.code, name=row.name, description=row.description, is_active=row.is_active, created_at=row.created_at, updated_at=row.updated_at, ) async def _attach_category(self, row: ItemTypeModel) -> ItemType: category_code = None category_name = None if row.category_id: category = await self._session.get(ItemCategoryModel, row.category_id) if category: category_code = category.code category_name = category.name return self._to_item_type(row, category_code, category_name) def _to_group( self, row: ItemGroupModel, category_code: str | None = None, category_name: str | None = None, item_type_code: str | None = None, item_type_name: str | None = None, ) -> ItemGroup: return ItemGroup( id=row.id, company_id=row.company_id, category_id=row.category_id, category_code=category_code, category_name=category_name, item_type_id=row.item_type_id, item_type_code=item_type_code, item_type_name=item_type_name, code=row.code, name=row.name, description=row.description, is_active=row.is_active, sort_order=row.sort_order, icon=row.icon, remarks=row.remarks, created_at=row.created_at, updated_at=row.updated_at, ) async def _attach_group_relations(self, row: ItemGroupModel) -> ItemGroup: category_code = None category_name = None item_type_code = None item_type_name = None if row.category_id: category = await self._session.get(ItemCategoryModel, row.category_id) if category: category_code = category.code category_name = category.name if row.item_type_id: item_type = await self._session.get(ItemTypeModel, row.item_type_id) if item_type: item_type_code = item_type.code item_type_name = item_type.name return self._to_group( row, category_code, category_name, item_type_code, item_type_name ) def _to_unit_type( self, row: UnitTypeModel, base_unit_code: str | None = None, base_unit_name: str | None = None, ) -> UnitType: return UnitType( id=row.id, company_id=row.company_id, code=row.code, name=row.name, base_unit_id=row.base_unit_id, base_unit_code=base_unit_code, base_unit_name=base_unit_name, unit_kind=UnitKind(row.unit_kind), conversion_rate=to_decimal(row.conversion_rate), decimal_places=row.decimal_places, is_active=row.is_active, sort_order=row.sort_order, description=row.description, remarks=row.remarks, created_at=row.created_at, updated_at=row.updated_at, ) async def _attach_base_unit(self, row: UnitTypeModel) -> UnitType: base_unit_code = None base_unit_name = None if row.base_unit_id: base = await self._session.get(BaseUnitModel, row.base_unit_id) if base: base_unit_code = base.code base_unit_name = base.name return self._to_unit_type(row, base_unit_code, base_unit_name) def _to_item( self, row: ItemModel, category_code: str | None = None, category_name: str | None = None, item_type_code: str | None = None, item_type_name: str | None = None, group_code: str | None = None, group_name: str | None = None, base_unit_code: str | None = None, base_unit_name: str | None = None, warehouse_code: str | None = None, warehouse_name: str | None = None, ) -> Item: return Item( id=row.id, company_id=row.company_id, sku=row.sku, name=row.name, barcode=row.barcode, image=row.image, description=row.description, specifications=row.specifications, remarks=row.remarks, category_id=row.category_id, category_code=category_code, category_name=category_name, item_type_id=row.item_type_id, item_type_code=item_type_code, item_type_name=item_type_name, group_id=row.group_id, group_code=group_code, group_name=group_name, brand_name=row.brand_name, base_unit_id=row.base_unit_id, base_unit_code=base_unit_code, base_unit_name=base_unit_name, warehouse_id=row.warehouse_id, warehouse_code=warehouse_code, warehouse_name=warehouse_name, unit_type_id=row.unit_type_id, pricing_model=PricingModel(row.pricing_model), purchase_price=to_decimal(row.purchase_price), sale_price=to_decimal(row.sale_price), tax_percent=to_decimal(row.tax_percent), reorder_level=to_decimal(row.reorder_level), track_inventory=row.track_inventory, inventory_account_id=row.inventory_account_id, expense_account_id=row.expense_account_id, cogs_account_id=row.cogs_account_id, is_active=row.is_active, created_at=row.created_at, updated_at=row.updated_at, ) def _item_select_query(self): return ( select( ItemModel, ItemCategoryModel.code.label("joined_category_code"), ItemCategoryModel.name.label("joined_category_name"), ItemTypeModel.code.label("joined_item_type_code"), ItemTypeModel.name.label("joined_item_type_name"), ItemGroupModel.code.label("joined_group_code"), ItemGroupModel.name.label("joined_group_name"), BaseUnitModel.code.label("joined_base_unit_code"), BaseUnitModel.name.label("joined_base_unit_name"), WarehouseModel.code.label("joined_warehouse_code"), WarehouseModel.name.label("joined_warehouse_name"), ) .select_from(ItemModel) .join(ItemCategoryModel, ItemCategoryModel.id == ItemModel.category_id) .join(ItemTypeModel, ItemTypeModel.id == ItemModel.item_type_id) .join(ItemGroupModel, ItemGroupModel.id == ItemModel.group_id) .join(BaseUnitModel, BaseUnitModel.id == ItemModel.base_unit_id) .join(WarehouseModel, WarehouseModel.id == ItemModel.warehouse_id) ) def _item_from_row(self, row) -> Item: return self._to_item( row.ItemModel, category_code=row.joined_category_code, category_name=row.joined_category_name, item_type_code=row.joined_item_type_code, item_type_name=row.joined_item_type_name, group_code=row.joined_group_code, group_name=row.joined_group_name, base_unit_code=row.joined_base_unit_code, base_unit_name=row.joined_base_unit_name, warehouse_code=row.joined_warehouse_code, warehouse_name=row.joined_warehouse_name, ) def _to_balance(self, row: InventoryBalanceModel) -> InventoryBalance: return InventoryBalance( id=row.id, company_id=row.company_id, item_id=row.item_id, warehouse_id=row.warehouse_id, quantity_on_hand=to_decimal(row.quantity_on_hand), quantity_reserved=to_decimal(row.quantity_reserved), average_cost=to_decimal(row.average_cost), last_cost=to_decimal(row.last_cost), updated_at=row.updated_at, ) def _to_transaction(self, row: InventoryTransactionModel) -> InventoryTransaction: return InventoryTransaction( id=row.id, company_id=row.company_id, item_id=row.item_id, warehouse_id=row.warehouse_id, txn_type=InventoryTxnType(row.txn_type), txn_date=row.txn_date, quantity_in=to_decimal(row.quantity_in), quantity_out=to_decimal(row.quantity_out), unit_cost=to_decimal(row.unit_cost), total_cost=to_decimal(row.total_cost), balance_after=to_decimal(row.balance_after), reference_type=row.reference_type, reference_id=row.reference_id, reference_number=row.reference_number, notes=row.notes, created_at=row.created_at, ) # ---- Categories ---- async def create_category(self, category: ItemCategory) -> ItemCategory: row = ItemCategoryModel( id=category.id or new_id(), company_id=category.company_id, code=category.code, name=category.name, description=category.description, is_active=category.is_active, created_at=category.created_at, updated_at=category.updated_at, ) self._session.add(row) await self._session.commit() await self._session.refresh(row) return self._to_category(row) async def get_category(self, category_id: str, company_id: str) -> ItemCategory | None: result = await self._session.execute( select(ItemCategoryModel).where( ItemCategoryModel.id == category_id, ItemCategoryModel.company_id == company_id, ) ) row = result.scalar_one_or_none() return self._to_category(row) if row else None async def get_category_by_code(self, company_id: str, code: str) -> ItemCategory | None: result = await self._session.execute( select(ItemCategoryModel).where( ItemCategoryModel.company_id == company_id, ItemCategoryModel.code == code, ) ) row = result.scalar_one_or_none() return self._to_category(row) if row else None async def list_categories( self, company_id: str, is_active: bool | None = None, skip: int = 0, limit: int = 100 ) -> list[ItemCategory]: query = select(ItemCategoryModel).where(ItemCategoryModel.company_id == company_id) if is_active is not None: query = query.where(ItemCategoryModel.is_active == is_active) result = await self._session.execute( query.order_by(ItemCategoryModel.code.asc()).offset(skip).limit(limit) ) return [self._to_category(row) for row in result.scalars().all()] async def count_categories(self, company_id: str, is_active: bool | None = None) -> int: query = select(func.count()).select_from(ItemCategoryModel).where( ItemCategoryModel.company_id == company_id ) if is_active is not None: query = query.where(ItemCategoryModel.is_active == is_active) result = await self._session.execute(query) return int(result.scalar_one()) async def update_category(self, category_id: str, category: ItemCategory) -> ItemCategory | None: row = await self._session.get(ItemCategoryModel, category_id) if not row: return None row.code = category.code row.name = category.name row.description = category.description row.is_active = category.is_active row.updated_at = datetime.utcnow() await self._session.commit() await self._session.refresh(row) return self._to_category(row) async def delete_category(self, category_id: str) -> bool: row = await self._session.get(ItemCategoryModel, category_id) if not row: return False await self._session.delete(row) await self._session.commit() return True # ---- Base units ---- async def create_base_unit(self, base_unit: BaseUnit) -> BaseUnit: row = BaseUnitModel( id=base_unit.id or new_id(), company_id=base_unit.company_id, code=base_unit.code, name=base_unit.name, description=base_unit.description, is_active=base_unit.is_active, sort_order=base_unit.sort_order, created_at=base_unit.created_at, updated_at=base_unit.updated_at, ) self._session.add(row) await self._session.commit() await self._session.refresh(row) return self._to_base_unit(row) async def get_base_unit(self, base_unit_id: str, company_id: str) -> BaseUnit | None: result = await self._session.execute( select(BaseUnitModel).where( BaseUnitModel.id == base_unit_id, BaseUnitModel.company_id == company_id, ) ) row = result.scalar_one_or_none() return self._to_base_unit(row) if row else None async def get_base_unit_by_code(self, company_id: str, code: str) -> BaseUnit | None: result = await self._session.execute( select(BaseUnitModel).where( BaseUnitModel.company_id == company_id, BaseUnitModel.code == code, ) ) row = result.scalar_one_or_none() return self._to_base_unit(row) if row else None async def list_base_units( self, company_id: str, is_active: bool | None = None, skip: int = 0, limit: int = 100 ) -> list[BaseUnit]: query = select(BaseUnitModel).where(BaseUnitModel.company_id == company_id) if is_active is not None: query = query.where(BaseUnitModel.is_active == is_active) result = await self._session.execute( query.order_by( BaseUnitModel.sort_order.is_(None), BaseUnitModel.sort_order.asc(), BaseUnitModel.code.asc(), ) .offset(skip) .limit(limit) ) return [self._to_base_unit(row) for row in result.scalars().all()] async def count_base_units(self, company_id: str, is_active: bool | None = None) -> int: query = select(func.count()).select_from(BaseUnitModel).where( BaseUnitModel.company_id == company_id ) if is_active is not None: query = query.where(BaseUnitModel.is_active == is_active) result = await self._session.execute(query) return int(result.scalar_one()) async def update_base_unit(self, base_unit_id: str, base_unit: BaseUnit) -> BaseUnit | None: row = await self._session.get(BaseUnitModel, base_unit_id) if not row: return None row.code = base_unit.code row.name = base_unit.name row.description = base_unit.description row.is_active = base_unit.is_active row.sort_order = base_unit.sort_order row.updated_at = datetime.utcnow() await self._session.commit() await self._session.refresh(row) return self._to_base_unit(row) async def delete_base_unit(self, base_unit_id: str) -> bool: row = await self._session.get(BaseUnitModel, base_unit_id) if not row: return False await self._session.delete(row) await self._session.commit() return True # ---- Warehouses ---- async def user_exists(self, user_id: str) -> bool: row = await self._session.get(UserRegistrationModel, user_id) if row is not None: return True company_user = await self._session.get(CompanyUserModel, user_id) return company_user is not None async def create_warehouse(self, warehouse: Warehouse) -> Warehouse: row = WarehouseModel( id=warehouse.id or new_id(), company_id=warehouse.company_id, code=warehouse.code, name=warehouse.name, warehouse_type=warehouse.warehouse_type.value, status=warehouse.status.value, priority=warehouse.priority.value, parent_warehouse_id=warehouse.parent_warehouse_id, manager_id=warehouse.manager_id, contact_person=warehouse.contact_person, phone=warehouse.phone, email=warehouse.email, address_line1=warehouse.address_line1, address_line2=warehouse.address_line2, country=warehouse.country, state_province=warehouse.state_province, city=warehouse.city, postal_code=warehouse.postal_code, latitude=warehouse.latitude, longitude=warehouse.longitude, address=warehouse.address or warehouse.address_line1 or "", location=warehouse.location or warehouse.city or "", is_active=warehouse.is_active, description=warehouse.description, capacity=warehouse.capacity, capacity_uom=warehouse.capacity_uom, operating_hours=warehouse.operating_hours, notes=warehouse.notes, remarks=warehouse.remarks, allow_stock_in=warehouse.allow_stock_in, allow_stock_out=warehouse.allow_stock_out, allow_stock_transfer=warehouse.allow_stock_transfer, allow_returns=warehouse.allow_returns, attachments=self._dump_json_list(warehouse.attachments), created_at=warehouse.created_at, updated_at=warehouse.updated_at, ) self._session.add(row) await self._session.commit() await self._session.refresh(row) return await self._attach_warehouse_relations(row) async def get_warehouse(self, warehouse_id: str, company_id: str) -> Warehouse | None: result = await self._session.execute( select(WarehouseModel).where( WarehouseModel.id == warehouse_id, WarehouseModel.company_id == company_id, ) ) row = result.scalar_one_or_none() return await self._attach_warehouse_relations(row) if row else None async def get_warehouse_by_code(self, company_id: str, code: str) -> Warehouse | None: result = await self._session.execute( select(WarehouseModel).where( WarehouseModel.company_id == company_id, WarehouseModel.code == code, ) ) row = result.scalar_one_or_none() return await self._attach_warehouse_relations(row) if row else None async def list_warehouses( self, company_id: str, is_active: bool | None = None, warehouse_type: str | None = None, status: str | None = None, skip: int = 0, limit: int = 100, ) -> list[Warehouse]: parent = WarehouseModel.__table__.alias("parent_warehouse") query = ( select( WarehouseModel, UserRegistrationModel.full_name.label("joined_manager_name"), parent.c.code.label("joined_parent_code"), parent.c.name.label("joined_parent_name"), ) .select_from(WarehouseModel) .outerjoin( UserRegistrationModel, UserRegistrationModel.id == WarehouseModel.manager_id, ) .outerjoin(parent, parent.c.id == WarehouseModel.parent_warehouse_id) .where(WarehouseModel.company_id == company_id) ) if is_active is not None: query = query.where(WarehouseModel.is_active == is_active) if warehouse_type: query = query.where(WarehouseModel.warehouse_type == warehouse_type) if status: query = query.where(WarehouseModel.status == status) result = await self._session.execute( query.order_by(WarehouseModel.code.asc()).offset(skip).limit(limit) ) return [ self._to_warehouse( row.WarehouseModel, manager_name=row.joined_manager_name, parent_warehouse_code=row.joined_parent_code, parent_warehouse_name=row.joined_parent_name, ) for row in result.all() ] async def count_warehouses( self, company_id: str, is_active: bool | None = None, warehouse_type: str | None = None, status: str | None = None, ) -> int: query = select(func.count()).select_from(WarehouseModel).where( WarehouseModel.company_id == company_id ) if is_active is not None: query = query.where(WarehouseModel.is_active == is_active) if warehouse_type: query = query.where(WarehouseModel.warehouse_type == warehouse_type) if status: query = query.where(WarehouseModel.status == status) result = await self._session.execute(query) return int(result.scalar_one()) async def update_warehouse(self, warehouse_id: str, warehouse: Warehouse) -> Warehouse | None: row = await self._session.get(WarehouseModel, warehouse_id) if not row: return None row.code = warehouse.code row.name = warehouse.name row.warehouse_type = warehouse.warehouse_type.value row.status = warehouse.status.value row.priority = warehouse.priority.value row.parent_warehouse_id = warehouse.parent_warehouse_id row.manager_id = warehouse.manager_id row.contact_person = warehouse.contact_person row.phone = warehouse.phone row.email = warehouse.email row.address_line1 = warehouse.address_line1 row.address_line2 = warehouse.address_line2 row.country = warehouse.country row.state_province = warehouse.state_province row.city = warehouse.city row.postal_code = warehouse.postal_code row.latitude = warehouse.latitude row.longitude = warehouse.longitude row.address = warehouse.address or warehouse.address_line1 or "" row.location = warehouse.location or warehouse.city or "" row.is_active = warehouse.is_active row.description = warehouse.description row.capacity = warehouse.capacity row.capacity_uom = warehouse.capacity_uom row.operating_hours = warehouse.operating_hours row.notes = warehouse.notes row.remarks = warehouse.remarks row.allow_stock_in = warehouse.allow_stock_in row.allow_stock_out = warehouse.allow_stock_out row.allow_stock_transfer = warehouse.allow_stock_transfer row.allow_returns = warehouse.allow_returns row.attachments = self._dump_json_list(warehouse.attachments) row.updated_at = datetime.utcnow() await self._session.commit() await self._session.refresh(row) return await self._attach_warehouse_relations(row) async def delete_warehouse(self, warehouse_id: str) -> bool: row = await self._session.get(WarehouseModel, warehouse_id) if not row: return False await self._session.delete(row) await self._session.commit() return True # ---- Brands ---- async def create_brand(self, brand: Brand) -> Brand: row = BrandModel( id=brand.id or new_id(), company_id=brand.company_id, code=brand.code, name=brand.name, description=brand.description, logo=brand.logo, website=brand.website, is_active=brand.is_active, contact_person=brand.contact_person, email=brand.email, phone=brand.phone, address=brand.address, created_at=brand.created_at, updated_at=brand.updated_at, ) self._session.add(row) await self._session.commit() await self._session.refresh(row) return self._to_brand(row) async def get_brand(self, brand_id: str, company_id: str) -> Brand | None: result = await self._session.execute( select(BrandModel).where( BrandModel.id == brand_id, BrandModel.company_id == company_id, ) ) row = result.scalar_one_or_none() return self._to_brand(row) if row else None async def get_brand_by_code(self, company_id: str, code: str) -> Brand | None: result = await self._session.execute( select(BrandModel).where( BrandModel.company_id == company_id, BrandModel.code == code, ) ) row = result.scalar_one_or_none() return self._to_brand(row) if row else None async def get_next_brand_code(self, company_id: str) -> str: result = await self._session.execute( select(BrandModel.code).where( BrandModel.company_id == company_id, BrandModel.code.like("BR-%"), ) ) 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"BR-{max_num + 1:03d}" async def list_brands( self, company_id: str, is_active: bool | None = None, skip: int = 0, limit: int = 100 ) -> list[Brand]: query = select(BrandModel).where(BrandModel.company_id == company_id) if is_active is not None: query = query.where(BrandModel.is_active == is_active) result = await self._session.execute( query.order_by(BrandModel.code.asc()).offset(skip).limit(limit) ) return [self._to_brand(row) for row in result.scalars().all()] async def count_brands(self, company_id: str, is_active: bool | None = None) -> int: query = select(func.count()).select_from(BrandModel).where( BrandModel.company_id == company_id ) if is_active is not None: query = query.where(BrandModel.is_active == is_active) result = await self._session.execute(query) return int(result.scalar_one()) async def update_brand(self, brand_id: str, brand: Brand) -> Brand | None: row = await self._session.get(BrandModel, brand_id) if not row: return None row.code = brand.code row.name = brand.name row.description = brand.description row.logo = brand.logo row.website = brand.website row.is_active = brand.is_active row.contact_person = brand.contact_person row.email = brand.email row.phone = brand.phone row.address = brand.address row.updated_at = datetime.utcnow() await self._session.commit() await self._session.refresh(row) return self._to_brand(row) async def delete_brand(self, brand_id: str) -> bool: row = await self._session.get(BrandModel, brand_id) if not row: return False await self._session.delete(row) await self._session.commit() return True # ---- Item types ---- async def create_item_type(self, item_type: ItemType) -> ItemType: row = ItemTypeModel( id=item_type.id or new_id(), company_id=item_type.company_id, category_id=item_type.category_id, code=item_type.code, name=item_type.name, description=item_type.description, is_active=item_type.is_active, created_at=item_type.created_at, updated_at=item_type.updated_at, ) self._session.add(row) await self._session.commit() await self._session.refresh(row) return await self._attach_category(row) async def get_item_type(self, item_type_id: str, company_id: str) -> ItemType | None: result = await self._session.execute( select(ItemTypeModel).where( ItemTypeModel.id == item_type_id, ItemTypeModel.company_id == company_id, ) ) row = result.scalar_one_or_none() return await self._attach_category(row) if row else None async def get_item_type_by_code(self, company_id: str, code: str) -> ItemType | None: result = await self._session.execute( select(ItemTypeModel).where( ItemTypeModel.company_id == company_id, ItemTypeModel.code == code, ) ) row = result.scalar_one_or_none() return await self._attach_category(row) if row else None async def list_item_types( self, company_id: str, is_active: bool | None = None, category_id: str | None = None, skip: int = 0, limit: int = 100, ) -> list[ItemType]: query = ( select( ItemTypeModel, ItemCategoryModel.id.label("joined_category_id"), ItemCategoryModel.code.label("joined_category_code"), ItemCategoryModel.name.label("joined_category_name"), ) .select_from(ItemTypeModel) .outerjoin( ItemCategoryModel, ItemCategoryModel.id == ItemTypeModel.category_id, ) .where(ItemTypeModel.company_id == company_id) ) if is_active is not None: query = query.where(ItemTypeModel.is_active == is_active) if category_id: query = query.where(ItemTypeModel.category_id == category_id) result = await self._session.execute( query.order_by(ItemTypeModel.code.asc()).offset(skip).limit(limit) ) items: list[ItemType] = [] for row in result.all(): item_type_row = row[0] items.append( self._to_item_type( item_type_row, category_code=row.joined_category_code, category_name=row.joined_category_name, ) ) return items async def count_item_types( self, company_id: str, is_active: bool | None = None, category_id: str | None = None, ) -> int: query = select(func.count()).select_from(ItemTypeModel).where( ItemTypeModel.company_id == company_id ) if is_active is not None: query = query.where(ItemTypeModel.is_active == is_active) if category_id: query = query.where(ItemTypeModel.category_id == category_id) result = await self._session.execute(query) return int(result.scalar_one()) async def update_item_type(self, item_type_id: str, item_type: ItemType) -> ItemType | None: row = await self._session.get(ItemTypeModel, item_type_id) if not row: return None row.category_id = item_type.category_id row.code = item_type.code row.name = item_type.name row.description = item_type.description row.is_active = item_type.is_active row.updated_at = datetime.utcnow() await self._session.commit() await self._session.refresh(row) return await self._attach_category(row) async def delete_item_type(self, item_type_id: str) -> bool: row = await self._session.get(ItemTypeModel, item_type_id) if not row: return False await self._session.delete(row) await self._session.commit() return True # ---- Groups ---- async def create_group(self, group: ItemGroup) -> ItemGroup: row = ItemGroupModel( id=group.id or new_id(), company_id=group.company_id, category_id=group.category_id, item_type_id=group.item_type_id, code=group.code, name=group.name, description=group.description, is_active=group.is_active, sort_order=group.sort_order, icon=group.icon, remarks=group.remarks, created_at=group.created_at, updated_at=group.updated_at, ) self._session.add(row) await self._session.commit() await self._session.refresh(row) return await self._attach_group_relations(row) async def get_group(self, group_id: str, company_id: str) -> ItemGroup | None: result = await self._session.execute( select(ItemGroupModel).where( ItemGroupModel.id == group_id, ItemGroupModel.company_id == company_id, ) ) row = result.scalar_one_or_none() return await self._attach_group_relations(row) if row else None async def get_group_by_code(self, company_id: str, code: str) -> ItemGroup | None: result = await self._session.execute( select(ItemGroupModel).where( ItemGroupModel.company_id == company_id, ItemGroupModel.code == code, ) ) row = result.scalar_one_or_none() return await self._attach_group_relations(row) if row else None async def list_groups( self, company_id: str, is_active: bool | None = None, category_id: str | None = None, item_type_id: str | None = None, skip: int = 0, limit: int = 100, ) -> list[ItemGroup]: query = ( select( ItemGroupModel, ItemCategoryModel.code.label("joined_category_code"), ItemCategoryModel.name.label("joined_category_name"), ItemTypeModel.code.label("joined_item_type_code"), ItemTypeModel.name.label("joined_item_type_name"), ) .select_from(ItemGroupModel) .outerjoin( ItemCategoryModel, ItemCategoryModel.id == ItemGroupModel.category_id, ) .outerjoin( ItemTypeModel, ItemTypeModel.id == ItemGroupModel.item_type_id, ) .where(ItemGroupModel.company_id == company_id) ) if is_active is not None: query = query.where(ItemGroupModel.is_active == is_active) if category_id: query = query.where(ItemGroupModel.category_id == category_id) if item_type_id: query = query.where(ItemGroupModel.item_type_id == item_type_id) result = await self._session.execute( query.order_by( ItemGroupModel.sort_order.is_(None), ItemGroupModel.sort_order.asc(), ItemGroupModel.code.asc(), ) .offset(skip) .limit(limit) ) return [ self._to_group( row.ItemGroupModel, category_code=row.joined_category_code, category_name=row.joined_category_name, item_type_code=row.joined_item_type_code, item_type_name=row.joined_item_type_name, ) for row in result.all() ] async def count_groups( self, company_id: str, is_active: bool | None = None, category_id: str | None = None, item_type_id: str | None = None, ) -> int: query = select(func.count()).select_from(ItemGroupModel).where( ItemGroupModel.company_id == company_id ) if is_active is not None: query = query.where(ItemGroupModel.is_active == is_active) if category_id: query = query.where(ItemGroupModel.category_id == category_id) if item_type_id: query = query.where(ItemGroupModel.item_type_id == item_type_id) result = await self._session.execute(query) return int(result.scalar_one()) async def update_group(self, group_id: str, group: ItemGroup) -> ItemGroup | None: row = await self._session.get(ItemGroupModel, group_id) if not row: return None row.category_id = group.category_id row.item_type_id = group.item_type_id row.code = group.code row.name = group.name row.description = group.description row.is_active = group.is_active row.sort_order = group.sort_order row.icon = group.icon row.remarks = group.remarks row.updated_at = datetime.utcnow() await self._session.commit() await self._session.refresh(row) return await self._attach_group_relations(row) async def delete_group(self, group_id: str) -> bool: row = await self._session.get(ItemGroupModel, group_id) if not row: return False await self._session.delete(row) await self._session.commit() return True # ---- Unit types ---- async def create_unit_type(self, unit_type: UnitType) -> UnitType: row = UnitTypeModel( id=unit_type.id or new_id(), company_id=unit_type.company_id, code=unit_type.code, name=unit_type.name, base_unit_id=unit_type.base_unit_id, unit_kind=unit_type.unit_kind.value, conversion_rate=unit_type.conversion_rate, decimal_places=unit_type.decimal_places, is_active=unit_type.is_active, sort_order=unit_type.sort_order, description=unit_type.description, remarks=unit_type.remarks, created_at=unit_type.created_at, updated_at=unit_type.updated_at, ) self._session.add(row) await self._session.commit() await self._session.refresh(row) return await self._attach_base_unit(row) async def get_unit_type(self, unit_type_id: str, company_id: str) -> UnitType | None: result = await self._session.execute( select(UnitTypeModel).where( UnitTypeModel.id == unit_type_id, UnitTypeModel.company_id == company_id, ) ) row = result.scalar_one_or_none() return await self._attach_base_unit(row) if row else None async def get_unit_type_by_code(self, company_id: str, code: str) -> UnitType | None: result = await self._session.execute( select(UnitTypeModel).where( UnitTypeModel.company_id == company_id, UnitTypeModel.code == code, ) ) row = result.scalar_one_or_none() return await self._attach_base_unit(row) if row else None async def list_unit_types( self, company_id: str, is_active: bool | None = None, skip: int = 0, limit: int = 100 ) -> list[UnitType]: query = ( select( UnitTypeModel, BaseUnitModel.code.label("joined_base_unit_code"), BaseUnitModel.name.label("joined_base_unit_name"), ) .select_from(UnitTypeModel) .outerjoin(BaseUnitModel, BaseUnitModel.id == UnitTypeModel.base_unit_id) .where(UnitTypeModel.company_id == company_id) ) if is_active is not None: query = query.where(UnitTypeModel.is_active == is_active) result = await self._session.execute( query.order_by( UnitTypeModel.sort_order.is_(None), UnitTypeModel.sort_order.asc(), UnitTypeModel.code.asc(), ) .offset(skip) .limit(limit) ) return [ self._to_unit_type( row.UnitTypeModel, base_unit_code=row.joined_base_unit_code, base_unit_name=row.joined_base_unit_name, ) for row in result.all() ] async def count_unit_types(self, company_id: str, is_active: bool | None = None) -> int: query = select(func.count()).select_from(UnitTypeModel).where( UnitTypeModel.company_id == company_id ) if is_active is not None: query = query.where(UnitTypeModel.is_active == is_active) result = await self._session.execute(query) return int(result.scalar_one()) async def update_unit_type(self, unit_type_id: str, unit_type: UnitType) -> UnitType | None: row = await self._session.get(UnitTypeModel, unit_type_id) if not row: return None row.code = unit_type.code row.name = unit_type.name row.base_unit_id = unit_type.base_unit_id row.unit_kind = unit_type.unit_kind.value row.conversion_rate = unit_type.conversion_rate row.decimal_places = unit_type.decimal_places row.is_active = unit_type.is_active row.sort_order = unit_type.sort_order row.description = unit_type.description row.remarks = unit_type.remarks row.updated_at = datetime.utcnow() await self._session.commit() await self._session.refresh(row) return await self._attach_base_unit(row) async def delete_unit_type(self, unit_type_id: str) -> bool: row = await self._session.get(UnitTypeModel, unit_type_id) if not row: return False await self._session.delete(row) await self._session.commit() return True # ---- Items ---- async def create_item(self, item: Item) -> Item: row = ItemModel( id=item.id or new_id(), company_id=item.company_id, sku=item.sku, name=item.name, barcode=item.barcode, image=item.image, description=item.description, specifications=item.specifications, remarks=item.remarks, category_id=item.category_id, item_type_id=item.item_type_id, group_id=item.group_id, brand_name=item.brand_name, base_unit_id=item.base_unit_id, warehouse_id=item.warehouse_id, unit_type_id=item.unit_type_id, pricing_model=item.pricing_model.value, purchase_price=item.purchase_price, sale_price=item.sale_price, tax_percent=item.tax_percent, reorder_level=item.reorder_level, track_inventory=item.track_inventory, inventory_account_id=item.inventory_account_id, expense_account_id=item.expense_account_id, cogs_account_id=item.cogs_account_id, is_active=item.is_active, created_at=item.created_at, updated_at=item.updated_at, ) self._session.add(row) await self._session.commit() await self._session.refresh(row) return await self.get_item(row.id, item.company_id) or self._to_item(row) async def get_item(self, item_id: str, company_id: str) -> Item | None: query = self._item_select_query().where( ItemModel.id == item_id, ItemModel.company_id == company_id, ) result = await self._session.execute(query) row = result.first() return self._item_from_row(row) if row else None async def get_item_by_sku(self, company_id: str, sku: str) -> Item | None: query = self._item_select_query().where( ItemModel.company_id == company_id, ItemModel.sku == sku, ) result = await self._session.execute(query) row = result.first() return self._item_from_row(row) if row else None async def get_item_by_barcode(self, company_id: str, barcode: str) -> Item | None: query = self._item_select_query().where( ItemModel.company_id == company_id, ItemModel.barcode == barcode, ) result = await self._session.execute(query) row = result.first() return self._item_from_row(row) if row else None async def get_next_item_code(self, company_id: str) -> str: result = await self._session.execute( select(ItemModel.sku).where( ItemModel.company_id == company_id, ItemModel.sku.like("ITM-%"), ) ) max_num = 0 for (sku,) in result.all(): suffix = sku.rsplit("-", 1)[-1] if suffix.isdigit(): max_num = max(max_num, int(suffix)) return f"ITM-{max_num + 1:04d}" async def list_items( self, company_id: str, category_id: str | None = None, item_type_id: str | None = None, group_id: str | None = None, warehouse_id: str | None = None, is_active: bool | None = None, skip: int = 0, limit: int = 100, ) -> list[Item]: query = self._item_select_query().where(ItemModel.company_id == company_id) if category_id: query = query.where(ItemModel.category_id == category_id) if item_type_id: query = query.where(ItemModel.item_type_id == item_type_id) if group_id: query = query.where(ItemModel.group_id == group_id) if warehouse_id: query = query.where(ItemModel.warehouse_id == warehouse_id) if is_active is not None: query = query.where(ItemModel.is_active == is_active) result = await self._session.execute( query.order_by(ItemModel.sku.asc()).offset(skip).limit(limit) ) return [self._item_from_row(row) for row in result.all()] async def count_items( self, company_id: str, category_id: str | None = None, item_type_id: str | None = None, group_id: str | None = None, warehouse_id: str | None = None, is_active: bool | None = None, ) -> int: query = select(func.count()).select_from(ItemModel).where(ItemModel.company_id == company_id) if category_id: query = query.where(ItemModel.category_id == category_id) if item_type_id: query = query.where(ItemModel.item_type_id == item_type_id) if group_id: query = query.where(ItemModel.group_id == group_id) if warehouse_id: query = query.where(ItemModel.warehouse_id == warehouse_id) if is_active is not None: query = query.where(ItemModel.is_active == is_active) result = await self._session.execute(query) return int(result.scalar_one()) async def update_item(self, item_id: str, item: Item) -> Item | None: row = await self._session.get(ItemModel, item_id) if not row: return None row.sku = item.sku row.name = item.name row.barcode = item.barcode row.image = item.image row.description = item.description row.specifications = item.specifications row.remarks = item.remarks row.category_id = item.category_id row.item_type_id = item.item_type_id row.group_id = item.group_id row.brand_name = item.brand_name row.base_unit_id = item.base_unit_id row.warehouse_id = item.warehouse_id row.unit_type_id = item.unit_type_id row.pricing_model = item.pricing_model.value row.purchase_price = item.purchase_price row.sale_price = item.sale_price row.tax_percent = item.tax_percent row.reorder_level = item.reorder_level row.track_inventory = item.track_inventory row.inventory_account_id = item.inventory_account_id row.expense_account_id = item.expense_account_id row.cogs_account_id = item.cogs_account_id row.is_active = item.is_active row.updated_at = datetime.utcnow() await self._session.commit() await self._session.refresh(row) return await self.get_item(item_id, row.company_id) async def delete_item(self, item_id: str) -> bool: row = await self._session.get(ItemModel, item_id) if not row: return False await self._session.delete(row) await self._session.commit() return True # ---- Balances ---- async def get_balance( self, company_id: str, item_id: str, warehouse_id: str | None = None ) -> InventoryBalance | None: resolved_warehouse_id = warehouse_id if not resolved_warehouse_id: item = await self.get_item(item_id, company_id) resolved_warehouse_id = item.warehouse_id if item else None if not resolved_warehouse_id: return None result = await self._session.execute( select(InventoryBalanceModel).where( InventoryBalanceModel.company_id == company_id, InventoryBalanceModel.item_id == item_id, InventoryBalanceModel.warehouse_id == resolved_warehouse_id, ) ) row = result.scalar_one_or_none() return self._to_balance(row) if row else None async def list_balances( self, company_id: str, skip: int = 0, limit: int = 100, warehouse_id: str | None = None, item_id: str | None = None, ) -> list[InventoryBalance]: query = select(InventoryBalanceModel).where( InventoryBalanceModel.company_id == company_id ) if warehouse_id: query = query.where(InventoryBalanceModel.warehouse_id == warehouse_id) if item_id: query = query.where(InventoryBalanceModel.item_id == item_id) result = await self._session.execute( query.order_by(InventoryBalanceModel.updated_at.desc()) .offset(skip) .limit(limit) ) return [self._to_balance(row) for row in result.scalars().all()] async def count_balances( self, company_id: str, warehouse_id: str | None = None, item_id: str | None = None, ) -> int: query = ( select(func.count()) .select_from(InventoryBalanceModel) .where(InventoryBalanceModel.company_id == company_id) ) if warehouse_id: query = query.where(InventoryBalanceModel.warehouse_id == warehouse_id) if item_id: query = query.where(InventoryBalanceModel.item_id == item_id) result = await self._session.execute(query) return int(result.scalar_one()) async def upsert_balance(self, balance: InventoryBalance) -> InventoryBalance: warehouse_id = balance.warehouse_id if not warehouse_id: item = await self.get_item(balance.item_id, balance.company_id) warehouse_id = item.warehouse_id if item else None if not warehouse_id: raise ValueError("warehouse_id is required for inventory balance") balance.warehouse_id = warehouse_id existing = await self.get_balance( balance.company_id, balance.item_id, warehouse_id ) if existing and existing.id: row = await self._session.get(InventoryBalanceModel, existing.id) if row: row.quantity_on_hand = balance.quantity_on_hand row.quantity_reserved = balance.quantity_reserved row.average_cost = balance.average_cost row.last_cost = balance.last_cost row.updated_at = datetime.utcnow() await self._session.commit() await self._session.refresh(row) return self._to_balance(row) row = InventoryBalanceModel( id=balance.id or new_id(), company_id=balance.company_id, item_id=balance.item_id, warehouse_id=warehouse_id, quantity_on_hand=balance.quantity_on_hand, quantity_reserved=balance.quantity_reserved, average_cost=balance.average_cost, last_cost=balance.last_cost, updated_at=datetime.utcnow(), ) self._session.add(row) await self._session.commit() await self._session.refresh(row) return self._to_balance(row) # ---- Transactions ---- async def create_transaction( self, transaction: InventoryTransaction ) -> tuple[InventoryTransaction, InventoryBalance]: qty_in = to_decimal(transaction.quantity_in) qty_out = to_decimal(transaction.quantity_out) unit_cost = to_decimal(transaction.unit_cost) warehouse_id = transaction.warehouse_id if not warehouse_id: item = await self.get_item(transaction.item_id, transaction.company_id) warehouse_id = item.warehouse_id if item else None if not warehouse_id: raise ValueError("warehouse_id is required for inventory transaction") transaction.warehouse_id = warehouse_id balance = await self.get_balance( transaction.company_id, transaction.item_id, warehouse_id ) old_qty = balance.quantity_on_hand if balance else Decimal("0.0000") old_avg = balance.average_cost if balance else Decimal("0.0000") new_qty = old_qty + qty_in - qty_out if new_qty < 0: new_qty = Decimal("0.0000") new_avg = old_avg last_cost = balance.last_cost if balance else Decimal("0.0000") if qty_in > 0: last_cost = unit_cost denom = old_qty + qty_in if denom > 0: new_avg = ((old_qty * old_avg) + (qty_in * unit_cost)) / denom total_cost = (qty_in + qty_out) * unit_cost transaction.total_cost = total_cost.quantize(Decimal("0.0001")) transaction.balance_after = new_qty.quantize(Decimal("0.0001")) txn_row = InventoryTransactionModel( id=transaction.id or new_id(), company_id=transaction.company_id, item_id=transaction.item_id, warehouse_id=warehouse_id, txn_type=transaction.txn_type.value, txn_date=transaction.txn_date, quantity_in=qty_in, quantity_out=qty_out, unit_cost=unit_cost, total_cost=transaction.total_cost, balance_after=transaction.balance_after, reference_type=transaction.reference_type, reference_id=transaction.reference_id, reference_number=transaction.reference_number, notes=transaction.notes, created_at=transaction.created_at, ) self._session.add(txn_row) updated_balance = InventoryBalance( id=balance.id if balance else None, company_id=transaction.company_id, item_id=transaction.item_id, warehouse_id=warehouse_id, quantity_on_hand=new_qty.quantize(Decimal("0.0001")), quantity_reserved=balance.quantity_reserved if balance else Decimal("0.0000"), average_cost=new_avg.quantize(Decimal("0.0001")), last_cost=last_cost.quantize(Decimal("0.0001")), ) # Persist balance without intermediate commit so txn + balance share one commit if balance and balance.id: bal_row = await self._session.get(InventoryBalanceModel, balance.id) if bal_row: bal_row.quantity_on_hand = updated_balance.quantity_on_hand bal_row.quantity_reserved = updated_balance.quantity_reserved bal_row.average_cost = updated_balance.average_cost bal_row.last_cost = updated_balance.last_cost bal_row.updated_at = datetime.utcnow() else: bal_row = InventoryBalanceModel( id=new_id(), company_id=updated_balance.company_id, item_id=updated_balance.item_id, warehouse_id=warehouse_id, quantity_on_hand=updated_balance.quantity_on_hand, quantity_reserved=updated_balance.quantity_reserved, average_cost=updated_balance.average_cost, last_cost=updated_balance.last_cost, updated_at=datetime.utcnow(), ) self._session.add(bal_row) else: bal_row = InventoryBalanceModel( id=new_id(), company_id=updated_balance.company_id, item_id=updated_balance.item_id, warehouse_id=warehouse_id, quantity_on_hand=updated_balance.quantity_on_hand, quantity_reserved=updated_balance.quantity_reserved, average_cost=updated_balance.average_cost, last_cost=updated_balance.last_cost, updated_at=datetime.utcnow(), ) self._session.add(bal_row) await self._session.commit() await self._session.refresh(txn_row) await self._session.refresh(bal_row) return self._to_transaction(txn_row), self._to_balance(bal_row) async def list_transactions( self, company_id: str, item_id: str | None = None, skip: int = 0, limit: int = 100, ) -> list[InventoryTransaction]: query = select(InventoryTransactionModel).where( InventoryTransactionModel.company_id == company_id ) if item_id: query = query.where(InventoryTransactionModel.item_id == item_id) result = await self._session.execute( query.order_by( InventoryTransactionModel.txn_date.desc(), InventoryTransactionModel.created_at.desc(), ) .offset(skip) .limit(limit) ) return [self._to_transaction(row) for row in result.scalars().all()] async def count_transactions(self, company_id: str, item_id: str | None = None) -> int: query = select(func.count()).select_from(InventoryTransactionModel).where( InventoryTransactionModel.company_id == company_id ) if item_id: query = query.where(InventoryTransactionModel.item_id == item_id) result = await self._session.execute(query) return int(result.scalar_one()) # ---- Item transactions ---- def _parse_json_list(self, raw: str | None) -> list[str]: if not raw: return [] import json try: data = json.loads(raw) if isinstance(data, list): return [str(x) for x in data] except (TypeError, ValueError): pass return [] def _dump_json_list(self, values: list[str] | None) -> str | None: if not values: return None import json return json.dumps(values) def _to_item_txn_line( self, row: ItemTransactionLineModel, *, item_sku: str | None = None, item_name: str | None = None, base_unit_code: str | None = None, base_unit_name: str | None = None, ) -> ItemTransactionLine: return ItemTransactionLine( id=row.id, item_transaction_id=row.item_transaction_id, line_number=row.line_number, item_id=row.item_id, item_sku=item_sku, item_name=item_name, description=row.description, batch_lot_no=row.batch_lot_no, expiry_date=row.expiry_date, base_unit_id=row.base_unit_id, base_unit_code=base_unit_code, base_unit_name=base_unit_name, quantity=to_decimal(row.quantity), unit_cost=to_decimal(row.unit_cost), discount_type=AmountType(row.discount_type or AmountType.PERCENT.value), discount_value=to_decimal(row.discount_value), discount_amount=to_decimal(row.discount_amount), tax_type=AmountType(row.tax_type or AmountType.PERCENT.value), tax_rate=to_decimal(row.tax_rate), tax_amount=to_decimal(row.tax_amount), line_total=to_decimal(row.line_total), ) def _item_txn_line_model( self, txn_id: str, line: ItemTransactionLine ) -> ItemTransactionLineModel: return ItemTransactionLineModel( id=line.id or new_id(), item_transaction_id=txn_id, line_number=line.line_number, item_id=line.item_id, description=line.description, batch_lot_no=line.batch_lot_no, expiry_date=line.expiry_date, base_unit_id=line.base_unit_id, quantity=line.quantity, unit_cost=line.unit_cost, discount_type=line.discount_type.value, discount_value=line.discount_value, discount_amount=line.discount_amount, tax_type=line.tax_type.value, tax_rate=line.tax_rate, tax_amount=line.tax_amount, line_total=line.line_total, ) async def _enrich_item_txn(self, row: ItemTransactionModel) -> ItemTransaction: warehouse_code = warehouse_name = None if row.warehouse_id: wh = await self._session.get(WarehouseModel, row.warehouse_id) if wh: warehouse_code = wh.code warehouse_name = wh.name vendor_code = vendor_name = None if row.vendor_id: vendor = await self._session.get(VendorModel, row.vendor_id) if vendor: vendor_code = vendor.code vendor_name = vendor.name created_by_name = None if row.created_by: user = await self._session.get(UserRegistrationModel, row.created_by) if user: created_by_name = user.full_name line_meta: dict[str, dict] = {} item_ids = {line.item_id for line in (row.lines or [])} unit_ids = {line.base_unit_id for line in (row.lines or []) if line.base_unit_id} items_by_id: dict[str, ItemModel] = {} units_by_id: dict[str, BaseUnitModel] = {} if item_ids: result = await self._session.execute( select(ItemModel).where(ItemModel.id.in_(item_ids)) ) items_by_id = {i.id: i for i in result.scalars().all()} if unit_ids: result = await self._session.execute( select(BaseUnitModel).where(BaseUnitModel.id.in_(unit_ids)) ) units_by_id = {u.id: u for u in result.scalars().all()} lines: list[ItemTransactionLine] = [] for line in row.lines or []: item = items_by_id.get(line.item_id) unit = units_by_id.get(line.base_unit_id) if line.base_unit_id else None lines.append( self._to_item_txn_line( line, item_sku=item.sku if item else None, item_name=item.name if item else None, base_unit_code=unit.code if unit else None, base_unit_name=unit.name if unit else None, ) ) return ItemTransaction( id=row.id, company_id=row.company_id, txn_number=row.txn_number, txn_type=InventoryTxnType(row.txn_type), txn_date=row.txn_date, reference_type=( ItemTransactionReferenceType(row.reference_type) if row.reference_type else None ), reference_number=row.reference_number, direction=InventoryTxnDirection(row.direction), warehouse_id=row.warehouse_id, warehouse_code=warehouse_code, warehouse_name=warehouse_name, vendor_id=row.vendor_id, vendor_code=vendor_code, vendor_name=vendor_name, po_date=row.po_date, expected_date=row.expected_date, grn_number=row.grn_number, remarks=row.remarks, internal_note=row.internal_note, tags=self._parse_json_list(row.tags), attachments=self._parse_json_list(row.attachments), status=ItemTransactionStatus(row.status), total_quantity=to_decimal(row.total_quantity), subtotal=to_decimal(row.subtotal), discount_amount=to_decimal(row.discount_amount), tax_amount=to_decimal(row.tax_amount), transport_charges=to_decimal(row.transport_charges), rounding=to_decimal(row.rounding), grand_total=to_decimal(row.grand_total), created_by=row.created_by, created_by_name=created_by_name, posted_at=row.posted_at, cancelled_at=row.cancelled_at, lines=lines, created_at=row.created_at, updated_at=row.updated_at, ) async def create_item_transaction(self, doc: ItemTransaction) -> ItemTransaction: from sqlalchemy.orm import selectinload txn_id = doc.id or new_id() row = ItemTransactionModel( id=txn_id, company_id=doc.company_id, txn_number=doc.txn_number, txn_type=doc.txn_type.value, txn_date=doc.txn_date, reference_type=doc.reference_type.value if doc.reference_type else None, reference_number=doc.reference_number, direction=doc.direction.value, warehouse_id=doc.warehouse_id, vendor_id=doc.vendor_id, po_date=doc.po_date, expected_date=doc.expected_date, grn_number=doc.grn_number, remarks=doc.remarks, internal_note=doc.internal_note, tags=self._dump_json_list(doc.tags), attachments=self._dump_json_list(doc.attachments), status=doc.status.value, total_quantity=doc.total_quantity, subtotal=doc.subtotal, discount_amount=doc.discount_amount, tax_amount=doc.tax_amount, transport_charges=doc.transport_charges, rounding=doc.rounding, grand_total=doc.grand_total, created_by=doc.created_by, posted_at=doc.posted_at, cancelled_at=doc.cancelled_at, created_at=doc.created_at, updated_at=doc.updated_at, lines=[self._item_txn_line_model(txn_id, line) for line in doc.lines], ) self._session.add(row) await self._session.commit() return await self.get_item_transaction(txn_id, doc.company_id) # type: ignore[return-value] async def get_item_transaction( self, txn_id: str, company_id: str ) -> ItemTransaction | None: from sqlalchemy.orm import selectinload result = await self._session.execute( select(ItemTransactionModel) .options(selectinload(ItemTransactionModel.lines)) .where( ItemTransactionModel.id == txn_id, ItemTransactionModel.company_id == company_id, ) ) row = result.scalar_one_or_none() if not row: return None return await self._enrich_item_txn(row) async def list_item_transactions( self, company_id: str, status: str | None = None, direction: str | None = None, warehouse_id: str | None = None, skip: int = 0, limit: int = 100, ) -> list[ItemTransaction]: from sqlalchemy.orm import selectinload query = ( select(ItemTransactionModel) .options(selectinload(ItemTransactionModel.lines)) .where(ItemTransactionModel.company_id == company_id) ) if status: query = query.where(ItemTransactionModel.status == status) if direction: query = query.where(ItemTransactionModel.direction == direction) if warehouse_id: query = query.where(ItemTransactionModel.warehouse_id == warehouse_id) result = await self._session.execute( query.order_by(ItemTransactionModel.txn_date.desc()).offset(skip).limit(limit) ) rows = result.scalars().all() return [await self._enrich_item_txn(row) for row in rows] async def count_item_transactions( self, company_id: str, status: str | None = None, direction: str | None = None, warehouse_id: str | None = None, ) -> int: query = select(func.count()).select_from(ItemTransactionModel).where( ItemTransactionModel.company_id == company_id ) if status: query = query.where(ItemTransactionModel.status == status) if direction: query = query.where(ItemTransactionModel.direction == direction) if warehouse_id: query = query.where(ItemTransactionModel.warehouse_id == warehouse_id) result = await self._session.execute(query) return int(result.scalar_one()) async def update_item_transaction(self, doc: ItemTransaction) -> ItemTransaction | None: from sqlalchemy.orm import selectinload if not doc.id: return None result = await self._session.execute( select(ItemTransactionModel) .options(selectinload(ItemTransactionModel.lines)) .where( ItemTransactionModel.id == doc.id, ItemTransactionModel.company_id == doc.company_id, ) ) row = result.scalar_one_or_none() if not row: return None row.txn_type = doc.txn_type.value row.txn_date = doc.txn_date row.reference_type = doc.reference_type.value if doc.reference_type else None row.reference_number = doc.reference_number row.direction = doc.direction.value row.warehouse_id = doc.warehouse_id row.vendor_id = doc.vendor_id row.po_date = doc.po_date row.expected_date = doc.expected_date row.grn_number = doc.grn_number row.remarks = doc.remarks row.internal_note = doc.internal_note row.tags = self._dump_json_list(doc.tags) row.attachments = self._dump_json_list(doc.attachments) row.status = doc.status.value row.total_quantity = doc.total_quantity row.subtotal = doc.subtotal row.discount_amount = doc.discount_amount row.tax_amount = doc.tax_amount row.transport_charges = doc.transport_charges row.rounding = doc.rounding row.grand_total = doc.grand_total row.posted_at = doc.posted_at row.cancelled_at = doc.cancelled_at row.updated_at = datetime.utcnow() row.lines.clear() await self._session.flush() for line in doc.lines: row.lines.append(self._item_txn_line_model(doc.id, line)) await self._session.commit() return await self.get_item_transaction(doc.id, doc.company_id) async def get_next_item_transaction_number(self, company_id: str) -> str: result = await self._session.execute( select(ItemTransactionModel.txn_number) .where(ItemTransactionModel.company_id == company_id) .order_by(ItemTransactionModel.created_at.desc()) .limit(50) ) max_n = 0 for (number,) in result.all(): if not number: continue parts = str(number).rsplit("-", 1) if len(parts) == 2 and parts[1].isdigit(): max_n = max(max_n, int(parts[1])) return f"ITX-{max_n + 1:05d}" async def vendor_exists(self, vendor_id: str, company_id: str) -> bool: row = await self._session.get(VendorModel, vendor_id) return bool(row and row.company_id == company_id) # ---- Stock transfers ---- def _stock_transfer_line_model( self, transfer_id: str, line: StockTransferLine ) -> StockTransferLineModel: return StockTransferLineModel( id=line.id or new_id(), stock_transfer_id=transfer_id, line_number=line.line_number, item_id=line.item_id, available_qty=line.available_qty, transfer_qty=line.transfer_qty, base_unit_id=line.base_unit_id, batch_lot_no=line.batch_lot_no, unit_cost=line.unit_cost, line_value=line.line_value, ) def _warehouse_address(self, wh: WarehouseModel | None) -> str | None: if not wh: return None parts = [ wh.address_line1 or wh.address, wh.address_line2, wh.city or wh.location, wh.state_province, wh.country, wh.postal_code, ] text = ", ".join(p for p in parts if p) return text or None async def _enrich_stock_transfer(self, row: StockTransferModel) -> StockTransfer: from_wh = await self._session.get(WarehouseModel, row.from_warehouse_id) to_wh = await self._session.get(WarehouseModel, row.to_warehouse_id) created_by_name = None if row.created_by: user = await self._session.get(UserRegistrationModel, row.created_by) if user: created_by_name = user.full_name item_ids = {line.item_id for line in (row.lines or [])} unit_ids = {line.base_unit_id for line in (row.lines or []) if line.base_unit_id} items_by_id: dict[str, ItemModel] = {} units_by_id: dict[str, BaseUnitModel] = {} if item_ids: result = await self._session.execute( select(ItemModel).where(ItemModel.id.in_(item_ids)) ) items_by_id = {i.id: i for i in result.scalars().all()} if unit_ids: result = await self._session.execute( select(BaseUnitModel).where(BaseUnitModel.id.in_(unit_ids)) ) units_by_id = {u.id: u for u in result.scalars().all()} lines: list[StockTransferLine] = [] for line in row.lines or []: item = items_by_id.get(line.item_id) unit = units_by_id.get(line.base_unit_id) if line.base_unit_id else None lines.append( StockTransferLine( id=line.id, stock_transfer_id=line.stock_transfer_id, line_number=line.line_number, item_id=line.item_id, item_sku=item.sku if item else None, item_name=item.name if item else None, available_qty=to_decimal(line.available_qty), transfer_qty=to_decimal(line.transfer_qty), base_unit_id=line.base_unit_id, base_unit_code=unit.code if unit else None, base_unit_name=unit.name if unit else None, batch_lot_no=line.batch_lot_no, unit_cost=to_decimal(line.unit_cost), line_value=to_decimal(line.line_value), ) ) return StockTransfer( id=row.id, company_id=row.company_id, transfer_number=row.transfer_number, transfer_date=row.transfer_date, expected_delivery_date=row.expected_delivery_date, priority=StockTransferPriority(row.priority), reason=StockTransferReason(row.reason), reference=row.reference, notes=row.notes, from_warehouse_id=row.from_warehouse_id, from_warehouse_code=from_wh.code if from_wh else None, from_warehouse_name=from_wh.name if from_wh else None, from_warehouse_address=self._warehouse_address(from_wh), to_warehouse_id=row.to_warehouse_id, to_warehouse_code=to_wh.code if to_wh else None, to_warehouse_name=to_wh.name if to_wh else None, to_warehouse_address=self._warehouse_address(to_wh), status=StockTransferStatus(row.status), total_items=row.total_items, total_quantity=to_decimal(row.total_quantity), total_transfer_value=to_decimal(row.total_transfer_value), created_by=row.created_by, created_by_name=created_by_name, submitted_at=row.submitted_at, approved_at=row.approved_at, picked_at=row.picked_at, shipped_at=row.shipped_at, received_at=row.received_at, completed_at=row.completed_at, cancelled_at=row.cancelled_at, lines=lines, created_at=row.created_at, updated_at=row.updated_at, ) async def create_stock_transfer(self, doc: StockTransfer) -> StockTransfer: transfer_id = doc.id or new_id() row = StockTransferModel( id=transfer_id, company_id=doc.company_id, transfer_number=doc.transfer_number, transfer_date=doc.transfer_date, expected_delivery_date=doc.expected_delivery_date, priority=doc.priority.value, reason=doc.reason.value, reference=doc.reference, notes=doc.notes, from_warehouse_id=doc.from_warehouse_id, to_warehouse_id=doc.to_warehouse_id, status=doc.status.value, total_items=doc.total_items, total_quantity=doc.total_quantity, total_transfer_value=doc.total_transfer_value, created_by=doc.created_by, submitted_at=doc.submitted_at, approved_at=doc.approved_at, picked_at=doc.picked_at, shipped_at=doc.shipped_at, received_at=doc.received_at, completed_at=doc.completed_at, cancelled_at=doc.cancelled_at, created_at=doc.created_at, updated_at=doc.updated_at, lines=[self._stock_transfer_line_model(transfer_id, line) for line in doc.lines], ) self._session.add(row) await self._session.commit() return await self.get_stock_transfer(transfer_id, doc.company_id) # type: ignore[return-value] async def get_stock_transfer( self, transfer_id: str, company_id: str ) -> StockTransfer | None: from sqlalchemy.orm import selectinload result = await self._session.execute( select(StockTransferModel) .options(selectinload(StockTransferModel.lines)) .where( StockTransferModel.id == transfer_id, StockTransferModel.company_id == company_id, ) ) row = result.scalar_one_or_none() return await self._enrich_stock_transfer(row) if row else None async def list_stock_transfers( self, company_id: str, status: str | None = None, from_warehouse_id: str | None = None, to_warehouse_id: str | None = None, skip: int = 0, limit: int = 100, ) -> list[StockTransfer]: from sqlalchemy.orm import selectinload query = ( select(StockTransferModel) .options(selectinload(StockTransferModel.lines)) .where(StockTransferModel.company_id == company_id) ) if status: query = query.where(StockTransferModel.status == status) if from_warehouse_id: query = query.where(StockTransferModel.from_warehouse_id == from_warehouse_id) if to_warehouse_id: query = query.where(StockTransferModel.to_warehouse_id == to_warehouse_id) result = await self._session.execute( query.order_by(StockTransferModel.transfer_date.desc(), StockTransferModel.created_at.desc()) .offset(skip) .limit(limit) ) return [await self._enrich_stock_transfer(row) for row in result.scalars().all()] async def count_stock_transfers( self, company_id: str, status: str | None = None, from_warehouse_id: str | None = None, to_warehouse_id: str | None = None, ) -> int: query = select(func.count()).select_from(StockTransferModel).where( StockTransferModel.company_id == company_id ) if status: query = query.where(StockTransferModel.status == status) if from_warehouse_id: query = query.where(StockTransferModel.from_warehouse_id == from_warehouse_id) if to_warehouse_id: query = query.where(StockTransferModel.to_warehouse_id == to_warehouse_id) result = await self._session.execute(query) return int(result.scalar_one()) async def update_stock_transfer( self, doc: StockTransfer, *, replace_lines: bool = False ) -> StockTransfer | None: from sqlalchemy.orm import selectinload result = await self._session.execute( select(StockTransferModel) .options(selectinload(StockTransferModel.lines)) .where(StockTransferModel.id == doc.id) ) row = result.scalar_one_or_none() if not row: return None row.transfer_date = doc.transfer_date row.expected_delivery_date = doc.expected_delivery_date row.priority = doc.priority.value row.reason = doc.reason.value row.reference = doc.reference row.notes = doc.notes row.from_warehouse_id = doc.from_warehouse_id row.to_warehouse_id = doc.to_warehouse_id row.status = doc.status.value row.total_items = doc.total_items row.total_quantity = doc.total_quantity row.total_transfer_value = doc.total_transfer_value row.submitted_at = doc.submitted_at row.approved_at = doc.approved_at row.picked_at = doc.picked_at row.shipped_at = doc.shipped_at row.received_at = doc.received_at row.completed_at = doc.completed_at row.cancelled_at = doc.cancelled_at row.updated_at = datetime.utcnow() if replace_lines: row.lines.clear() await self._session.flush() for line in doc.lines: row.lines.append(self._stock_transfer_line_model(doc.id or row.id, line)) await self._session.commit() return await self.get_stock_transfer(row.id, doc.company_id) async def delete_stock_transfer(self, transfer_id: str) -> bool: row = await self._session.get(StockTransferModel, transfer_id) if not row: return False await self._session.delete(row) await self._session.commit() return True async def get_next_stock_transfer_number(self, company_id: str) -> str: year = datetime.utcnow().year pattern = f"ST-{year}-%" result = await self._session.execute( select(StockTransferModel.transfer_number) .where( StockTransferModel.company_id == company_id, StockTransferModel.transfer_number.like(pattern), ) .order_by(StockTransferModel.transfer_number.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"ST-{year}-{seq:05d}" # ---- Locations ---- def _to_location(self, row: LocationModel) -> Location: return Location( id=row.id, company_id=row.company_id, code=row.code, name=row.name, address=row.address, is_active=row.is_active, created_at=row.created_at, updated_at=row.updated_at, ) async def create_location(self, location: Location) -> Location: row = LocationModel( id=location.id or new_id(), company_id=location.company_id, code=location.code, name=location.name, address=location.address, is_active=location.is_active, created_at=location.created_at, updated_at=location.updated_at, ) self._session.add(row) await self._session.commit() await self._session.refresh(row) return self._to_location(row) async def get_location(self, location_id: str, company_id: str) -> Location | None: result = await self._session.execute( select(LocationModel).where( LocationModel.id == location_id, LocationModel.company_id == company_id, ) ) row = result.scalar_one_or_none() return self._to_location(row) if row else None async def get_location_by_code(self, company_id: str, code: str) -> Location | None: result = await self._session.execute( select(LocationModel).where( LocationModel.company_id == company_id, LocationModel.code == code, ) ) row = result.scalar_one_or_none() return self._to_location(row) if row else None async def list_locations( self, company_id: str, is_active: bool | None = None, skip: int = 0, limit: int = 100, ) -> list[Location]: query = select(LocationModel).where(LocationModel.company_id == company_id) if is_active is not None: query = query.where(LocationModel.is_active == is_active) result = await self._session.execute( query.order_by(LocationModel.code.asc()).offset(skip).limit(limit) ) return [self._to_location(row) for row in result.scalars().all()] async def count_locations( self, company_id: str, is_active: bool | None = None ) -> int: query = select(func.count()).select_from(LocationModel).where( LocationModel.company_id == company_id ) if is_active is not None: query = query.where(LocationModel.is_active == is_active) result = await self._session.execute(query) return int(result.scalar_one()) async def update_location(self, location_id: str, location: Location) -> Location | None: row = await self._session.get(LocationModel, location_id) if not row: return None row.code = location.code row.name = location.name row.address = location.address row.is_active = location.is_active row.updated_at = datetime.utcnow() await self._session.commit() await self._session.refresh(row) return self._to_location(row) async def delete_location(self, location_id: str) -> bool: row = await self._session.get(LocationModel, location_id) if not row: return False await self._session.delete(row) await self._session.commit() return True # ---- Departments ---- async def _enrich_department(self, row: DepartmentModel) -> Department: head_name = None if row.head_id: head = await self._session.get(UserRegistrationModel, row.head_id) if head: head_name = head.full_name else: company_head = await self._session.get(CompanyUserModel, row.head_id) if company_head: head_name = company_head.full_name location_code = location_name = None if row.location_id: loc = await self._session.get(LocationModel, row.location_id) if loc: location_code = loc.code location_name = loc.name parent_code = parent_name = None if row.parent_department_id: parent = await self._session.get(DepartmentModel, row.parent_department_id) if parent: parent_code = parent.code parent_name = parent.name return Department( id=row.id, company_id=row.company_id, code=row.code, name=row.name, head_id=row.head_id, head_name=head_name, location_id=row.location_id, location_code=location_code, location_name=location_name, parent_department_id=row.parent_department_id, parent_department_code=parent_code, parent_department_name=parent_name, monthly_issue_budget=( to_decimal(row.monthly_issue_budget) if row.monthly_issue_budget is not None else None ), description=row.description, status=DepartmentStatus(row.status), is_active=row.is_active, notify_email=row.notify_email, low_stock_alert=LowStockAlertTarget(row.low_stock_alert or "head"), created_at=row.created_at, updated_at=row.updated_at, ) async def create_department(self, department: Department) -> Department: row = DepartmentModel( id=department.id or new_id(), company_id=department.company_id, code=department.code, name=department.name, head_id=department.head_id, location_id=department.location_id, parent_department_id=department.parent_department_id, monthly_issue_budget=department.monthly_issue_budget, description=department.description, status=department.status.value, is_active=department.is_active, notify_email=department.notify_email, low_stock_alert=department.low_stock_alert.value, created_at=department.created_at, updated_at=department.updated_at, ) self._session.add(row) await self._session.commit() await self._session.refresh(row) return await self._enrich_department(row) async def get_department( self, department_id: str, company_id: str ) -> Department | None: result = await self._session.execute( select(DepartmentModel).where( DepartmentModel.id == department_id, DepartmentModel.company_id == company_id, ) ) row = result.scalar_one_or_none() return await self._enrich_department(row) if row else None async def get_department_by_code( self, company_id: str, code: str ) -> Department | None: result = await self._session.execute( select(DepartmentModel).where( DepartmentModel.company_id == company_id, DepartmentModel.code == code, ) ) row = result.scalar_one_or_none() return await self._enrich_department(row) if row else None async def list_departments( self, company_id: str, is_active: bool | None = None, status: str | None = None, location_id: str | None = None, skip: int = 0, limit: int = 100, ) -> list[Department]: query = select(DepartmentModel).where(DepartmentModel.company_id == company_id) if is_active is not None: query = query.where(DepartmentModel.is_active == is_active) if status: query = query.where(DepartmentModel.status == status) if location_id: query = query.where(DepartmentModel.location_id == location_id) result = await self._session.execute( query.order_by(DepartmentModel.code.asc()).offset(skip).limit(limit) ) return [await self._enrich_department(row) for row in result.scalars().all()] async def count_departments( self, company_id: str, is_active: bool | None = None, status: str | None = None, location_id: str | None = None, ) -> int: query = select(func.count()).select_from(DepartmentModel).where( DepartmentModel.company_id == company_id ) if is_active is not None: query = query.where(DepartmentModel.is_active == is_active) if status: query = query.where(DepartmentModel.status == status) if location_id: query = query.where(DepartmentModel.location_id == location_id) result = await self._session.execute(query) return int(result.scalar_one()) async def update_department( self, department_id: str, department: Department ) -> Department | None: row = await self._session.get(DepartmentModel, department_id) if not row: return None row.code = department.code row.name = department.name row.head_id = department.head_id row.location_id = department.location_id row.parent_department_id = department.parent_department_id row.monthly_issue_budget = department.monthly_issue_budget row.description = department.description row.status = department.status.value row.is_active = department.is_active row.notify_email = department.notify_email row.low_stock_alert = department.low_stock_alert.value row.updated_at = datetime.utcnow() await self._session.commit() await self._session.refresh(row) return await self._enrich_department(row) async def delete_department(self, department_id: str) -> bool: row = await self._session.get(DepartmentModel, department_id) if not row: return False await self._session.delete(row) await self._session.commit() return True # ---- Department issues ---- def _department_issue_line_model( self, issue_id: str, line: DepartmentIssueLine ) -> DepartmentIssueLineModel: return DepartmentIssueLineModel( id=line.id or new_id(), department_issue_id=issue_id, line_number=line.line_number, item_id=line.item_id, available_qty=line.available_qty, issue_qty=line.issue_qty, base_unit_id=line.base_unit_id, remarks=line.remarks, unit_cost=line.unit_cost, line_value=line.line_value, ) async def _enrich_department_issue(self, row: DepartmentIssueModel) -> DepartmentIssue: department_code = department_name = None dept = await self._session.get(DepartmentModel, row.department_id) if dept: department_code = dept.code department_name = dept.name warehouse_code = warehouse_name = None wh = await self._session.get(WarehouseModel, row.from_warehouse_id) if wh: warehouse_code = wh.code warehouse_name = wh.name requested_by_name = None if row.requested_by_id: requester = await self._session.get(UserRegistrationModel, row.requested_by_id) if requester: requested_by_name = requester.full_name created_by_name = None if row.created_by: creator = await self._session.get(UserRegistrationModel, row.created_by) if creator: created_by_name = creator.full_name item_ids = {line.item_id for line in (row.lines or [])} unit_ids = {line.base_unit_id for line in (row.lines or []) if line.base_unit_id} items_by_id: dict[str, ItemModel] = {} units_by_id: dict[str, BaseUnitModel] = {} if item_ids: result = await self._session.execute( select(ItemModel).where(ItemModel.id.in_(item_ids)) ) items_by_id = {i.id: i for i in result.scalars().all()} if unit_ids: result = await self._session.execute( select(BaseUnitModel).where(BaseUnitModel.id.in_(unit_ids)) ) units_by_id = {u.id: u for u in result.scalars().all()} lines: list[DepartmentIssueLine] = [] for line in row.lines or []: item = items_by_id.get(line.item_id) unit = units_by_id.get(line.base_unit_id) if line.base_unit_id else None lines.append( DepartmentIssueLine( id=line.id, department_issue_id=line.department_issue_id, line_number=line.line_number, item_id=line.item_id, item_sku=item.sku if item else None, item_name=item.name if item else None, available_qty=to_decimal(line.available_qty), issue_qty=to_decimal(line.issue_qty), base_unit_id=line.base_unit_id, base_unit_code=unit.code if unit else None, base_unit_name=unit.name if unit else None, remarks=line.remarks, unit_cost=to_decimal(line.unit_cost), line_value=to_decimal(line.line_value), ) ) return DepartmentIssue( id=row.id, company_id=row.company_id, issue_number=row.issue_number, issue_date=row.issue_date, required_date=row.required_date, priority=DepartmentIssuePriority(row.priority), department_id=row.department_id, department_code=department_code, department_name=department_name, requested_by_id=row.requested_by_id, requested_by_name=requested_by_name, designation=row.designation, cost_center=row.cost_center, issue_type=DepartmentIssueType(row.issue_type), reason=DepartmentIssueReason(row.reason), reference=row.reference, notes=row.notes, from_warehouse_id=row.from_warehouse_id, from_warehouse_code=warehouse_code, from_warehouse_name=warehouse_name, status=DepartmentIssueStatus(row.status), total_items=row.total_items, total_quantity=to_decimal(row.total_quantity), total_estimated_value=to_decimal(row.total_estimated_value), attachments=self._parse_json_list(row.attachments), created_by=row.created_by, created_by_name=created_by_name, submitted_at=row.submitted_at, approved_at=row.approved_at, issued_at=row.issued_at, completed_at=row.completed_at, cancelled_at=row.cancelled_at, lines=lines, created_at=row.created_at, updated_at=row.updated_at, ) async def create_department_issue(self, doc: DepartmentIssue) -> DepartmentIssue: issue_id = doc.id or new_id() row = DepartmentIssueModel( id=issue_id, company_id=doc.company_id, issue_number=doc.issue_number, issue_date=doc.issue_date, required_date=doc.required_date, priority=doc.priority.value, department_id=doc.department_id, requested_by_id=doc.requested_by_id, designation=doc.designation, cost_center=doc.cost_center, issue_type=doc.issue_type.value, reason=doc.reason.value, reference=doc.reference, notes=doc.notes, from_warehouse_id=doc.from_warehouse_id, status=doc.status.value, total_items=doc.total_items, total_quantity=doc.total_quantity, total_estimated_value=doc.total_estimated_value, attachments=self._dump_json_list(doc.attachments), created_by=doc.created_by, submitted_at=doc.submitted_at, approved_at=doc.approved_at, issued_at=doc.issued_at, completed_at=doc.completed_at, cancelled_at=doc.cancelled_at, created_at=doc.created_at, updated_at=doc.updated_at, lines=[self._department_issue_line_model(issue_id, line) for line in doc.lines], ) self._session.add(row) await self._session.commit() return await self.get_department_issue(issue_id, doc.company_id) # type: ignore[return-value] async def get_department_issue( self, issue_id: str, company_id: str ) -> DepartmentIssue | None: from sqlalchemy.orm import selectinload result = await self._session.execute( select(DepartmentIssueModel) .options(selectinload(DepartmentIssueModel.lines)) .where( DepartmentIssueModel.id == issue_id, DepartmentIssueModel.company_id == company_id, ) ) row = result.scalar_one_or_none() return await self._enrich_department_issue(row) if row else None async def list_department_issues( self, company_id: str, status: str | None = None, department_id: str | None = None, from_warehouse_id: str | None = None, skip: int = 0, limit: int = 100, ) -> list[DepartmentIssue]: from sqlalchemy.orm import selectinload query = ( select(DepartmentIssueModel) .options(selectinload(DepartmentIssueModel.lines)) .where(DepartmentIssueModel.company_id == company_id) ) if status: query = query.where(DepartmentIssueModel.status == status) if department_id: query = query.where(DepartmentIssueModel.department_id == department_id) if from_warehouse_id: query = query.where(DepartmentIssueModel.from_warehouse_id == from_warehouse_id) result = await self._session.execute( query.order_by( DepartmentIssueModel.issue_date.desc(), DepartmentIssueModel.created_at.desc(), ) .offset(skip) .limit(limit) ) return [await self._enrich_department_issue(row) for row in result.scalars().all()] async def count_department_issues( self, company_id: str, status: str | None = None, department_id: str | None = None, from_warehouse_id: str | None = None, ) -> int: query = select(func.count()).select_from(DepartmentIssueModel).where( DepartmentIssueModel.company_id == company_id ) if status: query = query.where(DepartmentIssueModel.status == status) if department_id: query = query.where(DepartmentIssueModel.department_id == department_id) if from_warehouse_id: query = query.where(DepartmentIssueModel.from_warehouse_id == from_warehouse_id) result = await self._session.execute(query) return int(result.scalar_one()) async def update_department_issue( self, doc: DepartmentIssue, *, replace_lines: bool = False ) -> DepartmentIssue | None: from sqlalchemy.orm import selectinload result = await self._session.execute( select(DepartmentIssueModel) .options(selectinload(DepartmentIssueModel.lines)) .where(DepartmentIssueModel.id == doc.id) ) row = result.scalar_one_or_none() if not row: return None row.issue_date = doc.issue_date row.required_date = doc.required_date row.priority = doc.priority.value row.department_id = doc.department_id row.requested_by_id = doc.requested_by_id row.designation = doc.designation row.cost_center = doc.cost_center row.issue_type = doc.issue_type.value row.reason = doc.reason.value row.reference = doc.reference row.notes = doc.notes row.from_warehouse_id = doc.from_warehouse_id row.status = doc.status.value row.total_items = doc.total_items row.total_quantity = doc.total_quantity row.total_estimated_value = doc.total_estimated_value row.attachments = self._dump_json_list(doc.attachments) row.submitted_at = doc.submitted_at row.approved_at = doc.approved_at row.issued_at = doc.issued_at row.completed_at = doc.completed_at row.cancelled_at = doc.cancelled_at row.updated_at = datetime.utcnow() if replace_lines: row.lines.clear() await self._session.flush() for line in doc.lines: row.lines.append(self._department_issue_line_model(doc.id or row.id, line)) await self._session.commit() return await self.get_department_issue(row.id, doc.company_id) async def delete_department_issue(self, issue_id: str) -> bool: row = await self._session.get(DepartmentIssueModel, issue_id) if not row: return False await self._session.delete(row) await self._session.commit() return True async def get_next_department_issue_number(self, company_id: str) -> str: year = datetime.utcnow().year pattern = f"ISS-{year}-%" result = await self._session.execute( select(DepartmentIssueModel.issue_number) .where( DepartmentIssueModel.company_id == company_id, DepartmentIssueModel.issue_number.like(pattern), ) .order_by(DepartmentIssueModel.issue_number.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"ISS-{year}-{seq:05d}" # ---- Department issue reports dashboard ---- def _department_issue_base_filters( self, company_id: str, *, from_date: date | None = None, to_date: date | None = None, department_id: str | None = None, from_warehouse_id: str | None = None, exclude_cancelled: bool = True, status: str | None = None, ): filters = [DepartmentIssueModel.company_id == company_id] if from_date is not None: filters.append(DepartmentIssueModel.issue_date >= from_date) if to_date is not None: filters.append(DepartmentIssueModel.issue_date <= to_date) if department_id: filters.append(DepartmentIssueModel.department_id == department_id) if from_warehouse_id: filters.append(DepartmentIssueModel.from_warehouse_id == from_warehouse_id) if status: filters.append(DepartmentIssueModel.status == status) elif exclude_cancelled: filters.append( DepartmentIssueModel.status != DepartmentIssueStatus.CANCELLED.value ) return filters async def get_department_issue_period_stats( self, company_id: str, from_date: date, to_date: date, *, department_id: str | None = None, from_warehouse_id: str | None = None, exclude_cancelled: bool = True, ) -> dict: filters = self._department_issue_base_filters( company_id, from_date=from_date, to_date=to_date, department_id=department_id, from_warehouse_id=from_warehouse_id, exclude_cancelled=exclude_cancelled, ) query = select( func.count(DepartmentIssueModel.id).label("issues_count"), func.coalesce(func.sum(DepartmentIssueModel.total_quantity), 0).label( "total_quantity" ), func.coalesce(func.sum(DepartmentIssueModel.total_estimated_value), 0).label( "total_value" ), func.count(func.distinct(DepartmentIssueModel.department_id)).label( "departments_count" ), func.count(func.distinct(DepartmentIssueModel.from_warehouse_id)).label( "warehouses_count" ), ).where(*filters) result = await self._session.execute(query) row = result.one() return { "issues_count": int(row.issues_count or 0), "total_quantity": to_decimal(row.total_quantity), "total_value": to_decimal(row.total_value), "departments_count": int(row.departments_count or 0), "warehouses_count": int(row.warehouses_count or 0), } async def get_department_issue_daily_trend( self, company_id: str, from_date: date, to_date: date, *, department_id: str | None = None, from_warehouse_id: str | None = None, exclude_cancelled: bool = True, ) -> list[dict]: filters = self._department_issue_base_filters( company_id, from_date=from_date, to_date=to_date, department_id=department_id, from_warehouse_id=from_warehouse_id, exclude_cancelled=exclude_cancelled, ) query = ( select( DepartmentIssueModel.issue_date.label("issue_date"), func.count(DepartmentIssueModel.id).label("issues_count"), func.coalesce(func.sum(DepartmentIssueModel.total_quantity), 0).label( "total_quantity" ), func.coalesce( func.sum(DepartmentIssueModel.total_estimated_value), 0 ).label("total_value"), ) .where(*filters) .group_by(DepartmentIssueModel.issue_date) .order_by(DepartmentIssueModel.issue_date.asc()) ) result = await self._session.execute(query) return [ { "date": row.issue_date, "issues_count": int(row.issues_count or 0), "total_quantity": to_decimal(row.total_quantity), "total_value": to_decimal(row.total_value), } for row in result.all() ] async def get_department_issue_breakdown_by_department( self, company_id: str, from_date: date, to_date: date, *, department_id: str | None = None, from_warehouse_id: str | None = None, exclude_cancelled: bool = True, ) -> list[dict]: filters = self._department_issue_base_filters( company_id, from_date=from_date, to_date=to_date, department_id=department_id, from_warehouse_id=from_warehouse_id, exclude_cancelled=exclude_cancelled, ) query = ( select( DepartmentIssueModel.department_id.label("id"), DepartmentModel.code.label("code"), DepartmentModel.name.label("name"), func.count(DepartmentIssueModel.id).label("issues_count"), func.coalesce(func.sum(DepartmentIssueModel.total_quantity), 0).label( "total_quantity" ), func.coalesce( func.sum(DepartmentIssueModel.total_estimated_value), 0 ).label("total_value"), ) .outerjoin( DepartmentModel, DepartmentModel.id == DepartmentIssueModel.department_id, ) .where(*filters) .group_by( DepartmentIssueModel.department_id, DepartmentModel.code, DepartmentModel.name, ) .order_by(func.sum(DepartmentIssueModel.total_quantity).desc()) ) result = await self._session.execute(query) return [ { "id": row.id, "code": row.code, "name": row.name or "Unknown", "issues_count": int(row.issues_count or 0), "total_quantity": to_decimal(row.total_quantity), "total_value": to_decimal(row.total_value), } for row in result.all() ] async def get_department_issue_breakdown_by_warehouse( self, company_id: str, from_date: date, to_date: date, *, department_id: str | None = None, from_warehouse_id: str | None = None, exclude_cancelled: bool = True, ) -> list[dict]: filters = self._department_issue_base_filters( company_id, from_date=from_date, to_date=to_date, department_id=department_id, from_warehouse_id=from_warehouse_id, exclude_cancelled=exclude_cancelled, ) query = ( select( DepartmentIssueModel.from_warehouse_id.label("id"), WarehouseModel.code.label("code"), WarehouseModel.name.label("name"), func.count(DepartmentIssueModel.id).label("issues_count"), func.coalesce(func.sum(DepartmentIssueModel.total_quantity), 0).label( "total_quantity" ), func.coalesce( func.sum(DepartmentIssueModel.total_estimated_value), 0 ).label("total_value"), ) .outerjoin( WarehouseModel, WarehouseModel.id == DepartmentIssueModel.from_warehouse_id, ) .where(*filters) .group_by( DepartmentIssueModel.from_warehouse_id, WarehouseModel.code, WarehouseModel.name, ) .order_by(func.sum(DepartmentIssueModel.total_quantity).desc()) ) result = await self._session.execute(query) return [ { "id": row.id, "code": row.code, "name": row.name or "Unknown", "issues_count": int(row.issues_count or 0), "total_quantity": to_decimal(row.total_quantity), "total_value": to_decimal(row.total_value), } for row in result.all() ] async def list_department_issue_summaries( self, company_id: str, *, from_date: date | None = None, to_date: date | None = None, status: str | None = None, department_id: str | None = None, from_warehouse_id: str | None = None, exclude_cancelled: bool = True, order_by_submitted: bool = False, skip: int = 0, limit: int = 10, ) -> list[DepartmentIssueSummary]: filters = self._department_issue_base_filters( company_id, from_date=from_date, to_date=to_date, department_id=department_id, from_warehouse_id=from_warehouse_id, exclude_cancelled=exclude_cancelled, status=status, ) query = ( select( DepartmentIssueModel, DepartmentModel.code.label("department_code"), DepartmentModel.name.label("department_name"), WarehouseModel.code.label("warehouse_code"), WarehouseModel.name.label("warehouse_name"), UserRegistrationModel.full_name.label("requested_by_name"), ) .outerjoin( DepartmentModel, DepartmentModel.id == DepartmentIssueModel.department_id, ) .outerjoin( WarehouseModel, WarehouseModel.id == DepartmentIssueModel.from_warehouse_id, ) .outerjoin( UserRegistrationModel, UserRegistrationModel.id == DepartmentIssueModel.requested_by_id, ) .where(*filters) ) if order_by_submitted: query = query.order_by( DepartmentIssueModel.submitted_at.asc(), DepartmentIssueModel.created_at.asc(), ) else: query = query.order_by( DepartmentIssueModel.issue_date.desc(), DepartmentIssueModel.created_at.desc(), ) result = await self._session.execute(query.offset(skip).limit(limit)) items: list[DepartmentIssueSummary] = [] for row in result.all(): issue = row.DepartmentIssueModel items.append( DepartmentIssueSummary( id=issue.id, issue_number=issue.issue_number, issue_date=issue.issue_date, required_date=issue.required_date, status=DepartmentIssueStatus(issue.status), priority=DepartmentIssuePriority(issue.priority), department_id=issue.department_id, department_code=row.department_code, department_name=row.department_name, from_warehouse_id=issue.from_warehouse_id, from_warehouse_code=row.warehouse_code, from_warehouse_name=row.warehouse_name, requested_by_id=issue.requested_by_id, requested_by_name=row.requested_by_name, total_items=issue.total_items, total_quantity=to_decimal(issue.total_quantity), total_estimated_value=to_decimal(issue.total_estimated_value), submitted_at=issue.submitted_at, ) ) return items async def count_department_issues_pending( self, company_id: str, *, from_date: date | None = None, to_date: date | None = None, department_id: str | None = None, from_warehouse_id: str | None = None, ) -> int: filters = self._department_issue_base_filters( company_id, from_date=from_date, to_date=to_date, department_id=department_id, from_warehouse_id=from_warehouse_id, exclude_cancelled=False, status=DepartmentIssueStatus.REVIEW.value, ) result = await self._session.execute( select(func.count()).select_from(DepartmentIssueModel).where(*filters) ) return int(result.scalar_one()) def _stock_status_expr(self): return case( (ItemModel.is_active.is_(False), StockReportStatus.INACTIVE.value), ( InventoryBalanceModel.quantity_on_hand <= 0, StockReportStatus.OUT_OF_STOCK.value, ), ( and_( ItemModel.reorder_level > 0, InventoryBalanceModel.quantity_on_hand > 0, InventoryBalanceModel.quantity_on_hand <= ItemModel.reorder_level, ), StockReportStatus.LOW_STOCK.value, ), else_=StockReportStatus.IN_STOCK.value, ) def _stock_value_expr(self): return InventoryBalanceModel.quantity_on_hand * InventoryBalanceModel.average_cost async def get_stock_report( self, company_id: str, *, from_date: date | None = None, to_date: date | None = None, warehouse_id: str | None = None, category_id: str | None = None, item_type_id: str | None = None, search: str | None = None, stock_status: str | None = None, min_stock_value: Decimal | None = None, page: int = 1, page_size: int = 20, ) -> StockReport: status_expr = self._stock_status_expr().label("stock_status") stock_value_expr = self._stock_value_expr().label("stock_value") opening_filters = [InventoryTransactionModel.company_id == company_id] if from_date is not None: opening_filters.append(InventoryTransactionModel.txn_date < from_date) else: # No period start => opening qty is 0 (exclude all txn rows). opening_filters.append(InventoryTransactionModel.id.is_(None)) opening_subq = ( select( InventoryTransactionModel.item_id.label("item_id"), InventoryTransactionModel.warehouse_id.label("warehouse_id"), func.coalesce( func.sum( InventoryTransactionModel.quantity_in - InventoryTransactionModel.quantity_out ), 0, ).label("opening_qty"), ) .where(*opening_filters) .group_by( InventoryTransactionModel.item_id, InventoryTransactionModel.warehouse_id, ) .subquery() ) base_filters = [ InventoryBalanceModel.company_id == company_id, ItemModel.company_id == company_id, ] if warehouse_id: base_filters.append(InventoryBalanceModel.warehouse_id == warehouse_id) if category_id: base_filters.append(ItemModel.category_id == category_id) if item_type_id: base_filters.append(ItemModel.item_type_id == item_type_id) if search: term = f"%{search.strip().lower()}%" base_filters.append( or_( func.lower(ItemModel.sku).like(term), func.lower(ItemModel.name).like(term), func.lower(func.coalesce(ItemModel.barcode, "")).like(term), ) ) if min_stock_value is not None: base_filters.append(self._stock_value_expr() >= min_stock_value) if stock_status: base_filters.append(self._stock_status_expr() == stock_status) from_clause = ( InventoryBalanceModel.__table__.join( ItemModel.__table__, InventoryBalanceModel.item_id == ItemModel.id, ) .join( WarehouseModel.__table__, InventoryBalanceModel.warehouse_id == WarehouseModel.id, ) .outerjoin( ItemCategoryModel.__table__, ItemModel.category_id == ItemCategoryModel.id, ) .outerjoin( ItemTypeModel.__table__, ItemModel.item_type_id == ItemTypeModel.id, ) .outerjoin( BaseUnitModel.__table__, ItemModel.base_unit_id == BaseUnitModel.id, ) .outerjoin( opening_subq, and_( opening_subq.c.item_id == InventoryBalanceModel.item_id, opening_subq.c.warehouse_id == InventoryBalanceModel.warehouse_id, ), ) ) summary_result = await self._session.execute( select( func.count(func.distinct(InventoryBalanceModel.item_id)), func.coalesce(func.sum(InventoryBalanceModel.quantity_on_hand), 0), func.coalesce(func.sum(self._stock_value_expr()), 0), func.count(func.distinct(InventoryBalanceModel.warehouse_id)), func.sum( case( ( self._stock_status_expr() == StockReportStatus.LOW_STOCK.value, 1, ), else_=0, ) ), ) .select_from(from_clause) .where(*base_filters) ) summary_row = summary_result.one() total_items = int(summary_row[0] or 0) total_quantity = to_decimal(summary_row[1]) total_stock_value = to_decimal(summary_row[2]) total_warehouses = int(summary_row[3] or 0) low_stock_items = int(summary_row[4] or 0) by_wh_result = await self._session.execute( select( WarehouseModel.id, WarehouseModel.code, WarehouseModel.name, func.count(InventoryBalanceModel.id).label("issues_count"), func.coalesce(func.sum(InventoryBalanceModel.quantity_on_hand), 0).label( "total_quantity" ), func.coalesce(func.sum(self._stock_value_expr()), 0).label("total_value"), ) .select_from(from_clause) .where(*base_filters) .group_by(WarehouseModel.id, WarehouseModel.code, WarehouseModel.name) .order_by(func.sum(self._stock_value_expr()).desc()) ) by_warehouse_value: list[DashboardBreakdownSlice] = [] for row in by_wh_result.all(): value = to_decimal(row.total_value) by_warehouse_value.append( DashboardBreakdownSlice( id=row.id, code=row.code, name=row.name, issues_count=int(row.issues_count or 0), total_quantity=to_decimal(row.total_quantity), total_value=value, percent_of_total=( (value / total_stock_value * Decimal("100")).quantize( Decimal("0.01") ) if total_stock_value > 0 else Decimal("0.00") ), ) ) status_result = await self._session.execute( select( status_expr, func.count(InventoryBalanceModel.id), ) .select_from(from_clause) .where(*base_filters) .group_by(status_expr) ) status_counts = { StockReportStatus.IN_STOCK.value: 0, StockReportStatus.LOW_STOCK.value: 0, StockReportStatus.OUT_OF_STOCK.value: 0, StockReportStatus.INACTIVE.value: 0, } status_total = 0 for status_val, cnt in status_result.all(): status_counts[str(status_val)] = int(cnt or 0) status_total += int(cnt or 0) status_overview = [ StockStatusSlice( status=StockReportStatus(key), count=status_counts[key], percent_of_total=( ( Decimal(status_counts[key]) / Decimal(status_total) * Decimal("100") ).quantize(Decimal("0.01")) if status_total > 0 else Decimal("0.00") ), ) for key in status_counts ] top_result = await self._session.execute( select( ItemModel.id, ItemModel.sku, ItemModel.name, WarehouseModel.id.label("warehouse_id"), WarehouseModel.name.label("warehouse_name"), InventoryBalanceModel.quantity_on_hand, InventoryBalanceModel.average_cost, stock_value_expr, ) .select_from(from_clause) .where(*base_filters) .order_by(self._stock_value_expr().desc()) .limit(5) ) top_items_by_value = [ StockTopItem( item_id=row.id, item_sku=row.sku, item_name=row.name, warehouse_id=row.warehouse_id, warehouse_name=row.warehouse_name, quantity_on_hand=to_decimal(row.quantity_on_hand), unit_cost=to_decimal(row.average_cost), stock_value=to_decimal(row.stock_value), ) for row in top_result.all() ] count_result = await self._session.execute( select(func.count()) .select_from(from_clause) .where(*base_filters) ) rows_total = int(count_result.scalar_one() or 0) total_pages = max(1, ceil(rows_total / page_size)) if page_size else 1 page = max(1, min(page, total_pages)) skip = (page - 1) * page_size rows_result = await self._session.execute( select( ItemModel.id.label("item_id"), ItemModel.sku.label("item_sku"), ItemModel.name.label("item_name"), ItemModel.category_id, ItemCategoryModel.code.label("category_code"), ItemCategoryModel.name.label("category_name"), ItemModel.item_type_id, ItemTypeModel.code.label("item_type_code"), ItemTypeModel.name.label("item_type_name"), WarehouseModel.id.label("warehouse_id"), WarehouseModel.code.label("warehouse_code"), WarehouseModel.name.label("warehouse_name"), func.coalesce(opening_subq.c.opening_qty, 0).label("opening_qty"), InventoryBalanceModel.quantity_on_hand, InventoryBalanceModel.quantity_reserved, ( InventoryBalanceModel.quantity_on_hand - InventoryBalanceModel.quantity_reserved ).label("available_qty"), ItemModel.base_unit_id, BaseUnitModel.code.label("base_unit_code"), BaseUnitModel.name.label("base_unit_name"), InventoryBalanceModel.average_cost.label("unit_cost"), stock_value_expr, ItemModel.reorder_level, status_expr, ItemModel.is_active, ) .select_from(from_clause) .where(*base_filters) .order_by(ItemModel.sku.asc(), WarehouseModel.code.asc()) .offset(skip) .limit(page_size) ) rows = [ StockReportRow( item_id=row.item_id, item_sku=row.item_sku, item_name=row.item_name, category_id=row.category_id, category_code=row.category_code, category_name=row.category_name, item_type_id=row.item_type_id, item_type_code=row.item_type_code, item_type_name=row.item_type_name, warehouse_id=row.warehouse_id, warehouse_code=row.warehouse_code, warehouse_name=row.warehouse_name, opening_qty=to_decimal(row.opening_qty), quantity_on_hand=to_decimal(row.quantity_on_hand), quantity_reserved=to_decimal(row.quantity_reserved), available_qty=to_decimal(row.available_qty), base_unit_id=row.base_unit_id, base_unit_code=row.base_unit_code, base_unit_name=row.base_unit_name, unit_cost=to_decimal(row.unit_cost), stock_value=to_decimal(row.stock_value), reorder_level=to_decimal(row.reorder_level), stock_status=StockReportStatus(row.stock_status), is_active=bool(row.is_active), ) for row in rows_result.all() ] return StockReport( from_date=from_date, to_date=to_date, report_basis="current_stock", total_items=total_items, total_quantity=total_quantity, total_stock_value=total_stock_value, total_warehouses=total_warehouses, low_stock_items=low_stock_items, by_warehouse_value=by_warehouse_value, status_overview=status_overview, top_items_by_value=top_items_by_value, rows=rows, rows_total=rows_total, page=page, page_size=page_size, total_pages=total_pages, )