amo-server/docs/amocrm-service-design.md

AMO CRM Data Collection Service - Design Document

Overview

This service collects data from AMO CRM API (wecheap.amocrm.ru) and stores it in SQLite database for further export to Google Sheets. The service handles AMO CRM's HAL+JSON format and maintains proper relationships between entities.

Key Features:

• Simple API without authentication
• Direct AMO CRM integration with long-term tokens
• Real-time data fetching and testing
• Unified export configuration for all entities
• Background job processing for exports

Architecture

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   AMO CRM API   │    │   Our Service   │    │  Google Sheets  │
│                 │    │                 │    │                 │
│ HAL+JSON Format │◄──►│ SQLite Database │───►│   Export API    │
│                 │    │                 │    │                 │
└─────────────────┘    └─────────────────┘    └─────────────────┘

Components

• Adapters: Database connections, AMO CRM API client, Google Sheets client
• Routers: API endpoints (no authentication required)
• Servers: Business logic for data processing and export
• Workers: Background data collection and processing
• Scripts: Utilities for fetching real AMO CRM data

Data Entities

1. Deals (Сделки)

• Primary entity representing sales opportunities
• Links to: Contacts, Companies, Users, Pipelines
• Contains custom fields and standard fields

2. Contacts (Контакты)

• Individual people in the CRM
• Links to: Companies, Deals, Users
• Contains communication history

3. Companies (Компании)

• Organizations in the CRM
• Links to: Contacts, Deals, Users
• Contains company-specific custom fields

4. Pipelines (Воронки)

• Sales process definitions
• Contains stages and automation rules
• Referenced by Deals

5. Users (Пользователи)

• CRM users (employees)
• Referenced by all other entities as owners/responsible persons

6. Events (События)

• Activity logs (filtered types only):
  • incoming_call
  • outgoing_call
  • lead_status_changed
• Links to related entities

Database Schema

Core Tables

-- Users table
CREATE TABLE amo_users (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    email TEXT,
    is_active BOOLEAN DEFAULT TRUE,
    created_at INTEGER,
    updated_at INTEGER,
    raw_data JSON -- Original AMO CRM data
);

-- Pipelines table
CREATE TABLE amo_pipelines (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    sort INTEGER,
    is_main BOOLEAN DEFAULT FALSE,
    is_unsorted BOOLEAN DEFAULT FALSE,
    is_archive BOOLEAN DEFAULT FALSE,
    account_id INTEGER,
    created_at INTEGER,
    updated_at INTEGER,
    raw_data JSON
);

-- Pipeline stages
CREATE TABLE amo_pipeline_stages (
    id INTEGER PRIMARY KEY,
    pipeline_id INTEGER REFERENCES amo_pipelines(id),
    name TEXT NOT NULL,
    sort INTEGER,
    is_editable BOOLEAN DEFAULT TRUE,
    color TEXT,
    created_at INTEGER,
    updated_at INTEGER,
    raw_data JSON
);

-- Companies table
CREATE TABLE amo_companies (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    responsible_user_id INTEGER REFERENCES amo_users(id),
    group_id INTEGER,
    created_by INTEGER REFERENCES amo_users(id),
    updated_by INTEGER REFERENCES amo_users(id),
    created_at INTEGER,
    updated_at INTEGER,
    closest_task_at INTEGER,
    is_deleted BOOLEAN DEFAULT FALSE,
    raw_data JSON
);

-- Contacts table
CREATE TABLE amo_contacts (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    first_name TEXT,
    last_name TEXT,
    responsible_user_id INTEGER REFERENCES amo_users(id),
    group_id INTEGER,
    created_by INTEGER REFERENCES amo_users(id),
    updated_by INTEGER REFERENCES amo_users(id),
    created_at INTEGER,
    updated_at INTEGER,
    closest_task_at INTEGER,
    is_deleted BOOLEAN DEFAULT FALSE,
    raw_data JSON
);

