""" Shared batch number generator utility """ from datetime import datetime from typing import Optional, Protocol, Dict, Any from sqlalchemy.ext.asyncio import AsyncSession import structlog logger = structlog.get_logger() class BatchCountProvider(Protocol): """Protocol for providing batch counts for a specific tenant and date range""" async def get_daily_batch_count( self, tenant_id: str, date_start: datetime, date_end: datetime, prefix: Optional[str] = None ) -> int: """Get the count of batches created today for the given tenant""" ... class BatchNumberGenerator: """Generates unique batch numbers across different services""" def __init__(self, batch_provider: BatchCountProvider): self.batch_provider = batch_provider async def generate_batch_number( self, tenant_id: str, prefix: str = "BATCH", date: Optional[datetime] = None ) -> str: """ Generate a unique batch number with format: {PREFIX}-{YYYYMMDD}-{XXX} Args: tenant_id: The tenant ID prefix: Prefix for the batch number (e.g., "INV", "PROD", "BATCH") date: Date to use for the batch number (defaults to today) Returns: Unique batch number string """ try: # Use provided date or current date target_date = date or datetime.utcnow() date_prefix = target_date.strftime("%Y%m%d") # Calculate date range for the day today_start = datetime.combine(target_date.date(), datetime.min.time()) today_end = datetime.combine(target_date.date(), datetime.max.time()) # Get count of batches created today with this prefix daily_count = await self.batch_provider.get_daily_batch_count( tenant_id=tenant_id, date_start=today_start, date_end=today_end, prefix=prefix ) # Generate sequential number (starting from 1) sequence = daily_count + 1 batch_number = f"{prefix}-{date_prefix}-{sequence:03d}" logger.info( "Generated batch number", tenant_id=tenant_id, prefix=prefix, date=target_date.date(), sequence=sequence, batch_number=batch_number ) return batch_number except Exception as e: logger.error( "Failed to generate batch number", tenant_id=tenant_id, prefix=prefix, error=str(e) ) raise def create_fallback_batch_number( prefix: str = "BATCH", date: Optional[datetime] = None, sequence: int = 1 ) -> str: """ Create a fallback batch number when database access fails Args: prefix: Prefix for the batch number date: Date to use (defaults to now) sequence: Sequence number to use Returns: Fallback batch number string """ target_date = date or datetime.utcnow() date_prefix = target_date.strftime("%Y%m%d") return f"{prefix}-{date_prefix}-{sequence:03d}"