""" Address Services for customer address management. """ from typing import Dict, List, Optional, Tuple, Union from datetime import datetime, timezone import logging from fastapi import HTTPException, status from sqlalchemy import select, and_, or_, func, desc, asc from sqlalchemy.orm import Session, selectinload from src.apps.base.models.address import Address from src.apps.customers.models.customer import Customer from src.apps.base.schemas.common import AddressSchema, AddressCreateRequestSchema from src.helpers.pagination import QueryPaginator from src.core.exceptions import APIException logger = logging.getLogger(__name__) async def create_customer_address( db: Session, customer_id: str, address_data: AddressCreateRequestSchema ) -> AddressSchema: """Create a new address for a customer.""" try: # Validate customer exists customer_stmt = select(Customer).where(Customer.customer_id == customer_id) if hasattr(Customer, 'deleted_at'): customer_stmt = customer_stmt.where(Customer.deleted_at.is_(None)) customer = db.execute(customer_stmt).scalar_one_or_none() if not customer: raise APIException( message="Customer not found", status_code=404, module="customers.address_services" ) # Convert schema to dict, excluding None values address_dict = address_data.model_dump(exclude_none=True) # Add customer relationship fields address_dict["customer_id"] = customer.id address_dict["merchant_id"] = customer.merchant_id # Use customer's merchant # Check if customer has any existing addresses existing_addresses_stmt = select(Address).where( and_( Address.customer_id == customer.id, Address.deleted_at.is_(None) if hasattr(Address, 'deleted_at') else True ) ) existing_addresses = db.execute(existing_addresses_stmt).scalars().all() # If this is the first address, it must be default if len(existing_addresses) == 0: address_dict["use_as_default"] = True else: # If user explicitly sets use_as_default=True, unset others if address_dict.get("use_as_default", False): # Set all other addresses to use_as_default=False for addr in existing_addresses: if addr.use_as_default: addr.use_as_default = False db.add(addr) else: # Ensure use_as_default is False for additional addresses by default address_dict["use_as_default"] = False # Set defaults for optional fields if not provided if "is_active" not in address_dict: address_dict["is_active"] = True # Create address object address_obj = Address(**address_dict) db.add(address_obj) db.commit() db.refresh(address_obj) logger.info(f"Address created for customer {customer_id} with ID: {address_obj.id}") return AddressSchema.model_validate(address_obj) except APIException: raise # Re-raise APIException as-is except Exception as e: logger.error(f"Error creating address for customer {customer_id}: {e}") raise APIException( message=f"Failed to create address: {str(e)}", status_code=500, module="customers.address_services" ) async def get_customer_addresses( db: Session, customer_id: str, limit: int = 10, offset: int = 0, address_type: Optional[str] = None, is_active: Optional[bool] = None, use_as_default: Optional[bool] = None, search: Optional[str] = None ) -> Dict: """Get paginated list of customer addresses.""" try: # Validate customer exists customer_stmt = select(Customer).where(Customer.customer_id == customer_id) if hasattr(Customer, 'deleted_at'): customer_stmt = customer_stmt.where(Customer.deleted_at.is_(None)) customer = db.execute(customer_stmt).scalar_one_or_none() if not customer: raise APIException( message="Customer not found", status_code=404, module="customers.address_services" ) # Base query stmt = select(Address).where(Address.customer_id == customer.id) # Filter out deleted records if hasattr(Address, 'deleted_at'): stmt = stmt.where(Address.deleted_at.is_(None)) # Apply filters if address_type: stmt = stmt.where(Address.address_type == address_type) if is_active is not None: stmt = stmt.where(Address.is_active == is_active) if use_as_default is not None: stmt = stmt.where(Address.use_as_default == use_as_default) if search: search_term = f"%{search}%" stmt = stmt.where( or_( Address.address_line_1.ilike(search_term), Address.address_line_2.ilike(search_term), Address.city.ilike(search_term), Address.state.ilike(search_term), Address.country.ilike(search_term), Address.zipcode.ilike(search_term) ) ) # Apply default sorting (created_at descending) stmt = stmt.order_by(desc(Address.created_at)) # Use QueryPaginator for pagination paginator = QueryPaginator( query=stmt, schema=AddressSchema, db=db, model=Address, url=f"/api/v1/customers/{customer_id}/addresses", offset=offset, limit=limit, use_orm=True, ) pagination_result = paginator.paginate() return { "data": pagination_result.result, "total": pagination_result.total, "page": pagination_result.page, "per_page": pagination_result.per_page, "next": pagination_result.next, "previous": pagination_result.previous, "first": pagination_result.first, "last": pagination_result.last, } except APIException: raise # Re-raise APIException as-is except Exception as e: logger.error(f"Error getting addresses for customer {customer_id}: {e}") return { "data": [], "total": 0, "page": 1, "per_page": limit, "next": None, "previous": None, "first": None, "last": None, } async def get_customer_address( db: Session, customer_id: str, address_id: int ) -> AddressSchema: """Get a specific address by ID for a customer.""" try: # Validate customer exists customer_stmt = select(Customer).where(Customer.customer_id == customer_id) if hasattr(Customer, 'deleted_at'): customer_stmt = customer_stmt.where(Customer.deleted_at.is_(None)) customer = db.execute(customer_stmt).scalar_one_or_none() if not customer: raise APIException( message="Customer not found", status_code=404, module="customers.address_services" ) # Query for specific address stmt = select(Address).where( and_( Address.id == address_id, Address.customer_id == customer.id ) ) if hasattr(Address, 'deleted_at'): stmt = stmt.where(Address.deleted_at.is_(None)) address = db.execute(stmt).scalar_one_or_none() if not address: raise APIException( message="Address not found", status_code=404, module="customers.address_services" ) return AddressSchema.model_validate(address) except APIException: raise # Re-raise APIException as-is except Exception as e: logger.error(f"Error getting address {address_id} for customer {customer_id}: {e}") raise APIException( message=f"Failed to retrieve address: {str(e)}", status_code=500, module="customers.address_services" ) async def update_customer_address( db: Session, customer_id: str, address_id: int, address_data: dict ) -> AddressSchema: """Update a customer address.""" try: # Validate customer exists customer_stmt = select(Customer).where(Customer.customer_id == customer_id) if hasattr(Customer, 'deleted_at'): customer_stmt = customer_stmt.where(Customer.deleted_at.is_(None)) customer = db.execute(customer_stmt).scalar_one_or_none() if not customer: raise APIException( message="Customer not found", status_code=404, module="customers.address_services" ) # Query for specific address stmt = select(Address).where( and_( Address.id == address_id, Address.customer_id == customer.id ) ) if hasattr(Address, 'deleted_at'): stmt = stmt.where(Address.deleted_at.is_(None)) address = db.execute(stmt).scalar_one_or_none() if not address: raise APIException( message="Address not found", status_code=404, module="customers.address_services" ) # Handle use_as_default logic for updates if "use_as_default" in address_data and address_data["use_as_default"]: # If setting this as default, unset others other_addresses_stmt = select(Address).where( and_( Address.customer_id == customer.id, Address.id != address.id, Address.deleted_at.is_(None) if hasattr(Address, 'deleted_at') else True ) ) other_addresses = db.execute(other_addresses_stmt).scalars().all() for addr in other_addresses: if addr.use_as_default: addr.use_as_default = False addr.updated_at = datetime.now(timezone.utc) db.add(addr) # Update fields for field, value in address_data.items(): if hasattr(address, field) and value is not None: setattr(address, field, value) address.updated_at = datetime.now(timezone.utc) db.commit() db.refresh(address) logger.info(f"Address {address_id} updated for customer {customer_id}") return AddressSchema.model_validate(address) except APIException: raise except Exception as e: logger.error(f"Error updating address {address_id} for customer {customer_id}: {e}") raise APIException( message=f"Failed to update address: {str(e)}", status_code=500, module="customers.address_services" ) async def delete_customer_address( db: Session, customer_id: str, address_id: int ) -> bool: """Soft delete a customer address.""" try: # Validate customer exists customer_stmt = select(Customer).where(Customer.customer_id == customer_id) if hasattr(Customer, 'deleted_at'): customer_stmt = customer_stmt.where(Customer.deleted_at.is_(None)) customer = db.execute(customer_stmt).scalar_one_or_none() if not customer: raise APIException( message="Customer not found", status_code=404, module="customers.address_services" ) # Query for specific address stmt = select(Address).where( and_( Address.id == address_id, Address.customer_id == customer.id ) ) if hasattr(Address, 'deleted_at'): stmt = stmt.where(Address.deleted_at.is_(None)) address = db.execute(stmt).scalar_one_or_none() if not address: raise APIException( message="Address not found", status_code=404, module="customers.address_services" ) # Check if this is the only address or the only default address all_addresses_stmt = select(Address).where( and_( Address.customer_id == customer.id, Address.deleted_at.is_(None) if hasattr(Address, 'deleted_at') else True ) ) all_addresses = db.execute(all_addresses_stmt).scalars().all() # Prevent deletion if this is the only address if len(all_addresses) <= 1: raise APIException( message="Cannot delete the only address. Customer must have at least one address.", status_code=400, module="customers.address_services" ) # If deleting the default address, make another one default if address.use_as_default: other_addresses = [addr for addr in all_addresses if addr.id != address.id] if other_addresses: # Make the first active address the new default active_addresses = [addr for addr in other_addresses if addr.is_active] new_default = active_addresses[0] if active_addresses else other_addresses[0] new_default.use_as_default = True new_default.updated_at = datetime.now(timezone.utc) db.add(new_default) address.deleted_at = datetime.now(timezone.utc) address.updated_at = datetime.now(timezone.utc) db.commit() logger.info(f"Address {address_id} deleted for customer {customer_id}") return True except APIException: raise except Exception as e: logger.error(f"Error deleting address {address_id} for customer {customer_id}: {e}") raise APIException( message=f"Failed to delete address: {str(e)}", status_code=500, module="customers.address_services" )