-- Deals table
CREATE TABLE amo_deals (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    price INTEGER DEFAULT 0,
    responsible_user_id INTEGER REFERENCES amo_users(id),
    group_id INTEGER,
    status_id INTEGER REFERENCES amo_pipeline_stages(id),
    pipeline_id INTEGER REFERENCES amo_pipelines(id),
    loss_reason_id INTEGER,
    created_by INTEGER REFERENCES amo_users(id),
    updated_by INTEGER REFERENCES amo_users(id),
    closed_at INTEGER,
    created_at INTEGER,
    updated_at INTEGER,
    closest_task_at INTEGER,
    is_deleted BOOLEAN DEFAULT FALSE,
    raw_data JSON
);

-- Events table
CREATE TABLE amo_events (
    id INTEGER PRIMARY KEY,
    type TEXT NOT NULL CHECK (type IN ('incoming_call', 'outgoing_call', 'lead_status_changed')),
    entity_id INTEGER,
    entity_type TEXT,
    created_by INTEGER REFERENCES amo_users(id),
    created_at INTEGER,
    value_after JSON,
    value_before JSON,
    account_id INTEGER,
    raw_data JSON
);

-- Relations tables
CREATE TABLE amo_deal_contacts (
    deal_id INTEGER REFERENCES amo_deals(id),
    contact_id INTEGER REFERENCES amo_contacts(id),
    is_main BOOLEAN DEFAULT FALSE,
    PRIMARY KEY (deal_id, contact_id)
);

CREATE TABLE amo_deal_companies (
    deal_id INTEGER REFERENCES amo_deals(id),
    company_id INTEGER REFERENCES amo_companies(id),
    is_main BOOLEAN DEFAULT FALSE,
    PRIMARY KEY (deal_id, company_id)
);

CREATE TABLE amo_contact_companies (
    contact_id INTEGER REFERENCES amo_contacts(id),
    company_id INTEGER REFERENCES amo_companies(id),
    is_main BOOLEAN DEFAULT FALSE,
    PRIMARY KEY (contact_id, company_id)
);

Custom Fields Handling

-- Universal custom fields table
CREATE TABLE amo_custom_fields (
    id INTEGER PRIMARY KEY,
    entity_type TEXT NOT NULL, -- 'deals', 'contacts', 'companies'
    entity_id INTEGER NOT NULL,
    field_id INTEGER NOT NULL,
    field_name TEXT NOT NULL,
    field_type TEXT NOT NULL, -- 'text', 'numeric', 'checkbox', 'select', 'multiselect', 'date', 'url', 'textarea'
    field_value TEXT,
    field_value_numeric REAL,
    field_value_date INTEGER,
    is_custom BOOLEAN DEFAULT TRUE,
    created_at INTEGER DEFAULT (strftime('%s', 'now')),
    updated_at INTEGER DEFAULT (strftime('%s', 'now'))
);

-- Indexes for performance
CREATE INDEX idx_custom_fields_entity ON amo_custom_fields(entity_type, entity_id);
CREATE INDEX idx_custom_fields_field ON amo_custom_fields(field_id);
CREATE INDEX idx_custom_fields_name ON amo_custom_fields(field_name);

Google Sheets Export Configuration

-- Single unified export configuration
CREATE TABLE export_configuration (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    sheet_id TEXT NOT NULL,
    date_range_start INTEGER,
    date_range_end INTEGER,
    is_active BOOLEAN DEFAULT TRUE,
    created_at INTEGER DEFAULT (strftime('%s', 'now')),
    updated_at INTEGER DEFAULT (strftime('%s', 'now'))
);

-- Entity-specific sheet mappings within the configuration
CREATE TABLE export_entity_mappings (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    configuration_id INTEGER REFERENCES export_configuration(id),
    entity_type TEXT NOT NULL, -- 'deals', 'contacts', 'companies', 'pipelines', 'users', 'events'
    sheet_name TEXT NOT NULL, -- Sheet tab name for this entity
    field_mapping JSON NOT NULL, -- [{"field_name": "name", "column": "A", "order": 1}]
    is_enabled BOOLEAN DEFAULT TRUE,
    created_at INTEGER DEFAULT (strftime('%s', 'now')),
    updated_at INTEGER DEFAULT (strftime('%s', 'now'))
);

