from typing import Dict, List, Union, Optional import logging from starlette import status from sqlalchemy.orm import Session from sqlalchemy import select, and_, or_, func, desc, asc from src.core.exceptions import APIException from src.apps.site_masters.models.site_master import SiteMaster from src.apps.users.schemas.user_common import UserSchema from src.apps.site_masters.schemas.site_master_common import SiteMasterSchema from src.apps.site_masters.schemas.site_master_requests import SiteMasterCreateRequestSchema from src.apps.site_masters.seeders.site_master import DEFAULT_SITE_MASTER logger = logging.getLogger(__name__) async def create_master_entry( db: Session, payload: SiteMasterCreateRequestSchema, created_by: UserSchema ) -> SiteMasterSchema: """ Create a new site master entry using SQLAlchemy 2.0 """ try: # Create new SiteMaster instance site_master = SiteMaster( key=payload.key, value=payload.value, label=payload.label, group=payload.group, # Add created_by if the model supports it # created_by_id=created_by.id if hasattr(SiteMaster, 'created_by_id') else None ) db.add(site_master) db.commit() db.refresh(site_master) return SiteMasterSchema.model_validate(site_master) except Exception as e: db.rollback() logger.error(f"Error creating site master entry: {e}") raise APIException( module=__name__, error={}, status_code=status.HTTP_400_BAD_REQUEST, message=f"Failed to create site master entry: {str(e)}", ) def get_by_keys(db: Session, keys: List[str]) -> List[SiteMasterSchema]: """ Get site master entries by keys using SQLAlchemy 2.0 """ try: stmt = select(SiteMaster).where( and_( SiteMaster.key.in_(keys) ) ) result = db.execute(stmt).scalars().all() return [SiteMasterSchema.model_validate(item) for item in result] except Exception as e: logger.error(f"Error getting site master entries by keys: {e}") return [] def get_all(db: Session, search: Optional[str] = None) -> List[SiteMasterSchema]: """ Get all site master entries using SQLAlchemy 2.0 """ try: stmt = select(SiteMaster) if hasattr(SiteMaster, 'deleted_at'): stmt = stmt.where(SiteMaster.deleted_at.is_(None)) # Apply search filter if provided if search: search_conditions = [ SiteMaster.key.ilike(f"%{search}%"), SiteMaster.value.ilike(f"%{search}%"), SiteMaster.label.ilike(f"%{search}%"), SiteMaster.group.ilike(f"%{search}%") ] stmt = stmt.where(or_(*search_conditions)) # Order by display_order field and then by key stmt = stmt.order_by(asc(SiteMaster.display_order), asc(SiteMaster.key)) result = db.execute(stmt).scalars().all() return [SiteMasterSchema.model_validate(item) for item in result] except Exception as e: logger.error(f"Error getting all site master entries: {e}") return [] async def list_master_entries( db: Session, keys: str, search: str = None ) -> Union[List[SiteMasterSchema], Dict]: """ List site master entries with SQLAlchemy 2.0 """ items: List[SiteMasterSchema] = [] if len(keys) > 0: entries = keys.split(",") if len(entries) > 10: raise APIException( module=__name__, error={}, status_code=status.HTTP_400_BAD_REQUEST, message="A maximum of 10 entries can be requested at once", ) items = get_by_keys(db=db, keys=entries) # Handle industries if requested industries = None if "industries" in keys: # TODO: Implement site_master_industry_repository.get_all(db) using SQLAlchemy 2.0 # industries = get_all_industries(db) pass else: items = get_all(db, search=search) # Group items by key response: Dict = {} for entry in items: if entry.key not in response: response[entry.key] = [] response[entry.key].append(entry) # Add industries if they were requested and retrieved if "industries" in keys and industries: response["industries"] = industries return response def populate_default_site_master_data(db: Session) -> Dict[str, Union[int, str, List[str]]]: """ Populate SiteMaster table with default data from seeder using SQLAlchemy 2.0 Args: db: Database session Returns: Dictionary containing operation results with counts and messages """ try: logger.info("Starting to populate default site master data...") # Check current total records total_stmt = select(func.count(SiteMaster.id)) current_total = db.execute(total_stmt).scalar() or 0 logger.info(f"Current SiteMaster records in database: {current_total}") # Track operations to_insert = [] inserted_records = [] skipped_records = [] error_records = [] # Process each item from the default dataset for item in DEFAULT_SITE_MASTER: try: # Check if this specific key-value combination already exists exists_stmt = select(SiteMaster).where( and_( SiteMaster.key == item.get("key"), SiteMaster.value == item.get("value") ) ) existing_record = db.execute(exists_stmt).scalar_one_or_none() if not existing_record: # Create new site master record site_master = SiteMaster(**item) to_insert.append(site_master) inserted_records.append(f"{item.get('key')} -> {item.get('value')}") logger.debug(f"Will insert: {item.get('key')} -> {item.get('value')}") else: skipped_records.append(f"{item.get('key')} -> {item.get('value')}") logger.debug(f"Skipped (exists): {item.get('key')} -> {item.get('value')}") except Exception as item_error: error_msg = f"Error processing item {item.get('key', 'unknown')}: {str(item_error)}" error_records.append(error_msg) logger.error(error_msg) # Bulk insert new records if to_insert: db.add_all(to_insert) db.commit() logger.info(f"Successfully inserted {len(to_insert)} new SiteMaster records") else: logger.info("No new records to insert - all data already exists") # Get final count final_total = db.execute(total_stmt).scalar() or 0 # Prepare result summary result = { "success": True, "message": "Site master data population completed", "records_inserted": len(inserted_records), "records_skipped": len(skipped_records), "records_with_errors": len(error_records), "total_processed": len(DEFAULT_SITE_MASTER), "database_total_before": current_total, "database_total_after": final_total, "inserted_items": inserted_records[:10] if inserted_records else [], # Show first 10 "skipped_items": skipped_records[:5] if skipped_records else [], # Show first 5 "error_items": error_records, } logger.info(f"Population summary: {len(inserted_records)} inserted, {len(skipped_records)} skipped, {len(error_records)} errors") return result except Exception as e: db.rollback() error_msg = f"Failed to populate site master data: {str(e)}" logger.error(error_msg) return { "success": False, "message": error_msg, "records_inserted": 0, "records_skipped": 0, "records_with_errors": 1, "total_processed": 0, "database_total_before": 0, "database_total_after": 0, "inserted_items": [], "skipped_items": [], "error_items": [error_msg], } def seed_site_master_data_if_empty(db: Session) -> Dict[str, Union[int, str, List[str]]]: """ Populate site master data only if the table is empty or has fewer records than the default dataset Args: db: Database session Returns: Dictionary containing operation results """ try: # Check current record count total_stmt = select(func.count(SiteMaster.id)) current_total = db.execute(total_stmt).scalar() or 0 expected_total = len(DEFAULT_SITE_MASTER) if current_total >= expected_total: return { "success": True, "message": f"Site master table already has sufficient data ({current_total}/{expected_total} records)", "records_inserted": 0, "records_skipped": current_total, "records_with_errors": 0, "total_processed": 0, "database_total_before": current_total, "database_total_after": current_total, "inserted_items": [], "skipped_items": [f"Table has {current_total} records, expected {expected_total}"], "error_items": [], } logger.info(f"Table has {current_total} records, expected {expected_total}. Proceeding with population...") return populate_default_site_master_data(db) except Exception as e: error_msg = f"Failed to check site master data: {str(e)}" logger.error(error_msg) return { "success": False, "message": error_msg, "records_inserted": 0, "records_skipped": 0, "records_with_errors": 1, "total_processed": 0, "database_total_before": 0, "database_total_after": 0, "inserted_items": [], "skipped_items": [], "error_items": [error_msg], }