from datetime import date, datetime, timedelta from decimal import Decimal from app.application.company_access import resolve_company_for_user from app.application.exceptions import ConflictError, NotFoundError, ValidationError from app.domain.entities.company import Company from app.domain.entities.inventory import ( BaseUnit, Brand, Department, DepartmentIssue, DepartmentIssueDashboard, DepartmentIssueLine, DepartmentIssueSummary, DashboardBreakdownSlice, DashboardKpiMetric, DashboardTrendPoint, InventoryBalance, InventoryTransaction, Item, ItemCategory, ItemGroup, ItemTransaction, ItemTransactionLine, ItemType, Location, QuickReportLink, StockReport, StockTransfer, StockTransferLine, UnitType, Warehouse, ) from app.domain.entities.user_registration import UserRegistration from app.domain.enums import ( AmountType, DepartmentIssuePriority, DepartmentIssueReason, DepartmentIssueStatus, DepartmentIssueType, DepartmentStatus, InventoryTxnDirection, InventoryTxnType, ItemTransactionReferenceType, ItemTransactionStatus, LowStockAlertTarget, PricingModel, StockReportBasis, StockReportStatus, StockTransferPriority, StockTransferReason, StockTransferStatus, UnitKind, WarehousePriority, WarehouseStatus, WarehouseType, ) from app.domain.repositories.company_repository import CompanyRepository from app.domain.repositories.inventory_repository import InventoryRepository class InventoryService: def __init__( self, repository: InventoryRepository, company_repository: CompanyRepository, ) -> None: self._repository = repository self._companies = company_repository # ---- Categories ---- async def create_category( self, user: UserRegistration, company_id: str, data: dict ) -> ItemCategory: await self._get_user_company(user, company_id) if await self._repository.get_category_by_code(company_id, data["code"]): raise ConflictError(f"Category code '{data['code']}' already exists") category = ItemCategory( company_id=company_id, code=data["code"], name=data["name"], description=data.get("description"), is_active=data.get("is_active", True), ) return await self._repository.create_category(category) async def get_category( self, user: UserRegistration, company_id: str, category_id: str ) -> ItemCategory: await self._get_user_company(user, company_id) category = await self._repository.get_category(category_id, company_id) if not category: raise NotFoundError("Category not found") return category async def list_categories( self, user: UserRegistration, company_id: str, is_active: bool | None = None, skip: int = 0, limit: int = 100, ) -> tuple[list[ItemCategory], int]: await self._get_user_company(user, company_id) items = await self._repository.list_categories(company_id, is_active, skip, limit) total = await self._repository.count_categories(company_id, is_active) return items, total async def update_category( self, user: UserRegistration, company_id: str, category_id: str, data: dict ) -> ItemCategory: category = await self.get_category(user, company_id, category_id) if "code" in data and data["code"] != category.code: if await self._repository.get_category_by_code(company_id, data["code"]): raise ConflictError(f"Category code '{data['code']}' already exists") category.code = data["code"] if "name" in data: category.name = data["name"] if "description" in data: category.description = data["description"] if "is_active" in data: category.is_active = data["is_active"] category.updated_at = datetime.utcnow() updated = await self._repository.update_category(category_id, category) if not updated: raise NotFoundError("Category not found") return updated async def delete_category( self, user: UserRegistration, company_id: str, category_id: str ) -> None: await self.get_category(user, company_id, category_id) if not await self._repository.delete_category(category_id): raise NotFoundError("Category not found") # ---- Base units ---- async def create_base_unit( self, user: UserRegistration, company_id: str, data: dict ) -> BaseUnit: await self._get_user_company(user, company_id) if await self._repository.get_base_unit_by_code(company_id, data["code"]): raise ConflictError(f"Base unit code '{data['code']}' already exists") base_unit = BaseUnit( company_id=company_id, code=data["code"], name=data["name"], description=data.get("description"), is_active=data.get("is_active", True), sort_order=data.get("sort_order"), ) return await self._repository.create_base_unit(base_unit) async def get_base_unit( self, user: UserRegistration, company_id: str, base_unit_id: str ) -> BaseUnit: await self._get_user_company(user, company_id) base_unit = await self._repository.get_base_unit(base_unit_id, company_id) if not base_unit: raise NotFoundError("Base unit not found") return base_unit async def list_base_units( self, user: UserRegistration, company_id: str, is_active: bool | None = None, skip: int = 0, limit: int = 100, ) -> tuple[list[BaseUnit], int]: await self._get_user_company(user, company_id) items = await self._repository.list_base_units(company_id, is_active, skip, limit) total = await self._repository.count_base_units(company_id, is_active) return items, total async def update_base_unit( self, user: UserRegistration, company_id: str, base_unit_id: str, data: dict ) -> BaseUnit: base_unit = await self.get_base_unit(user, company_id, base_unit_id) if "code" in data and data["code"] != base_unit.code: if await self._repository.get_base_unit_by_code(company_id, data["code"]): raise ConflictError(f"Base unit code '{data['code']}' already exists") base_unit.code = data["code"] if "name" in data: base_unit.name = data["name"] if "description" in data: base_unit.description = data["description"] if "is_active" in data: base_unit.is_active = data["is_active"] if "sort_order" in data: base_unit.sort_order = data["sort_order"] base_unit.updated_at = datetime.utcnow() updated = await self._repository.update_base_unit(base_unit_id, base_unit) if not updated: raise NotFoundError("Base unit not found") return updated async def delete_base_unit( self, user: UserRegistration, company_id: str, base_unit_id: str ) -> None: await self.get_base_unit(user, company_id, base_unit_id) if not await self._repository.delete_base_unit(base_unit_id): raise NotFoundError("Base unit not found") # ---- Warehouses ---- def _resolve_warehouse_status(self, data: dict, current: Warehouse | None = None) -> WarehouseStatus: if "status" in data and data["status"] is not None: return WarehouseStatus(data["status"]) if "is_active" in data and data["is_active"] is not None: return WarehouseStatus.ACTIVE if data["is_active"] else WarehouseStatus.INACTIVE if current is not None: return current.status return WarehouseStatus.ACTIVE def _validate_warehouse_location(self, status: WarehouseStatus, data: dict, current: Warehouse | None = None) -> None: if status != WarehouseStatus.ACTIVE: return address_line1 = data.get("address_line1") if address_line1 is None and current is not None: address_line1 = current.address_line1 or current.address country = data.get("country") if country is None and current is not None: country = current.country state_province = data.get("state_province") if state_province is None and current is not None: state_province = current.state_province city = data.get("city") if city is None and current is not None: city = current.city or current.location missing = [] if not (address_line1 or "").strip(): missing.append("address_line1") if not (country or "").strip(): missing.append("country") if not (state_province or "").strip(): missing.append("state_province") if not (city or "").strip(): missing.append("city") if missing: raise ValidationError( f"Required when status is active: {', '.join(missing)}" ) async def create_warehouse( self, user: UserRegistration, company_id: str, data: dict ) -> Warehouse: await self._get_user_company(user, company_id) if await self._repository.get_warehouse_by_code(company_id, data["code"]): raise ConflictError(f"Warehouse code '{data['code']}' already exists") status = self._resolve_warehouse_status(data) self._validate_warehouse_location(status, data) parent_warehouse_id = data.get("parent_warehouse_id") or None if parent_warehouse_id: parent = await self._repository.get_warehouse(parent_warehouse_id, company_id) if not parent: raise NotFoundError("Parent warehouse not found") manager_id = data.get("manager_id") or None if manager_id and not await self._repository.user_exists(manager_id): raise NotFoundError("Manager not found") address_line1 = data.get("address_line1") or data.get("address") city = data.get("city") or data.get("location") notes = data.get("notes") or data.get("remarks") priority = WarehousePriority(data.get("priority") or WarehousePriority.NORMAL.value) warehouse = Warehouse( company_id=company_id, code=data["code"], name=data["name"], warehouse_type=WarehouseType(data["warehouse_type"]), status=status, priority=priority, parent_warehouse_id=parent_warehouse_id, manager_id=manager_id, contact_person=data.get("contact_person"), phone=data.get("phone"), email=data.get("email"), address_line1=address_line1, address_line2=data.get("address_line2"), country=data.get("country"), state_province=data.get("state_province"), city=city, postal_code=data.get("postal_code"), latitude=data.get("latitude"), longitude=data.get("longitude"), address=address_line1 or "", location=city or "", is_active=status == WarehouseStatus.ACTIVE, description=data.get("description"), capacity=data.get("capacity"), capacity_uom=data.get("capacity_uom"), operating_hours=data.get("operating_hours"), notes=notes, remarks=data.get("remarks"), allow_stock_in=bool(data.get("allow_stock_in", True)), allow_stock_out=bool(data.get("allow_stock_out", True)), allow_stock_transfer=bool(data.get("allow_stock_transfer", True)), allow_returns=bool(data.get("allow_returns", True)), attachments=list(data.get("attachments") or []), ) return await self._repository.create_warehouse(warehouse) async def get_warehouse( self, user: UserRegistration, company_id: str, warehouse_id: str ) -> Warehouse: await self._get_user_company(user, company_id) warehouse = await self._repository.get_warehouse(warehouse_id, company_id) if not warehouse: raise NotFoundError("Warehouse not found") return warehouse async def list_warehouses( self, user: UserRegistration, company_id: str, is_active: bool | None = None, warehouse_type: str | None = None, status: str | None = None, skip: int = 0, limit: int = 100, ) -> tuple[list[Warehouse], int]: await self._get_user_company(user, company_id) items = await self._repository.list_warehouses( company_id, is_active, warehouse_type, status, skip, limit ) total = await self._repository.count_warehouses( company_id, is_active, warehouse_type, status ) return items, total async def update_warehouse( self, user: UserRegistration, company_id: str, warehouse_id: str, data: dict ) -> Warehouse: warehouse = await self.get_warehouse(user, company_id, warehouse_id) if "code" in data and data["code"] != warehouse.code: if await self._repository.get_warehouse_by_code(company_id, data["code"]): raise ConflictError(f"Warehouse code '{data['code']}' already exists") warehouse.code = data["code"] if "name" in data: warehouse.name = data["name"] if "warehouse_type" in data: warehouse.warehouse_type = WarehouseType(data["warehouse_type"]) if "status" in data or "is_active" in data: warehouse.status = self._resolve_warehouse_status(data, warehouse) warehouse.is_active = warehouse.status == WarehouseStatus.ACTIVE self._validate_warehouse_location(warehouse.status, data, warehouse) if "priority" in data and data["priority"] is not None: warehouse.priority = WarehousePriority(data["priority"]) if "parent_warehouse_id" in data: parent_warehouse_id = data["parent_warehouse_id"] or None if parent_warehouse_id == warehouse_id: raise ValidationError("Warehouse cannot be its own parent") if parent_warehouse_id: parent = await self._repository.get_warehouse(parent_warehouse_id, company_id) if not parent: raise NotFoundError("Parent warehouse not found") warehouse.parent_warehouse_id = parent_warehouse_id if "manager_id" in data: manager_id = data["manager_id"] or None if manager_id and not await self._repository.user_exists(manager_id): raise NotFoundError("Manager not found") warehouse.manager_id = manager_id if "contact_person" in data: warehouse.contact_person = data["contact_person"] if "phone" in data: warehouse.phone = data["phone"] if "email" in data: warehouse.email = data["email"] if "address_line1" in data or "address" in data: warehouse.address_line1 = data.get("address_line1", data.get("address")) warehouse.address = warehouse.address_line1 or "" if "address_line2" in data: warehouse.address_line2 = data["address_line2"] if "country" in data: warehouse.country = data["country"] if "state_province" in data: warehouse.state_province = data["state_province"] if "city" in data or "location" in data: warehouse.city = data.get("city", data.get("location")) warehouse.location = warehouse.city or "" if "postal_code" in data: warehouse.postal_code = data["postal_code"] if "latitude" in data: warehouse.latitude = data["latitude"] if "longitude" in data: warehouse.longitude = data["longitude"] if "description" in data: warehouse.description = data["description"] if "capacity" in data: warehouse.capacity = data["capacity"] if "capacity_uom" in data: warehouse.capacity_uom = data["capacity_uom"] if "operating_hours" in data: warehouse.operating_hours = data["operating_hours"] if "notes" in data: warehouse.notes = data["notes"] if "remarks" in data: warehouse.remarks = data["remarks"] if "notes" not in data and not warehouse.notes: warehouse.notes = data["remarks"] if "allow_stock_in" in data and data["allow_stock_in"] is not None: warehouse.allow_stock_in = bool(data["allow_stock_in"]) if "allow_stock_out" in data and data["allow_stock_out"] is not None: warehouse.allow_stock_out = bool(data["allow_stock_out"]) if "allow_stock_transfer" in data and data["allow_stock_transfer"] is not None: warehouse.allow_stock_transfer = bool(data["allow_stock_transfer"]) if "allow_returns" in data and data["allow_returns"] is not None: warehouse.allow_returns = bool(data["allow_returns"]) if "attachments" in data and data["attachments"] is not None: warehouse.attachments = list(data["attachments"]) warehouse.updated_at = datetime.utcnow() updated = await self._repository.update_warehouse(warehouse_id, warehouse) if not updated: raise NotFoundError("Warehouse not found") return updated async def add_warehouse_attachments( self, user: UserRegistration, company_id: str, warehouse_id: str, paths: list[str], ) -> Warehouse: warehouse = await self.get_warehouse(user, company_id, warehouse_id) warehouse.attachments = list(warehouse.attachments or []) + paths warehouse.updated_at = datetime.utcnow() updated = await self._repository.update_warehouse(warehouse_id, warehouse) if not updated: raise NotFoundError("Warehouse not found") return updated async def delete_warehouse( self, user: UserRegistration, company_id: str, warehouse_id: str ) -> None: await self.get_warehouse(user, company_id, warehouse_id) if not await self._repository.delete_warehouse(warehouse_id): raise NotFoundError("Warehouse not found") # ---- Brands ---- async def create_brand(self, user: UserRegistration, company_id: str, data: dict) -> Brand: await self._get_user_company(user, company_id) code = (data.get("code") or "").strip() or await self._repository.get_next_brand_code( company_id ) if await self._repository.get_brand_by_code(company_id, code): raise ConflictError(f"Brand code '{code}' already exists") brand = Brand( company_id=company_id, code=code, name=data["name"], description=data.get("description"), logo=data.get("logo"), website=data.get("website"), is_active=data.get("is_active", True), contact_person=data.get("contact_person"), email=data.get("email"), phone=data.get("phone"), address=data.get("address"), ) return await self._repository.create_brand(brand) async def get_brand(self, user: UserRegistration, company_id: str, brand_id: str) -> Brand: await self._get_user_company(user, company_id) brand = await self._repository.get_brand(brand_id, company_id) if not brand: raise NotFoundError("Brand not found") return brand async def list_brands( self, user: UserRegistration, company_id: str, is_active: bool | None = None, skip: int = 0, limit: int = 100, ) -> tuple[list[Brand], int]: await self._get_user_company(user, company_id) items = await self._repository.list_brands(company_id, is_active, skip, limit) total = await self._repository.count_brands(company_id, is_active) return items, total async def update_brand( self, user: UserRegistration, company_id: str, brand_id: str, data: dict ) -> Brand: brand = await self.get_brand(user, company_id, brand_id) if "code" in data and data["code"] and data["code"] != brand.code: if await self._repository.get_brand_by_code(company_id, data["code"]): raise ConflictError(f"Brand code '{data['code']}' already exists") brand.code = data["code"] for field in ( "name", "description", "logo", "website", "is_active", "contact_person", "email", "phone", "address", ): if field in data: setattr(brand, field, data[field]) brand.updated_at = datetime.utcnow() updated = await self._repository.update_brand(brand_id, brand) if not updated: raise NotFoundError("Brand not found") return updated async def delete_brand( self, user: UserRegistration, company_id: str, brand_id: str ) -> None: await self.get_brand(user, company_id, brand_id) if not await self._repository.delete_brand(brand_id): raise NotFoundError("Brand not found") # ---- Item types ---- async def create_item_type( self, user: UserRegistration, company_id: str, data: dict ) -> ItemType: await self._get_user_company(user, company_id) if await self._repository.get_item_type_by_code(company_id, data["code"]): raise ConflictError(f"Item type code '{data['code']}' already exists") category_id = data.get("category_id") or None if category_id: await self.get_category(user, company_id, category_id) item_type = ItemType( company_id=company_id, category_id=category_id, code=data["code"], name=data["name"], description=data.get("description"), is_active=data.get("is_active", True), ) return await self._repository.create_item_type(item_type) async def get_item_type( self, user: UserRegistration, company_id: str, item_type_id: str ) -> ItemType: await self._get_user_company(user, company_id) item_type = await self._repository.get_item_type(item_type_id, company_id) if not item_type: raise NotFoundError("Item type not found") return item_type async def list_item_types( self, user: UserRegistration, company_id: str, is_active: bool | None = None, category_id: str | None = None, skip: int = 0, limit: int = 100, ) -> tuple[list[ItemType], int]: await self._get_user_company(user, company_id) items = await self._repository.list_item_types( company_id, is_active, category_id, skip, limit ) total = await self._repository.count_item_types(company_id, is_active, category_id) return items, total async def update_item_type( self, user: UserRegistration, company_id: str, item_type_id: str, data: dict ) -> ItemType: item_type = await self.get_item_type(user, company_id, item_type_id) if "code" in data and data["code"] != item_type.code: if await self._repository.get_item_type_by_code(company_id, data["code"]): raise ConflictError(f"Item type code '{data['code']}' already exists") item_type.code = data["code"] if "name" in data: item_type.name = data["name"] if "description" in data: item_type.description = data["description"] if "is_active" in data: item_type.is_active = data["is_active"] if "category_id" in data: category_id = data["category_id"] or None if category_id: await self.get_category(user, company_id, category_id) item_type.category_id = category_id item_type.updated_at = datetime.utcnow() updated = await self._repository.update_item_type(item_type_id, item_type) if not updated: raise NotFoundError("Item type not found") return updated async def delete_item_type( self, user: UserRegistration, company_id: str, item_type_id: str ) -> None: await self.get_item_type(user, company_id, item_type_id) if not await self._repository.delete_item_type(item_type_id): raise NotFoundError("Item type not found") # ---- Groups ---- async def _validate_group_category_and_type( self, user: UserRegistration, company_id: str, category_id: str, item_type_id: str ) -> None: await self.get_category(user, company_id, category_id) item_type = await self.get_item_type(user, company_id, item_type_id) if item_type.category_id and item_type.category_id != category_id: raise ValidationError("Item type does not belong to the selected category") async def create_group( self, user: UserRegistration, company_id: str, data: dict ) -> ItemGroup: await self._get_user_company(user, company_id) if await self._repository.get_group_by_code(company_id, data["code"]): raise ConflictError(f"Group code '{data['code']}' already exists") category_id = data["category_id"] item_type_id = data["item_type_id"] await self._validate_group_category_and_type( user, company_id, category_id, item_type_id ) group = ItemGroup( company_id=company_id, category_id=category_id, item_type_id=item_type_id, code=data["code"], name=data["name"], description=data["description"], is_active=data.get("is_active", True), sort_order=data.get("sort_order"), icon=data.get("icon"), remarks=data.get("remarks"), ) return await self._repository.create_group(group) async def get_group( self, user: UserRegistration, company_id: str, group_id: str ) -> ItemGroup: await self._get_user_company(user, company_id) group = await self._repository.get_group(group_id, company_id) if not group: raise NotFoundError("Group not found") return group async def list_groups( self, user: UserRegistration, 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, ) -> tuple[list[ItemGroup], int]: await self._get_user_company(user, company_id) items = await self._repository.list_groups( company_id, is_active, category_id, item_type_id, skip, limit ) total = await self._repository.count_groups( company_id, is_active, category_id, item_type_id ) return items, total async def update_group( self, user: UserRegistration, company_id: str, group_id: str, data: dict ) -> ItemGroup: group = await self.get_group(user, company_id, group_id) if "code" in data and data["code"] != group.code: if await self._repository.get_group_by_code(company_id, data["code"]): raise ConflictError(f"Group code '{data['code']}' already exists") group.code = data["code"] if "category_id" in data and data["category_id"]: group.category_id = data["category_id"] if "item_type_id" in data and data["item_type_id"]: group.item_type_id = data["item_type_id"] if "category_id" in data or "item_type_id" in data: await self._validate_group_category_and_type( user, company_id, group.category_id, group.item_type_id ) if "name" in data: group.name = data["name"] if "description" in data: group.description = data["description"] if "is_active" in data: group.is_active = data["is_active"] if "sort_order" in data: group.sort_order = data["sort_order"] if "icon" in data: group.icon = data["icon"] if "remarks" in data: group.remarks = data["remarks"] group.updated_at = datetime.utcnow() updated = await self._repository.update_group(group_id, group) if not updated: raise NotFoundError("Group not found") return updated async def delete_group( self, user: UserRegistration, company_id: str, group_id: str ) -> None: await self.get_group(user, company_id, group_id) if not await self._repository.delete_group(group_id): raise NotFoundError("Group not found") # ---- Unit types ---- async def create_unit_type( self, user: UserRegistration, company_id: str, data: dict ) -> UnitType: await self._get_user_company(user, company_id) if await self._repository.get_unit_type_by_code(company_id, data["code"]): raise ConflictError(f"Unit type code '{data['code']}' already exists") base_unit_id = data.get("base_unit_id") or None if base_unit_id: await self.get_base_unit(user, company_id, base_unit_id) conversion_rate = data.get("conversion_rate", data.get("pack_size", "1")) unit_type = UnitType( company_id=company_id, code=data["code"], name=data["name"], base_unit_id=base_unit_id, unit_kind=UnitKind(data.get("unit_kind", UnitKind.INDIVIDUAL.value)), conversion_rate=Decimal(str(conversion_rate)), decimal_places=int(data.get("decimal_places", 2)), is_active=data.get("is_active", True), sort_order=data.get("sort_order"), description=data.get("description"), remarks=data.get("remarks"), ) return await self._repository.create_unit_type(unit_type) async def get_unit_type( self, user: UserRegistration, company_id: str, unit_type_id: str ) -> UnitType: await self._get_user_company(user, company_id) unit_type = await self._repository.get_unit_type(unit_type_id, company_id) if not unit_type: raise NotFoundError("Unit type not found") return unit_type async def list_unit_types( self, user: UserRegistration, company_id: str, is_active: bool | None = None, skip: int = 0, limit: int = 100, ) -> tuple[list[UnitType], int]: await self._get_user_company(user, company_id) items = await self._repository.list_unit_types(company_id, is_active, skip, limit) total = await self._repository.count_unit_types(company_id, is_active) return items, total async def update_unit_type( self, user: UserRegistration, company_id: str, unit_type_id: str, data: dict ) -> UnitType: unit_type = await self.get_unit_type(user, company_id, unit_type_id) if "code" in data and data["code"] != unit_type.code: if await self._repository.get_unit_type_by_code(company_id, data["code"]): raise ConflictError(f"Unit type code '{data['code']}' already exists") unit_type.code = data["code"] if "name" in data: unit_type.name = data["name"] if "unit_kind" in data: unit_type.unit_kind = UnitKind(data["unit_kind"]) if "conversion_rate" in data: unit_type.conversion_rate = Decimal(str(data["conversion_rate"])) elif "pack_size" in data: unit_type.conversion_rate = Decimal(str(data["pack_size"])) if "decimal_places" in data: unit_type.decimal_places = int(data["decimal_places"]) if "base_unit_id" in data: base_unit_id = data["base_unit_id"] or None if base_unit_id: await self.get_base_unit(user, company_id, base_unit_id) unit_type.base_unit_id = base_unit_id if "is_active" in data: unit_type.is_active = data["is_active"] if "sort_order" in data: unit_type.sort_order = data["sort_order"] if "description" in data: unit_type.description = data["description"] if "remarks" in data: unit_type.remarks = data["remarks"] unit_type.updated_at = datetime.utcnow() updated = await self._repository.update_unit_type(unit_type_id, unit_type) if not updated: raise NotFoundError("Unit type not found") return updated async def delete_unit_type( self, user: UserRegistration, company_id: str, unit_type_id: str ) -> None: await self.get_unit_type(user, company_id, unit_type_id) if not await self._repository.delete_unit_type(unit_type_id): raise NotFoundError("Unit type not found") # ---- Items ---- async def create_item(self, user: UserRegistration, company_id: str, data: dict) -> Item: await self._get_user_company(user, company_id) sku = (data.get("sku") or "").strip() or await self._repository.get_next_item_code( company_id ) if await self._repository.get_item_by_sku(company_id, sku): raise ConflictError(f"Item code '{sku}' already exists") barcode = (data.get("barcode") or "").strip() or None if barcode and await self._repository.get_item_by_barcode(company_id, barcode): raise ConflictError(f"Barcode '{barcode}' already exists") await self.get_category(user, company_id, data["category_id"]) await self.get_item_type(user, company_id, data["item_type_id"]) await self.get_group(user, company_id, data["group_id"]) await self.get_base_unit(user, company_id, data["base_unit_id"]) await self.get_warehouse(user, company_id, data["warehouse_id"]) unit_type_id = data.get("unit_type_id") if unit_type_id: unit = await self._repository.get_unit_type(unit_type_id, company_id) if not unit: raise NotFoundError("Unit type not found") item = Item( company_id=company_id, sku=sku, name=data["name"], barcode=barcode, image=data.get("image"), description=data.get("description"), specifications=data.get("specifications"), remarks=data.get("remarks"), category_id=data["category_id"], item_type_id=data["item_type_id"], group_id=data["group_id"], brand_name=data.get("brand_name"), base_unit_id=data["base_unit_id"], warehouse_id=data["warehouse_id"], unit_type_id=unit_type_id, pricing_model=PricingModel(data.get("pricing_model", PricingModel.AVERAGE_COST.value)), purchase_price=Decimal(str(data.get("purchase_price", 0))), sale_price=Decimal(str(data["sale_price"])), tax_percent=Decimal(str(data.get("tax_percent", 0))), reorder_level=Decimal(str(data.get("reorder_level", 0))), track_inventory=data.get("track_inventory", True), inventory_account_id=data.get("inventory_account_id"), expense_account_id=data.get("expense_account_id"), cogs_account_id=data.get("cogs_account_id"), is_active=data.get("is_active", True), ) created = await self._repository.create_item(item) opening_qty = Decimal(str(data.get("opening_stock_qty", 0))) if opening_qty > 0: await self._repository.create_transaction( InventoryTransaction( company_id=company_id, item_id=created.id or "", warehouse_id=created.warehouse_id, txn_type=InventoryTxnType.OPENING, txn_date=date.today(), quantity_in=opening_qty, quantity_out=Decimal("0"), unit_cost=created.purchase_price, reference_type="item", reference_id=created.id, reference_number=created.sku, notes="Opening stock", ) ) return await self.get_item(user, company_id, created.id or "") async def get_item(self, user: UserRegistration, company_id: str, item_id: str) -> Item: await self._get_user_company(user, company_id) item = await self._repository.get_item(item_id, company_id) if not item: raise NotFoundError("Item not found") return item async def list_items( self, user: UserRegistration, 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, ) -> tuple[list[Item], int]: await self._get_user_company(user, company_id) items = await self._repository.list_items( company_id, category_id, item_type_id, group_id, warehouse_id, is_active, skip, limit ) total = await self._repository.count_items( company_id, category_id, item_type_id, group_id, warehouse_id, is_active ) return items, total async def update_item( self, user: UserRegistration, company_id: str, item_id: str, data: dict ) -> Item: item = await self.get_item(user, company_id, item_id) if "sku" in data and data["sku"] and data["sku"] != item.sku: if await self._repository.get_item_by_sku(company_id, data["sku"]): raise ConflictError(f"Item code '{data['sku']}' already exists") item.sku = data["sku"] if "barcode" in data: barcode = (data["barcode"] or "").strip() or None if barcode and barcode != item.barcode: if await self._repository.get_item_by_barcode(company_id, barcode): raise ConflictError(f"Barcode '{barcode}' already exists") item.barcode = barcode if "image" in data: item.image = data["image"] if "category_id" in data and data["category_id"]: await self.get_category(user, company_id, data["category_id"]) item.category_id = data["category_id"] if "item_type_id" in data and data["item_type_id"]: await self.get_item_type(user, company_id, data["item_type_id"]) item.item_type_id = data["item_type_id"] if "group_id" in data and data["group_id"]: await self.get_group(user, company_id, data["group_id"]) item.group_id = data["group_id"] if "base_unit_id" in data and data["base_unit_id"]: await self.get_base_unit(user, company_id, data["base_unit_id"]) item.base_unit_id = data["base_unit_id"] if "warehouse_id" in data and data["warehouse_id"]: await self.get_warehouse(user, company_id, data["warehouse_id"]) item.warehouse_id = data["warehouse_id"] if "unit_type_id" in data: unit_type_id = data["unit_type_id"] or None if unit_type_id: unit = await self._repository.get_unit_type(unit_type_id, company_id) if not unit: raise NotFoundError("Unit type not found") item.unit_type_id = unit_type_id for field in ( "name", "description", "specifications", "remarks", "brand_name", "track_inventory", "inventory_account_id", "expense_account_id", "cogs_account_id", "is_active", ): if field in data: setattr(item, field, data[field]) if "pricing_model" in data: item.pricing_model = PricingModel(data["pricing_model"]) if "purchase_price" in data: item.purchase_price = Decimal(str(data["purchase_price"])) if "sale_price" in data: item.sale_price = Decimal(str(data["sale_price"])) if "tax_percent" in data: item.tax_percent = Decimal(str(data["tax_percent"])) if "reorder_level" in data: item.reorder_level = Decimal(str(data["reorder_level"])) item.updated_at = datetime.utcnow() updated = await self._repository.update_item(item_id, item) if not updated: raise NotFoundError("Item not found") return updated async def delete_item(self, user: UserRegistration, company_id: str, item_id: str) -> None: await self.get_item(user, company_id, item_id) if not await self._repository.delete_item(item_id): raise NotFoundError("Item not found") # ---- Stock / transactions ---- async def get_balance( self, user: UserRegistration, company_id: str, item_id: str, warehouse_id: str | None = None, ) -> InventoryBalance: await self._get_user_company(user, company_id) item = await self.get_item(user, company_id, item_id) resolved_warehouse_id = warehouse_id or item.warehouse_id if warehouse_id: warehouse = await self._repository.get_warehouse(warehouse_id, company_id) if not warehouse: raise NotFoundError("Warehouse not found") balance = await self._repository.get_balance( company_id, item_id, resolved_warehouse_id ) if not balance: return InventoryBalance( company_id=company_id, item_id=item_id, warehouse_id=resolved_warehouse_id, ) return balance async def list_balances( self, user: UserRegistration, company_id: str, skip: int = 0, limit: int = 100, warehouse_id: str | None = None, item_id: str | None = None, ) -> tuple[list[InventoryBalance], int]: await self._get_user_company(user, company_id) if warehouse_id: warehouse = await self._repository.get_warehouse(warehouse_id, company_id) if not warehouse: raise NotFoundError("Warehouse not found") items = await self._repository.list_balances( company_id, skip, limit, warehouse_id, item_id ) total = await self._repository.count_balances(company_id, warehouse_id, item_id) return items, total async def create_transaction( self, user: UserRegistration, company_id: str, data: dict ) -> InventoryTransaction: await self._get_user_company(user, company_id) item = await self.get_item(user, company_id, data["item_id"]) if not item.track_inventory: raise ValidationError("Item does not track inventory") warehouse_id = data.get("warehouse_id") or item.warehouse_id warehouse = await self._repository.get_warehouse(warehouse_id, company_id) if not warehouse: raise NotFoundError("Warehouse not found") qty_in = Decimal(str(data.get("quantity_in", 0))) qty_out = Decimal(str(data.get("quantity_out", 0))) if qty_in < 0 or qty_out < 0: raise ValidationError("Quantities must be non-negative") if qty_in == 0 and qty_out == 0: raise ValidationError("Either quantity_in or quantity_out is required") if qty_in > 0 and qty_out > 0: raise ValidationError("Cannot have both quantity_in and quantity_out") txn = InventoryTransaction( company_id=company_id, item_id=data["item_id"], warehouse_id=warehouse_id, txn_type=InventoryTxnType(data["txn_type"]), txn_date=data["txn_date"], quantity_in=qty_in, quantity_out=qty_out, unit_cost=Decimal(str(data.get("unit_cost", 0))), reference_type=data.get("reference_type"), reference_id=data.get("reference_id"), reference_number=data.get("reference_number"), notes=data.get("notes"), ) created, _ = await self._repository.create_transaction(txn) return created async def list_transactions( self, user: UserRegistration, company_id: str, item_id: str | None = None, skip: int = 0, limit: int = 100, ) -> tuple[list[InventoryTransaction], int]: await self._get_user_company(user, company_id) items = await self._repository.list_transactions(company_id, item_id, skip, limit) total = await self._repository.count_transactions(company_id, item_id) return items, total # ---- Item transactions (Add Item Transaction form) ---- _IN_TYPES = { InventoryTxnType.OPENING, InventoryTxnType.PURCHASE_RECEIPT, InventoryTxnType.ADJUSTMENT_IN, InventoryTxnType.TRANSFER_IN, } _OUT_TYPES = { InventoryTxnType.SALE, InventoryTxnType.PURCHASE_RETURN, InventoryTxnType.ADJUSTMENT_OUT, InventoryTxnType.TRANSFER_OUT, InventoryTxnType.ISSUE_TO_DEPARTMENT, } async def create_item_transaction( self, user: UserRegistration, company_id: str, data: dict ) -> ItemTransaction: await self._get_user_company(user, company_id) warehouse = await self._repository.get_warehouse(data["warehouse_id"], company_id) if not warehouse: raise NotFoundError("Warehouse not found") vendor_id = data.get("vendor_id") or None if vendor_id and not await self._repository.vendor_exists(vendor_id, company_id): raise NotFoundError("Vendor not found") txn_type = InventoryTxnType(data["txn_type"]) direction = InventoryTxnDirection(data["direction"]) self._validate_direction(txn_type, direction) lines = await self._build_item_txn_lines(company_id, data.get("lines") or []) totals = self._sum_item_txn_totals( lines, transport_charges=Decimal(str(data.get("transport_charges", 0))), rounding=Decimal(str(data.get("rounding", 0))), ) txn_number = await self._repository.get_next_item_transaction_number(company_id) reference_type = data.get("reference_type") if isinstance(reference_type, str): reference_type = ItemTransactionReferenceType(reference_type) doc = ItemTransaction( company_id=company_id, txn_number=txn_number, txn_type=txn_type, txn_date=data["txn_date"], reference_type=reference_type, reference_number=data.get("reference_number"), direction=direction, warehouse_id=data["warehouse_id"], vendor_id=vendor_id, po_date=data.get("po_date"), expected_date=data.get("expected_date"), grn_number=data.get("grn_number"), remarks=data.get("remarks"), internal_note=data.get("internal_note"), tags=list(data.get("tags") or []), attachments=list(data.get("attachments") or []), status=ItemTransactionStatus.DRAFT, total_quantity=totals["total_quantity"], subtotal=totals["subtotal"], discount_amount=totals["discount_amount"], tax_amount=totals["tax_amount"], transport_charges=totals["transport_charges"], rounding=totals["rounding"], grand_total=totals["grand_total"], created_by=user.id, lines=lines, ) created = await self._repository.create_item_transaction(doc) if data.get("post_now"): return await self.post_item_transaction(user, company_id, created.id or "") return created async def get_item_transaction( self, user: UserRegistration, company_id: str, txn_id: str ) -> ItemTransaction: await self._get_user_company(user, company_id) doc = await self._repository.get_item_transaction(txn_id, company_id) if not doc: raise NotFoundError("Item transaction not found") return doc async def list_item_transactions( self, user: UserRegistration, company_id: str, status: ItemTransactionStatus | None = None, direction: InventoryTxnDirection | None = None, warehouse_id: str | None = None, skip: int = 0, limit: int = 100, ) -> tuple[list[ItemTransaction], int]: await self._get_user_company(user, company_id) status_val = status.value if status else None direction_val = direction.value if direction else None items = await self._repository.list_item_transactions( company_id, status_val, direction_val, warehouse_id, skip, limit ) total = await self._repository.count_item_transactions( company_id, status_val, direction_val, warehouse_id ) return items, total async def update_item_transaction( self, user: UserRegistration, company_id: str, txn_id: str, data: dict ) -> ItemTransaction: doc = await self.get_item_transaction(user, company_id, txn_id) if doc.status != ItemTransactionStatus.DRAFT: raise ValidationError("Only draft item transactions can be updated") if "warehouse_id" in data and data["warehouse_id"]: warehouse = await self._repository.get_warehouse(data["warehouse_id"], company_id) if not warehouse: raise NotFoundError("Warehouse not found") doc.warehouse_id = data["warehouse_id"] if "vendor_id" in data: vendor_id = data["vendor_id"] or None if vendor_id and not await self._repository.vendor_exists(vendor_id, company_id): raise NotFoundError("Vendor not found") doc.vendor_id = vendor_id if "txn_type" in data and data["txn_type"] is not None: doc.txn_type = InventoryTxnType(data["txn_type"]) if "direction" in data and data["direction"] is not None: doc.direction = InventoryTxnDirection(data["direction"]) self._validate_direction(doc.txn_type, doc.direction) if "reference_type" in data: value = data["reference_type"] doc.reference_type = ( ItemTransactionReferenceType(value) if value else None ) for field in ( "txn_date", "reference_number", "po_date", "expected_date", "grn_number", "remarks", "internal_note", ): if field in data: setattr(doc, field, data[field]) if "tags" in data and data["tags"] is not None: doc.tags = list(data["tags"]) if "attachments" in data and data["attachments"] is not None: doc.attachments = list(data["attachments"]) if "lines" in data and data["lines"] is not None: doc.lines = await self._build_item_txn_lines(company_id, data["lines"]) transport = ( Decimal(str(data["transport_charges"])) if "transport_charges" in data and data["transport_charges"] is not None else doc.transport_charges ) rounding = ( Decimal(str(data["rounding"])) if "rounding" in data and data["rounding"] is not None else doc.rounding ) totals = self._sum_item_txn_totals( doc.lines, transport_charges=transport, rounding=rounding ) doc.total_quantity = totals["total_quantity"] doc.subtotal = totals["subtotal"] doc.discount_amount = totals["discount_amount"] doc.tax_amount = totals["tax_amount"] doc.transport_charges = totals["transport_charges"] doc.rounding = totals["rounding"] doc.grand_total = totals["grand_total"] doc.updated_at = datetime.utcnow() updated = await self._repository.update_item_transaction(doc) if not updated: raise NotFoundError("Item transaction not found") return updated async def add_item_transaction_attachments( self, user: UserRegistration, company_id: str, txn_id: str, paths: list[str] ) -> ItemTransaction: doc = await self.get_item_transaction(user, company_id, txn_id) if doc.status == ItemTransactionStatus.CANCELLED: raise ValidationError("Cannot attach files to a cancelled transaction") doc.attachments = list(doc.attachments or []) + paths doc.updated_at = datetime.utcnow() updated = await self._repository.update_item_transaction(doc) if not updated: raise NotFoundError("Item transaction not found") return updated async def post_item_transaction( self, user: UserRegistration, company_id: str, txn_id: str ) -> ItemTransaction: doc = await self.get_item_transaction(user, company_id, txn_id) if doc.status != ItemTransactionStatus.DRAFT: raise ValidationError("Only draft item transactions can be posted") if not doc.lines: raise ValidationError("Item transaction must have at least one line") txn_date = doc.txn_date.date() if isinstance(doc.txn_date, datetime) else doc.txn_date for line in doc.lines: item = await self._repository.get_item(line.item_id, company_id) if not item: raise NotFoundError(f"Item '{line.item_id}' not found") if not item.track_inventory: continue qty_in = line.quantity if doc.direction == InventoryTxnDirection.IN else Decimal("0") qty_out = ( line.quantity if doc.direction == InventoryTxnDirection.OUT else Decimal("0") ) ledger = InventoryTransaction( company_id=company_id, item_id=line.item_id, warehouse_id=doc.warehouse_id, txn_type=doc.txn_type, txn_date=txn_date, quantity_in=qty_in, quantity_out=qty_out, unit_cost=line.unit_cost, reference_type="item_transaction", reference_id=doc.id, reference_number=doc.txn_number, notes=doc.remarks, ) await self._repository.create_transaction(ledger) doc.status = ItemTransactionStatus.POSTED doc.posted_at = datetime.utcnow() doc.updated_at = datetime.utcnow() updated = await self._repository.update_item_transaction(doc) if not updated: raise NotFoundError("Item transaction not found") return updated async def cancel_item_transaction( self, user: UserRegistration, company_id: str, txn_id: str ) -> ItemTransaction: doc = await self.get_item_transaction(user, company_id, txn_id) if doc.status != ItemTransactionStatus.DRAFT: raise ValidationError("Only draft item transactions can be cancelled") doc.status = ItemTransactionStatus.CANCELLED doc.cancelled_at = datetime.utcnow() doc.updated_at = datetime.utcnow() updated = await self._repository.update_item_transaction(doc) if not updated: raise NotFoundError("Item transaction not found") return updated def _validate_direction( self, txn_type: InventoryTxnType, direction: InventoryTxnDirection ) -> None: if direction == InventoryTxnDirection.IN and txn_type in self._OUT_TYPES: raise ValidationError( f"Transaction type '{txn_type.value}' requires direction 'out'" ) if direction == InventoryTxnDirection.OUT and txn_type in self._IN_TYPES: raise ValidationError( f"Transaction type '{txn_type.value}' requires direction 'in'" ) async def _build_item_txn_lines( self, company_id: str, raw_lines: list[dict] ) -> list[ItemTransactionLine]: if not raw_lines: raise ValidationError("Item transaction must have at least one line") lines: list[ItemTransactionLine] = [] for index, raw in enumerate(raw_lines, start=1): item = await self._repository.get_item(raw["item_id"], company_id) if not item: raise NotFoundError(f"Item '{raw['item_id']}' not found") qty = Decimal(str(raw["quantity"])) unit_cost = Decimal(str(raw.get("unit_cost", 0))) if qty <= 0: raise ValidationError("Line quantity must be greater than zero") if unit_cost < 0: raise ValidationError("Unit cost cannot be negative") discount_type = raw.get("discount_type") or AmountType.PERCENT if isinstance(discount_type, str): discount_type = AmountType(discount_type) tax_type = raw.get("tax_type") or AmountType.PERCENT if isinstance(tax_type, str): tax_type = AmountType(tax_type) discount_value = Decimal(str(raw.get("discount_value", 0))) tax_rate = Decimal(str(raw.get("tax_rate", 0))) base_unit_id = raw.get("base_unit_id") or item.base_unit_id if base_unit_id: unit = await self._repository.get_base_unit(base_unit_id, company_id) if not unit: raise NotFoundError(f"Base unit '{base_unit_id}' not found") base = (qty * unit_cost).quantize(Decimal("0.01")) if discount_type == AmountType.PERCENT: discount_amount = (base * discount_value / Decimal("100")).quantize( Decimal("0.01") ) else: discount_amount = discount_value.quantize(Decimal("0.01")) if discount_amount > base: raise ValidationError("Discount cannot exceed line amount") after_discount = (base - discount_amount).quantize(Decimal("0.01")) if tax_type == AmountType.PERCENT: tax_amount = (after_discount * tax_rate / Decimal("100")).quantize( Decimal("0.01") ) else: tax_amount = tax_rate.quantize(Decimal("0.01")) lines.append( ItemTransactionLine( line_number=index, item_id=raw["item_id"], item_sku=item.sku, item_name=item.name, description=raw.get("description") or item.name, batch_lot_no=raw.get("batch_lot_no"), expiry_date=raw.get("expiry_date"), base_unit_id=base_unit_id, quantity=qty, unit_cost=unit_cost, discount_type=discount_type, discount_value=discount_value, discount_amount=discount_amount, tax_type=tax_type, tax_rate=tax_rate, tax_amount=tax_amount, line_total=(after_discount + tax_amount).quantize(Decimal("0.01")), ) ) return lines def _sum_item_txn_totals( self, lines: list[ItemTransactionLine], *, transport_charges: Decimal, rounding: Decimal, ) -> dict[str, Decimal]: total_quantity = Decimal("0.0000") subtotal = Decimal("0.00") discount_amount = Decimal("0.00") tax_amount = Decimal("0.00") for line in lines: total_quantity += line.quantity base = (line.quantity * line.unit_cost).quantize(Decimal("0.01")) subtotal += base discount_amount += line.discount_amount tax_amount += line.tax_amount transport_charges = transport_charges.quantize(Decimal("0.01")) rounding = rounding.quantize(Decimal("0.01")) grand_total = ( subtotal - discount_amount + tax_amount + transport_charges + rounding ).quantize(Decimal("0.01")) return { "total_quantity": total_quantity.quantize(Decimal("0.0001")), "subtotal": subtotal, "discount_amount": discount_amount, "tax_amount": tax_amount, "transport_charges": transport_charges, "rounding": rounding, "grand_total": grand_total, } # ---- Stock transfers (Create Stock Transfer form) ---- _STOCK_TRANSFER_EDITABLE = {StockTransferStatus.DRAFT, StockTransferStatus.REVIEW} _STOCK_TRANSFER_FLOW = { StockTransferStatus.DRAFT: StockTransferStatus.REVIEW, StockTransferStatus.REVIEW: StockTransferStatus.APPROVED, StockTransferStatus.APPROVED: StockTransferStatus.PICK_PACK, StockTransferStatus.PICK_PACK: StockTransferStatus.IN_TRANSIT, StockTransferStatus.IN_TRANSIT: StockTransferStatus.RECEIVE, StockTransferStatus.RECEIVE: StockTransferStatus.COMPLETED, } async def create_stock_transfer( self, user: UserRegistration, company_id: str, data: dict ) -> StockTransfer: await self._get_user_company(user, company_id) from_wh, to_wh = await self._validate_stock_transfer_warehouses( company_id, data["from_warehouse_id"], data["to_warehouse_id"] ) lines = await self._build_stock_transfer_lines( company_id, data.get("lines") or [], data["from_warehouse_id"] ) totals = self._sum_stock_transfer_totals(lines) transfer_number = await self._repository.get_next_stock_transfer_number(company_id) status = ( StockTransferStatus.REVIEW if data.get("submit_for_review") else StockTransferStatus.DRAFT ) now = datetime.utcnow() doc = StockTransfer( company_id=company_id, transfer_number=transfer_number, transfer_date=data["transfer_date"], expected_delivery_date=data["expected_delivery_date"], priority=StockTransferPriority( data.get("priority") or StockTransferPriority.NORMAL.value ), reason=StockTransferReason(data["reason"]), reference=data.get("reference"), notes=data.get("notes"), from_warehouse_id=from_wh.id or data["from_warehouse_id"], to_warehouse_id=to_wh.id or data["to_warehouse_id"], status=status, total_items=totals["total_items"], total_quantity=totals["total_quantity"], total_transfer_value=totals["total_transfer_value"], created_by=user.id, submitted_at=now if status == StockTransferStatus.REVIEW else None, lines=lines, ) return await self._repository.create_stock_transfer(doc) async def get_stock_transfer( self, user: UserRegistration, company_id: str, transfer_id: str ) -> StockTransfer: await self._get_user_company(user, company_id) doc = await self._repository.get_stock_transfer(transfer_id, company_id) if not doc: raise NotFoundError("Stock transfer not found") return doc async def list_stock_transfers( self, user: UserRegistration, 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, ) -> tuple[list[StockTransfer], int]: await self._get_user_company(user, company_id) items = await self._repository.list_stock_transfers( company_id, status, from_warehouse_id, to_warehouse_id, skip, limit ) total = await self._repository.count_stock_transfers( company_id, status, from_warehouse_id, to_warehouse_id ) return items, total async def update_stock_transfer( self, user: UserRegistration, company_id: str, transfer_id: str, data: dict ) -> StockTransfer: doc = await self.get_stock_transfer(user, company_id, transfer_id) if doc.status not in self._STOCK_TRANSFER_EDITABLE: raise ValidationError("Only draft or review stock transfers can be edited") replace_lines = False if "transfer_date" in data and data["transfer_date"] is not None: doc.transfer_date = data["transfer_date"] if "expected_delivery_date" in data and data["expected_delivery_date"] is not None: doc.expected_delivery_date = data["expected_delivery_date"] if "priority" in data and data["priority"] is not None: doc.priority = StockTransferPriority(data["priority"]) if "reason" in data and data["reason"] is not None: doc.reason = StockTransferReason(data["reason"]) if "reference" in data: doc.reference = data["reference"] if "notes" in data: doc.notes = data["notes"] from_id = data.get("from_warehouse_id", doc.from_warehouse_id) to_id = data.get("to_warehouse_id", doc.to_warehouse_id) if "from_warehouse_id" in data or "to_warehouse_id" in data: await self._validate_stock_transfer_warehouses(company_id, from_id, to_id) doc.from_warehouse_id = from_id doc.to_warehouse_id = to_id if "lines" in data and data["lines"] is not None: doc.lines = await self._build_stock_transfer_lines( company_id, data["lines"], doc.from_warehouse_id ) totals = self._sum_stock_transfer_totals(doc.lines) doc.total_items = totals["total_items"] doc.total_quantity = totals["total_quantity"] doc.total_transfer_value = totals["total_transfer_value"] replace_lines = True doc.updated_at = datetime.utcnow() updated = await self._repository.update_stock_transfer(doc, replace_lines=replace_lines) if not updated: raise NotFoundError("Stock transfer not found") return updated async def delete_stock_transfer( self, user: UserRegistration, company_id: str, transfer_id: str ) -> None: doc = await self.get_stock_transfer(user, company_id, transfer_id) if doc.status != StockTransferStatus.DRAFT: raise ValidationError("Only draft stock transfers can be deleted") if not await self._repository.delete_stock_transfer(transfer_id): raise NotFoundError("Stock transfer not found") async def submit_stock_transfer_for_review( self, user: UserRegistration, company_id: str, transfer_id: str ) -> StockTransfer: return await self._advance_stock_transfer( user, company_id, transfer_id, StockTransferStatus.DRAFT, StockTransferStatus.REVIEW ) async def approve_stock_transfer( self, user: UserRegistration, company_id: str, transfer_id: str ) -> StockTransfer: return await self._advance_stock_transfer( user, company_id, transfer_id, StockTransferStatus.REVIEW, StockTransferStatus.APPROVED ) async def start_pick_pack_stock_transfer( self, user: UserRegistration, company_id: str, transfer_id: str ) -> StockTransfer: return await self._advance_stock_transfer( user, company_id, transfer_id, StockTransferStatus.APPROVED, StockTransferStatus.PICK_PACK, ) async def mark_stock_transfer_in_transit( self, user: UserRegistration, company_id: str, transfer_id: str ) -> StockTransfer: return await self._advance_stock_transfer( user, company_id, transfer_id, StockTransferStatus.PICK_PACK, StockTransferStatus.IN_TRANSIT, ) async def start_receive_stock_transfer( self, user: UserRegistration, company_id: str, transfer_id: str ) -> StockTransfer: return await self._advance_stock_transfer( user, company_id, transfer_id, StockTransferStatus.IN_TRANSIT, StockTransferStatus.RECEIVE, ) async def complete_stock_transfer( self, user: UserRegistration, company_id: str, transfer_id: str ) -> StockTransfer: doc = await self.get_stock_transfer(user, company_id, transfer_id) if doc.status != StockTransferStatus.RECEIVE: raise ValidationError("Only transfers in receive status can be completed") if not doc.lines: raise ValidationError("Stock transfer must have at least one line") if not doc.from_warehouse_id or not doc.to_warehouse_id: raise ValidationError("Stock transfer warehouses are required") if doc.from_warehouse_id == doc.to_warehouse_id: raise ValidationError("Source and destination warehouses must be different") for line in doc.lines: item = await self._repository.get_item(line.item_id, company_id) if not item: raise NotFoundError(f"Item '{line.item_id}' not found") if not item.track_inventory: continue await self._repository.create_transaction( InventoryTransaction( company_id=company_id, item_id=line.item_id, warehouse_id=doc.from_warehouse_id, txn_type=InventoryTxnType.TRANSFER_OUT, txn_date=doc.transfer_date, quantity_in=Decimal("0"), quantity_out=line.transfer_qty, unit_cost=line.unit_cost, reference_type="stock_transfer", reference_id=doc.id, reference_number=doc.transfer_number, notes=f"Transfer out to {doc.to_warehouse_code or doc.to_warehouse_id}", ) ) await self._repository.create_transaction( InventoryTransaction( company_id=company_id, item_id=line.item_id, warehouse_id=doc.to_warehouse_id, txn_type=InventoryTxnType.TRANSFER_IN, txn_date=doc.transfer_date, quantity_in=line.transfer_qty, quantity_out=Decimal("0"), unit_cost=line.unit_cost, reference_type="stock_transfer", reference_id=doc.id, reference_number=doc.transfer_number, notes=f"Transfer in from {doc.from_warehouse_code or doc.from_warehouse_id}", ) ) now = datetime.utcnow() doc.status = StockTransferStatus.COMPLETED doc.completed_at = now if not doc.received_at: doc.received_at = now doc.updated_at = now updated = await self._repository.update_stock_transfer(doc) if not updated: raise NotFoundError("Stock transfer not found") return updated async def cancel_stock_transfer( self, user: UserRegistration, company_id: str, transfer_id: str ) -> StockTransfer: doc = await self.get_stock_transfer(user, company_id, transfer_id) if doc.status in { StockTransferStatus.COMPLETED, StockTransferStatus.CANCELLED, StockTransferStatus.IN_TRANSIT, StockTransferStatus.RECEIVE, }: raise ValidationError("Cannot cancel a completed, cancelled, or in-transit transfer") doc.status = StockTransferStatus.CANCELLED doc.cancelled_at = datetime.utcnow() doc.updated_at = datetime.utcnow() updated = await self._repository.update_stock_transfer(doc) if not updated: raise NotFoundError("Stock transfer not found") return updated async def _advance_stock_transfer( self, user: UserRegistration, company_id: str, transfer_id: str, expected: StockTransferStatus, target: StockTransferStatus, ) -> StockTransfer: doc = await self.get_stock_transfer(user, company_id, transfer_id) if doc.status != expected: raise ValidationError( f"Stock transfer must be '{expected.value}' to move to '{target.value}'" ) if not doc.lines: raise ValidationError("Stock transfer must have at least one line") now = datetime.utcnow() doc.status = target if target == StockTransferStatus.REVIEW: doc.submitted_at = now elif target == StockTransferStatus.APPROVED: doc.approved_at = now elif target == StockTransferStatus.PICK_PACK: doc.picked_at = now elif target == StockTransferStatus.IN_TRANSIT: doc.shipped_at = now elif target == StockTransferStatus.RECEIVE: doc.received_at = now doc.updated_at = now updated = await self._repository.update_stock_transfer(doc) if not updated: raise NotFoundError("Stock transfer not found") return updated async def _validate_stock_transfer_warehouses( self, company_id: str, from_warehouse_id: str, to_warehouse_id: str ) -> tuple[Warehouse, Warehouse]: if from_warehouse_id == to_warehouse_id: raise ValidationError("Source and destination warehouses must be different") from_wh = await self._repository.get_warehouse(from_warehouse_id, company_id) if not from_wh: raise NotFoundError("Source warehouse not found") to_wh = await self._repository.get_warehouse(to_warehouse_id, company_id) if not to_wh: raise NotFoundError("Destination warehouse not found") if not from_wh.allow_stock_transfer: raise ValidationError("Source warehouse does not allow stock transfers") if not to_wh.allow_stock_transfer: raise ValidationError("Destination warehouse does not allow stock transfers") if from_wh.status == WarehouseStatus.INACTIVE or not from_wh.is_active: raise ValidationError("Source warehouse is inactive") if to_wh.status == WarehouseStatus.INACTIVE or not to_wh.is_active: raise ValidationError("Destination warehouse is inactive") return from_wh, to_wh async def _build_stock_transfer_lines( self, company_id: str, raw_lines: list[dict], from_warehouse_id: str ) -> list[StockTransferLine]: if not raw_lines: raise ValidationError("At least one transfer item is required") lines: list[StockTransferLine] = [] for idx, raw in enumerate(raw_lines, start=1): item = await self._repository.get_item(raw["item_id"], company_id) if not item: raise NotFoundError(f"Item '{raw['item_id']}' not found") transfer_qty = Decimal(str(raw["transfer_qty"])) if transfer_qty <= 0: raise ValidationError(f"Transfer qty must be > 0 for item '{item.sku}'") balance = await self._repository.get_balance( company_id, item.id or raw["item_id"], from_warehouse_id ) available = ( (balance.quantity_on_hand - balance.quantity_reserved) if balance else Decimal("0.0000") ) if item.track_inventory and transfer_qty > available: raise ValidationError( f"Transfer qty ({transfer_qty}) exceeds available qty ({available}) " f"for item '{item.sku}' at source warehouse" ) base_unit_id = raw.get("base_unit_id") or item.base_unit_id if base_unit_id: unit = await self._repository.get_base_unit(base_unit_id, company_id) if not unit: raise NotFoundError(f"UOM '{base_unit_id}' not found") if raw.get("unit_cost") is not None: unit_cost = Decimal(str(raw["unit_cost"])) elif balance and balance.average_cost > 0: unit_cost = balance.average_cost else: unit_cost = item.purchase_price line_value = (transfer_qty * unit_cost).quantize(Decimal("0.01")) lines.append( StockTransferLine( line_number=idx, item_id=item.id or raw["item_id"], available_qty=available.quantize(Decimal("0.0001")), transfer_qty=transfer_qty.quantize(Decimal("0.0001")), base_unit_id=base_unit_id, batch_lot_no=raw.get("batch_lot_no"), unit_cost=unit_cost.quantize(Decimal("0.0001")), line_value=line_value, ) ) return lines def _sum_stock_transfer_totals(self, lines: list[StockTransferLine]) -> dict: total_quantity = sum((line.transfer_qty for line in lines), Decimal("0")) total_value = sum((line.line_value for line in lines), Decimal("0")) return { "total_items": len(lines), "total_quantity": total_quantity.quantize(Decimal("0.0001")), "total_transfer_value": total_value.quantize(Decimal("0.01")), } # ---- Locations ---- async def create_location( self, user: UserRegistration, company_id: str, data: dict ) -> Location: await self._get_user_company(user, company_id) if await self._repository.get_location_by_code(company_id, data["code"]): raise ConflictError(f"Location code '{data['code']}' already exists") location = Location( company_id=company_id, code=data["code"], name=data["name"], address=data.get("address"), is_active=bool(data.get("is_active", True)), ) return await self._repository.create_location(location) async def get_location( self, user: UserRegistration, company_id: str, location_id: str ) -> Location: await self._get_user_company(user, company_id) location = await self._repository.get_location(location_id, company_id) if not location: raise NotFoundError("Location not found") return location async def list_locations( self, user: UserRegistration, company_id: str, is_active: bool | None = None, skip: int = 0, limit: int = 100, ) -> tuple[list[Location], int]: await self._get_user_company(user, company_id) items = await self._repository.list_locations(company_id, is_active, skip, limit) total = await self._repository.count_locations(company_id, is_active) return items, total async def update_location( self, user: UserRegistration, company_id: str, location_id: str, data: dict ) -> Location: location = await self.get_location(user, company_id, location_id) if "code" in data and data["code"] != location.code: if await self._repository.get_location_by_code(company_id, data["code"]): raise ConflictError(f"Location code '{data['code']}' already exists") location.code = data["code"] if "name" in data: location.name = data["name"] if "address" in data: location.address = data["address"] if "is_active" in data and data["is_active"] is not None: location.is_active = bool(data["is_active"]) location.updated_at = datetime.utcnow() updated = await self._repository.update_location(location_id, location) if not updated: raise NotFoundError("Location not found") return updated async def delete_location( self, user: UserRegistration, company_id: str, location_id: str ) -> None: await self.get_location(user, company_id, location_id) if not await self._repository.delete_location(location_id): raise NotFoundError("Location not found") # ---- Departments ---- def _resolve_department_status( self, data: dict, current: Department | None = None ) -> DepartmentStatus: if "status" in data and data["status"] is not None: return DepartmentStatus(data["status"]) if "is_active" in data and data["is_active"] is not None: return ( DepartmentStatus.ACTIVE if data["is_active"] else DepartmentStatus.INACTIVE ) if current is not None: return current.status return DepartmentStatus.ACTIVE def _validate_department_required( self, status: DepartmentStatus, data: dict, current: Department | None = None ) -> None: if status != DepartmentStatus.ACTIVE: return head_id = data.get("head_id") if head_id is None and current is not None: head_id = current.head_id location_id = data.get("location_id") if location_id is None and current is not None: location_id = current.location_id missing = [] if not head_id: missing.append("head_id") if not location_id: missing.append("location_id") if missing: raise ValidationError( f"Required when status is active: {', '.join(missing)}" ) async def create_department( self, user: UserRegistration, company_id: str, data: dict ) -> Department: await self._get_user_company(user, company_id) code = str(data["code"]).strip().upper() if len(code) > 6: raise ValidationError("Department code must be at most 6 characters") if await self._repository.get_department_by_code(company_id, code): raise ConflictError(f"Department code '{code}' already exists") status = self._resolve_department_status(data) self._validate_department_required(status, data) head_id = data.get("head_id") or None if head_id and not await self._repository.user_exists(head_id): raise NotFoundError("Head / Incharge not found") location_id = data.get("location_id") or None if location_id: location = await self._repository.get_location(location_id, company_id) if not location: raise NotFoundError("Location not found") parent_department_id = data.get("parent_department_id") or None if parent_department_id: parent = await self._repository.get_department(parent_department_id, company_id) if not parent: raise NotFoundError("Parent department not found") budget = data.get("monthly_issue_budget") department = Department( company_id=company_id, code=code, name=data["name"], head_id=head_id, location_id=location_id, parent_department_id=parent_department_id, monthly_issue_budget=Decimal(str(budget)) if budget is not None else None, description=data.get("description"), status=status, is_active=status == DepartmentStatus.ACTIVE, notify_email=data.get("notify_email"), low_stock_alert=LowStockAlertTarget( data.get("low_stock_alert") or LowStockAlertTarget.HEAD.value ), ) return await self._repository.create_department(department) async def get_department( self, user: UserRegistration, company_id: str, department_id: str ) -> Department: await self._get_user_company(user, company_id) department = await self._repository.get_department(department_id, company_id) if not department: raise NotFoundError("Department not found") return department async def list_departments( self, user: UserRegistration, company_id: str, is_active: bool | None = None, status: str | None = None, location_id: str | None = None, skip: int = 0, limit: int = 100, ) -> tuple[list[Department], int]: await self._get_user_company(user, company_id) items = await self._repository.list_departments( company_id, is_active, status, location_id, skip, limit ) total = await self._repository.count_departments( company_id, is_active, status, location_id ) return items, total async def update_department( self, user: UserRegistration, company_id: str, department_id: str, data: dict ) -> Department: department = await self.get_department(user, company_id, department_id) if "code" in data and data["code"] is not None: code = str(data["code"]).strip().upper() if len(code) > 6: raise ValidationError("Department code must be at most 6 characters") if code != department.code: if await self._repository.get_department_by_code(company_id, code): raise ConflictError(f"Department code '{code}' already exists") department.code = code if "name" in data and data["name"] is not None: department.name = data["name"] if "status" in data or "is_active" in data: department.status = self._resolve_department_status(data, department) department.is_active = department.status == DepartmentStatus.ACTIVE self._validate_department_required(department.status, data, department) if "head_id" in data: head_id = data["head_id"] or None if head_id and not await self._repository.user_exists(head_id): raise NotFoundError("Head / Incharge not found") department.head_id = head_id if "location_id" in data: location_id = data["location_id"] or None if location_id: location = await self._repository.get_location(location_id, company_id) if not location: raise NotFoundError("Location not found") department.location_id = location_id if "parent_department_id" in data: parent_department_id = data["parent_department_id"] or None if parent_department_id == department_id: raise ValidationError("Department cannot be its own parent") if parent_department_id: parent = await self._repository.get_department( parent_department_id, company_id ) if not parent: raise NotFoundError("Parent department not found") department.parent_department_id = parent_department_id if "monthly_issue_budget" in data: budget = data["monthly_issue_budget"] department.monthly_issue_budget = ( Decimal(str(budget)) if budget is not None else None ) if "description" in data: department.description = data["description"] if "notify_email" in data: department.notify_email = data["notify_email"] if "low_stock_alert" in data and data["low_stock_alert"] is not None: department.low_stock_alert = LowStockAlertTarget(data["low_stock_alert"]) department.updated_at = datetime.utcnow() updated = await self._repository.update_department(department_id, department) if not updated: raise NotFoundError("Department not found") return updated async def delete_department( self, user: UserRegistration, company_id: str, department_id: str ) -> None: await self.get_department(user, company_id, department_id) if not await self._repository.delete_department(department_id): raise NotFoundError("Department not found") # ---- Department issues (Issue Items to Department) ---- _DEPARTMENT_ISSUE_EDITABLE = { DepartmentIssueStatus.DRAFT, DepartmentIssueStatus.REVIEW, } async def create_department_issue( self, user: UserRegistration, company_id: str, data: dict ) -> DepartmentIssue: await self._get_user_company(user, company_id) department = await self._repository.get_department(data["department_id"], company_id) if not department: raise NotFoundError("Department not found") if department.status != DepartmentStatus.ACTIVE or not department.is_active: raise ValidationError("Department must be active for item issuance") warehouse = await self._repository.get_warehouse(data["from_warehouse_id"], company_id) if not warehouse: raise NotFoundError("From warehouse not found") if warehouse.status == WarehouseStatus.INACTIVE or not warehouse.is_active: raise ValidationError("From warehouse is inactive") if not warehouse.allow_stock_out: raise ValidationError("From warehouse does not allow stock out") requested_by_id = data.get("requested_by_id") or user.id if requested_by_id and not await self._repository.user_exists(requested_by_id): raise NotFoundError("Requested by user not found") lines = await self._build_department_issue_lines( company_id, data.get("lines") or [], data["from_warehouse_id"] ) totals = self._sum_department_issue_totals(lines) if department.monthly_issue_budget is not None: if totals["total_quantity"] > department.monthly_issue_budget: raise ValidationError( f"Total issue qty ({totals['total_quantity']}) exceeds department " f"monthly issue budget ({department.monthly_issue_budget})" ) status = ( DepartmentIssueStatus.REVIEW if data.get("submit_for_approval") else DepartmentIssueStatus.DRAFT ) now = datetime.utcnow() doc = DepartmentIssue( company_id=company_id, issue_number=await self._repository.get_next_department_issue_number(company_id), issue_date=data["issue_date"], required_date=data["required_date"], priority=DepartmentIssuePriority( data.get("priority") or DepartmentIssuePriority.NORMAL.value ), department_id=data["department_id"], requested_by_id=requested_by_id, designation=data.get("designation"), cost_center=data.get("cost_center"), issue_type=DepartmentIssueType( data.get("issue_type") or DepartmentIssueType.REGULAR.value ), reason=DepartmentIssueReason(data["reason"]), reference=data.get("reference"), notes=data.get("notes"), from_warehouse_id=data["from_warehouse_id"], status=status, total_items=totals["total_items"], total_quantity=totals["total_quantity"], total_estimated_value=totals["total_estimated_value"], attachments=list(data.get("attachments") or []), created_by=user.id, submitted_at=now if status == DepartmentIssueStatus.REVIEW else None, lines=lines, ) return await self._repository.create_department_issue(doc) async def get_department_issue( self, user: UserRegistration, company_id: str, issue_id: str ) -> DepartmentIssue: await self._get_user_company(user, company_id) doc = await self._repository.get_department_issue(issue_id, company_id) if not doc: raise NotFoundError("Department issue not found") return doc async def list_department_issues( self, user: UserRegistration, company_id: str, status: str | None = None, department_id: str | None = None, from_warehouse_id: str | None = None, skip: int = 0, limit: int = 100, ) -> tuple[list[DepartmentIssue], int]: await self._get_user_company(user, company_id) items = await self._repository.list_department_issues( company_id, status, department_id, from_warehouse_id, skip, limit ) total = await self._repository.count_department_issues( company_id, status, department_id, from_warehouse_id ) return items, total async def update_department_issue( self, user: UserRegistration, company_id: str, issue_id: str, data: dict ) -> DepartmentIssue: doc = await self.get_department_issue(user, company_id, issue_id) if doc.status not in self._DEPARTMENT_ISSUE_EDITABLE: raise ValidationError("Only draft or review department issues can be edited") replace_lines = False if "issue_date" in data and data["issue_date"] is not None: doc.issue_date = data["issue_date"] if "required_date" in data and data["required_date"] is not None: doc.required_date = data["required_date"] if "priority" in data and data["priority"] is not None: doc.priority = DepartmentIssuePriority(data["priority"]) if "issue_type" in data and data["issue_type"] is not None: doc.issue_type = DepartmentIssueType(data["issue_type"]) if "reason" in data and data["reason"] is not None: doc.reason = DepartmentIssueReason(data["reason"]) if "designation" in data: doc.designation = data["designation"] if "cost_center" in data: doc.cost_center = data["cost_center"] if "reference" in data: doc.reference = data["reference"] if "notes" in data: doc.notes = data["notes"] if "attachments" in data and data["attachments"] is not None: doc.attachments = list(data["attachments"]) if "requested_by_id" in data: requested_by_id = data["requested_by_id"] or None if requested_by_id and not await self._repository.user_exists(requested_by_id): raise NotFoundError("Requested by user not found") doc.requested_by_id = requested_by_id if "department_id" in data and data["department_id"]: department = await self._repository.get_department(data["department_id"], company_id) if not department: raise NotFoundError("Department not found") if department.status != DepartmentStatus.ACTIVE or not department.is_active: raise ValidationError("Department must be active for item issuance") doc.department_id = data["department_id"] if "from_warehouse_id" in data and data["from_warehouse_id"]: warehouse = await self._repository.get_warehouse( data["from_warehouse_id"], company_id ) if not warehouse: raise NotFoundError("From warehouse not found") if warehouse.status == WarehouseStatus.INACTIVE or not warehouse.is_active: raise ValidationError("From warehouse is inactive") if not warehouse.allow_stock_out: raise ValidationError("From warehouse does not allow stock out") doc.from_warehouse_id = data["from_warehouse_id"] if "lines" in data and data["lines"] is not None: doc.lines = await self._build_department_issue_lines( company_id, data["lines"], doc.from_warehouse_id ) totals = self._sum_department_issue_totals(doc.lines) department = await self._repository.get_department(doc.department_id, company_id) if department and department.monthly_issue_budget is not None: if totals["total_quantity"] > department.monthly_issue_budget: raise ValidationError( f"Total issue qty ({totals['total_quantity']}) exceeds department " f"monthly issue budget ({department.monthly_issue_budget})" ) doc.total_items = totals["total_items"] doc.total_quantity = totals["total_quantity"] doc.total_estimated_value = totals["total_estimated_value"] replace_lines = True doc.updated_at = datetime.utcnow() updated = await self._repository.update_department_issue( doc, replace_lines=replace_lines ) if not updated: raise NotFoundError("Department issue not found") return updated async def delete_department_issue( self, user: UserRegistration, company_id: str, issue_id: str ) -> None: doc = await self.get_department_issue(user, company_id, issue_id) if doc.status != DepartmentIssueStatus.DRAFT: raise ValidationError("Only draft department issues can be deleted") if not await self._repository.delete_department_issue(issue_id): raise NotFoundError("Department issue not found") async def add_department_issue_attachments( self, user: UserRegistration, company_id: str, issue_id: str, paths: list[str] ) -> DepartmentIssue: doc = await self.get_department_issue(user, company_id, issue_id) if doc.status == DepartmentIssueStatus.CANCELLED: raise ValidationError("Cannot attach files to a cancelled issue") if doc.status == DepartmentIssueStatus.COMPLETED: raise ValidationError("Cannot attach files to a completed issue") doc.attachments = list(doc.attachments or []) + paths doc.updated_at = datetime.utcnow() updated = await self._repository.update_department_issue(doc) if not updated: raise NotFoundError("Department issue not found") return updated async def submit_department_issue_for_approval( self, user: UserRegistration, company_id: str, issue_id: str ) -> DepartmentIssue: return await self._advance_department_issue( user, company_id, issue_id, DepartmentIssueStatus.DRAFT, DepartmentIssueStatus.REVIEW, ) async def approve_department_issue( self, user: UserRegistration, company_id: str, issue_id: str ) -> DepartmentIssue: return await self._advance_department_issue( user, company_id, issue_id, DepartmentIssueStatus.REVIEW, DepartmentIssueStatus.APPROVED, ) async def start_issue_department_items( self, user: UserRegistration, company_id: str, issue_id: str ) -> DepartmentIssue: return await self._advance_department_issue( user, company_id, issue_id, DepartmentIssueStatus.APPROVED, DepartmentIssueStatus.ISSUE_ITEMS, ) async def complete_department_issue( self, user: UserRegistration, company_id: str, issue_id: str ) -> DepartmentIssue: doc = await self.get_department_issue(user, company_id, issue_id) if doc.status != DepartmentIssueStatus.ISSUE_ITEMS: raise ValidationError("Only issues in 'issue_items' status can be completed") if not doc.lines: raise ValidationError("Department issue must have at least one line") for line in doc.lines: item = await self._repository.get_item(line.item_id, company_id) if not item: raise NotFoundError(f"Item '{line.item_id}' not found") if not item.track_inventory: continue await self._repository.create_transaction( InventoryTransaction( company_id=company_id, item_id=line.item_id, warehouse_id=doc.from_warehouse_id, txn_type=InventoryTxnType.ISSUE_TO_DEPARTMENT, txn_date=doc.issue_date, quantity_in=Decimal("0"), quantity_out=line.issue_qty, unit_cost=line.unit_cost, reference_type="department_issue", reference_id=doc.id, reference_number=doc.issue_number, notes=( f"Issue to department {doc.department_code or doc.department_id}" ), ) ) now = datetime.utcnow() doc.status = DepartmentIssueStatus.COMPLETED doc.completed_at = now if not doc.issued_at: doc.issued_at = now doc.updated_at = now updated = await self._repository.update_department_issue(doc) if not updated: raise NotFoundError("Department issue not found") return updated async def cancel_department_issue( self, user: UserRegistration, company_id: str, issue_id: str ) -> DepartmentIssue: doc = await self.get_department_issue(user, company_id, issue_id) if doc.status in { DepartmentIssueStatus.COMPLETED, DepartmentIssueStatus.CANCELLED, DepartmentIssueStatus.ISSUE_ITEMS, }: raise ValidationError( "Cannot cancel a completed, cancelled, or in-progress issue" ) doc.status = DepartmentIssueStatus.CANCELLED doc.cancelled_at = datetime.utcnow() doc.updated_at = datetime.utcnow() updated = await self._repository.update_department_issue(doc) if not updated: raise NotFoundError("Department issue not found") return updated # ---- Reports Dashboard (Issue to Department) ---- def _kpi_metric(self, current: Decimal | int, previous: Decimal | int) -> DashboardKpiMetric: cur = Decimal(str(current)) prev = Decimal(str(previous)) change = None if prev != 0: change = ((cur - prev) / prev * Decimal("100")).quantize(Decimal("0.1")) return DashboardKpiMetric(current=cur, previous=prev, change_percent=change) def _previous_period( self, from_date: date, to_date: date ) -> tuple[date, date]: days = (to_date - from_date).days + 1 previous_to = from_date - timedelta(days=1) previous_from = previous_to - timedelta(days=days - 1) return previous_from, previous_to def _fill_daily_trend( self, from_date: date, to_date: date, rows: list[dict] ) -> list[DashboardTrendPoint]: by_date = {row["date"]: row for row in rows} points: list[DashboardTrendPoint] = [] cursor = from_date while cursor <= to_date: row = by_date.get(cursor) if row: points.append( DashboardTrendPoint( date=cursor, issues_count=row["issues_count"], total_quantity=row["total_quantity"], total_value=row["total_value"], ) ) else: points.append(DashboardTrendPoint(date=cursor)) cursor += timedelta(days=1) return points def _top_n_with_others( self, rows: list[dict], *, top_n: int = 5, quantity_based: bool = True ) -> list[DashboardBreakdownSlice]: if not rows: return [] metric_key = "total_quantity" if quantity_based else "total_value" total = sum((Decimal(str(r[metric_key])) for r in rows), Decimal("0")) top = rows[:top_n] rest = rows[top_n:] slices: list[DashboardBreakdownSlice] = [] for row in top: metric = Decimal(str(row[metric_key])) percent = ( (metric / total * Decimal("100")).quantize(Decimal("0.1")) if total > 0 else Decimal("0.0") ) slices.append( DashboardBreakdownSlice( id=row["id"], code=row["code"], name=row["name"], issues_count=row["issues_count"], total_quantity=row["total_quantity"], total_value=row["total_value"], percent_of_total=percent, ) ) if rest: others_qty = sum((r["total_quantity"] for r in rest), Decimal("0")) others_value = sum((r["total_value"] for r in rest), Decimal("0")) others_count = sum((r["issues_count"] for r in rest), 0) metric = others_qty if quantity_based else others_value percent = ( (metric / total * Decimal("100")).quantize(Decimal("0.1")) if total > 0 else Decimal("0.0") ) slices.append( DashboardBreakdownSlice( id=None, code=None, name="Others", issues_count=others_count, total_quantity=others_qty, total_value=others_value, percent_of_total=percent, ) ) return slices def _quick_report_links(self) -> list[QuickReportLink]: return [ QuickReportLink( key="issue_summary", title="Issue Summary Report", description="Overview of all issues in the selected period", path="/api/v1/reports/dashboard", ), QuickReportLink( key="issue_by_department", title="Issue by Department", description="Breakdown of issued items by department", path="/api/v1/reports/dashboard", ), QuickReportLink( key="issue_by_item", title="Issue by Item", description="Item-wise issue quantities and values", path="/api/v1/inventory/department-issues", ), QuickReportLink( key="issue_by_warehouse", title="Issue by Warehouse", description="Warehouse-wise issue quantities", path="/api/v1/reports/dashboard", ), QuickReportLink( key="pending_approvals", title="Pending Approvals Report", description="Issues awaiting review and approval", path="/api/v1/inventory/department-issues?status=review", ), QuickReportLink( key="financial_report", title="Financial Report", description="Value issued and financial impact", path="/api/v1/reports/dashboard", ), ] async def get_department_issue_dashboard( self, user: UserRegistration, company_id: str, from_date: date, to_date: date, *, department_id: str | None = None, from_warehouse_id: str | None = None, recent_limit: int = 10, pending_limit: int = 10, ) -> DepartmentIssueDashboard: await self._get_user_company(user, company_id) if to_date < from_date: raise ValidationError("to_date must be on or after from_date") previous_from, previous_to = self._previous_period(from_date, to_date) filters = { "department_id": department_id, "from_warehouse_id": from_warehouse_id, "exclude_cancelled": True, } current_stats = await self._repository.get_department_issue_period_stats( company_id, from_date, to_date, **filters ) previous_stats = await self._repository.get_department_issue_period_stats( company_id, previous_from, previous_to, **filters ) current_pending = await self._repository.count_department_issues_pending( company_id, from_date=from_date, to_date=to_date, department_id=department_id, from_warehouse_id=from_warehouse_id, ) previous_pending = await self._repository.count_department_issues_pending( company_id, from_date=previous_from, to_date=previous_to, department_id=department_id, from_warehouse_id=from_warehouse_id, ) trend_rows = await self._repository.get_department_issue_daily_trend( company_id, from_date, to_date, **filters ) by_dept_rows = await self._repository.get_department_issue_breakdown_by_department( company_id, from_date, to_date, **filters ) by_wh_rows = await self._repository.get_department_issue_breakdown_by_warehouse( company_id, from_date, to_date, **filters ) recent = await self._repository.list_department_issue_summaries( company_id, from_date=from_date, to_date=to_date, department_id=department_id, from_warehouse_id=from_warehouse_id, exclude_cancelled=True, limit=recent_limit, ) pending_list = await self._repository.list_department_issue_summaries( company_id, from_date=from_date, to_date=to_date, status=DepartmentIssueStatus.REVIEW.value, department_id=department_id, from_warehouse_id=from_warehouse_id, exclude_cancelled=False, order_by_submitted=True, limit=pending_limit, ) return DepartmentIssueDashboard( from_date=from_date, to_date=to_date, previous_from_date=previous_from, previous_to_date=previous_to, total_issues=self._kpi_metric( current_stats["issues_count"], previous_stats["issues_count"] ), items_issued=self._kpi_metric( current_stats["total_quantity"], previous_stats["total_quantity"] ), total_departments=self._kpi_metric( current_stats["departments_count"], previous_stats["departments_count"] ), total_warehouses=self._kpi_metric( current_stats["warehouses_count"], previous_stats["warehouses_count"] ), total_value_issued=self._kpi_metric( current_stats["total_value"], previous_stats["total_value"] ), pending_approvals=self._kpi_metric(current_pending, previous_pending), trend=self._fill_daily_trend(from_date, to_date, trend_rows), by_department=self._top_n_with_others(by_dept_rows, top_n=5), by_warehouse=[ DashboardBreakdownSlice( id=row["id"], code=row["code"], name=row["name"], issues_count=row["issues_count"], total_quantity=row["total_quantity"], total_value=row["total_value"], percent_of_total=( ( row["total_quantity"] / current_stats["total_quantity"] * Decimal("100") ).quantize(Decimal("0.1")) if current_stats["total_quantity"] > 0 else Decimal("0.0") ), ) for row in by_wh_rows ], recent_issues=recent, pending_approvals_list=pending_list, quick_reports=self._quick_report_links(), ) async def get_stock_report( self, user: UserRegistration, company_id: str, *, from_date: date | None = None, to_date: date | None = None, report_basis: StockReportBasis = StockReportBasis.CURRENT_STOCK, warehouse_id: str | None = None, department_id: str | None = None, category_id: str | None = None, item_type_id: str | None = None, search: str | None = None, stock_status: StockReportStatus | None = None, min_stock_value: Decimal | None = None, page: int = 1, page_size: int = 20, ) -> StockReport: await self._get_user_company(user, company_id) if from_date and to_date and to_date < from_date: raise ValidationError("to_date must be on or after from_date") if report_basis != StockReportBasis.CURRENT_STOCK: raise ValidationError("Only report_basis=current_stock is supported") # department_id accepted for UI parity; stock balances are warehouse-scoped. _ = department_id report = await self._repository.get_stock_report( company_id, from_date=from_date, to_date=to_date, warehouse_id=warehouse_id, category_id=category_id, item_type_id=item_type_id, search=search, stock_status=stock_status.value if stock_status else None, min_stock_value=min_stock_value, page=page, page_size=page_size, ) report.report_basis = report_basis.value return report async def _advance_department_issue( self, user: UserRegistration, company_id: str, issue_id: str, expected: DepartmentIssueStatus, target: DepartmentIssueStatus, ) -> DepartmentIssue: doc = await self.get_department_issue(user, company_id, issue_id) if doc.status != expected: raise ValidationError( f"Department issue must be '{expected.value}' to move to '{target.value}'" ) if not doc.lines: raise ValidationError("Department issue must have at least one line") now = datetime.utcnow() doc.status = target if target == DepartmentIssueStatus.REVIEW: doc.submitted_at = now elif target == DepartmentIssueStatus.APPROVED: doc.approved_at = now elif target == DepartmentIssueStatus.ISSUE_ITEMS: doc.issued_at = now doc.updated_at = now updated = await self._repository.update_department_issue(doc) if not updated: raise NotFoundError("Department issue not found") return updated async def _build_department_issue_lines( self, company_id: str, raw_lines: list[dict], from_warehouse_id: str ) -> list[DepartmentIssueLine]: if not raw_lines: raise ValidationError("At least one issue item is required") lines: list[DepartmentIssueLine] = [] for idx, raw in enumerate(raw_lines, start=1): item = await self._repository.get_item(raw["item_id"], company_id) if not item: raise NotFoundError(f"Item '{raw['item_id']}' not found") issue_qty = Decimal(str(raw["issue_qty"])) if issue_qty <= 0: raise ValidationError(f"Issue qty must be > 0 for item '{item.sku}'") balance = await self._repository.get_balance( company_id, item.id or raw["item_id"], from_warehouse_id ) available = ( (balance.quantity_on_hand - balance.quantity_reserved) if balance else Decimal("0.0000") ) if item.track_inventory and issue_qty > available: raise ValidationError( f"Issue qty ({issue_qty}) exceeds available qty ({available}) " f"for item '{item.sku}' at warehouse" ) base_unit_id = raw.get("base_unit_id") or item.base_unit_id if base_unit_id: unit = await self._repository.get_base_unit(base_unit_id, company_id) if not unit: raise NotFoundError(f"UOM '{base_unit_id}' not found") if raw.get("unit_cost") is not None: unit_cost = Decimal(str(raw["unit_cost"])) elif balance and balance.average_cost > 0: unit_cost = balance.average_cost else: unit_cost = item.purchase_price line_value = (issue_qty * unit_cost).quantize(Decimal("0.01")) lines.append( DepartmentIssueLine( line_number=idx, item_id=item.id or raw["item_id"], available_qty=available.quantize(Decimal("0.0001")), issue_qty=issue_qty.quantize(Decimal("0.0001")), base_unit_id=base_unit_id, remarks=raw.get("remarks"), unit_cost=unit_cost.quantize(Decimal("0.0001")), line_value=line_value, ) ) return lines def _sum_department_issue_totals(self, lines: list[DepartmentIssueLine]) -> dict: total_quantity = sum((line.issue_qty for line in lines), Decimal("0")) total_value = sum((line.line_value for line in lines), Decimal("0")) return { "total_items": len(lines), "total_quantity": total_quantity.quantize(Decimal("0.0001")), "total_estimated_value": total_value.quantize(Decimal("0.01")), } async def _get_user_company(self, user: UserRegistration, company_id: str) -> Company: return await resolve_company_for_user(self._companies, user, company_id)