-- Export jobs tracking
CREATE TABLE export_jobs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    configuration_id INTEGER REFERENCES export_configuration(id),
    status TEXT NOT NULL DEFAULT 'pending', -- 'pending', 'running', 'completed', 'failed'
    records_processed INTEGER DEFAULT 0,
    total_records INTEGER DEFAULT 0,
    error_message TEXT,
    started_at INTEGER,
    completed_at INTEGER,
    created_at INTEGER DEFAULT (strftime('%s', 'now'))
);

API Endpoints

Base URL: http://localhost:8000/api/v1 Authentication: None required

1. Entity Management

GET /entities

List all entities with statistics

{
  "entities": [
    {
      "type": "deals",
      "count": 1500,
      "last_updated": "2024-01-15T10:30:00Z",
      "date_range": {
        "earliest": "2017-01-01T00:00:00Z",
        "latest": "2024-01-15T10:30:00Z"
      }
    },
    {
      "type": "contacts",
      "count": 3200,
      "last_updated": "2024-01-15T09:45:00Z",
      "date_range": {
        "earliest": "2017-03-15T00:00:00Z",
        "latest": "2024-01-15T09:45:00Z"
      }
    }
  ]
}

GET /entities/{entity_type}/fields

List fields with statistics and examples

{
  "entity_type": "deals",
  "fields": [
    {
      "name": "name",
      "type": "text",
      "is_custom": false,
      "usage_count": 1500,
      "examples": ["Deal #1", "Important client deal", "Q4 opportunity"],
      "null_count": 0
    },
    {
      "name": "custom_priority",
      "type": "select",
      "is_custom": true,
      "field_id": 123456,
      "usage_count": 1200,
      "examples": ["High", "Medium", "Low"],
      "null_count": 300,
      "possible_values": ["High", "Medium", "Low"]
    }
  ]
}

2. AMO CRM Integration

GET /amocrm/info

Get AMO CRM connection information

{
  "domain": "wecheap.amocrm.ru",
  "has_token": true,
  "base_url": "https://wecheap.amocrm.ru/api/v4",
  "supported_entities": [
    "users", "pipelines", "companies",
    "contacts", "deals", "events"
  ]
}

GET /amocrm/fetch/{entity_type}

Fetch data directly from AMO CRM API

• Parameters: limit (default: 10), page (default: 1)
• Supported entities: users, pipelines, companies, contacts, deals, events
• Returns: Raw AMO CRM HAL+JSON response

GET /amocrm/fetch/custom_fields/{entity_type}

Fetch custom fields metadata from AMO CRM

• Supported entities: leads, contacts, companies
• Returns: Custom fields definitions with enums

GET /amocrm/fetch/all

Fetch all data from AMO CRM (limited amounts for testing)

• Returns: Complete data set from all entities

POST /amocrm/sync/{entity_type}

Fetch data from AMO CRM and store it in database

• Parameters: limit (default: 250), page (default: 1)
• Returns: Sync status and fetched data

3. Google Sheets Export

POST /export/configure

Create or update unified export configuration for all entities

{
  "name": "Q1 2024 AMO CRM Export",
  "sheet_id": "1BxiMVs0XRA5nFMdKvBdBZjgmUUqptlbs74OgvE2upms",
  "date_range_start": "2024-01-01T00:00:00Z",
  "date_range_end": "2024-03-31T23:59:59Z",
  "entity_mappings": {
    "deals": {
      "sheet_name": "Deals",
      "is_enabled": true,
      "field_mapping": [
        {"field_name": "name", "column": "A", "order": 1},
        {"field_name": "price", "column": "B", "order": 2},
        {"field_name": "created_at", "column": "C", "order": 3},
        {"field_name": "custom_priority", "column": "D", "order": 4}
      ]
    },
    "contacts": {
      "sheet_name": "Contacts",
      "is_enabled": true,
      "field_mapping": [
        {"field_name": "name", "column": "A", "order": 1},
        {"field_name": "first_name", "column": "B", "order": 2},
        {"field_name": "last_name", "column": "C", "order": 3},
        {"field_name": "created_at", "column": "D", "order": 4}
      ]
    },
    "companies": {
      "sheet_name": "Companies",
      "is_enabled": true,
      "field_mapping": [
        {"field_name": "name", "column": "A", "order": 1},
        {"field_name": "created_at", "column": "B", "order": 2}
      ]
    },
    "pipelines": {
      "sheet_name": "Pipelines",
      "is_enabled": false,
      "field_mapping": [
        {"field_name": "name", "column": "A", "order": 1},
        {"field_name": "is_main", "column": "B", "order": 2}
      ]
    },
    "users": {
      "sheet_name": "Users",
      "is_enabled": false,
      "field_mapping": [
        {"field_name": "name", "column": "A", "order": 1},
        {"field_name": "email", "column": "B", "order": 2}
      ]
    },
    "events": {
      "sheet_name": "Events",
      "is_enabled": true,
      "field_mapping": [
        {"field_name": "type", "column": "A", "order": 1},
        {"field_name": "entity_type", "column": "B", "order": 2},
        {"field_name": "created_at", "column": "C", "order": 3}
      ]
    }
  }
}

