Mini-CRM

Lead distribution service that automatically assigns incoming requests to operators based on configurable weights, availability, and workload limits.

Features

• Weighted distribution — operators receive leads proportionally to their assigned weights
• Workload management — respects per-operator active request limits
• Multi-source support — different weight configurations per lead source (bot, website, etc.)
• Lead deduplication — identifies leads by external ID, phone, or email
• Async architecture — built on FastAPI + async SQLAlchemy 2.x for high throughput
• Full CRUD — manage operators, sources, leads, and requests via REST API

Tech Stack

• Framework: FastAPI (async)
• ORM: SQLAlchemy 2.x with async support
• Validation: Pydantic v2
• Database: SQLite (dev) / PostgreSQL (prod)
• Migrations: Alembic
• Testing: Pytest + pytest-asyncio + httpx
• Containerization: Docker + Docker Compose

Quick Start

Local

pip install -r requirements.txt
uvicorn app.main:app --reload
# API docs: http://localhost:8000/docs

Docker

docker-compose up --build

Data Model

┌─────────────┐       ┌─────────────────────┐       ┌─────────────┐
│  Operator   │───────│ OperatorSourceWeight │───────│   Source    │
│             │  1:N  │                      │  N:1  │             │
│ - name      │       │ - operator_id        │       │ - name      │
│ - is_active │       │ - source_id          │       │ - code      │
│ - max_load  │       │ - weight             │       │ - is_active │
└──────┬──────┘       └──────────────────────┘       └──────┬──────┘
       │                                                    │
       │ 1:N                                          1:N   │
       │              ┌─────────────┐                       │
       └──────────────│   Request   │───────────────────────┘
                      │             │
                      │ - lead_id   │       ┌─────────────┐
                      │ - source_id │───────│    Lead     │
                      │ - operator  │  N:1  │             │
                      │ - status    │       │ - external_id│
                      │ - message   │       │ - phone     │
                      └─────────────┘       │ - email     │
                                            └─────────────┘

Distribution Algorithm

1. Lead identification — match by external_id, phone, or email; create if new
2. Operator filtering — only active operators under their workload limit
3. Weighted random selection — probability = operator_weight / total_weights
4. Graceful fallback — if no operators available, request is created unassigned for manual handling

API Endpoints

Operators

Method	Endpoint	Description
POST	/operators/	Create operator
GET	/operators/	List operators with current load
GET	/operators/{id}	Get operator details
PATCH	/operators/{id}	Update (activity, limits)
DELETE	/operators/{id}	Delete operator

Sources

Method	Endpoint	Description
POST	/sources/	Create source
GET	/sources/	List sources
PUT	/sources/{id}/operators	Configure operator weights

Requests

Method	Endpoint	Description
POST	/requests/	Create request (auto-distributes)
GET	/requests/	List requests
PATCH	/requests/{id}/status	Update status

Leads

Method	Endpoint	Description
GET	/leads/	List leads with request counts
GET	/leads/{id}	Lead details with request history

Usage Example

# Create operators
curl -X POST localhost:8000/operators/ \
  -H "Content-Type: application/json" \
  -d '{"name": "Alice", "max_active_requests": 10}'

curl -X POST localhost:8000/operators/ \
  -H "Content-Type: application/json" \
  -d '{"name": "Bob", "max_active_requests": 15}'

# Create source and set weights (Alice 25%, Bob 75%)
curl -X POST localhost:8000/sources/ \
  -H "Content-Type: application/json" \
  -d '{"name": "Telegram Bot", "code": "tg_bot"}'

curl -X PUT localhost:8000/sources/1/operators \
  -H "Content-Type: application/json" \
  -d '{"weights": [{"operator_id": 1, "weight": 1}, {"operator_id": 2, "weight": 3}]}'

# Submit a request — automatically assigned to an operator
curl -X POST localhost:8000/requests/ \
  -H "Content-Type: application/json" \
  -d '{"source_code": "tg_bot", "phone": "+1234567890", "message": "Interested in your product"}'

Testing

pytest

Tests cover weighted distribution, inactive operator exclusion, workload limits, and CRUD operations.

Project Structure

app/
├── api/            # Route handlers
├── models/         # SQLAlchemy models
├── schemas/        # Pydantic schemas
├── services/       # Business logic
├── config.py
├── database.py
└── main.py
tests/
├── conftest.py     # Fixtures
├── test_distribution.py
├── test_operators.py
├── test_requests.py
└── test_sources.py

License

MIT