""" Customer router for handling customer-related API endpoints. """ from typing import List, Optional, Dict, Union, Any from fastapi import APIRouter, Depends, HTTPException, status, Query, Path, Body from sqlalchemy.orm import Session import logging from src.core.database import get_session from src.core.exceptions import APIException from src.apps.customers.schemas.customer_common import ( CustomerSchema, CustomerListSchema ) from src.apps.customers.schemas.customer_requests import ( CustomerCreateRequestSchema, CustomerUpdateRequestSchema, CustomerActionRequestSchema, CustomerBulkActionRequestSchema, CustomerFilterSchema, CustomerPaginationSchema ) from src.apps.base.schemas.common import AddressSchema, AddressCreateRequestSchema from src.apps.customers.schemas.customer_contact import ( CustomerContactSchema, CustomerContactCreateRequestSchema, CustomerContactUpdateRequestSchema ) from src.apps.users.models.user import User # Import specific service functions using from attribute from src.apps.customers.services import ( get_customer as get_customer_service, create_customer as create_customer_service, get_customers_list as get_customers_list_service, get_customer_by_email as get_customer_by_email_service, update_customer as update_customer_service, delete_customer as delete_customer_service, activate_customer as activate_customer_service, deactivate_customer as deactivate_customer_service, check_customer_id_exists as check_customer_id_exists, check_account_literal_exists as check_account_literal_exists, get_customer_activities as get_customer_activities_service, get_customer_summary as get_customer_summary_service, bulk_customer_actions as bulk_customer_actions_service ) # Import address services from src.apps.customers.address_services import ( create_customer_address, get_customer_addresses, get_customer_address, update_customer_address, delete_customer_address ) # Import contact services from src.apps.customers.contact_services import ( create_customer_contact, get_customer_contacts, get_customer_contact, update_customer_contact, delete_customer_contact ) # Import note services from src.apps.notes.services import ( get_customer_notes as get_customer_notes_service, create_customer_note as create_customer_note_service, update_customer_note as update_customer_note_service, delete_customer_note as delete_customer_note_service, get_customer_note as get_customer_note_service ) # Import note schemas from src.apps.notes.schemas.note_common import ( NoteResponseSchema, NoteListResponseSchema ) from src.apps.notes.schemas.note_requests import ( NoteCreateRequestSchema, NoteUpdateRequestSchema ) from src.core.functions import generate_id router = APIRouter() logger = logging.getLogger(__name__) @router.post("", response_model=Dict[str, Any], status_code=status.HTTP_201_CREATED) async def create_customer( *, db: Session = Depends(get_session), customer_in: CustomerCreateRequestSchema, ) -> Any: """ Create new customer. The system supports two ways to handle account literals: 1. User-provided: Include 'account_literal' field (15 characters) - system will validate uniqueness 2. Auto-generated: Leave 'account_literal' empty - system will generate a unique account literal Use /customers/account-suggestions to get suggested account literals before creation. Note: All validation (email uniqueness, account literal format, etc.) is handled by Pydantic validators. Note: customer_id is for internal use only, account_literal is the public-facing account number. """ # Create customer using services customer, message = await create_customer_service(db, customer_data=customer_in) if customer is None: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=message ) return { "success": True, "message": message, "data": customer.model_dump() if hasattr(customer, 'model_dump') else customer } @router.post("/actions", response_model=Dict[str, Any]) async def bulk_customer_actions( *, db: Session = Depends(get_session), action_request: CustomerBulkActionRequestSchema, ) -> Any: """ Perform bulk actions on multiple customers. Limit- Maximum 10 customers per request for performance optimization. Available actions: - activate: Enable customers and restore soft-deleted ones - deactivate: Disable active customers - delete: Soft delete customers For single customers, pass one ID in the ids array. Returns detailed success/failure results for each customer. """ try: # Perform bulk actions using service results = await bulk_customer_actions_service( db=db, action=action_request.action, customer_ids=action_request.ids ) # Determine overall success status overall_success = results["failed_count"] == 0 if overall_success: message = f"All {results['success_count']} customers were successfully {action_request.action}d" else: message = f"Bulk action completed: {results['success_count']} succeeded, {results['failed_count']} failed" return { "success": overall_success, "message": message, "data": results } except APIException as e: raise HTTPException( status_code=e.status_code, detail=e.message ) except Exception as e: logging.error(f"Unexpected error in bulk customer actions: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="An unexpected error occurred while processing bulk actions" ) @router.get("", response_model=Dict[str, Any]) async def get_customers_list( pagination: CustomerPaginationSchema = Depends(), filters: CustomerFilterSchema = Depends(), db: Session = Depends(get_session), ) -> Any: """ Retrieve customers with filtering and pagination using schema-based query parameters. """ # Calculate offset from page offset = (pagination.page - 1) * pagination.per_page # Process filter values for service layer (convert 0 to None for optional filters) filters.is_active = filters.is_active if filters.is_active != 0 else None filters.is_vendor = filters.is_vendor if filters.is_vendor != 0 else None # Get customers using QueryPaginator from service pagination_result = await get_customers_list_service( db=db, filters=filters, limit=pagination.per_page, offset=offset, paginate=True ) return { "success": True, "data": pagination_result["data"], "meta": { "total": pagination_result["total"], "page": pagination_result["page"], "per_page": pagination_result["per_page"], "total_pages": (pagination_result["total"] + pagination_result["per_page"] - 1) // pagination_result["per_page"], "has_next": pagination_result["next"] is not None, "has_previous": pagination_result["previous"] is not None }, "links": { "next": pagination_result["next"], "previous": pagination_result["previous"], "first": pagination_result["first"], "last": pagination_result["last"] } } @router.get("/account-suggestions", response_model=Dict[str, Any]) async def get_account_suggestions( db: Session = Depends(get_session), ) -> Any: """ Generate four unique 15-character account literal suggestions. Returns four automatically generated account literals that are: - Exactly 15 characters long - Follow a consistent pattern - Validated against database for uniqueness - Used as public-facing account numbers """ try: # Generate 1 unique account literal suggestions suggestions = [] max_attempts = 20 attempts = 0 while len(suggestions) < 1 and attempts < max_attempts: # Generate a new account literal account_literal = generate_id('customer') # Check if it's unique in database (check account_literal field) if not await check_account_literal_exists(db, account_literal): # Also check if it's not already in our suggestions list if account_literal not in suggestions: suggestions.append(account_literal) attempts += 1 if len(suggestions) < 1: logger.warning(f"Could only generate {len(suggestions)} unique suggestions after {max_attempts} attempts") return { "success": True, "message": f"Generated {len(suggestions)} account literal suggestions", "data": { "suggestions": suggestions, "pattern_info": { "length": 15, "format": "ACC + 3-char identifier + 6-digit timestamp + 3-char UUID", "example": "ACCM7X123456ABC" } } } except Exception as e: logger.error(f"Error generating account suggestions: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to generate account suggestions" ) @router.get("/summary", response_model=Dict[str, Any]) async def get_customer_summary_endpoint( db: Session = Depends(get_session), is_vendor: bool = Query(False, description="Include vendor statistics: true=vendor summary, false=customer summary", example=False), date_from: Optional[str] = Query(None, description="Start date for summary statistics (YYYY-MM-DD format)", example="2024-01-01"), date_to: Optional[str] = Query(None, description="End date for summary statistics (YYYY-MM-DD format)", example="2024-12-31"), limit: int = Query(10, ge=1, le=50, description="Number of top/lowest customers to return", example=10) ) -> Any: """ Get customer summary statistics. Returns summary data including top customers, lowest customers, and most frequent customers. Currently returns static data for demonstration purposes. """ try: summary_data = await get_customer_summary_service(db, is_vendor) return summary_data except APIException as e: raise HTTPException( status_code=e.status_code, detail=e.message ) except Exception as e: logger.error(f"Error fetching customer summary: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Internal server error: {str(e)}" ) @router.get("/{customer_id}", response_model=Dict[str, Any]) async def get_customer( customer_id: str, db: Session = Depends(get_session), fields: Optional[str] = Query(None, description="Comma-separated list of related fields to include: addresses, contacts, merchant, attachments", example="addresses,contacts"), ) -> Any: """ Get customer by ID with optional related fields. By default, no related fields are loaded for better performance. Use the 'fields' parameter to specify which relationships to include. """ try: customer = await get_customer_service(db, customer_id=customer_id, fields=fields) return { "success": True, "message": "Customer found", "data": customer.model_dump() if hasattr(customer, 'model_dump') else customer } except APIException as api_exc: raise HTTPException( status_code=api_exc.status_code, detail=api_exc.message ) @router.put("/{customer_id}", response_model=Dict[str, Any]) async def update_customer( customer_id: str, customer_update: CustomerUpdateRequestSchema, db: Session = Depends(get_session), ) -> Any: """ Update customer by ID. """ try: updated_customer = await update_customer_service(db, customer_id=customer_id, customer_data=customer_update) return { "success": True, "message": "Customer updated successfully", "data": updated_customer.model_dump() if hasattr(updated_customer, 'model_dump') else updated_customer } except APIException as api_exc: raise HTTPException( status_code=api_exc.status_code, detail=api_exc.message ) @router.delete("/{customer_id}", response_model=Dict[str, Any]) async def delete_customer( customer_id: str, db: Session = Depends(get_session), ) -> Any: """ Delete (soft delete) customer by ID. """ # Check if customer exists first try: existing_customer = await get_customer_service(db, customer_id=customer_id, fields=None) except APIException as e: if e.status_code == 404: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Customer not found" ) raise HTTPException( status_code=e.status_code, detail=e.message ) # Use ORM service for delete with await success = await delete_customer_service(db, customer_id=customer_id) if not success: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Failed to delete customer" ) return { "success": True, "message": "Customer deleted successfully" } @router.post("/{customer_id}/addresses", response_model=Dict[str, Any], status_code=status.HTTP_201_CREATED) async def create_customer_address_endpoint( customer_id: str, address_data: AddressCreateRequestSchema, db: Session = Depends(get_session) ) -> Dict[str, Any]: """Create a new address for a customer with use_as_default=False.""" try: address = await create_customer_address(db, customer_id, address_data) return { "success": True, "message": "Address created successfully", "data": address.model_dump() if hasattr(address, 'model_dump') else address } except APIException as e: raise HTTPException( status_code=e.status_code, detail=e.message ) except Exception as e: logger.error(f"Error creating address for customer {customer_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Internal server error: {str(e)}" ) @router.get("/{customer_id}/addresses", response_model=Dict[str, Any]) async def list_customer_addresses( customer_id: str, db: Session = Depends(get_session), page: int = Query(1, ge=1, description="Page number for pagination", example=1), per_page: int = Query(10, ge=1, le=100, description="Number of items per page (max 100)", example=10), address_type: Optional[str] = Query(None, description="Filter by address type", example="billing", enum=["billing", "shipping", "legal", "mailing"]), is_active: Optional[bool] = Query(None, description="Filter by active status: true=active only, false=inactive only", example=True), use_as_default: Optional[bool] = Query(None, description="Filter by default address status", example=True), search: Optional[str] = Query(None, description="Search in address lines, city, state, or country", example="New York") ) -> Dict[str, Any]: """Get paginated list of customer addresses with optional filtering.""" try: # Calculate offset from page limit = per_page offset = (page - 1) * per_page result = await get_customer_addresses( db=db, customer_id=customer_id, limit=limit, offset=offset, address_type=address_type, is_active=is_active, use_as_default=use_as_default, search=search ) return { "success": True, "message": f"Found {result['total']} addresses", "data": result["data"], "meta": { "total": result["total"], "page": result["page"], "per_page": result["per_page"], "next": result["next"], "previous": result["previous"], "first": result["first"], "last": result["last"] } } except APIException as e: raise HTTPException( status_code=e.status_code, detail=e.message ) except Exception as e: logger.error(f"Error listing addresses for customer {customer_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Internal server error: {str(e)}" ) @router.get("/{customer_id}/addresses/{address_id}", response_model=Dict[str, Any]) async def get_customer_address_endpoint( customer_id: str, address_id: int, db: Session = Depends(get_session) ) -> Dict[str, Any]: """Get a specific address by ID for a customer.""" try: address = await get_customer_address(db, customer_id, address_id) return { "success": True, "message": "Address found", "data": address.model_dump() if hasattr(address, 'model_dump') else address } except APIException as e: raise HTTPException( status_code=e.status_code, detail=e.message ) except Exception as e: logger.error(f"Error getting address {address_id} for customer {customer_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Internal server error: {str(e)}" ) @router.put("/{customer_id}/addresses/{address_id}", response_model=Dict[str, Any]) async def update_customer_address_endpoint( customer_id: str, address_id: int, address_data: dict, db: Session = Depends(get_session) ) -> Dict[str, Any]: """Update a customer address with partial or full data.""" try: address = await update_customer_address(db, customer_id, address_id, address_data) return { "success": True, "message": "Address updated successfully", "data": address.model_dump() if hasattr(address, 'model_dump') else address } except APIException as e: raise HTTPException( status_code=e.status_code, detail=e.message ) except Exception as e: logger.error(f"Error updating address {address_id} for customer {customer_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Internal server error: {str(e)}" ) @router.delete("/{customer_id}/addresses/{address_id}", response_model=Dict[str, Any]) async def delete_customer_address_endpoint( customer_id: str, address_id: int, db: Session = Depends(get_session) ) -> Dict[str, Any]: """Soft delete a customer address with validation.""" try: await delete_customer_address(db, customer_id, address_id) return { "success": True, "message": "Address deleted successfully" } except APIException as e: raise HTTPException( status_code=e.status_code, detail=e.message ) except Exception as e: logger.error(f"Error deleting address {address_id} for customer {customer_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Internal server error: {str(e)}" ) @router.post("/{customer_id}/contacts", response_model=Dict[str, Any], status_code=status.HTTP_201_CREATED) async def create_customer_contact_endpoint( customer_id: str, contact_data: CustomerContactCreateRequestSchema, db: Session = Depends(get_session) ) -> Dict[str, Any]: """Create a new contact for a customer.""" try: contact = await create_customer_contact(db, customer_id, contact_data) return { "success": True, "message": "Contact created successfully", "data": contact.model_dump() if hasattr(contact, 'model_dump') else contact } except APIException as e: raise HTTPException( status_code=e.status_code, detail=e.message ) except Exception as e: logger.error(f"Error creating contact for customer {customer_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Internal server error: {str(e)}" ) @router.get("/{customer_id}/contacts", response_model=Dict[str, Any]) async def list_customer_contacts( customer_id: str, db: Session = Depends(get_session), page: int = Query(1, ge=1, description="Page number for pagination", example=1), per_page: int = Query(10, ge=1, le=100, description="Number of items per page (max 100)", example=10), is_active: Optional[bool] = Query(None, description="Filter by active status: true=active only, false=inactive only", example=True), search: Optional[str] = Query(None, description="Search across contact name, email, title, or phone number", example="john"), title: Optional[str] = Query(None, description="Filter by contact job title (partial match)", example="Manager"), relation: Optional[str] = Query(None, description="Filter by contact relation type", example="primary") ) -> Dict[str, Any]: """Get paginated list of customer contacts with optional filtering.""" try: # Calculate offset from page limit = per_page offset = (page - 1) * per_page result = await get_customer_contacts( db=db, customer_id=customer_id, limit=limit, offset=offset, is_active=is_active, search=search, title=title, relation=relation ) return { "success": True, "message": f"Found {result['total']} contacts", "data": result["data"], "meta": { "total": result["total"], "page": result["page"], "per_page": result["per_page"], "next": result["next"], "previous": result["previous"], "first": result["first"], "last": result["last"] } } except APIException as e: raise HTTPException( status_code=e.status_code, detail=e.message ) except Exception as e: logger.error(f"Error listing contacts for customer {customer_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Internal server error: {str(e)}" ) @router.get("/{customer_id}/contacts/{contact_id}", response_model=Dict[str, Any]) async def get_customer_contact_endpoint( customer_id: str, contact_id: str, db: Session = Depends(get_session) ) -> Dict[str, Any]: """Get a specific contact by contact_id for a customer.""" try: contact = await get_customer_contact(db, customer_id, contact_id) return { "success": True, "message": "Contact found", "data": contact.model_dump() if hasattr(contact, 'model_dump') else contact } except APIException as e: raise HTTPException( status_code=e.status_code, detail=e.message ) except Exception as e: logger.error(f"Error getting contact {contact_id} for customer {customer_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Internal server error: {str(e)}" ) @router.put("/{customer_id}/contacts/{contact_id}", response_model=Dict[str, Any]) async def update_customer_contact_endpoint( customer_id: str, contact_id: str, contact_data: CustomerContactUpdateRequestSchema, db: Session = Depends(get_session) ) -> Dict[str, Any]: """Update a customer contact with partial or full data.""" try: contact = await update_customer_contact(db, customer_id, contact_id, contact_data) return { "success": True, "message": "Contact updated successfully", "data": contact.model_dump() if hasattr(contact, 'model_dump') else contact } except APIException as e: raise HTTPException( status_code=e.status_code, detail=e.message ) except Exception as e: logger.error(f"Error updating contact {contact_id} for customer {customer_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Internal server error: {str(e)}" ) @router.delete("/{customer_id}/contacts/{contact_id}", response_model=Dict[str, Any]) async def delete_customer_contact_endpoint( customer_id: str, contact_id: str, db: Session = Depends(get_session) ) -> Dict[str, Any]: """Soft delete a customer contact with validation.""" try: await delete_customer_contact(db, customer_id, contact_id) return { "success": True, "message": "Contact deleted successfully" } except APIException as e: raise HTTPException( status_code=e.status_code, detail=e.message ) except Exception as e: logger.error(f"Error deleting contact {contact_id} for customer {customer_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Internal server error: {str(e)}" ) @router.get("/{customer_id}/activities", response_model=Dict[str, Any]) async def get_customer_activities( *, db: Session = Depends(get_session), customer_id: str ) -> Any: """ Get customer activities. Returns a list of activities for the specified customer. Currently returns static data for demonstration purposes. """ try: activities_data = await get_customer_activities_service(db, customer_id) return activities_data except APIException as e: raise HTTPException( status_code=e.status_code, detail=e.message ) except Exception as e: logger.error(f"Error fetching activities for customer {customer_id}: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Internal server error: {str(e)}" ) # Notes endpoints @router.get( "/{customer_id}/notes", response_model=Dict[str, Any], summary="Get Customer Notes", description="Retrieve paginated list of notes for a specific customer", response_description="Paginated list of customer notes with user details", responses={ 200: { "description": "Notes retrieved successfully", "content": { "application/json": { "example": { "success": True, "message": "Notes retrieved successfully", "data": [ { "id": 1, "description": "Customer called regarding payment inquiry", "created_at": "2025-12-02T15:30:00.000Z", "updated_at": "2025-12-02T15:30:00.000Z", "deleted_at": None, "created_by_id": 1, "created_by": { "id": 1, "first_name": "Admin", "last_name": "User", "email": "admin@example.com" } } ], "pagination": { "total": 1, "page": 1, "per_page": 10, "next": None, "previous": None } } } } }, 404: {"description": "Customer not found"}, 500: {"description": "Internal server error"} } ) async def get_customer_notes( customer_id: str = Path(..., description="Customer ID (customer_id field)", example="ACCKDU1537895E7"), limit: int = Query(10, ge=1, le=100, description="Number of items per page (max 100)", example=10), offset: int = Query(0, ge=0, description="Number of items to skip for pagination", example=0), search: Optional[str] = Query(None, description="Search term to filter notes by description content", example="payment"), db: Session = Depends(get_session) ) -> Dict[str, Any]: """ Get paginated list of notes for a specific customer. Returns paginated notes associated with the customer, ordered by creation date (most recent first). Each note includes full details of the user who created it. **Features:** - Pagination support with configurable page size - Search functionality in note descriptions - Full user details for note creators - Ordered by creation date (newest first) **Parameters:** - **customer_id**: The customer's unique identifier - **limit**: Number of notes per page (1-100, default: 10) - **offset**: Number of notes to skip (default: 0) - **search**: Optional search term to filter notes """ try: result = await get_customer_notes_service( db, customer_id, limit=limit, offset=offset, search=search ) return { "success": True, "message": "Notes retrieved successfully", "data": result["data"], "pagination": { "total": result["total"], "page": result["page"], "per_page": result["per_page"], "next": result["next"], "previous": result["previous"], "first": result["first"], "last": result["last"] }, "status_code": 200 } except APIException as api_exc: raise HTTPException( status_code=api_exc.status_code, detail=api_exc.message ) except Exception as e: logger.error(f"Unexpected error in get_customer_notes: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal server error" ) @router.post( "/{customer_id}/notes", response_model=Dict[str, Any], status_code=status.HTTP_201_CREATED, summary="Create Customer Note", description="Create a new note for a specific customer", response_description="Note created successfully with full details", responses={ 201: { "description": "Note created successfully", "content": { "application/json": { "example": { "success": True, "message": "Note created successfully", "data": { "id": 1, "description": "Customer called regarding payment inquiry", "created_at": "2025-12-02T15:30:00.000Z", "updated_at": "2025-12-02T15:30:00.000Z", "deleted_at": None, "created_by_id": 1, "created_by": { "id": 1, "first_name": "Admin", "last_name": "User", "email": "admin@example.com" } } } } } }, 400: {"description": "Invalid input data or validation error"}, 404: {"description": "Customer not found"}, 500: {"description": "Internal server error"} } ) async def create_customer_note( customer_id: str = Path(..., description="Customer ID (customer_id field)", example="ACCKDU1537895E7"), note_data: NoteCreateRequestSchema = Body(..., description="Note creation data with description (1-2000 characters)"), db: Session = Depends(get_session), # TODO: Add current_user dependency when auth is implemented # current_user: User = Depends(get_current_user) ) -> Dict[str, Any]: """ Create a new note for a customer. Creates a new note associated with the specified customer. The note will be attributed to the currently authenticated user. **Features:** - Automatic user attribution to authenticated user - Description validation (1-2000 characters) - Customer existence validation - Returns full note details with user information **Request Body Example:** ```json { "description": "Customer called regarding payment inquiry and requested status update" } ``` **Authorization Required:** Valid JWT token (currently uses placeholder user ID) """ try: # TODO: Replace with actual current_user when auth is implemented # For now, creating a temporary user object (remove when auth is implemented) created_by_user = User( id=1, email="temp@example.com", username="temp_user", first_name="Temp", last_name="User" ) note = await create_customer_note_service( db, customer_id, note_data, created_by_user ) return { "success": True, "message": "Note created successfully", "data": note.model_dump(), "status_code": 201 } except APIException as api_exc: raise HTTPException( status_code=api_exc.status_code, detail=api_exc.message ) except Exception as e: logger.error(f"Unexpected error in create_customer_note: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal server error" ) @router.get( "/{customer_id}/notes/{note_id}", response_model=Dict[str, Any], summary="Get Customer Note", description="Get a specific note by ID for a customer", response_description="Note details with user information", responses={ 200: { "description": "Note found successfully", "content": { "application/json": { "example": { "success": True, "message": "Note found", "data": { "id": 1, "description": "Customer called regarding payment inquiry", "created_at": "2025-12-02T15:30:00.000Z", "updated_at": "2025-12-02T15:30:00.000Z", "deleted_at": None, "created_by_id": 1, "created_by": { "id": 1, "first_name": "Admin", "last_name": "User", "email": "admin@example.com" } } } } } }, 404: {"description": "Customer or note not found"}, 500: {"description": "Internal server error"} } ) async def get_customer_note( customer_id: str = Path(..., description="Customer ID (customer_id field)", example="ACCKDU1537895E7"), note_id: int = Path(..., description="Note ID to retrieve", example=1), db: Session = Depends(get_session) ) -> Dict[str, Any]: """ Get a specific note by ID for a customer. Returns full details of the specified note, including the user who created it. **Features:** - Full note details with user information - Customer and note existence validation - Soft-delete awareness (won't return deleted notes) **Parameters:** - **customer_id**: The customer's unique identifier - **note_id**: The note's unique ID to retrieve """ try: note = await get_customer_note_service(db, customer_id, note_id) return { "success": True, "message": "Note found", "data": note.model_dump(), "status_code": 200 } except APIException as api_exc: raise HTTPException( status_code=api_exc.status_code, detail=api_exc.message ) except Exception as e: logger.error(f"Unexpected error in get_customer_note: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal server error" ) @router.put( "/{customer_id}/notes/{note_id}", response_model=Dict[str, Any], summary="Update Customer Note", description="Update an existing note for a customer", response_description="Note updated successfully with full details", responses={ 200: { "description": "Note updated successfully", "content": { "application/json": { "example": { "success": True, "message": "Note updated successfully", "data": { "id": 1, "description": "Updated note: Customer called regarding payment inquiry and provided additional details", "created_at": "2025-12-02T15:30:00.000Z", "updated_at": "2025-12-02T16:15:00.000Z", "deleted_at": None, "created_by_id": 1, "created_by": { "id": 1, "first_name": "Admin", "last_name": "User", "email": "admin@example.com" } } } } } }, 400: {"description": "Invalid input data or validation error"}, 404: {"description": "Customer or note not found"}, 500: {"description": "Internal server error"} } ) async def update_customer_note( customer_id: str = Path(..., description="Customer ID (customer_id field)", example="ACCKDU1537895E7"), note_id: int = Path(..., description="Note ID to update", example=1), note_data: NoteUpdateRequestSchema = Body(..., description="Note update data with description (1-2000 characters)"), db: Session = Depends(get_session), # TODO: Add current_user dependency when auth is implemented # current_user: User = Depends(get_current_user) ) -> Dict[str, Any]: """ Update an existing note for a customer. Updates the specified note with new data. Only the user who created the note or admin users should be able to update it. **Features:** - Description validation (1-2000 characters) - Customer and note existence validation - Automatic timestamp update - Full note details in response **Request Body Example:** ```json { "description": "Updated note: Customer called regarding payment inquiry and provided additional details" } ``` **Authorization Required:** Valid JWT token (currently uses placeholder user ID) **Permissions:** Note creator or admin users """ try: # TODO: Replace with actual current_user when auth is implemented # For now, creating a temporary user object (remove when auth is implemented) updated_by_user = User( id=1, email="temp@example.com", username="temp_user", first_name="Temp", last_name="User" ) note = await update_customer_note_service( db, customer_id, note_id, note_data, updated_by_user ) return { "success": True, "message": "Note updated successfully", "data": note.model_dump(), "status_code": 200 } except APIException as api_exc: raise HTTPException( status_code=api_exc.status_code, detail=api_exc.message ) except Exception as e: logger.error(f"Unexpected error in update_customer_note: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal server error" ) @router.delete( "/{customer_id}/notes/{note_id}", response_model=Dict[str, Any], summary="Delete Customer Note", description="Soft delete a note for a customer", response_description="Note deleted successfully", responses={ 200: { "description": "Note deleted successfully", "content": { "application/json": { "example": { "success": True, "message": "Note deleted successfully", "data": None } } } }, 404: {"description": "Customer or note not found"}, 500: {"description": "Internal server error"} } ) async def delete_customer_note( customer_id: str = Path(..., description="Customer ID (customer_id field)", example="ACCKDU1537895E7"), note_id: int = Path(..., description="Note ID to delete", example=1), db: Session = Depends(get_session), # TODO: Add current_user dependency when auth is implemented # current_user: User = Depends(get_current_user) ) -> Dict[str, Any]: """ Delete a note for a customer. Soft deletes the specified note (sets deleted_at timestamp). Only the user who created the note or admin users should be able to delete it. **Features:** - Soft delete (preserves data with deleted_at timestamp) - Customer and note existence validation - Returns success confirmation **Parameters:** - **customer_id**: The customer's unique identifier - **note_id**: The note's unique ID to delete **Authorization Required:** Valid JWT token (currently uses placeholder user ID) **Permissions:** Note creator or admin users **Note:** This is a soft delete operation - the note will be marked as deleted but data is preserved in the database. """ try: # TODO: Replace with actual current_user when auth is implemented # For now, creating a temporary user object (remove when auth is implemented) deleted_by_user = User( id=1, email="temp@example.com", username="temp_user", first_name="Temp", last_name="User" ) result = await delete_customer_note_service( db, customer_id, note_id, deleted_by_user ) return result except APIException as api_exc: raise HTTPException( status_code=api_exc.status_code, detail=api_exc.message ) except Exception as e: logger.error(f"Unexpected error in delete_customer_note: {e}") raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Internal server error" )