POST /export/start

Start async export job

{
  "configuration_id": 1
}

Response:

{
  "job_id": 42,
  "status": "pending",
  "message": "Export job queued successfully"
}

GET /export/status/{job_id}

Get export job status

{
  "job_id": 42,
  "status": "running",
  "progress": {
    "records_processed": 2100,
    "total_records": 4200,
    "percentage": 50,
    "entities": {
      "deals": {"processed": 1500, "total": 1500, "status": "completed"},
      "contacts": {"processed": 600, "total": 2200, "status": "running"},
      "companies": {"processed": 0, "total": 300, "status": "pending"},
      "events": {"processed": 0, "total": 200, "status": "pending"}
    }
  },
  "started_at": "2024-01-15T10:30:00Z",
  "estimated_completion": "2024-01-15T10:35:00Z"
}

GET /export/jobs

List all export jobs

{
  "jobs": [
    {
      "job_id": 42,
      "configuration_name": "Q1 2024 AMO CRM Export",
      "status": "completed",
      "records_processed": 4200,
      "total_records": 4200,
      "entities_processed": {
        "deals": 1500,
        "contacts": 2200,
        "companies": 300,
        "events": 200
      },
      "started_at": "2024-01-15T10:30:00Z",
      "completed_at": "2024-01-15T10:35:00Z"
    }
  ],
  "total": 1,
  "page": 1,
  "per_page": 20
}

4. Data Ingestion (Worker API)

POST /data/{entity_type}

Put data into database (used by workers)

{
  "data": [
    {
      "id": 123456,
      "name": "Important Deal",
      "price": 50000,
      "responsible_user_id": 789,
      "pipeline_id": 456,
      "status_id": 142,
      "created_at": 1640995200,
      "updated_at": 1704067200,
      "custom_fields": [
        {
          "field_id": 123456,
          "field_name": "Priority",
          "field_type": "select",
          "values": [{"value": "High", "enum_id": 1}]
        }
      ],
      "_embedded": {
        "contacts": [{"id": 111, "is_main": true}],
        "companies": [{"id": 222, "is_main": true}]
      }
    }
  ],
  "sync_mode": "upsert" // "insert", "upsert", "replace"
}

Data Processing Logic

1. HAL+JSON Parsing

• Extract main entity data from root level
• Process _embedded section for related entities
• Handle _links section for additional relationships
• Parse custom fields from custom_fields_values array

2. Date Inference

• Convert Unix timestamps to readable dates
• Validate dates are between 2017-2026
• Handle null/missing dates appropriately

3. Custom Fields Processing

