# Example External API Integration Implementation # This file demonstrates how to use the organization sync system from organization.models import SyncConfiguration, SyncLog, Organization from organization.services.external_api_service import ExternalAPIService from organization.services.sync_service import OrganizationSyncService import logging logger = logging.getLogger(__name__) # ============================================================================ # EXAMPLE 1: Create Sync Configuration Programmatically # ============================================================================ def create_sync_config_example(): """ Create a sync configuration for a sample external API. """ config, created = SyncConfiguration.objects.get_or_create( source_name="example_university_api", defaults={ "base_url": "https://api.example-university.edu", "api_endpoint": "/v1/organizations/", "auth_type": "api_key", "auth_key": "sk_live_abc123def456ghi789", "timeout_seconds": 30, "max_retries": 3, "retry_delay_seconds": 5, "batch_size": 50, "cache_enabled": True, "cache_ttl_seconds": 3600, "auto_sync_enabled": False, "field_mapping": { "name": "organization_name", "code": "organization_code", "email": "primary_email", "phone": "phone_number", "website": "website_url", "city": "location_city", "country": "location_country", } } ) if created: logger.info(f"Created new sync config: {config.source_name}") else: logger.info(f"Using existing sync config: {config.source_name}") return config # ============================================================================ # EXAMPLE 2: Test API Connection # ============================================================================ def test_api_connection_example(): """ Test connection to external API. """ try: config = SyncConfiguration.objects.get(source_name="example_university_api") success, message = config.test_connection() if success: print(f"✓ Connection successful: {message}") return True else: print(f"✗ Connection failed: {message}") return False except SyncConfiguration.DoesNotExist: print("Sync configuration not found") return False # ============================================================================ # EXAMPLE 3: Fetch Organizations with ExternalAPIService # ============================================================================ def fetch_organizations_example(): """ Directly fetch organizations from external API. """ config = SyncConfiguration.objects.get(source_name="example_university_api") api_service = ExternalAPIService(config) try: # Fetch first page response = api_service.fetch_organizations(limit=10, offset=0) print(f"Fetched response: {response}") # Or fetch all with pagination print("\nFetching all organizations with pagination:") for org_data in api_service.fetch_paginated(batch_size=50): print(f" - {org_data.get('organization_name', 'Unknown')}") except Exception as e: logger.error(f"Error fetching organizations: {str(e)}") # ============================================================================ # EXAMPLE 4: Manually Trigger Sync # ============================================================================ def sync_organizations_example(): """ Manually trigger organization sync. """ # Get config config = SyncConfiguration.objects.get(source_name="example_university_api") # Create sync log entry sync_log = SyncLog.objects.create( external_source=config.source_name, triggered_by="example_script", sync_type="full" ) # Perform sync sync_service = OrganizationSyncService(config) try: sync_service.sync_organizations(sync_log, max_records=None) # Display results print(f"\n✓ Sync completed successfully!") print(f" Total fetched: {sync_log.total_fetched}") print(f" Created: {sync_log.created_count}") print(f" Updated: {sync_log.updated_count}") print(f" Errors: {sync_log.error_count}") print(f" Duration: {sync_log.duration_seconds:.2f}s") except Exception as e: print(f"\n✗ Sync failed: {str(e)}") print(f" Error: {sync_log.error_message}") # ============================================================================ # EXAMPLE 5: Sync with Limited Records # ============================================================================ def sync_with_limit_example(max_records: int = 10): """ Sync a limited number of organizations (useful for testing). """ config = SyncConfiguration.objects.get(source_name="example_university_api") sync_log = SyncLog.objects.create( external_source=config.source_name, triggered_by="test_script", sync_type="partial" ) sync_service = OrganizationSyncService(config) sync_service.sync_organizations(sync_log, max_records=max_records) return sync_log # ============================================================================ # EXAMPLE 6: Query Synced Organizations # ============================================================================ def query_organizations_example(): """ Query organizations in database. """ # Get all organizations all_orgs = Organization.objects.all() print(f"Total organizations: {all_orgs.count()}") # Get specific source example_orgs = Organization.objects.filter(external_source="example_university_api") print(f"\nOrganizations from 'example_university_api': {example_orgs.count()}") for org in example_orgs[:5]: print(f" - {org.name} ({org.code})") # Get recently synced recently_synced = Organization.objects.filter( last_synced__isnull=False ).order_by("-last_synced")[:5] print(f"\nRecently synced organizations:") for org in recently_synced: print(f" - {org.name} (last synced: {org.last_synced})") # ============================================================================ # EXAMPLE 7: View Sync History # ============================================================================ def view_sync_history_example(limit: int = 10): """ View sync operation history. """ sync_logs = SyncLog.objects.all().order_by("-started_at")[:limit] print(f"\n=== Recent Sync Operations ({limit} items) ===\n") for log in sync_logs: status_symbol = "✓" if log.is_successful else "✗" print(f"{status_symbol} Sync {log.sync_id}") print(f" Source: {log.external_source}") print(f" Status: {log.status}") print(f" Started: {log.started_at}") print(f" Created: {log.created_count}, Updated: {log.updated_count}, Errors: {log.error_count}") print(f" Duration: {log.duration_seconds:.2f}s" if log.duration_seconds else " Duration: N/A") if log.error_message: print(f" Error: {log.error_message[:100]}") print() # ============================================================================ # EXAMPLE 8: Manual Data Mapping (Custom Use Case) # ============================================================================ def custom_data_mapping_example(): """ Example of accessing data mapping functionality directly. """ config = SyncConfiguration.objects.get(source_name="example_university_api") sync_service = OrganizationSyncService(config) # Example external data from API external_record = { "id": "ext_org_001", "organization_name": "Example University", "organization_code": "EU", "primary_email": "contact@example.edu", "phone_number": "555-0000", "website_url": "https://www.example.edu", "location_city": "Boston", "location_country": "USA" } # Map to internal format try: mapped_data = sync_service._map_external_data(external_record) print("Mapped data:") for key, value in mapped_data.items(): print(f" {key}: {value}") except Exception as e: print(f"Mapping failed: {str(e)}") # ============================================================================ # Main Script - Run All Examples # ============================================================================ if __name__ == "__main__": import django django.setup() # Initialize Django print("=" * 80) print("EXTERNAL API INTEGRATION EXAMPLES") print("=" * 80) # Create configuration print("\n1. Creating sync configuration...") config = create_sync_config_example() # Test connection print("\n2. Testing API connection...") test_api_connection_example() # Query organizations print("\n3. Querying organizations...") query_organizations_example() # View sync history print("\n4. Viewing sync history...") view_sync_history_example() print("\n" + "=" * 80) print("EXAMPLES COMPLETED") print("=" * 80)