Spaces:
Sleeping
Sleeping
| from datetime import datetime | |
| from typing import Optional, List, Dict, Any | |
| from bson import ObjectId | |
| import logging | |
| import uuid | |
| from app.core.nosql_client import db | |
| logger = logging.getLogger(__name__) | |
| class AddressModel: | |
| """Model for managing user delivery addresses embedded under customer documents""" | |
| async def create_address(customer_id: str, address_data: Dict[str, Any]) -> Optional[str]: | |
| """Create a new embedded address for a user inside customers collection""" | |
| try: | |
| from app.models.user_model import BookMyServiceUserModel | |
| address_id = str(uuid.uuid4()) | |
| current_time = datetime.utcnow() | |
| address_doc = { | |
| "address_id": address_id, # New field for address identification | |
| "address_line_1": address_data.get("address_line_1"), | |
| "address_line_2": address_data.get("address_line_2", ""), | |
| "city": address_data.get("city"), | |
| "state": address_data.get("state"), | |
| "postal_code": address_data.get("postal_code"), | |
| "country": address_data.get("country", "India"), | |
| "address_type": address_data.get("address_type", "home"), # home, work, other | |
| "is_default": address_data.get("is_default", False), | |
| "landmark": address_data.get("landmark", ""), | |
| "created_at": current_time, | |
| "updated_at": current_time, | |
| } | |
| # Fetch user doc | |
| user = await BookMyServiceUserModel.collection.find_one({"customer_id": customer_id}) | |
| if not user: | |
| logger.error(f"User not found for customer_id {customer_id}") | |
| return None | |
| addresses = user.get("addresses", []) | |
| # If setting default, unset any existing defaults | |
| if address_doc.get("is_default"): | |
| for a in addresses: | |
| if a.get("is_default"): | |
| a["is_default"] = False | |
| a["updated_at"] = datetime.utcnow() | |
| else: | |
| # If this is the first address, set default | |
| if len(addresses) == 0: | |
| address_doc["is_default"] = True | |
| addresses.append(address_doc) | |
| update_result = await BookMyServiceUserModel.collection.update_one( | |
| {"customer_id": customer_id}, | |
| {"$set": {"addresses": addresses}} | |
| ) | |
| if update_result.modified_count == 0: | |
| logger.error(f"Failed to insert embedded address for user {customer_id}") | |
| return None | |
| logger.info(f"Created embedded address for user {customer_id}") | |
| return address_doc["address_id"] # Return the address_id field instead of _id | |
| except Exception as e: | |
| logger.error(f"Error creating embedded address for user {customer_id}: {str(e)}") | |
| return None | |
| async def get_user_addresses(customer_id: str) -> List[Dict[str, Any]]: | |
| """Get all embedded addresses for a user from customers collection""" | |
| try: | |
| from app.models.user_model import BookMyServiceUserModel | |
| user = await BookMyServiceUserModel.collection.find_one({"customer_id": customer_id}) | |
| if not user: | |
| return [] | |
| addresses = user.get("addresses", []) | |
| # Sort by created_at desc and return as-is (no _id field) | |
| addresses.sort(key=lambda x: x.get("created_at", datetime.utcnow()), reverse=True) | |
| return addresses | |
| except Exception as e: | |
| logger.error(f"Error getting embedded addresses for user {customer_id}: {str(e)}") | |
| return [] | |
| async def get_address_by_id(customer_id: str, address_id: str) -> Optional[Dict[str, Any]]: | |
| """Get a specific embedded address by ID for a user""" | |
| try: | |
| from app.models.user_model import BookMyServiceUserModel | |
| user = await BookMyServiceUserModel.collection.find_one({"customer_id": customer_id}) | |
| if not user: | |
| return None | |
| addresses = user.get("addresses", []) | |
| for a in addresses: | |
| if a.get("address_id") == address_id: | |
| a_copy = dict(a) | |
| # Inject customer_id for backward-compat where used in router | |
| a_copy["customer_id"] = customer_id | |
| return a_copy | |
| return None | |
| except Exception as e: | |
| logger.error(f"Error getting embedded address {address_id} for user {customer_id}: {str(e)}") | |
| return None | |
| async def update_address(customer_id: str, address_id: str, update_data: Dict[str, Any]) -> bool: | |
| """Update an existing embedded address""" | |
| try: | |
| from app.models.user_model import BookMyServiceUserModel | |
| user = await BookMyServiceUserModel.collection.find_one({"customer_id": customer_id}) | |
| if not user: | |
| return False | |
| addresses = user.get("addresses", []) | |
| updated = False | |
| for a in addresses: | |
| if a.get("address_id") == address_id: | |
| allowed_fields = [ | |
| "address_line_1", "address_line_2", "city", "state", "postal_code", | |
| "country", "address_type", "is_default", "landmark" | |
| ] | |
| for field in allowed_fields: | |
| if field in update_data: | |
| a[field] = update_data[field] | |
| a["updated_at"] = datetime.utcnow() | |
| updated = True | |
| break | |
| if not updated: | |
| return False | |
| # If setting as default, unset other defaults | |
| if update_data.get("is_default"): | |
| for a in addresses: | |
| if a.get("address_id") != address_id and a.get("is_default"): | |
| a["is_default"] = False | |
| a["updated_at"] = datetime.utcnow() | |
| result = await BookMyServiceUserModel.collection.update_one( | |
| {"customer_id": customer_id}, | |
| {"$set": {"addresses": addresses}} | |
| ) | |
| return result.modified_count > 0 | |
| except Exception as e: | |
| logger.error(f"Error updating embedded address {address_id} for user {customer_id}: {str(e)}") | |
| return False | |
| async def delete_address(customer_id: str, address_id: str) -> bool: | |
| """Delete an embedded address""" | |
| try: | |
| from app.models.user_model import BookMyServiceUserModel | |
| user = await BookMyServiceUserModel.collection.find_one({"customer_id": customer_id}) | |
| if not user: | |
| return False | |
| addresses = user.get("addresses", []) | |
| # Filter out by new domain id field 'address_id' | |
| new_addresses = [a for a in addresses if a.get("address_id") != address_id] | |
| result = await BookMyServiceUserModel.collection.update_one( | |
| {"customer_id": customer_id}, | |
| {"$set": {"addresses": new_addresses}} | |
| ) | |
| logger.info(f"Deleted embedded address {address_id} for user {customer_id}") | |
| return result.modified_count > 0 | |
| except Exception as e: | |
| logger.error(f"Error deleting embedded address {address_id} for user {customer_id}: {str(e)}") | |
| return False | |
| async def get_default_address(customer_id: str) -> Optional[Dict[str, Any]]: | |
| """Get the default embedded address for a user""" | |
| try: | |
| from app.models.user_model import BookMyServiceUserModel | |
| user = await BookMyServiceUserModel.collection.find_one({"customer_id": customer_id}) | |
| if not user: | |
| return None | |
| addresses = user.get("addresses", []) | |
| for a in addresses: | |
| if a.get("is_default"): | |
| a_copy = dict(a) | |
| a_copy["customer_id"] = customer_id | |
| return a_copy | |
| return None | |
| except Exception as e: | |
| logger.error(f"Error getting default embedded address for user {customer_id}: {str(e)}") | |
| return None | |
| async def set_default_address(customer_id: str, address_id: str) -> bool: | |
| """Set an embedded address as default and unset others""" | |
| try: | |
| from app.models.user_model import BookMyServiceUserModel | |
| user = await BookMyServiceUserModel.collection.find_one({"customer_id": customer_id}) | |
| if not user: | |
| return False | |
| addresses = user.get("addresses", []) | |
| changed = False | |
| for a in addresses: | |
| if a.get("address_id") == address_id: | |
| if not a.get("is_default"): | |
| a["is_default"] = True | |
| a["updated_at"] = datetime.utcnow() | |
| changed = True | |
| else: | |
| if a.get("is_default"): | |
| a["is_default"] = False | |
| a["updated_at"] = datetime.utcnow() | |
| changed = True | |
| if not changed: | |
| # Even if nothing changed, ensure persistence | |
| pass | |
| result = await BookMyServiceUserModel.collection.update_one( | |
| {"customer_id": customer_id}, | |
| {"$set": {"addresses": addresses}} | |
| ) | |
| return result.modified_count >= 0 | |
| except Exception as e: | |
| logger.error(f"Error setting default embedded address {address_id} for user {customer_id}: {str(e)}") | |
| return False |