def process_custom_field(field_data, entity_type, entity_id):
    """
    Process custom field from AMO CRM format
    """
    field_record = {
        'entity_type': entity_type,
        'entity_id': entity_id,
        'field_id': field_data['field_id'],
        'field_name': field_data['field_name'],
        'field_type': field_data['field_type'],
        'is_custom': True
    }

    # Handle different field types
    if field_data['field_type'] in ['text', 'textarea', 'url']:
        field_record['field_value'] = field_data['values'][0]['value']
    elif field_data['field_type'] == 'numeric':
        field_record['field_value_numeric'] = float(field_data['values'][0]['value'])
        field_record['field_value'] = str(field_data['values'][0]['value'])
    elif field_data['field_type'] == 'date':
        field_record['field_value_date'] = int(field_data['values'][0]['value'])
        field_record['field_value'] = str(field_data['values'][0]['value'])
    elif field_data['field_type'] in ['select', 'multiselect']:
        # Handle enum values
        values = [v['value'] for v in field_data['values']]
        field_record['field_value'] = ', '.join(values) if len(values) > 1 else values[0]

    return field_record

4. Relationship Mapping

• Extract relationships from _embedded section
• Create junction table records for many-to-many relationships
• Handle is_main flags for primary relationships

Implementation Status

✅ Completed Features

Phase 1: Core Infrastructure

• ✅ FastAPI application structure with uv
• ✅ SQLite adapter with migrations (Alembic)
• ✅ Complete entity models with relationships
• ✅ AMO CRM API client for wecheap.amocrm.ru

Phase 2: Data Ingestion

• ✅ HAL+JSON parser implementation
• ✅ Data ingestion endpoints for all entities
• ✅ Custom fields processing with type detection
• ✅ Relationship mapping (many-to-many)

Phase 3: API Development

• ✅ Entity listing endpoints with statistics
• ✅ Field statistics endpoints with examples
• ✅ Export configuration management
• ✅ Direct AMO CRM integration endpoints

Testing & Utilities

• ✅ Real AMO CRM response examples
• ✅ Comprehensive test suite
• ✅ Data fetching script for real responses
• ✅ API documentation (OpenAPI/Swagger)

🔄 In Progress / TODO

Phase 4: Google Sheets Integration

• 🔄 Google Sheets API client implementation
• 🔄 Actual export processing
• ✅ Export job management structure
• ✅ Job status tracking

Phase 5: Workers & Background Processing

• 🔄 FastStream worker implementation
• 🔄 Data synchronization workers with Redis broker
• 🔄 Error handling and retry logic
• 🔄 Monitoring and logging

Technology Stack

• Framework: FastAPI (with uvicorn)
• Package Manager: uv (modern Python package management)
• Database: SQLite with SQLAlchemy 2.0 ORM
• Migrations: Alembic
• HTTP Client: httpx for AMO CRM API
• Data Validation: Pydantic v2
• Testing: pytest with real AMO CRM fixtures
• Background Jobs: FastStream with Redis broker
• Google Sheets: google-api-python-client (planned)
• Development: Black, isort, mypy, flake8

Configuration

Environment Variables

# Database
DATABASE_URL=sqlite:///./amo_data.db

# AMO CRM API (wecheap.amocrm.ru)
AMO_CRM_DOMAIN=wecheap.amocrm.ru
AMO_CRM_ACCESS_TOKEN=your-longterm-access-token

# Google Sheets API (optional)
GOOGLE_SERVICE_ACCOUNT_FILE=path/to/service-account.json
GOOGLE_SCOPES=https://www.googleapis.com/auth/spreadsheets

# Redis (for FastStream message broker)
REDIS_URL=redis://localhost:6379/0

# API Settings
API_V1_STR=/api/v1
LOG_LEVEL=INFO

Key Features

• No Authentication: Simple public API
• Long-term Tokens: Uses AMO CRM long-term access tokens
• Real-time Testing: Direct AMO CRM data fetching
• Comprehensive Fixtures: Real API response examples
• Modern Workers: FastStream async message processing
• Redis Integration: Reliable message broker with persistence
• Built-in Observability: Prometheus metrics and OpenTelemetry tracing

Quick Start

1. Setup

# Clone and setup
git clone <repository>
cd amo-server
uv sync

# Install FastStream with Redis support
uv add 'faststream[redis]'

# Configure environment
cp env.example .env
# Edit .env with your AMO CRM token and Redis URL

2. Run Service

# Start Redis (required for FastStream workers)
redis-server

# Start development server
uv run uvicorn app:app --reload

# Start FastStream workers (in separate terminal)
uv run faststream run workers.broker:app --reload

