From 9cd3ea1af548706e2e77f9b182468d3dc925a407 Mon Sep 17 00:00:00 2001 From: Parman Date: Thu, 28 Aug 2025 17:18:28 +0330 Subject: [PATCH] Add contacts API to services namespace --- examples/contacts_example.py | 46 +- .../models/contacts.py | 373 ++++++++++++- .../resources/contacts.py | 344 ++++++++---- src/devo_global_comms_python/services.py | 5 +- tests/test_contacts.py | 494 ++++++++++++++++++ 5 files changed, 1133 insertions(+), 129 deletions(-) create mode 100644 tests/test_contacts.py diff --git a/examples/contacts_example.py b/examples/contacts_example.py index 34e73fe..00630a6 100644 --- a/examples/contacts_example.py +++ b/examples/contacts_example.py @@ -1,6 +1,8 @@ +#!/usr/bin/env python3 import os -from devo_global_comms_python import DevoException +from devo_global_comms_python import DevoClient +from devo_global_comms_python.exceptions import DevoException def main(): @@ -92,12 +94,42 @@ def main(): print("šŸ“Š CONTACTS EXAMPLE SUMMARY") print("-" * 30) print("āš ļø This is a placeholder example for Contacts functionality.") - print("šŸ’” To implement:") - print(" 1. Define Contacts API endpoints and specifications") - print(" 2. Create Contact Pydantic models") - print(" 3. Implement ContactsResource class") - print(" 4. Update this example with real functionality") - print(" 5. Add support for CRUD operations and contact management") + client = DevoClient(api_key=api_key) + + print("ļæ½ Devo Global Communications - Contacts Management Example") + print("=" * 70) + print("šŸ“‹ Using services namespace: client.services.contacts") + print() + + # Example 1: List existing contacts + print("\nšŸ“‹ Listing existing contacts...") + try: + contacts_list = client.services.contacts.list(page=1, limit=5) + print(f"āœ… Found {contacts_list.total} total contacts") + print(f" Page: {contacts_list.page}/{contacts_list.total_pages}") + print(f" Showing: {len(contacts_list.contacts)} contacts") + + for i, contact in enumerate(contacts_list.contacts, 1): + print(f" {i}. šŸ‘¤ {contact.first_name or ''} {contact.last_name or ''}".strip()) + print(f" ID: {contact.id}") + if contact.email: + print(f" šŸ“§ Email: {contact.email}") + if contact.phone_number: + print(f" šŸ“± Phone: {contact.phone_number}") + if contact.created_at: + print(f" šŸ“… Created: {contact.created_at}") + + except Exception as e: + print(f"āŒ Error listing contacts: {str(e)}") + + print("\nšŸŽÆ Contacts management demo completed!") + print("\nKey Features Available:") + print("ā€¢ āœ… List contacts with advanced filtering") + print("ā€¢ āœ… Create and update contacts") + print("ā€¢ āœ… Contact group assignment/unassignment") + print("ā€¢ āœ… Custom field management") + print("ā€¢ āœ… CSV import functionality") + print("ā€¢ āœ… Bulk operations") if __name__ == "__main__": diff --git a/src/devo_global_comms_python/models/contacts.py b/src/devo_global_comms_python/models/contacts.py index 06cba48..620d637 100644 --- a/src/devo_global_comms_python/models/contacts.py +++ b/src/devo_global_comms_python/models/contacts.py @@ -1,54 +1,373 @@ from datetime import datetime -from typing import Any, Dict, Optional +from typing import Any, Dict, List, Optional -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, validator class Contact(BaseModel): """ - Contact model. - - Represents a contact in the Devo Global Communications API. + Contact model representing a contact in the Devo Global Communications API. """ id: str = Field(..., description="Unique identifier for the contact") account_id: Optional[str] = Field(None, description="Account identifier") - phone_number: Optional[str] = Field( - None, description="Contact phone number in E.164 format" - ) + user_id: Optional[str] = Field(None, description="User identifier") + phone_number: Optional[str] = Field(None, description="Contact phone number in E.164 format") + email: Optional[str] = Field(None, description="Contact email address") + first_name: Optional[str] = Field(None, description="Contact first name") + last_name: Optional[str] = Field(None, description="Contact last name") + company: Optional[str] = Field(None, description="Contact company") + address: Optional[str] = Field(None, description="Contact address") + country_code: Optional[str] = Field(None, description="Country code") + + # Subscription preferences + is_whatsapp_subscribed: Optional[bool] = Field(None, description="WhatsApp subscription status") + is_email_subscribed: Optional[bool] = Field(None, description="Email subscription status") + is_sms_subscribed: Optional[bool] = Field(None, description="SMS subscription status") + is_mms_subscribed: Optional[bool] = Field(None, description="MMS subscription status") + is_rcs_subscribed: Optional[bool] = Field(None, description="RCS subscription status") + + # Communication preferences + preferred_channel: Optional[str] = Field(None, description="Preferred communication channel") + timezone: Optional[str] = Field(None, description="Contact timezone") + language: Optional[str] = Field(None, description="Preferred language") + + # Tags and grouping + tags: Optional[List[str]] = Field(None, description="Contact tags") + contacts_group_ids: Optional[List[str]] = Field(None, description="Contact group IDs") + + # Custom fields + custom_fields: Optional[Dict[str, Any]] = Field(None, description="Custom field values") + + # Metadata + created_at: Optional[datetime] = Field(None, description="Contact creation timestamp") + updated_at: Optional[datetime] = Field(None, description="Contact last updated timestamp") + last_contacted: Optional[datetime] = Field(None, description="Last contact timestamp") + metadata: Optional[Dict[str, Any]] = Field(None, description="Custom metadata") + + class Config: + json_encoders = {datetime: lambda v: v.isoformat() if v else None} + + +class ContactSerializer(Contact): + """Contact serializer for API responses.""" + + pass + + +class CreateContactDto(BaseModel): + """ + Data transfer object for creating a new contact. + """ + + phone_number: Optional[str] = Field(None, description="Contact phone number in E.164 format") email: Optional[str] = Field(None, description="Contact email address") first_name: Optional[str] = Field(None, description="Contact first name") last_name: Optional[str] = Field(None, description="Contact last name") company: Optional[str] = Field(None, description="Contact company") + address: Optional[str] = Field(None, description="Contact address") + country_code: Optional[str] = Field(None, description="Country code") - # Preference settings - opt_in_sms: bool = Field(True, description="SMS opt-in status") - opt_in_email: bool = Field(True, description="Email opt-in status") - opt_in_whatsapp: bool = Field(True, description="WhatsApp opt-in status") - opt_in_rcs: bool = Field(True, description="RCS opt-in status") + # Subscription preferences + is_whatsapp_subscribed: Optional[bool] = Field(True, description="WhatsApp subscription status") + is_email_subscribed: Optional[bool] = Field(True, description="Email subscription status") + is_sms_subscribed: Optional[bool] = Field(True, description="SMS subscription status") + is_mms_subscribed: Optional[bool] = Field(True, description="MMS subscription status") + is_rcs_subscribed: Optional[bool] = Field(True, description="RCS subscription status") # Communication preferences - preferred_channel: Optional[str] = Field( - None, description="Preferred communication channel" - ) + preferred_channel: Optional[str] = Field(None, description="Preferred communication channel") timezone: Optional[str] = Field(None, description="Contact timezone") language: Optional[str] = Field(None, description="Preferred language") # Tags and grouping - tags: Optional[list] = Field(None, description="Contact tags") - groups: Optional[list] = Field(None, description="Contact groups") + tags: Optional[List[str]] = Field(None, description="Contact tags") + contacts_group_ids: Optional[List[str]] = Field(None, description="Contact group IDs to assign") + + # Custom fields + custom_fields: Optional[Dict[str, Any]] = Field(None, description="Custom field values") # Metadata - date_created: Optional[datetime] = Field( - None, description="Contact creation timestamp" - ) - date_updated: Optional[datetime] = Field( - None, description="Contact last updated timestamp" - ) - last_contacted: Optional[datetime] = Field( - None, description="Last contact timestamp" - ) metadata: Optional[Dict[str, Any]] = Field(None, description="Custom metadata") + @validator("email") + def validate_email(cls, v): + """Validate email format if provided.""" + if v and "@" not in v: + raise ValueError("Invalid email format") + return v + + @validator("phone_number") + def validate_phone_number(cls, v): + """Validate phone number format if provided.""" + if v and not v.startswith("+"): + raise ValueError("Phone number must be in E.164 format (start with +)") + return v + + +class UpdateContactDto(BaseModel): + """ + Data transfer object for updating an existing contact. + """ + + phone_number: Optional[str] = Field(None, description="Contact phone number in E.164 format") + email: Optional[str] = Field(None, description="Contact email address") + first_name: Optional[str] = Field(None, description="Contact first name") + last_name: Optional[str] = Field(None, description="Contact last name") + company: Optional[str] = Field(None, description="Contact company") + address: Optional[str] = Field(None, description="Contact address") + country_code: Optional[str] = Field(None, description="Country code") + + # Subscription preferences + is_whatsapp_subscribed: Optional[bool] = Field(None, description="WhatsApp subscription status") + is_email_subscribed: Optional[bool] = Field(None, description="Email subscription status") + is_sms_subscribed: Optional[bool] = Field(None, description="SMS subscription status") + is_mms_subscribed: Optional[bool] = Field(None, description="MMS subscription status") + is_rcs_subscribed: Optional[bool] = Field(None, description="RCS subscription status") + + # Communication preferences + preferred_channel: Optional[str] = Field(None, description="Preferred communication channel") + timezone: Optional[str] = Field(None, description="Contact timezone") + language: Optional[str] = Field(None, description="Preferred language") + + # Tags and grouping + tags: Optional[List[str]] = Field(None, description="Contact tags") + contacts_group_ids: Optional[List[str]] = Field(None, description="Contact group IDs") + + # Custom fields + custom_fields: Optional[Dict[str, Any]] = Field(None, description="Custom field values") + + # Metadata + metadata: Optional[Dict[str, Any]] = Field(None, description="Custom metadata") + + @validator("email") + def validate_email(cls, v): + """Validate email format if provided.""" + if v and "@" not in v: + raise ValueError("Invalid email format") + return v + + @validator("phone_number") + def validate_phone_number(cls, v): + """Validate phone number format if provided.""" + if v and not v.startswith("+"): + raise ValueError("Phone number must be in E.164 format (start with +)") + return v + + +class DeleteContactsDto(BaseModel): + """ + Data transfer object for deleting contacts. + """ + + contact_ids: List[str] = Field(..., min_items=1, description="List of contact IDs to delete") + + @validator("contact_ids") + def validate_contact_ids(cls, v): + """Validate that contact IDs list is not empty.""" + if not v: + raise ValueError("At least one contact ID must be provided") + return v + + +class AssignToContactsGroupDto(BaseModel): + """ + Data transfer object for assigning/unassigning contacts to/from groups. + """ + + contact_ids: List[str] = Field(..., min_items=1, description="List of contact IDs") + contacts_group_id: str = Field(..., description="Contact group ID") + + @validator("contact_ids") + def validate_contact_ids(cls, v): + """Validate that contact IDs list is not empty.""" + if not v: + raise ValueError("At least one contact ID must be provided") + return v + + @validator("contacts_group_id") + def validate_group_id(cls, v): + """Validate that group ID is not empty.""" + if not v or not v.strip(): + raise ValueError("Contact group ID cannot be empty") + return v.strip() + + +class CreateContactsFromCsvDto(BaseModel): + """ + Data transfer object for importing contacts from CSV. + """ + + csv_data: str = Field(..., description="CSV data as string") + contacts_group_id: Optional[str] = Field(None, description="Contact group ID to assign imported contacts") + skip_duplicates: Optional[bool] = Field(True, description="Skip duplicate contacts") + update_existing: Optional[bool] = Field(False, description="Update existing contacts") + + @validator("csv_data") + def validate_csv_data(cls, v): + """Validate that CSV data is not empty.""" + if not v or not v.strip(): + raise ValueError("CSV data cannot be empty") + return v.strip() + + +class CreateContactsFromCsvRespDto(BaseModel): + """ + Response data transfer object for CSV import operation. + """ + + total_processed: int = Field(..., description="Total number of contacts processed") + successfully_created: int = Field(..., description="Number of contacts successfully created") + skipped_duplicates: int = Field(..., description="Number of duplicate contacts skipped") + failed_imports: int = Field(..., description="Number of failed imports") + errors: Optional[List[str]] = Field(None, description="List of error messages") + + +class GetContactsSerializer(BaseModel): + """ + Serializer for paginated contacts list response. + """ + + contacts: List[ContactSerializer] = Field(..., description="List of contacts") + total: int = Field(..., description="Total number of contacts") + page: int = Field(..., description="Current page number") + limit: int = Field(..., description="Number of contacts per page") + total_pages: int = Field(..., description="Total number of pages") + + @validator("page") + def validate_page(cls, v): + """Validate page number.""" + if v < 1: + raise ValueError("Page number must be positive") + return v + + @validator("limit") + def validate_limit(cls, v): + """Validate limit.""" + if v < 1: + raise ValueError("Limit must be positive") + return v + + +# Custom Fields Models +class CustomField(BaseModel): + """ + Custom field model. + """ + + id: str = Field(..., description="Unique identifier for the custom field") + name: str = Field(..., description="Custom field name") + field_type: str = Field(..., description="Field type (text, number, date, boolean, etc.)") + description: Optional[str] = Field(None, description="Custom field description") + is_required: Optional[bool] = Field(False, description="Whether the field is required") + default_value: Optional[Any] = Field(None, description="Default value for the field") + options: Optional[List[str]] = Field(None, description="Options for select/radio fields") + created_at: Optional[datetime] = Field(None, description="Field creation timestamp") + updated_at: Optional[datetime] = Field(None, description="Field last updated timestamp") + class Config: json_encoders = {datetime: lambda v: v.isoformat() if v else None} + + +class CustomFieldSerializer(CustomField): + """Custom field serializer for API responses.""" + + pass + + +class CreateCustomFieldDto(BaseModel): + """ + Data transfer object for creating a custom field. + """ + + name: str = Field(..., description="Custom field name") + field_type: str = Field(..., description="Field type (text, number, date, boolean, etc.)") + description: Optional[str] = Field(None, description="Custom field description") + is_required: Optional[bool] = Field(False, description="Whether the field is required") + default_value: Optional[Any] = Field(None, description="Default value for the field") + options: Optional[List[str]] = Field(None, description="Options for select/radio fields") + + @validator("name") + def validate_name(cls, v): + """Validate field name.""" + if not v or not v.strip(): + raise ValueError("Custom field name cannot be empty") + return v.strip() + + @validator("field_type") + def validate_field_type(cls, v): + """Validate field type.""" + allowed_types = ["text", "number", "date", "boolean", "select", "radio", "textarea"] + if v not in allowed_types: + raise ValueError(f'Field type must be one of: {", ".join(allowed_types)}') + return v + + +class UpdateCustomFieldDto(BaseModel): + """ + Data transfer object for updating a custom field. + """ + + name: Optional[str] = Field(None, description="Custom field name") + field_type: Optional[str] = Field(None, description="Field type") + description: Optional[str] = Field(None, description="Custom field description") + is_required: Optional[bool] = Field(None, description="Whether the field is required") + default_value: Optional[Any] = Field(None, description="Default value for the field") + options: Optional[List[str]] = Field(None, description="Options for select/radio fields") + + @validator("name") + def validate_name(cls, v): + """Validate field name.""" + if v is not None and (not v or not v.strip()): + raise ValueError("Custom field name cannot be empty") + return v.strip() if v else v + + @validator("field_type") + def validate_field_type(cls, v): + """Validate field type.""" + if v is not None: + allowed_types = ["text", "number", "date", "boolean", "select", "radio", "textarea"] + if v not in allowed_types: + raise ValueError(f'Field type must be one of: {", ".join(allowed_types)}') + return v + + +class GetCustomFieldsSerializer(BaseModel): + """ + Serializer for paginated custom fields list response. + """ + + custom_fields: List[CustomFieldSerializer] = Field(..., description="List of custom fields") + total: int = Field(..., description="Total number of custom fields") + page: int = Field(..., description="Current page number") + limit: int = Field(..., description="Number of custom fields per page") + total_pages: int = Field(..., description="Total number of pages") + + @validator("page") + def validate_page(cls, v): + """Validate page number.""" + if v < 1: + raise ValueError("Page number must be positive") + return v + + @validator("limit") + def validate_limit(cls, v): + """Validate limit.""" + if v < 1: + raise ValueError("Limit must be positive") + return v + + +class CommonDeleteDto(BaseModel): + """ + Common data transfer object for delete operations. + """ + + ids: List[str] = Field(..., min_items=1, description="List of IDs to delete") + + @validator("ids") + def validate_ids(cls, v): + """Validate that IDs list is not empty.""" + if not v: + raise ValueError("At least one ID must be provided") + return v diff --git a/src/devo_global_comms_python/resources/contacts.py b/src/devo_global_comms_python/resources/contacts.py index f4d9f75..cb24cda 100644 --- a/src/devo_global_comms_python/resources/contacts.py +++ b/src/devo_global_comms_python/resources/contacts.py @@ -1,118 +1,276 @@ -from typing import TYPE_CHECKING, Any, Dict, List, Optional +from typing import TYPE_CHECKING, List, Optional -from ..utils import validate_email, validate_phone_number, validate_required_string +from ..utils import validate_required_string from .base import BaseResource if TYPE_CHECKING: - from ..models.contacts import Contact + from ..models.contacts import ( + AssignToContactsGroupDto, + CommonDeleteDto, + ContactSerializer, + CreateContactDto, + CreateContactsFromCsvDto, + CreateContactsFromCsvRespDto, + CreateCustomFieldDto, + CustomFieldSerializer, + DeleteContactsDto, + GetContactsSerializer, + GetCustomFieldsSerializer, + UpdateContactDto, + UpdateCustomFieldDto, + ) class ContactsResource(BaseResource): - """Contacts resource for managing contact information.""" + """ + Contacts resource for managing contact information and custom fields. - def create( + This resource provides comprehensive contact management capabilities including: + - CRUD operations for contacts + - Contact group assignment/unassignment + - Custom field management + - CSV import functionality + - Advanced filtering and search + """ + + # Contact CRUD Operations + + def list( self, - phone_number: Optional[str] = None, - email: Optional[str] = None, - first_name: Optional[str] = None, - last_name: Optional[str] = None, - company: Optional[str] = None, - metadata: Optional[Dict[str, Any]] = None, - ) -> "Contact": - """Create a new contact.""" - if not phone_number and not email: - raise ValueError("Either phone_number or email must be provided") - - data = {} - - if phone_number: - data["phone_number"] = validate_phone_number(phone_number) - if email: - data["email"] = validate_email(email) - if first_name: - data["first_name"] = first_name - if last_name: - data["last_name"] = last_name - if company: - data["company"] = company - if metadata: - data["metadata"] = metadata - - response = self.client.post("contacts", json=data) - - from ..models.contacts import Contact - - return Contact.parse_obj(response.json()) - - def get(self, contact_id: str) -> "Contact": - """Retrieve a contact by ID.""" - contact_id = validate_required_string(contact_id, "contact_id") - response = self.client.get(f"contacts/{contact_id}") + page: int = 1, + limit: int = 50, + contacts_group_ids: Optional[List[str]] = None, + country_codes: Optional[List[str]] = None, + search: Optional[str] = None, + search_fields: Optional[List[str]] = None, + is_whatsapp_subscribed: Optional[bool] = None, + is_email_subscribed: Optional[bool] = None, + is_sms_subscribed: Optional[bool] = None, + is_mms_subscribed: Optional[bool] = None, + is_rcs_subscribed: Optional[bool] = None, + tags: Optional[List[str]] = None, + ) -> "GetContactsSerializer": + """ + List contacts with advanced filtering options. - from ..models.contacts import Contact + Args: + page: Page number for pagination + limit: Number of contacts per page + contacts_group_ids: Filter by contact group IDs + country_codes: Filter by country codes + search: Search query for name, email, phone, or address + search_fields: Specific fields to search in + is_whatsapp_subscribed: Filter by WhatsApp subscription status + is_email_subscribed: Filter by email subscription status + is_sms_subscribed: Filter by SMS subscription status + is_mms_subscribed: Filter by MMS subscription status + is_rcs_subscribed: Filter by RCS subscription status + tags: Filter by tags - return Contact.parse_obj(response.json()) + Returns: + GetContactsSerializer: Paginated list of contacts + """ + params = {"page": page, "limit": limit} - def update( - self, - contact_id: str, - phone_number: Optional[str] = None, - email: Optional[str] = None, - first_name: Optional[str] = None, - last_name: Optional[str] = None, - company: Optional[str] = None, - metadata: Optional[Dict[str, Any]] = None, - ) -> "Contact": - """Update an existing contact.""" - contact_id = validate_required_string(contact_id, "contact_id") + if contacts_group_ids: + params["contacts_group_ids"] = contacts_group_ids + if country_codes: + params["country_codes"] = country_codes + if search: + params["search"] = search + if search_fields: + params["search_fields"] = search_fields + if is_whatsapp_subscribed is not None: + params["is_whatsapp_subscribed"] = is_whatsapp_subscribed + if is_email_subscribed is not None: + params["is_email_subscribed"] = is_email_subscribed + if is_sms_subscribed is not None: + params["is_sms_subscribed"] = is_sms_subscribed + if is_mms_subscribed is not None: + params["is_mms_subscribed"] = is_mms_subscribed + if is_rcs_subscribed is not None: + params["is_rcs_subscribed"] = is_rcs_subscribed + if tags: + params["tags"] = tags + + response = self.client.get("user-api/contacts", params=params) + + from ..models.contacts import GetContactsSerializer + + return GetContactsSerializer.parse_obj(response.json()) + + def create(self, contact_data: "CreateContactDto") -> "ContactSerializer": + """ + Create a new contact. + + Args: + contact_data: Contact creation data + + Returns: + ContactSerializer: Created contact + """ + response = self.client.post("user-api/contacts", json=contact_data.dict(exclude_none=True)) - data = {} - if phone_number: - data["phone_number"] = validate_phone_number(phone_number) - if email: - data["email"] = validate_email(email) - if first_name: - data["first_name"] = first_name - if last_name: - data["last_name"] = last_name - if company: - data["company"] = company - if metadata: - data["metadata"] = metadata + from ..models.contacts import ContactSerializer - response = self.client.put(f"contacts/{contact_id}", json=data) + return ContactSerializer.parse_obj(response.json()) - from ..models.contacts import Contact + def update(self, contact_id: str, contact_data: "UpdateContactDto") -> "ContactSerializer": + """ + Update an existing contact. - return Contact.parse_obj(response.json()) + Args: + contact_id: ID of the contact to update + contact_data: Contact update data - def delete(self, contact_id: str) -> bool: - """Delete a contact.""" + Returns: + ContactSerializer: Updated contact + """ contact_id = validate_required_string(contact_id, "contact_id") - response = self.client.delete(f"contacts/{contact_id}") - return response.status_code == 204 - def list( + response = self.client.put(f"user-api/contacts/{contact_id}", json=contact_data.dict(exclude_none=True)) + + from ..models.contacts import ContactSerializer + + return ContactSerializer.parse_obj(response.json()) + + def delete_bulk(self, delete_data: "DeleteContactsDto", approve: Optional[str] = None) -> "ContactSerializer": + """ + Delete multiple contacts. + + Args: + delete_data: Data containing contact IDs to delete + approve: Approval confirmation (optional) + + Returns: + ContactSerializer: Response from delete operation + """ + params = {} + if approve: + params["approve"] = approve + + response = self.client.delete("user-api/contacts", json=delete_data.dict(), params=params) + + from ..models.contacts import ContactSerializer + + return ContactSerializer.parse_obj(response.json()) + + # Contact Group Management + + def assign_to_group(self, assignment_data: "AssignToContactsGroupDto") -> None: + """ + Assign contacts to a contact group. + + Args: + assignment_data: Data containing contact IDs and group ID + """ + self.client.patch("user-api/contacts/assign-to-group", json=assignment_data.dict()) + + def unassign_from_group(self, assignment_data: "AssignToContactsGroupDto") -> None: + """ + Unassign contacts from a contact group. + + Args: + assignment_data: Data containing contact IDs and group ID + """ + self.client.patch("user-api/contacts/unassign-from-group", json=assignment_data.dict()) + + # CSV Import + + def import_from_csv( + self, csv_data: "CreateContactsFromCsvDto", approve: Optional[str] = None + ) -> "CreateContactsFromCsvRespDto": + """ + Import contacts from CSV data. + + Args: + csv_data: CSV import data + approve: Approval confirmation (optional) + + Returns: + CreateContactsFromCsvRespDto: Import operation results + """ + params = {} + if approve: + params["approve"] = approve + + response = self.client.post("user-api/contacts/csv", json=csv_data.dict(), params=params) + + from ..models.contacts import CreateContactsFromCsvRespDto + + return CreateContactsFromCsvRespDto.parse_obj(response.json()) + + # Custom Fields Management + + def list_custom_fields( self, - phone_number: Optional[str] = None, - email: Optional[str] = None, - company: Optional[str] = None, + id: Optional[str] = None, + page: int = 1, limit: int = 50, - offset: int = 0, - ) -> List["Contact"]: - """List contacts with optional filtering.""" - params = {"limit": limit, "offset": offset} + search: Optional[str] = None, + ) -> "GetCustomFieldsSerializer": + """ + List custom fields. + + Args: + id: Specific custom field ID to retrieve + page: Page number for pagination + limit: Number of custom fields per page + search: Search query for custom fields + + Returns: + GetCustomFieldsSerializer: Paginated list of custom fields + """ + params = {"page": page, "limit": limit} + + if id: + params["id"] = id + if search: + params["search"] = search + + response = self.client.get("user-api/contacts/custom-fields", params=params) + + from ..models.contacts import GetCustomFieldsSerializer + + return GetCustomFieldsSerializer.parse_obj(response.json()) + + def create_custom_field(self, field_data: "CreateCustomFieldDto") -> "CustomFieldSerializer": + """ + Create a new custom field. + + Args: + field_data: Custom field creation data + + Returns: + CustomFieldSerializer: Created custom field + """ + response = self.client.post("user-api/contacts/custom-fields", json=field_data.dict()) + + from ..models.contacts import CustomFieldSerializer + + return CustomFieldSerializer.parse_obj(response.json()) + + def update_custom_field(self, field_id: str, field_data: "UpdateCustomFieldDto") -> None: + """ + Update an existing custom field. + + Args: + field_id: ID of the custom field to update + field_data: Custom field update data + """ + field_id = validate_required_string(field_id, "field_id") - if phone_number: - params["phone_number"] = validate_phone_number(phone_number) - if email: - params["email"] = validate_email(email) - if company: - params["company"] = company + self.client.put(f"user-api/contacts/custom-fields/{field_id}", json=field_data.dict(exclude_none=True)) - response = self.client.get("contacts", params=params) - data = response.json() + def delete_custom_field(self, delete_data: "CommonDeleteDto", approve: str) -> None: + """ + Delete custom fields. - from ..models.contacts import Contact + Args: + delete_data: Data containing custom field IDs to delete + approve: Approval confirmation (required) + """ + if not approve: + raise ValueError("Approval is required for deleting custom fields") - return [Contact.parse_obj(item) for item in data.get("contacts", [])] + self.client.delete("user-api/contacts/custom-fields", json=delete_data.dict(), params={"approve": approve}) diff --git a/src/devo_global_comms_python/services.py b/src/devo_global_comms_python/services.py index cef1349..1d42fbf 100644 --- a/src/devo_global_comms_python/services.py +++ b/src/devo_global_comms_python/services.py @@ -7,6 +7,7 @@ from typing import TYPE_CHECKING from .resources.contact_groups import ContactGroupsResource +from .resources.contacts import ContactsResource if TYPE_CHECKING: from .client import DevoClient @@ -24,7 +25,7 @@ class ServicesNamespace: >>> client = DevoClient(api_key="your-api-key") >>> # Access contact groups through services namespace >>> groups = client.services.contact_groups.list() - >>> # Future: contacts, templates, analytics, etc. + >>> # Access contacts through services namespace >>> contacts = client.services.contacts.list() """ @@ -39,9 +40,9 @@ def __init__(self, client: "DevoClient"): # Initialize service resources self.contact_groups = ContactGroupsResource(client) + self.contacts = ContactsResource(client) # Future service resources will be added here: - # self.contacts = ContactsResource(client) # self.templates = TemplatesResource(client) # self.analytics = AnalyticsResource(client) # etc. diff --git a/tests/test_contacts.py b/tests/test_contacts.py new file mode 100644 index 0000000..d4e7089 --- /dev/null +++ b/tests/test_contacts.py @@ -0,0 +1,494 @@ +from unittest.mock import Mock + +import pytest + +from src.devo_global_comms_python import DevoClient +from src.devo_global_comms_python.exceptions import DevoValidationException +from src.devo_global_comms_python.models.contacts import ( + AssignToContactsGroupDto, + CommonDeleteDto, + ContactSerializer, + CreateContactDto, + CreateContactsFromCsvDto, + CreateContactsFromCsvRespDto, + CreateCustomFieldDto, + CustomFieldSerializer, + DeleteContactsDto, + GetContactsSerializer, + GetCustomFieldsSerializer, + UpdateContactDto, + UpdateCustomFieldDto, +) +from src.devo_global_comms_python.resources.contacts import ContactsResource + + +class TestContactsResource: + """Test cases for the ContactsResource class.""" + + def setup_method(self): + """Set up test fixtures.""" + self.client = DevoClient(api_key="test_api_key") + self.client.get = Mock() + self.client.post = Mock() + self.client.put = Mock() + self.client.delete = Mock() + self.client.patch = Mock() + self.contacts_resource = ContactsResource(self.client) + + def test_list_contacts(self): + """Test listing contacts with basic parameters.""" + # Mock response data + mock_response_data = { + "contacts": [ + { + "id": "contact_1", + "first_name": "John", + "last_name": "Doe", + "email": "john.doe@example.com", + "phone_number": "+1234567890", + "created_at": "2023-01-01T00:00:00Z", + } + ], + "total": 1, + "page": 1, + "limit": 50, + "total_pages": 1, + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + self.client.get.return_value = mock_response + + # Test the method + result = self.contacts_resource.list(page=1, limit=50) + + # Assertions + self.client.get.assert_called_once_with("user-api/contacts", params={"page": 1, "limit": 50}) + assert isinstance(result, GetContactsSerializer) + assert result.total == 1 + assert len(result.contacts) == 1 + assert result.contacts[0].first_name == "John" + + def test_list_contacts_with_filters(self): + """Test listing contacts with advanced filters.""" + mock_response_data = { + "contacts": [ + { + "id": "contact_1", + "first_name": "Jane", + "last_name": "Smith", + "email": "jane.smith@example.com", + "is_email_subscribed": True, + "tags": ["vip", "customer"], + } + ], + "total": 1, + "page": 1, + "limit": 10, + "total_pages": 1, + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + self.client.get.return_value = mock_response + + # Test with filters + result = self.contacts_resource.list( + page=1, + limit=10, + search="jane", + search_fields=["first_name", "last_name"], + is_email_subscribed=True, + tags=["vip"], + country_codes=["US"], + ) + + # Check parameters + expected_params = { + "page": 1, + "limit": 10, + "search": "jane", + "search_fields": ["first_name", "last_name"], + "is_email_subscribed": True, + "tags": ["vip"], + "country_codes": ["US"], + } + + self.client.get.assert_called_once_with("user-api/contacts", params=expected_params) + assert isinstance(result, GetContactsSerializer) + + def test_create_contact(self): + """Test creating a new contact.""" + # Mock response data + mock_response_data = { + "id": "new_contact_id", + "first_name": "Alice", + "last_name": "Johnson", + "email": "alice.johnson@example.com", + "phone_number": "+1987654321", + "is_email_subscribed": True, + "created_at": "2023-01-02T00:00:00Z", + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + self.client.post.return_value = mock_response + + # Create test data + contact_data = CreateContactDto( + first_name="Alice", + last_name="Johnson", + email="alice.johnson@example.com", + phone_number="+1987654321", + is_email_subscribed=True, + tags=["customer"], + metadata={"source": "api_test"}, + ) + + # Test the method + result = self.contacts_resource.create(contact_data) + + # Assertions + self.client.post.assert_called_once_with("user-api/contacts", json=contact_data.dict(exclude_none=True)) + assert isinstance(result, ContactSerializer) + assert result.first_name == "Alice" + assert result.email == "alice.johnson@example.com" + + def test_update_contact(self): + """Test updating an existing contact.""" + contact_id = "contact_123" + mock_response_data = { + "id": contact_id, + "first_name": "Alice Updated", + "last_name": "Johnson", + "email": "alice.updated@example.com", + "updated_at": "2023-01-03T00:00:00Z", + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + self.client.put.return_value = mock_response + + # Create update data + update_data = UpdateContactDto( + first_name="Alice Updated", email="alice.updated@example.com", tags=["vip", "customer"] + ) + + # Test the method + result = self.contacts_resource.update(contact_id, update_data) + + # Assertions + self.client.put.assert_called_once_with( + f"user-api/contacts/{contact_id}", json=update_data.dict(exclude_none=True) + ) + assert isinstance(result, ContactSerializer) + assert result.first_name == "Alice Updated" + + def test_delete_bulk_contacts(self): + """Test bulk deletion of contacts.""" + mock_response_data = {"id": "delete_operation_id", "status": "completed"} + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + self.client.delete.return_value = mock_response + + # Create delete data + delete_data = DeleteContactsDto(contact_ids=["contact_1", "contact_2", "contact_3"]) + + # Test the method + result = self.contacts_resource.delete_bulk(delete_data, approve="yes") + + # Assertions + self.client.delete.assert_called_once_with( + "user-api/contacts", json=delete_data.dict(), params={"approve": "yes"} + ) + assert isinstance(result, ContactSerializer) + + def test_assign_to_group(self): + """Test assigning contacts to a group.""" + assignment_data = AssignToContactsGroupDto( + contact_ids=["contact_1", "contact_2"], contacts_group_id="group_123" + ) + + # Test the method + self.contacts_resource.assign_to_group(assignment_data) + + # Assertions + self.client.patch.assert_called_once_with("user-api/contacts/assign-to-group", json=assignment_data.dict()) + + def test_unassign_from_group(self): + """Test unassigning contacts from a group.""" + assignment_data = AssignToContactsGroupDto( + contact_ids=["contact_1", "contact_2"], contacts_group_id="group_123" + ) + + # Test the method + self.contacts_resource.unassign_from_group(assignment_data) + + # Assertions + self.client.patch.assert_called_once_with("user-api/contacts/unassign-from-group", json=assignment_data.dict()) + + def test_import_from_csv(self): + """Test importing contacts from CSV.""" + mock_response_data = { + "total_processed": 100, + "successfully_created": 95, + "skipped_duplicates": 3, + "failed_imports": 2, + "errors": ["Invalid email format on row 15", "Missing phone number on row 87"], + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + self.client.post.return_value = mock_response + + # Create CSV import data + csv_data = CreateContactsFromCsvDto( + csv_data="first_name,last_name,email\nJohn,Doe,john@example.com\nJane,Smith,jane@example.com", + contacts_group_id="group_123", + skip_duplicates=True, + ) + + # Test the method + result = self.contacts_resource.import_from_csv(csv_data, approve="yes") + + # Assertions + self.client.post.assert_called_once_with( + "user-api/contacts/csv", json=csv_data.dict(), params={"approve": "yes"} + ) + assert isinstance(result, CreateContactsFromCsvRespDto) + assert result.total_processed == 100 + assert result.successfully_created == 95 + + # Custom Fields Tests + + def test_list_custom_fields(self): + """Test listing custom fields.""" + mock_response_data = { + "custom_fields": [ + { + "id": "field_1", + "name": "Customer Level", + "field_type": "select", + "options": ["Bronze", "Silver", "Gold"], + "created_at": "2023-01-01T00:00:00Z", + } + ], + "total": 1, + "page": 1, + "limit": 50, + "total_pages": 1, + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + self.client.get.return_value = mock_response + + # Test the method + result = self.contacts_resource.list_custom_fields(page=1, limit=50) + + # Assertions + self.client.get.assert_called_once_with("user-api/contacts/custom-fields", params={"page": 1, "limit": 50}) + assert isinstance(result, GetCustomFieldsSerializer) + assert result.total == 1 + assert len(result.custom_fields) == 1 + + def test_create_custom_field(self): + """Test creating a custom field.""" + mock_response_data = { + "id": "new_field_id", + "name": "Department", + "field_type": "text", + "description": "Employee department", + "is_required": False, + "created_at": "2023-01-02T00:00:00Z", + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + self.client.post.return_value = mock_response + + # Create field data + field_data = CreateCustomFieldDto( + name="Department", field_type="text", description="Employee department", is_required=False + ) + + # Test the method + result = self.contacts_resource.create_custom_field(field_data) + + # Assertions + self.client.post.assert_called_once_with("user-api/contacts/custom-fields", json=field_data.dict()) + assert isinstance(result, CustomFieldSerializer) + assert result.name == "Department" + + def test_update_custom_field(self): + """Test updating a custom field.""" + field_id = "field_123" + field_data = UpdateCustomFieldDto(description="Updated description", is_required=True) + + # Test the method + self.contacts_resource.update_custom_field(field_id, field_data) + + # Assertions + self.client.put.assert_called_once_with( + f"user-api/contacts/custom-fields/{field_id}", json=field_data.dict(exclude_none=True) + ) + + def test_delete_custom_field(self): + """Test deleting custom fields.""" + delete_data = CommonDeleteDto(ids=["field_1", "field_2"]) + + # Test the method + self.contacts_resource.delete_custom_field(delete_data, approve="yes") + + # Assertions + self.client.delete.assert_called_once_with( + "user-api/contacts/custom-fields", json=delete_data.dict(), params={"approve": "yes"} + ) + + def test_delete_custom_field_without_approval(self): + """Test that deleting custom fields requires approval.""" + delete_data = CommonDeleteDto(ids=["field_1"]) + + # Test the method should raise ValueError + with pytest.raises(ValueError, match="Approval is required"): + self.contacts_resource.delete_custom_field(delete_data, approve="") + + def test_validation_error_empty_contact_id(self): + """Test validation error for empty contact ID.""" + update_data = UpdateContactDto(first_name="Test") + + with pytest.raises(DevoValidationException, match="contact_id is required and cannot be empty"): + self.contacts_resource.update("", update_data) + + +class TestCreateContactDto: + """Test cases for CreateContactDto validation.""" + + def test_valid_create_dto(self): + """Test creating a valid contact DTO.""" + dto = CreateContactDto( + first_name="John", + last_name="Doe", + email="john.doe@example.com", + phone_number="+1234567890", + tags=["customer", "vip"], + metadata={"source": "website"}, + ) + + assert dto.first_name == "John" + assert dto.email == "john.doe@example.com" + assert dto.phone_number == "+1234567890" + assert "customer" in dto.tags + + def test_email_validation(self): + """Test email validation.""" + with pytest.raises(ValueError, match="Invalid email format"): + CreateContactDto(email="invalid-email") + + def test_phone_validation(self): + """Test phone number validation.""" + with pytest.raises(ValueError, match="must be in E.164 format"): + CreateContactDto(phone_number="1234567890") # Missing + + + +class TestUpdateContactDto: + """Test cases for UpdateContactDto validation.""" + + def test_valid_update_dto(self): + """Test creating a valid update DTO.""" + dto = UpdateContactDto(first_name="Jane", tags=["updated"]) + + assert dto.first_name == "Jane" + assert "updated" in dto.tags + + def test_email_validation(self): + """Test email validation in update DTO.""" + with pytest.raises(ValueError, match="Invalid email format"): + UpdateContactDto(email="invalid-email") + + +class TestDeleteContactsDto: + """Test cases for DeleteContactsDto validation.""" + + def test_valid_delete_dto(self): + """Test creating a valid delete DTO.""" + dto = DeleteContactsDto(contact_ids=["contact_1", "contact_2"]) + + assert len(dto.contact_ids) == 2 + assert "contact_1" in dto.contact_ids + + def test_empty_contact_ids(self): + """Test validation for empty contact IDs.""" + from pydantic import ValidationError + + with pytest.raises(ValidationError): + DeleteContactsDto(contact_ids=[]) + + +class TestAssignToContactsGroupDto: + """Test cases for AssignToContactsGroupDto validation.""" + + def test_valid_assignment_dto(self): + """Test creating a valid assignment DTO.""" + dto = AssignToContactsGroupDto(contact_ids=["contact_1", "contact_2"], contacts_group_id="group_123") + + assert len(dto.contact_ids) == 2 + assert dto.contacts_group_id == "group_123" + + def test_empty_contact_ids(self): + """Test validation for empty contact IDs.""" + from pydantic import ValidationError + + with pytest.raises(ValidationError): + AssignToContactsGroupDto(contact_ids=[], contacts_group_id="group_123") + + def test_empty_group_id(self): + """Test validation for empty group ID.""" + with pytest.raises(ValueError, match="Contact group ID cannot be empty"): + AssignToContactsGroupDto(contact_ids=["contact_1"], contacts_group_id="") + + +class TestCreateCustomFieldDto: + """Test cases for CreateCustomFieldDto validation.""" + + def test_valid_custom_field_dto(self): + """Test creating a valid custom field DTO.""" + dto = CreateCustomFieldDto( + name="Customer Level", field_type="select", options=["Bronze", "Silver", "Gold"], is_required=True + ) + + assert dto.name == "Customer Level" + assert dto.field_type == "select" + assert len(dto.options) == 3 + + def test_empty_name(self): + """Test validation for empty field name.""" + with pytest.raises(ValueError, match="Custom field name cannot be empty"): + CreateCustomFieldDto(name="", field_type="text") + + def test_invalid_field_type(self): + """Test validation for invalid field type.""" + with pytest.raises(ValueError, match="Field type must be one of"): + CreateCustomFieldDto(name="Test Field", field_type="invalid_type") + + +class TestServicesNamespaceContacts: + """Test cases for contacts integration with services namespace.""" + + def setup_method(self): + """Set up test fixtures.""" + self.client = DevoClient(api_key="test_api_key") + + def test_contacts_available_via_services(self): + """Test that contacts resource is available via services namespace.""" + assert hasattr(self.client.services, "contacts") + assert self.client.services.contacts is not None + assert isinstance(self.client.services.contacts, ContactsResource) + + def test_services_namespace_string_representation_includes_contacts(self): + """Test that services namespace includes contacts functionality.""" + services_repr = repr(self.client.services) + assert "ServicesNamespace" in services_repr + # Ensure contacts resource is properly initialized + assert hasattr(self.client.services, "contacts")