From 8721c10211ed094841553e4e40f3a4445ce425f1 Mon Sep 17 00:00:00 2001 From: Parman Date: Thu, 28 Aug 2025 16:12:54 +0330 Subject: [PATCH] feat: implement complete contact groups API --- examples/README.md | 75 +++ examples/basic_usage.py | 38 +- examples/contact_groups_example.py | 349 ++++++++++++ src/devo_global_comms_python/client.py | 2 + .../models/__init__.py | 13 + .../models/contact_groups.py | 62 +++ .../resources/__init__.py | 2 + .../resources/contact_groups.py | 224 ++++++++ tests/test_contact_groups.py | 498 ++++++++++++++++++ 9 files changed, 1249 insertions(+), 14 deletions(-) create mode 100644 examples/contact_groups_example.py create mode 100644 src/devo_global_comms_python/models/contact_groups.py create mode 100644 src/devo_global_comms_python/resources/contact_groups.py create mode 100644 tests/test_contact_groups.py diff --git a/examples/README.md b/examples/README.md index 810fed8..f152e71 100644 --- a/examples/README.md +++ b/examples/README.md @@ -26,6 +26,11 @@ This directory contains comprehensive examples for using the Devo Global Communi ### ๐Ÿ‘ฅ Management Resources - **`contacts_example.py`** - ๐Ÿšง **Placeholder** (Contact management) +- **`contact_groups_example.py`** - โœ… **Complete Contact Groups API implementation** + - CRUD operations for contact groups + - Bulk operations and contact transfer + - Search and pagination features + - Metadata management and workflow examples ## ๐Ÿš€ Getting Started @@ -66,6 +71,9 @@ python examples/rcs_example.py python examples/email_example.py python examples/whatsapp_example.py python examples/contacts_example.py + +# Contact groups functionality (fully implemented) +python examples/contact_groups_example.py ``` ## ๐ŸŒ Omni-channel Messaging Examples (Fully Implemented) @@ -145,6 +153,73 @@ The following examples show the structure and planned functionality but are not - **RCS**: Rich messaging, cards, carousels, capability checks - **Contacts**: CRUD operations, contact management +## ๐Ÿ“ Contact Groups Examples (Fully Implemented) + +The contact groups resource is fully implemented with all CRUD operations: + +### ๐Ÿ”ง Available Functions +1. **List Groups** - `client.contact_groups.list()` + - Uses GET `/api/v1/contacts-groups` + - Pagination and search support + - Field filtering capabilities + +2. **Create Group** - `client.contact_groups.create()` + - Uses POST `/api/v1/contacts-groups` + - Metadata and contact assignment + - Validation and error handling + +3. **Update Group** - `client.contact_groups.update()` + - Uses PUT `/api/v1/contacts-groups/{group_id}` + - Partial updates with metadata + - Flexible field modification + +4. **Get Group** - `client.contact_groups.get_by_id()` + - Uses GET `/api/v1/contacts-groups/{group_id}` + - Complete group information retrieval + +5. **Delete Group** - `client.contact_groups.delete_by_id()` + - Uses DELETE `/api/v1/contacts-groups/{group_id}` + - Individual group deletion with approval + +6. **Bulk Delete** - `client.contact_groups.delete_bulk()` + - Uses DELETE `/api/v1/contacts-groups` + - Multiple group deletion with contact transfer + +7. **Search Groups** - `client.contact_groups.search()` + - Uses GET `/api/v1/contacts-groups` + - Advanced search with field filtering + +### ๐Ÿ’ก Key Features +- **Complete CRUD Operations**: Full lifecycle management +- **Bulk Operations**: Efficient multi-group operations +- **Contact Transfer**: Safe deletion with contact preservation +- **Metadata Support**: Custom metadata for business logic +- **Search & Filter**: Advanced query capabilities +- **Pagination**: Efficient large dataset handling + +### ๐Ÿ“ Example Usage +```python +from devo_global_comms_python.models.contact_groups import CreateContactsGroupDto + +# Create new contact group +group_data = CreateContactsGroupDto( + name="VIP Customers", + description="High-value customers", + contact_ids=["contact1", "contact2"], + metadata={"priority": "high"} +) +group = client.contact_groups.create(group_data) + +# List with pagination +groups = client.contact_groups.list(page=1, limit=10, search="VIP") + +# Search groups +search_results = client.contact_groups.search( + query="priority", + fields=["name", "description"] +) +``` + ## ๐Ÿ”ง Configuration Notes ### Phone Numbers diff --git a/examples/basic_usage.py b/examples/basic_usage.py index 5bc68cc..aa79778 100644 --- a/examples/basic_usage.py +++ b/examples/basic_usage.py @@ -37,8 +37,12 @@ def main(): resources.append(("๐Ÿ’ฌ WhatsApp", "Placeholder", "whatsapp_example.py")) if hasattr(client, "contacts"): resources.append(("๐Ÿ‘ฅ Contacts", "Placeholder", "contacts_example.py")) + if hasattr(client, "contact_groups"): + resources.append(("๐Ÿ—‚๏ธ Contact Groups", "Implemented", "contact_groups_example.py")) if hasattr(client, "rcs"): resources.append(("๐ŸŽด RCS", "Placeholder", "rcs_example.py")) + if hasattr(client, "messages"): + resources.append(("๐Ÿ“ฌ Messages", "Implemented", "omni_channel_example.py")) for resource, status, example_file in resources: print(f" {resource:<12} - {status:<12} -> {example_file}") @@ -66,11 +70,13 @@ def main(): print("\n๐Ÿ’ก Getting Started:") print("-" * 30) print("1. Run individual resource examples:") - print(" python examples/sms_example.py # Complete SMS functionality") - print(" python examples/email_example.py # Email examples (placeholder)") - print(" python examples/whatsapp_example.py # WhatsApp examples (placeholder)") - print(" python examples/contacts_example.py # Contact management (placeholder)") - print(" python examples/rcs_example.py # RCS examples (placeholder)") + print(" python examples/sms_example.py # Complete SMS functionality") + print(" python examples/contact_groups_example.py # Complete Contact Groups functionality") + print(" python examples/omni_channel_example.py # Complete Omni-channel messaging") + print(" python examples/email_example.py # Email examples (placeholder)") + print(" python examples/whatsapp_example.py # WhatsApp examples (placeholder)") + print(" python examples/contacts_example.py # Contact management (placeholder)") + print(" python examples/rcs_example.py # RCS examples (placeholder)") print() print("2. Quick SMS example:") print(" from devo_global_comms_python import DevoClient") @@ -86,20 +92,24 @@ def main(): print("-" * 30) print("Would you like to run a specific example?") print("1. SMS Example (full functionality)") - print("2. Email Example (placeholder)") - print("3. WhatsApp Example (placeholder)") - print("4. Contacts Example (placeholder)") - print("5. RCS Example (placeholder)") + print("2. Contact Groups Example (full functionality)") + print("3. Omni-channel Messaging Example (full functionality)") + print("4. Email Example (placeholder)") + print("5. WhatsApp Example (placeholder)") + print("6. Contacts Example (placeholder)") + print("7. RCS Example (placeholder)") print("0. Exit") try: - choice = input("\nEnter your choice (0-5): ").strip() + choice = input("\nEnter your choice (0-7): ").strip() example_files = { "1": "sms_example.py", - "2": "email_example.py", - "3": "whatsapp_example.py", - "4": "contacts_example.py", - "5": "rcs_example.py", + "2": "contact_groups_example.py", + "3": "omni_channel_example.py", + "4": "email_example.py", + "5": "whatsapp_example.py", + "6": "contacts_example.py", + "7": "rcs_example.py", } if choice in example_files: diff --git a/examples/contact_groups_example.py b/examples/contact_groups_example.py new file mode 100644 index 0000000..474f938 --- /dev/null +++ b/examples/contact_groups_example.py @@ -0,0 +1,349 @@ +#!/usr/bin/env python3 +import os +from datetime import datetime + +from devo_global_comms_python import DevoClient +from devo_global_comms_python.models.contact_groups import ( + CreateContactsGroupDto, + DeleteContactsGroupsDto, + UpdateContactsGroupDto, +) + + +def main(): + """ + Demonstrate contact groups management capabilities. + + Shows how to create, read, update and delete contact groups + using the Contact Groups API. + """ + + # Initialize the client + api_key = os.getenv("DEVO_API_KEY") + if not api_key: + print("โŒ Error: DEVO_API_KEY environment variable not set") + return + + client = DevoClient(api_key=api_key) + + print("๐Ÿ—‚๏ธ Devo Global Communications - Contact Groups Management Example") + print("=" * 75) + + # Example 1: List existing contact groups + print("\n๐Ÿ“‹ Listing existing contact groups...") + try: + groups_list = client.contact_groups.list(page=1, limit=5) + print(f"โœ… Found {groups_list.total} total groups") + print(f" Page: {groups_list.page}/{groups_list.total_pages}") + print(f" Showing: {len(groups_list.groups)} groups") + + for i, group in enumerate(groups_list.groups, 1): + print(f" {i}. ๐Ÿ“ {group.name}") + print(f" ID: {group.id}") + if group.description: + print(f" Description: {group.description}") + print(f" Contacts: {group.contacts_count or 0}") + if group.created_at: + print(f" Created: {group.created_at}") + + except Exception as e: + print(f"โŒ Error listing groups: {str(e)}") + + # Example 2: Create a new contact group + print("\nโž• Creating a new contact group...") + try: + new_group_data = CreateContactsGroupDto( + name=f"API Demo Group {datetime.now().strftime('%Y%m%d_%H%M%S')}", + description="A demonstration group created via the API", + contact_ids=["demo_contact_1", "demo_contact_2"], + metadata={ + "created_by": "api_example", + "purpose": "demonstration", + "created_at": datetime.now().isoformat(), + }, + ) + + new_group = client.contact_groups.create(new_group_data) + print("โœ… Contact group created successfully!") + print(f" ๐Ÿ“ Name: {new_group.name}") + print(f" ๐Ÿ†” ID: {new_group.id}") + print(f" ๐Ÿ“ Description: {new_group.description}") + print(f" ๐Ÿ‘ฅ Contacts: {new_group.contacts_count or 0}") + if new_group.created_at: + print(f" ๐Ÿ“… Created: {new_group.created_at}") + + created_group_id = new_group.id + + except Exception as e: + print(f"โŒ Error creating group: {str(e)}") + created_group_id = None + + # Example 3: Update the created group + if created_group_id: + print(f"\nโœ๏ธ Updating contact group {created_group_id}...") + try: + update_data = UpdateContactsGroupDto( + name=f"Updated API Demo Group {datetime.now().strftime('%H%M%S')}", + description="This group has been updated via the API", + metadata={"updated_by": "api_example", "updated_at": datetime.now().isoformat(), "version": "2.0"}, + ) + + updated_group = client.contact_groups.update(created_group_id, update_data) + print("โœ… Contact group updated successfully!") + print(f" ๐Ÿ“ New name: {updated_group.name}") + print(f" ๐Ÿ“ New description: {updated_group.description}") + if updated_group.updated_at: + print(f" ๐Ÿ“… Updated: {updated_group.updated_at}") + + except Exception as e: + print(f"โŒ Error updating group: {str(e)}") + + # Example 4: Get specific group by ID + if created_group_id: + print(f"\n๐Ÿ” Retrieving specific group {created_group_id}...") + try: + specific_group = client.contact_groups.get_by_id(created_group_id) + print("โœ… Group retrieved successfully!") + print(f" ๐Ÿ“ Name: {specific_group.name}") + print(f" ๐Ÿ“ Description: {specific_group.description}") + print(f" ๐Ÿ‘ฅ Contacts: {specific_group.contacts_count or 0}") + print(f" ๐Ÿ‘ค Owner: {specific_group.user_id}") + + except Exception as e: + print(f"โŒ Error retrieving group: {str(e)}") + + # Example 5: Search contact groups + print("\n๐Ÿ”Ž Searching contact groups...") + try: + search_results = client.contact_groups.search(query="demo", fields=["name", "description"], page=1, limit=10) + print(f"โœ… Search completed! Found {search_results.total} matching groups") + + for i, group in enumerate(search_results.groups, 1): + print(f" {i}. ๐Ÿ“ {group.name}") + print(f" ๐Ÿ†” ID: {group.id}") + if group.description: + print(f" ๐Ÿ“ Description: {group.description}") + + except Exception as e: + print(f"โŒ Error searching groups: {str(e)}") + + # Example 6: Advanced listing with filters + print("\n๐Ÿ”ง Advanced group listing with filters...") + try: + filtered_groups = client.contact_groups.list( + page=1, limit=3, search="demo", search_fields=["name", "description"] + ) + print("โœ… Filtered listing completed!") + print(f" Total groups matching 'demo': {filtered_groups.total}") + print(f" Showing page {filtered_groups.page} of {filtered_groups.total_pages}") + + for group in filtered_groups.groups: + print(f" ๐Ÿ“ {group.name} (ID: {group.id})") + + except Exception as e: + print(f"โŒ Error with filtered listing: {str(e)}") + + # Example 7: Bulk operations demonstration + print("\n๐Ÿ“ฆ Bulk operations example...") + + # First, let's create a few more groups for bulk operations + bulk_group_ids = [] + + try: + for i in range(3): + bulk_group_data = CreateContactsGroupDto( + name=f"Bulk Demo Group {i+1}", + description=f"Group {i+1} for bulk operations demo", + metadata={"bulk_demo": True, "group_number": i + 1}, + ) + + bulk_group = client.contact_groups.create(bulk_group_data) + bulk_group_ids.append(bulk_group.id) + print(f" โœ… Created bulk group {i+1}: {bulk_group.name}") + + print(f"๐Ÿ“Š Created {len(bulk_group_ids)} groups for bulk demo") + + except Exception as e: + print(f"โŒ Error creating bulk groups: {str(e)}") + + # Example 8: Individual group deletion + if created_group_id: + print(f"\n๐Ÿ—‘๏ธ Deleting individual group {created_group_id}...") + try: + deleted_group = client.contact_groups.delete_by_id(created_group_id, approve="yes") + print("โœ… Individual group deleted successfully!") + print(f" ๐Ÿ“ Deleted group: {deleted_group.name}") + + except Exception as e: + print(f"โŒ Error deleting individual group: {str(e)}") + + # Example 9: Bulk deletion + if bulk_group_ids: + print(f"\n๐Ÿ—‘๏ธ Performing bulk deletion of {len(bulk_group_ids)} groups...") + try: + # Create backup group first + backup_group_data = CreateContactsGroupDto( + name="Backup Group for Bulk Delete Demo", description="Temporary group to receive transferred contacts" + ) + backup_group = client.contact_groups.create(backup_group_data) + + # Perform bulk deletion + bulk_delete_data = DeleteContactsGroupsDto(group_ids=bulk_group_ids, transfer_contacts_to=backup_group.id) + + bulk_delete_result = client.contact_groups.delete_bulk(bulk_delete_data, approve="yes") + print("โœ… Bulk deletion completed successfully!") + print(f" ๐Ÿ“Š Operation result: {bulk_delete_result.name}") + + # Clean up backup group + client.contact_groups.delete_by_id(backup_group.id, approve="yes") + print(" ๐Ÿงน Cleaned up backup group") + + except Exception as e: + print(f"โŒ Error with bulk deletion: {str(e)}") + + # Example 10: Error handling demonstration + print("\nโš ๏ธ Error handling demonstration...") + try: + # Try to get a non-existent group + client.contact_groups.get_by_id("non_existent_group_id") + + except Exception as e: + print(f"โœ… Properly handled expected error: {type(e).__name__}") + print(f" Error message: {str(e)}") + + print("\n" + "=" * 75) + print("๐ŸŽฏ Contact Groups management demo completed!") + print("\nKey Features Demonstrated:") + print("โ€ข โœ… List groups with pagination and search") + print("โ€ข โœ… Create new contact groups with metadata") + print("โ€ข โœ… Update existing groups") + print("โ€ข โœ… Retrieve specific groups by ID") + print("โ€ข โœ… Search groups with field filtering") + print("โ€ข โœ… Individual group deletion") + print("โ€ข โœ… Bulk group deletion with contact transfer") + print("โ€ข โœ… Comprehensive error handling") + + +def contact_group_management_workflow(): + """ + Example of a complete contact group management workflow. + + Demonstrates a realistic scenario of managing contact groups + for different business purposes. + """ + + print("\n" + "=" * 60) + print("๐Ÿ“Š Contact Group Management Workflow Example") + print("=" * 60) + + api_key = os.getenv("DEVO_API_KEY") + if not api_key: + print("โŒ Error: DEVO_API_KEY environment variable not set") + return + + client = DevoClient(api_key=api_key) + + # Define different types of contact groups for a business + group_types = [ + { + "name": "VIP Customers", + "description": "High-value customers requiring priority support", + "metadata": {"priority": "high", "support_tier": "premium"}, + }, + { + "name": "Newsletter Subscribers", + "description": "Contacts subscribed to weekly newsletter", + "metadata": {"communication_type": "newsletter", "frequency": "weekly"}, + }, + { + "name": "Product Beta Testers", + "description": "Users participating in beta testing programs", + "metadata": {"program": "beta", "access_level": "testing"}, + }, + { + "name": "Sales Prospects", + "description": "Potential customers in the sales pipeline", + "metadata": {"stage": "prospect", "department": "sales"}, + }, + ] + + created_groups = [] + + # Create business contact groups + print("\n๐Ÿข Creating business contact groups...") + for group_type in group_types: + try: + group_data = CreateContactsGroupDto( + name=group_type["name"], description=group_type["description"], metadata=group_type["metadata"] + ) + + group = client.contact_groups.create(group_data) + created_groups.append(group) + print(f" โœ… Created: {group.name}") + + except Exception as e: + print(f" โŒ Error creating {group_type['name']}: {str(e)}") + + # Demonstrate group analytics + print("\n๐Ÿ“Š Group Analytics:") + print(f" Total groups created: {len(created_groups)}") + + for group in created_groups: + print(f" ๐Ÿ“ {group.name}") + print(f" ๐Ÿ“ˆ Current contacts: {group.contacts_count or 0}") + print(f" ๐Ÿท๏ธ Category: {group.metadata.get('priority', 'standard')}") + + # Simulate group reorganization + print("\n๐Ÿ”„ Reorganizing groups...") + + # Update VIP group to include more metadata + vip_group = next((g for g in created_groups if "VIP" in g.name), None) + if vip_group: + try: + update_data = UpdateContactsGroupDto( + description="Premium customers with dedicated account management", + metadata={ + **vip_group.metadata, + "account_manager": "assigned", + "response_time": "< 1 hour", + "last_updated": datetime.now().isoformat(), + }, + ) + + client.contact_groups.update(vip_group.id, update_data) + print(" โœ… Updated VIP group with enhanced metadata") + + except Exception as e: + print(f" โŒ Error updating VIP group: {str(e)}") + + # Clean up demonstration groups + print("\n๐Ÿงน Cleaning up demonstration groups...") + if created_groups: + try: + group_ids = [group.id for group in created_groups] + + # Create a temporary group for contact transfer + temp_group_data = CreateContactsGroupDto( + name="Temporary Archive", description="Temporary group for workflow cleanup" + ) + temp_group = client.contact_groups.create(temp_group_data) + + # Bulk delete with contact transfer + delete_data = DeleteContactsGroupsDto(group_ids=group_ids, transfer_contacts_to=temp_group.id) + + client.contact_groups.delete_bulk(delete_data, approve="yes") + print(f" โœ… Bulk deleted {len(group_ids)} demonstration groups") + + # Delete temporary group + client.contact_groups.delete_by_id(temp_group.id, approve="yes") + print(" โœ… Cleaned up temporary archive group") + + except Exception as e: + print(f" โŒ Error during cleanup: {str(e)}") + + print("\n๐ŸŽฏ Workflow demonstration completed!") + + +if __name__ == "__main__": + main() + contact_group_management_workflow() diff --git a/src/devo_global_comms_python/client.py b/src/devo_global_comms_python/client.py index 1511032..71b04ed 100644 --- a/src/devo_global_comms_python/client.py +++ b/src/devo_global_comms_python/client.py @@ -6,6 +6,7 @@ from .auth import APIKeyAuth from .exceptions import DevoAPIException, DevoAuthenticationException, DevoException, DevoMissingAPIKeyException +from .resources.contact_groups import ContactGroupsResource from .resources.contacts import ContactsResource from .resources.email import EmailResource from .resources.messages import MessagesResource @@ -87,6 +88,7 @@ def __init__( self.whatsapp = WhatsAppResource(self) self.rcs = RCSResource(self) self.contacts = ContactsResource(self) + self.contact_groups = ContactGroupsResource(self) self.messages = MessagesResource(self) def _create_session(self, max_retries: int) -> requests.Session: diff --git a/src/devo_global_comms_python/models/__init__.py b/src/devo_global_comms_python/models/__init__.py index 94a4502..d9f58d0 100644 --- a/src/devo_global_comms_python/models/__init__.py +++ b/src/devo_global_comms_python/models/__init__.py @@ -1,3 +1,10 @@ +from .contact_groups import ( + ContactsGroup, + ContactsGroupListResponse, + CreateContactsGroupDto, + DeleteContactsGroupsDto, + UpdateContactsGroupDto, +) from .contacts import Contact from .email import EmailMessage from .messages import Message, SendMessageDto, SendMessageSerializer @@ -29,6 +36,12 @@ # Omni-channel messaging models "SendMessageDto", "SendMessageSerializer", + # Contact groups models + "ContactsGroup", + "ContactsGroupListResponse", + "CreateContactsGroupDto", + "DeleteContactsGroupsDto", + "UpdateContactsGroupDto", # New SMS API models "SMSQuickSendRequest", "SMSQuickSendResponse", diff --git a/src/devo_global_comms_python/models/contact_groups.py b/src/devo_global_comms_python/models/contact_groups.py new file mode 100644 index 0000000..4cc9290 --- /dev/null +++ b/src/devo_global_comms_python/models/contact_groups.py @@ -0,0 +1,62 @@ +from datetime import datetime +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, Field + + +class ContactsGroup(BaseModel): + """ + Contact group model representing a collection of contacts. + """ + + id: str = Field(..., description="Unique identifier for the contact group") + name: str = Field(..., description="Name of the contact group") + description: Optional[str] = Field(None, description="Description of the contact group") + contacts_count: Optional[int] = Field(None, description="Number of contacts in the group") + created_at: Optional[datetime] = Field(None, description="Creation timestamp") + updated_at: Optional[datetime] = Field(None, description="Last update timestamp") + user_id: Optional[str] = Field(None, description="ID of the user who owns the group") + metadata: Optional[Dict[str, Any]] = Field(None, description="Additional metadata") + + +class CreateContactsGroupDto(BaseModel): + """ + DTO for creating a new contact group. + """ + + name: str = Field(..., description="Name of the contact group", min_length=1, max_length=255) + description: Optional[str] = Field(None, description="Description of the contact group", max_length=1000) + contact_ids: Optional[List[str]] = Field(None, description="List of contact IDs to add to the group") + metadata: Optional[Dict[str, Any]] = Field(None, description="Additional metadata") + + +class UpdateContactsGroupDto(BaseModel): + """ + DTO for updating an existing contact group. + """ + + name: Optional[str] = Field(None, description="Updated name of the contact group", min_length=1, max_length=255) + description: Optional[str] = Field(None, description="Updated description of the contact group", max_length=1000) + contact_ids: Optional[List[str]] = Field(None, description="Updated list of contact IDs") + metadata: Optional[Dict[str, Any]] = Field(None, description="Updated metadata") + + +class DeleteContactsGroupsDto(BaseModel): + """ + DTO for bulk deleting contact groups. + """ + + group_ids: List[str] = Field(..., description="List of contact group IDs to delete", min_items=1) + transfer_contacts_to: Optional[str] = Field(None, description="Group ID to transfer contacts to before deletion") + + +class ContactsGroupListResponse(BaseModel): + """ + Response model for listing contact groups with pagination. + """ + + groups: List[ContactsGroup] = Field(..., description="List of contact groups") + total: int = Field(..., description="Total number of groups") + page: int = Field(..., description="Current page number") + limit: int = Field(..., description="Number of items per page") + total_pages: int = Field(..., description="Total number of pages") diff --git a/src/devo_global_comms_python/resources/__init__.py b/src/devo_global_comms_python/resources/__init__.py index 3630ff1..36c9b12 100644 --- a/src/devo_global_comms_python/resources/__init__.py +++ b/src/devo_global_comms_python/resources/__init__.py @@ -1,3 +1,4 @@ +from .contact_groups import ContactGroupsResource from .contacts import ContactsResource from .email import EmailResource from .messages import MessagesResource @@ -11,5 +12,6 @@ "WhatsAppResource", "RCSResource", "ContactsResource", + "ContactGroupsResource", "MessagesResource", ] diff --git a/src/devo_global_comms_python/resources/contact_groups.py b/src/devo_global_comms_python/resources/contact_groups.py new file mode 100644 index 0000000..c66d01c --- /dev/null +++ b/src/devo_global_comms_python/resources/contact_groups.py @@ -0,0 +1,224 @@ +from typing import TYPE_CHECKING, List, Optional + +from ..utils import validate_required_string, validate_response +from .base import BaseResource + +if TYPE_CHECKING: + from ..models.contact_groups import ( + ContactsGroup, + ContactsGroupListResponse, + CreateContactsGroupDto, + DeleteContactsGroupsDto, + UpdateContactsGroupDto, + ) + + +class ContactGroupsResource(BaseResource): + """ + Contact Groups Resource for managing contact group operations. + + This resource provides methods to create, read, update and delete contact groups, + as well as bulk operations for managing multiple groups. + """ + + def list( + self, + page: Optional[int] = None, + limit: Optional[int] = None, + search: Optional[str] = None, + search_fields: Optional[List[str]] = None, + ) -> "ContactsGroupListResponse": + """ + Get contact groups by user ID with pagination and search. + + Args: + page: Page number for pagination + limit: Number of items per page + search: Search term to filter groups + search_fields: Fields to search in + + Returns: + ContactsGroupListResponse with groups and pagination info + + Example: + groups = client.contact_groups.list(page=1, limit=10, search="marketing") + """ + params = {} + if page is not None: + params["page"] = page + if limit is not None: + params["limit"] = limit + if search: + params["search"] = search + if search_fields: + params["search_fields"] = search_fields + + response = self.client.get("contacts-groups", params=params) + + from ..models.contact_groups import ContactsGroupListResponse + + return validate_response(response, ContactsGroupListResponse) + + def create(self, data: "CreateContactsGroupDto") -> "ContactsGroup": + """ + Create a new contact group. + + Args: + data: CreateContactsGroupDto with group details + + Returns: + ContactsGroup: The created contact group + + Example: + from ..models.contact_groups import CreateContactsGroupDto + + group_data = CreateContactsGroupDto( + name="Marketing List", + description="Contacts for marketing campaigns", + contact_ids=["contact_1", "contact_2"] + ) + group = client.contact_groups.create(group_data) + """ + response = self.client.post("contacts-groups", data=data.model_dump(exclude_none=True)) + + from ..models.contact_groups import ContactsGroup + + return validate_response(response, ContactsGroup) + + def update(self, group_id: str, data: "UpdateContactsGroupDto") -> "ContactsGroup": + """ + Update an existing contact group. + + Args: + group_id: ID of the contact group to update + data: UpdateContactsGroupDto with updated details + + Returns: + ContactsGroup: The updated contact group + + Example: + from ..models.contact_groups import UpdateContactsGroupDto + + update_data = UpdateContactsGroupDto( + name="Updated Marketing List", + description="Updated description" + ) + group = client.contact_groups.update("group_123", update_data) + """ + group_id = validate_required_string(group_id, "group_id") + response = self.client.put(f"contacts-groups/{group_id}", data=data.model_dump(exclude_none=True)) + + from ..models.contact_groups import ContactsGroup + + return validate_response(response, ContactsGroup) + + def delete_by_id(self, group_id: str, approve: Optional[str] = None) -> "ContactsGroup": + """ + Delete a single contact group by ID. + + Args: + group_id: ID of the contact group to delete + approve: Approval confirmation for deletion + + Returns: + ContactsGroup: The deleted contact group details + + Example: + deleted_group = client.contact_groups.delete_by_id("group_123", approve="yes") + """ + group_id = validate_required_string(group_id, "group_id") + params = {} + if approve: + params["approve"] = approve + + response = self.client.delete(f"contacts-groups/{group_id}", params=params) + + from ..models.contact_groups import ContactsGroup + + return validate_response(response, ContactsGroup) + + def delete_bulk(self, data: "DeleteContactsGroupsDto", approve: Optional[str] = None) -> "ContactsGroup": + """ + Delete multiple contact groups in bulk. + + Args: + data: DeleteContactsGroupsDto with group IDs to delete + approve: Approval confirmation for deletion + + Returns: + ContactsGroup: Details of the bulk deletion operation + + Example: + from ..models.contact_groups import DeleteContactsGroupsDto + + delete_data = DeleteContactsGroupsDto( + group_ids=["group_1", "group_2", "group_3"], + transfer_contacts_to="group_backup" + ) + result = client.contact_groups.delete_bulk(delete_data, approve="yes") + """ + params = {} + if approve: + params["approve"] = approve + + response = self.client.delete( + "contacts-groups", + params=params, + data=data.model_dump(exclude_none=True), + ) + + from ..models.contact_groups import ContactsGroup + + return validate_response(response, ContactsGroup) + + def get_by_id(self, group_id: str) -> "ContactsGroup": + """ + Get a specific contact group by ID. + + Note: This method assumes the existence of a GET endpoint for individual groups, + which is common but not explicitly defined in the provided API spec. + + Args: + group_id: ID of the contact group to retrieve + + Returns: + ContactsGroup: The contact group details + + Example: + group = client.contact_groups.get_by_id("group_123") + """ + group_id = validate_required_string(group_id, "group_id") + response = self.client.get(f"contacts-groups/{group_id}") + + from ..models.contact_groups import ContactsGroup + + return validate_response(response, ContactsGroup) + + def search( + self, + query: str, + fields: Optional[List[str]] = None, + page: Optional[int] = None, + limit: Optional[int] = None, + ) -> "ContactsGroupListResponse": + """ + Search contact groups with specific criteria. + + Args: + query: Search query string + fields: Fields to search in (name, description, etc.) + page: Page number for pagination + limit: Number of items per page + + Returns: + ContactsGroupListResponse with matching groups + + Example: + results = client.contact_groups.search( + query="marketing", + fields=["name", "description"], + page=1, + limit=20 + ) + """ + return self.list(page=page, limit=limit, search=query, search_fields=fields) diff --git a/tests/test_contact_groups.py b/tests/test_contact_groups.py new file mode 100644 index 0000000..fc0898e --- /dev/null +++ b/tests/test_contact_groups.py @@ -0,0 +1,498 @@ +from datetime import datetime +from unittest.mock import Mock + +import pytest + +from src.devo_global_comms_python.models.contact_groups import ( + ContactsGroup, + ContactsGroupListResponse, + CreateContactsGroupDto, + DeleteContactsGroupsDto, + UpdateContactsGroupDto, +) +from src.devo_global_comms_python.resources.contact_groups import ContactGroupsResource + + +class TestContactGroupsResource: + """Test cases for the ContactGroupsResource class.""" + + def setup_method(self): + """Set up test fixtures.""" + self.mock_client = Mock() + self.contact_groups_resource = ContactGroupsResource(self.mock_client) + + def test_list_contact_groups(self): + """Test listing contact groups with pagination.""" + # Arrange + mock_response = Mock() + mock_response.json.return_value = { + "groups": [ + { + "id": "group_1", + "name": "Marketing List", + "description": "Marketing contacts", + "contacts_count": 150, + "created_at": "2024-01-01T12:00:00Z", + "user_id": "user_123", + }, + { + "id": "group_2", + "name": "Sales Team", + "description": "Sales contacts", + "contacts_count": 75, + "created_at": "2024-01-02T12:00:00Z", + "user_id": "user_123", + }, + ], + "total": 25, + "page": 1, + "limit": 10, + "total_pages": 3, + } + self.mock_client.get.return_value = mock_response + + # Act + result = self.contact_groups_resource.list(page=1, limit=10, search="marketing") + + # Assert + assert isinstance(result, ContactsGroupListResponse) + assert len(result.groups) == 2 + assert result.total == 25 + assert result.page == 1 + assert result.limit == 10 + assert result.total_pages == 3 + assert result.groups[0].name == "Marketing List" + + self.mock_client.get.assert_called_once_with( + "contacts-groups", params={"page": 1, "limit": 10, "search": "marketing"} + ) + + def test_list_contact_groups_with_search_fields(self): + """Test listing contact groups with search fields.""" + # Arrange + mock_response = Mock() + mock_response.json.return_value = { + "groups": [], + "total": 0, + "page": 1, + "limit": 10, + "total_pages": 0, + } + self.mock_client.get.return_value = mock_response + + # Act + self.contact_groups_resource.list(search="test", search_fields=["name", "description"]) + + # Assert + self.mock_client.get.assert_called_once_with( + "contacts-groups", + params={"search": "test", "search_fields": ["name", "description"]}, + ) + + def test_create_contact_group(self): + """Test creating a new contact group.""" + # Arrange + create_data = CreateContactsGroupDto( + name="New Marketing List", + description="A new marketing contact group", + contact_ids=["contact_1", "contact_2", "contact_3"], + metadata={"campaign": "summer_2024"}, + ) + + mock_response = Mock() + mock_response.json.return_value = { + "id": "group_new", + "name": "New Marketing List", + "description": "A new marketing contact group", + "contacts_count": 3, + "created_at": "2024-01-15T12:00:00Z", + "user_id": "user_123", + "metadata": {"campaign": "summer_2024"}, + } + self.mock_client.post.return_value = mock_response + + # Act + result = self.contact_groups_resource.create(create_data) + + # Assert + assert isinstance(result, ContactsGroup) + assert result.id == "group_new" + assert result.name == "New Marketing List" + assert result.description == "A new marketing contact group" + assert result.contacts_count == 3 + assert result.metadata == {"campaign": "summer_2024"} + + self.mock_client.post.assert_called_once_with( + "contacts-groups", + data={ + "name": "New Marketing List", + "description": "A new marketing contact group", + "contact_ids": ["contact_1", "contact_2", "contact_3"], + "metadata": {"campaign": "summer_2024"}, + }, + ) + + def test_update_contact_group(self): + """Test updating an existing contact group.""" + # Arrange + update_data = UpdateContactsGroupDto( + name="Updated Marketing List", + description="Updated description", + metadata={"updated": True}, + ) + + mock_response = Mock() + mock_response.json.return_value = { + "id": "group_123", + "name": "Updated Marketing List", + "description": "Updated description", + "contacts_count": 150, + "updated_at": "2024-01-15T14:00:00Z", + "user_id": "user_123", + "metadata": {"updated": True}, + } + self.mock_client.put.return_value = mock_response + + # Act + result = self.contact_groups_resource.update("group_123", update_data) + + # Assert + assert isinstance(result, ContactsGroup) + assert result.id == "group_123" + assert result.name == "Updated Marketing List" + assert result.description == "Updated description" + assert result.metadata == {"updated": True} + + self.mock_client.put.assert_called_once_with( + "contacts-groups/group_123", + data={ + "name": "Updated Marketing List", + "description": "Updated description", + "metadata": {"updated": True}, + }, + ) + + def test_delete_contact_group_by_id(self): + """Test deleting a contact group by ID.""" + # Arrange + mock_response = Mock() + mock_response.json.return_value = { + "id": "group_123", + "name": "Deleted Group", + "description": "This group was deleted", + "contacts_count": 0, + "user_id": "user_123", + } + self.mock_client.delete.return_value = mock_response + + # Act + result = self.contact_groups_resource.delete_by_id("group_123", approve="yes") + + # Assert + assert isinstance(result, ContactsGroup) + assert result.id == "group_123" + assert result.name == "Deleted Group" + + self.mock_client.delete.assert_called_once_with("contacts-groups/group_123", params={"approve": "yes"}) + + def test_delete_contact_group_by_id_without_approval(self): + """Test deleting a contact group by ID without approval.""" + # Arrange + mock_response = Mock() + mock_response.json.return_value = { + "id": "group_123", + "name": "Deleted Group", + "description": "This group was deleted", + "contacts_count": 0, + "user_id": "user_123", + } + self.mock_client.delete.return_value = mock_response + + # Act + result = self.contact_groups_resource.delete_by_id("group_123") + + # Assert + assert isinstance(result, ContactsGroup) + self.mock_client.delete.assert_called_once_with("contacts-groups/group_123", params={}) + + def test_delete_contact_groups_bulk(self): + """Test bulk deleting contact groups.""" + # Arrange + delete_data = DeleteContactsGroupsDto( + group_ids=["group_1", "group_2", "group_3"], + transfer_contacts_to="group_backup", + ) + + mock_response = Mock() + mock_response.json.return_value = { + "id": "bulk_delete_operation", + "name": "Bulk Delete Operation", + "description": "Deleted 3 groups", + "contacts_count": 0, + "user_id": "user_123", + } + self.mock_client.delete.return_value = mock_response + + # Act + result = self.contact_groups_resource.delete_bulk(delete_data, approve="yes") + + # Assert + assert isinstance(result, ContactsGroup) + assert result.name == "Bulk Delete Operation" + + self.mock_client.delete.assert_called_once_with( + "contacts-groups", + params={"approve": "yes"}, + data={ + "group_ids": ["group_1", "group_2", "group_3"], + "transfer_contacts_to": "group_backup", + }, + ) + + def test_get_contact_group_by_id(self): + """Test getting a specific contact group by ID.""" + # Arrange + mock_response = Mock() + mock_response.json.return_value = { + "id": "group_123", + "name": "Marketing List", + "description": "Marketing contacts", + "contacts_count": 150, + "created_at": "2024-01-01T12:00:00Z", + "user_id": "user_123", + } + self.mock_client.get.return_value = mock_response + + # Act + result = self.contact_groups_resource.get_by_id("group_123") + + # Assert + assert isinstance(result, ContactsGroup) + assert result.id == "group_123" + assert result.name == "Marketing List" + assert result.contacts_count == 150 + + self.mock_client.get.assert_called_once_with("contacts-groups/group_123") + + def test_search_contact_groups(self): + """Test searching contact groups.""" + # Arrange + mock_response = Mock() + mock_response.json.return_value = { + "groups": [ + { + "id": "group_1", + "name": "Marketing Team", + "description": "Marketing department contacts", + "contacts_count": 25, + "created_at": "2024-01-01T12:00:00Z", + "user_id": "user_123", + } + ], + "total": 1, + "page": 1, + "limit": 20, + "total_pages": 1, + } + self.mock_client.get.return_value = mock_response + + # Act + result = self.contact_groups_resource.search( + query="marketing", fields=["name", "description"], page=1, limit=20 + ) + + # Assert + assert isinstance(result, ContactsGroupListResponse) + assert len(result.groups) == 1 + assert result.groups[0].name == "Marketing Team" + + self.mock_client.get.assert_called_once_with( + "contacts-groups", + params={ + "page": 1, + "limit": 20, + "search": "marketing", + "search_fields": ["name", "description"], + }, + ) + + def test_validation_error_empty_group_id(self): + """Test validation error for empty group ID.""" + with pytest.raises(Exception): # DevoValidationException would be raised + self.contact_groups_resource.delete_by_id("") + + def test_api_error_handling(self): + """Test API error handling.""" + # Arrange + self.mock_client.get.side_effect = Exception("API Error") + + # Act & Assert + with pytest.raises(Exception, match="API Error"): + self.contact_groups_resource.list() + + +class TestCreateContactsGroupDto: + """Test cases for CreateContactsGroupDto model.""" + + def test_valid_create_dto(self): + """Test creating valid CreateContactsGroupDto.""" + dto = CreateContactsGroupDto( + name="Test Group", + description="A test group", + contact_ids=["contact_1", "contact_2"], + metadata={"test": True}, + ) + + assert dto.name == "Test Group" + assert dto.description == "A test group" + assert dto.contact_ids == ["contact_1", "contact_2"] + assert dto.metadata == {"test": True} + + def test_minimum_valid_create_dto(self): + """Test creating CreateContactsGroupDto with minimum required fields.""" + dto = CreateContactsGroupDto(name="Minimum Group") + + assert dto.name == "Minimum Group" + assert dto.description is None + assert dto.contact_ids is None + assert dto.metadata is None + + def test_invalid_empty_name(self): + """Test validation error for empty name.""" + with pytest.raises(ValueError): + CreateContactsGroupDto(name="") + + def test_invalid_long_name(self): + """Test validation error for name that's too long.""" + with pytest.raises(ValueError): + CreateContactsGroupDto(name="x" * 256) # Over 255 characters + + def test_invalid_long_description(self): + """Test validation error for description that's too long.""" + with pytest.raises(ValueError): + CreateContactsGroupDto(name="Test Group", description="x" * 1001) # Over 1000 characters + + +class TestUpdateContactsGroupDto: + """Test cases for UpdateContactsGroupDto model.""" + + def test_valid_update_dto(self): + """Test creating valid UpdateContactsGroupDto.""" + dto = UpdateContactsGroupDto( + name="Updated Group", + description="Updated description", + contact_ids=["contact_1"], + metadata={"updated": True}, + ) + + assert dto.name == "Updated Group" + assert dto.description == "Updated description" + assert dto.contact_ids == ["contact_1"] + assert dto.metadata == {"updated": True} + + def test_partial_update_dto(self): + """Test creating UpdateContactsGroupDto with partial fields.""" + dto = UpdateContactsGroupDto(name="Only Name Updated") + + assert dto.name == "Only Name Updated" + assert dto.description is None + assert dto.contact_ids is None + assert dto.metadata is None + + def test_empty_update_dto(self): + """Test creating empty UpdateContactsGroupDto.""" + dto = UpdateContactsGroupDto() + + assert dto.name is None + assert dto.description is None + assert dto.contact_ids is None + assert dto.metadata is None + + +class TestDeleteContactsGroupsDto: + """Test cases for DeleteContactsGroupsDto model.""" + + def test_valid_delete_dto(self): + """Test creating valid DeleteContactsGroupsDto.""" + dto = DeleteContactsGroupsDto( + group_ids=["group_1", "group_2"], + transfer_contacts_to="backup_group", + ) + + assert dto.group_ids == ["group_1", "group_2"] + assert dto.transfer_contacts_to == "backup_group" + + def test_delete_dto_without_transfer(self): + """Test creating DeleteContactsGroupsDto without transfer group.""" + dto = DeleteContactsGroupsDto(group_ids=["group_1", "group_2"]) + + assert dto.group_ids == ["group_1", "group_2"] + assert dto.transfer_contacts_to is None + + def test_invalid_empty_group_ids(self): + """Test validation error for empty group_ids list.""" + with pytest.raises(ValueError): + DeleteContactsGroupsDto(group_ids=[]) + + +class TestContactsGroup: + """Test cases for ContactsGroup model.""" + + def test_valid_contacts_group(self): + """Test creating valid ContactsGroup.""" + group = ContactsGroup( + id="group_123", + name="Test Group", + description="A test group", + contacts_count=50, + created_at=datetime(2024, 1, 1, 12, 0, 0), + user_id="user_123", + metadata={"test": True}, + ) + + assert group.id == "group_123" + assert group.name == "Test Group" + assert group.description == "A test group" + assert group.contacts_count == 50 + assert group.user_id == "user_123" + assert group.metadata == {"test": True} + + def test_minimum_valid_contacts_group(self): + """Test creating ContactsGroup with minimum required fields.""" + group = ContactsGroup(id="group_123", name="Minimum Group") + + assert group.id == "group_123" + assert group.name == "Minimum Group" + assert group.description is None + assert group.contacts_count is None + assert group.created_at is None + assert group.user_id is None + assert group.metadata is None + + +class TestContactsGroupListResponse: + """Test cases for ContactsGroupListResponse model.""" + + def test_valid_list_response(self): + """Test creating valid ContactsGroupListResponse.""" + groups = [ + ContactsGroup(id="group_1", name="Group 1"), + ContactsGroup(id="group_2", name="Group 2"), + ] + + response = ContactsGroupListResponse(groups=groups, total=25, page=1, limit=10, total_pages=3) + + assert len(response.groups) == 2 + assert response.total == 25 + assert response.page == 1 + assert response.limit == 10 + assert response.total_pages == 3 + + def test_empty_list_response(self): + """Test creating empty ContactsGroupListResponse.""" + response = ContactsGroupListResponse(groups=[], total=0, page=1, limit=10, total_pages=0) + + assert len(response.groups) == 0 + assert response.total == 0 + assert response.page == 1 + assert response.total_pages == 0