# Access API documentation
# http://localhost:8000/docs

3. Test AMO CRM Connection

# Check connection
curl http://localhost:8000/api/v1/amocrm/info

# Fetch real data
curl http://localhost:8000/api/v1/amocrm/fetch/users?limit=5

# Get all data for testing
uv run python scripts/fetch_amocrm_data.py

4. Example API Usage

# List entities
curl http://localhost:8000/api/v1/entities/

# Get deal fields
curl http://localhost:8000/api/v1/entities/deals/fields

# Fetch deals from AMO CRM
curl http://localhost:8000/api/v1/amocrm/fetch/deals?limit=10

# Create export configuration
curl -X POST http://localhost:8000/api/v1/export/configure \
  -H "Content-Type: application/json" \
  -d '{"name": "Test Export", "sheet_id": "123", "entity_mappings": {...}}'

Security Considerations

1. API Keys: Store AMO CRM tokens securely in environment variables
2. No Authentication: Current API has no authentication (as requested)
3. Rate Limiting: Consider implementing rate limiting for production
4. Data Validation: All incoming data is validated with Pydantic
5. CORS: Currently allows all origins (configure for production)

Performance Considerations

1. Database Indexing: Proper indexes on foreign keys and search fields
2. Batch Processing: Process data in batches to avoid memory issues
3. Caching: Use Redis for frequently accessed data
4. Async Processing: Use async/await for I/O operations
5. Pagination: Implement cursor-based pagination for large datasets

Monitoring & Logging

1. Application Metrics: Track API response times, error rates
2. Data Metrics: Monitor data freshness, sync status
3. Export Metrics: Track export job success rates, processing times
4. Error Tracking: Comprehensive error logging and alerting
5. Health Checks: Implement health check endpoints

Error Handling

1. AMO CRM API Errors: Handle rate limits, timeouts, authentication errors
2. Google Sheets API Errors: Handle quota limits, permission errors (planned)
3. Data Validation Errors: Proper error messages for invalid data
4. Database Errors: Handle connection issues, constraint violations
5. Retry Logic: Exponential backoff for transient failures (planned)

FastStream Workers & Background Processing

Worker Architecture

FastStream replaces Celery for background job processing, providing a modern async-first approach with Redis as the message broker.

Why FastStream over Celery?

1. Native Async Support: Built from ground-up for async/await patterns
2. Type Safety: Full Pydantic integration with automatic validation
3. Modern Python: Leverages Python 3.8+ features and type hints
4. Simplified Architecture: No separate result backend needed
5. FastAPI Integration: Seamless integration with existing FastAPI codebase
6. Built-in Observability: Prometheus and OpenTelemetry out-of-the-box
7. Better Error Handling: Structured error handling with automatic retries
8. AsyncAPI Documentation: Automatic API documentation for message flows

The architecture consists of:

1. Message Producers: API endpoints that queue jobs
2. Message Consumers: Worker functions that process jobs
3. Redis Broker: Message routing and persistence
4. Job Status Tracking: Database-backed status updates

Worker Implementation

1. FastStream Broker Setup

# workers/broker.py
from faststream import FastStream
from faststream.redis import RedisBroker
from utils.config import get_settings

settings = get_settings()
broker = RedisBroker(settings.redis_url)
app = FastStream(broker)

# Export job processing
@broker.subscriber("export-jobs")
async def process_export_job(job_data: dict):
    """Process Google Sheets export jobs"""
    from servers.export_server import ExportServer

    export_server = ExportServer()
    await export_server.process_export_job(job_data)

# AMO CRM data synchronization
@broker.subscriber("sync-jobs")
async def process_sync_job(sync_data: dict):
    """Synchronize data from AMO CRM"""
    from servers.sync_server import SyncServer

    sync_server = SyncServer()
    await sync_server.process_sync_job(sync_data)

# Scheduled data refresh
@broker.subscriber("refresh-jobs")
async def process_refresh_job(entity_type: str):
    """Refresh entity data from AMO CRM"""
    from servers.sync_server import SyncServer

    sync_server = SyncServer()
    await sync_server.refresh_entity_data(entity_type)

2. Job Publishers

# servers/job_server.py
from faststream.redis import RedisBroker
from typing import Dict, Any
import uuid
from datetime import datetime

class JobServer:
    def __init__(self, broker: RedisBroker):
        self.broker = broker

    async def queue_export_job(self, configuration_id: int) -> str:
        """Queue an export job for processing"""
        job_id = str(uuid.uuid4())
        job_data = {
            "job_id": job_id,
            "configuration_id": configuration_id,
            "created_at": datetime.utcnow().isoformat(),
            "status": "queued"
        }

        # Publish to export-jobs channel
        await self.broker.publish(job_data, "export-jobs")
        return job_id

    async def queue_sync_job(self, entity_type: str, **kwargs) -> str:
        """Queue a data synchronization job"""
        job_id = str(uuid.uuid4())
        sync_data = {
            "job_id": job_id,
            "entity_type": entity_type,
            "parameters": kwargs,
            "created_at": datetime.utcnow().isoformat(),
            "status": "queued"
        }

        await self.broker.publish(sync_data, "sync-jobs")
        return job_id

    async def schedule_refresh_job(self, entity_type: str):
        """Schedule periodic data refresh"""
        await self.broker.publish(entity_type, "refresh-jobs")

3. Export Job Processing

# servers/export_server.py
from typing import Dict, Any
from adapters.postgres.database import get_database
from adapters.google_sheets_client import GoogleSheetsClient
from datetime import datetime
import logging

logger = logging.getLogger(__name__)

class ExportServer:
    def __init__(self):
        self.db = get_database()
        self.sheets_client = GoogleSheetsClient()

    async def process_export_job(self, job_data: Dict[str, Any]):
        """Process a Google Sheets export job"""
        job_id = job_data["job_id"]
        configuration_id = job_data["configuration_id"]

        try:
            # Update job status to running
            await self._update_job_status(job_id, "running")

            # Get export configuration
            config = await self._get_export_configuration(configuration_id)

            # Process each enabled entity
            total_records = 0
            for entity_type, mapping in config["entity_mappings"].items():
                if mapping.get("is_enabled", False):
                    records_count = await self._export_entity(
                        entity_type,
                        config["sheet_id"],
                        mapping
                    )
                    total_records += records_count

            # Update job as completed
            await self._update_job_status(
                job_id,
                "completed",
                records_processed=total_records
            )

        except Exception as e:
            logger.error(f"Export job {job_id} failed: {str(e)}")
            await self._update_job_status(
                job_id,
                "failed",
                error_message=str(e)
            )

    async def _export_entity(self, entity_type: str, sheet_id: str, mapping: dict) -> int:
        """Export specific entity type to Google Sheets"""
        # Implementation details for entity export
        pass

    async def _update_job_status(self, job_id: str, status: str, **kwargs):
        """Update job status in database"""
        # Update export_jobs table
        pass

4. Data Synchronization Workers

# servers/sync_server.py
from adapters.amocrm_client import AMOCRMClient
from adapters.postgres.database import get_database
from typing import Dict, Any, List
import logging

logger = logging.getLogger(__name__)

class SyncServer:
    def __init__(self):
        self.amocrm = AMOCRMClient()
        self.db = get_database()

    async def process_sync_job(self, sync_data: Dict[str, Any]):
        """Process AMO CRM data synchronization"""
        job_id = sync_data["job_id"]
        entity_type = sync_data["entity_type"]
        parameters = sync_data.get("parameters", {})

        try:
            logger.info(f"Starting sync job {job_id} for {entity_type}")

            # Fetch data from AMO CRM
            data = await self.amocrm.fetch_entity_data(
                entity_type,
                **parameters
            )

            # Process and store data
            processed_count = await self._store_entity_data(entity_type, data)

            logger.info(f"Sync job {job_id} completed: {processed_count} records")

        except Exception as e:
            logger.error(f"Sync job {job_id} failed: {str(e)}")
            raise

    async def refresh_entity_data(self, entity_type: str):
        """Refresh all data for an entity type"""
        logger.info(f"Refreshing {entity_type} data")

        # Implement incremental refresh logic
        last_update = await self._get_last_update_timestamp(entity_type)

        data = await self.amocrm.fetch_entity_data(
            entity_type,
            updated_at=last_update,
            limit=250
        )

        await self._store_entity_data(entity_type, data)

FastStream Integration with FastAPI

# app.py (updated)
from faststream.redis.fastapi import RedisRouter
from workers.broker import broker

# Create FastStream router for FastAPI integration
redis_router = RedisRouter(broker)

# Include in FastAPI app
app.include_router(redis_router)

# Lifespan integration
app = FastAPI(lifespan=redis_router.lifespan_context)

Running Workers

Development

# Start FastStream worker
faststream run workers.broker:app

# Or with auto-reload
faststream run workers.broker:app --reload

Production

# Run multiple worker instances
faststream run workers.broker:app --workers 4

# With specific worker ID
WORKER_ID=worker-1 faststream run workers.broker:app

Monitoring and Observability

FastStream provides built-in observability features:

# workers/middleware.py
from faststream.prometheus import PrometheusMiddleware
from faststream.observability.middleware import TelemetryMiddleware

# Add Prometheus metrics
broker.add_middleware(PrometheusMiddleware)

# Add OpenTelemetry tracing
broker.add_middleware(TelemetryMiddleware)

Error Handling and Retry Logic

# workers/error_handling.py
from faststream.redis import RedisBroker
import asyncio
from typing import Any
import logging

logger = logging.getLogger(__name__)

@broker.subscriber("export-jobs", retry=3, retry_delay=60)
async def process_export_job_with_retry(job_data: dict):
    """Export job with automatic retry"""
    try:
        await process_export_job(job_data)
    except Exception as e:
        logger.error(f"Job failed: {e}")
        # FastStream will automatically retry based on retry settings
        raise

# Dead letter queue for failed jobs
@broker.subscriber("failed-jobs")
async def handle_failed_jobs(job_data: dict):
    """Handle permanently failed jobs"""
    logger.error(f"Job permanently failed: {job_data}")
    # Implement notification or manual intervention logic

Job Scheduling

For scheduled tasks, integrate with FastStream's scheduling capabilities:

# workers/scheduler.py
from taskiq_faststream import StreamScheduler
from taskiq.schedule_sources import LabelScheduleSource

# Schedule periodic data refresh
@broker.task(
    message={"entity_type": "deals"},
    channel="refresh-jobs",
    schedule=[{"cron": "0 */6 * * *"}]  # Every 6 hours
)
async def scheduled_deals_refresh():
    pass

@broker.task(
    message={"entity_type": "contacts"},
    channel="refresh-jobs",
    schedule=[{"cron": "0 */4 * * *"}]  # Every 4 hours
)
async def scheduled_contacts_refresh():
    pass

# Initialize scheduler
scheduler = StreamScheduler(
    broker=broker,
    sources=[LabelScheduleSource(broker)]
)

Scripts and Utilities

fetch_amocrm_data.py

• Purpose: Fetch real AMO CRM responses for testing
• Usage: uv run python scripts/fetch_amocrm_data.py
• Output:
  • Individual JSON files in tests/fixtures/real_responses/
  • Python fixtures in tests/fixtures/real_amocrm_responses.py
  • Summary of fetched data

Features

• Fetches all entity types from wecheap.amocrm.ru
• Includes custom fields metadata
• Proper error handling with troubleshooting tips
• Generates both JSON and Python fixtures

Testing

Test Structure

• Unit Tests: tests/test_amocrm_examples.py - Test AMO CRM response structure
• API Tests: tests/test_api.py - Test all API endpoints with real data
• Fixtures: Real AMO CRM responses in multiple formats

Running Tests

# Run all tests
uv run pytest

# Run with coverage
uv run pytest --cov=. --cov-report=html

# Run specific test file
uv run pytest tests/test_api.py -v

Development Workflow

Code Quality

# Format code
uv run black .
uv run isort .

# Lint code
uv run flake8

# Type checking
uv run mypy .

Database Migrations

# Generate migration
alembic revision --autogenerate -m "Description"

# Apply migrations
alembic upgrade head