docs/README.md

SwiftOps Backend API

Field Service Management Platform - FastAPI Backend

A scalable, production-ready FastAPI backend for managing field service operations including ticket management, payroll processing, inventory tracking, and financial workflows.

---

🏗️ Architecture Overview

backend/
├── app/
│   ├── __init__.py
│   ├── main.py                      # FastAPI application entry point
│   ├── config.py                    # Configuration management (env vars, settings)
│   │
│   ├── api/                         # API Layer - HTTP endpoints
│   │   ├── __init__.py
│   │   ├── deps.py                  # Shared dependencies (auth, db session)
│   │   ├── v1/                      # API version 1
│   │   │   ├── __init__.py
│   │   │   ├── router.py            # Main router aggregator
│   │   │   ├── auth.py              # Authentication endpoints
│   │   │   ├── users.py             # User management
│   │   │   ├── projects.py          # Project CRUD
│   │   │   ├── tickets.py           # Ticket operations
│   │   │   ├── assignments.py       # Ticket assignment logic
│   │   │   ├── payroll.py           # Payroll generation & management
│   │   │   ├── inventory.py         # Inventory tracking
│   │   │   ├── finance.py           # Financial transactions
│   │   │   ├── customers.py         # Customer management
│   │   │   ├── subscriptions.py     # Subscription lifecycle
│   │   │   ├── incidents.py         # Support incidents
│   │   │   ├── sales_orders.py      # Sales order processing
│   │   │   ├── analytics.py         # Dashboard & reporting
│   │   │   └── webhooks.py          # Payment gateway webhooks
│   │   │
│   │   └── websockets/              # WebSocket endpoints
│   │       ├── __init__.py
│   │       ├── location.py          # Real-time location tracking
│   │       └── notifications.py     # Real-time notifications
│   │
│   ├── core/                        # Core Business Logic
│   │   ├── __init__.py
│   │   ├── auth.py                  # JWT handling, password hashing
│   │   ├── security.py              # RLS, permissions, RBAC
│   │   ├── exceptions.py            # Custom exception classes
│   │   └── middleware.py            # Custom middleware (logging, CORS)
│   │
│   ├── services/                    # Business Logic Layer
│   │   ├── __init__.py
│   │   ├── ticket_service.py        # Ticket lifecycle, assignment logic
│   │   ├── payroll_service.py       # Payroll calculation, generation
│   │   ├── inventory_service.py     # Inventory allocation, tracking
│   │   ├── finance_service.py       # Transaction workflows, approvals
│   │   ├── sla_service.py           # SLA calculation, monitoring
│   │   ├── location_service.py      # GPS validation, distance calculation
│   │   ├── notification_service.py  # Email, SMS, push notifications
│   │   ├── payment_service.py       # Payment gateway integration
│   │   ├── subscription_service.py  # Subscription activation, lifecycle
│   │   └── analytics_service.py     # Metrics, dashboard data
│   │
│   ├── models/                      # SQLAlchemy ORM Models
│   │   ├── __init__.py
│   │   ├── base.py                  # Base model with common fields
│   │   ├── user.py                  # User, UserFinancialAccounts
│   │   ├── organization.py          # Clients, Contractors
│   │   ├── project.py               # Projects, ProjectTeam, ProjectRoles
│   │   ├── ticket.py                # Tickets, TicketAssignments, TicketStatusHistory
│   │   ├── customer.py              # Customers, SalesOrders, Subscriptions
│   │   ├── incident.py              # Incidents
│   │   ├── inventory.py             # ProjectInventory, InventoryAssignments
│   │   ├── finance.py               # ProjectFinance, UserPayroll, PaymentLogs
│   │   ├── timesheet.py             # Timesheets
│   │   └── document.py              # Documents
│   │
│   ├── schemas/                     # Pydantic Schemas (Request/Response)
│   │   ├── __init__.py
│   │   ├── user.py                  # UserCreate, UserUpdate, UserResponse
│   │   ├── project.py               # ProjectCreate, ProjectUpdate, ProjectResponse
│   │   ├── ticket.py                # TicketCreate, TicketUpdate, TicketResponse
│   │   ├── payroll.py               # PayrollCreate, PayrollCalculation
│   │   ├── inventory.py             # InventoryAssignment, InventoryDistribution
│   │   ├── finance.py               # TransactionCreate, TransactionApproval
│   │   └── common.py                # Shared schemas (Pagination, Filters)
│   │
│   ├── repositories/                # Data Access Layer (Repository Pattern)
│   │   ├── __init__.py
│   │   ├── base.py                  # Base repository with CRUD operations
│   │   ├── user_repository.py       # User-specific queries
│   │   ├── ticket_repository.py     # Ticket-specific queries
│   │   ├── payroll_repository.py    # Payroll-specific queries
│   │   └── finance_repository.py    # Finance-specific queries
│   │
│   ├── tasks/                       # Background Tasks (Celery)
│   │   ├── __init__.py
│   │   ├── celery_app.py            # Celery configuration
│   │   ├── payroll_tasks.py         # Weekly payroll generation
│   │   ├── sla_tasks.py             # SLA monitoring, alerts
│   │   ├── notification_tasks.py    # Send emails, SMS
│   │   ├── invoice_tasks.py         # Generate contractor invoices
│   │   └── analytics_tasks.py       # Pre-compute dashboard metrics
│   │
│   ├── integrations/                # External Service Integrations
│   │   ├── __init__.py
│   │   ├── supabase.py              # Supabase client (Auth, Storage, Realtime)
│   │   ├── mpesa.py                 # M-Pesa payment gateway
│   │   ├── google_maps.py           # Google Maps API (geocoding, distance)
│   │   ├── sms_provider.py          # SMS gateway (Africa's Talking, Twilio)
│   │   ├── email_provider.py        # Email service (SendGrid, AWS SES)
│   │   └── storage.py               # File storage (Supabase Storage, S3)
│   │
│   ├── utils/                       # Utility Functions
│   │   ├── __init__.py
│   │   ├── date_utils.py            # Date/time helpers
│   │   ├── location_utils.py        # GPS calculations, distance
│   │   ├── validation.py            # Custom validators
│   │   ├── formatters.py            # Data formatting helpers
│   │   └── constants.py             # Application constants
│   │
│   └── db/                          # Database Configuration
│       ├── __init__.py
│       ├── session.py               # Database session management
│       ├── base.py                  # Base class for all models
│       └── migrations/              # Alembic migrations
│           └── versions/
│
├── tests/                           # Test Suite
│   ├── __init__.py
│   ├── conftest.py                  # Pytest fixtures
│   ├── unit/                        # Unit tests
│   │   ├── test_services/
│   │   ├── test_repositories/
│   │   └── test_utils/
│   ├── integration/                 # Integration tests
│   │   ├── test_api/
│   │   └── test_tasks/
│   └── e2e/                         # End-to-end tests
│
├── scripts/                         # Utility Scripts
│   ├── seed_data.py                 # Seed database with test data
│   ├── generate_payroll.py          # Manual payroll generation
│   └── migrate_data.py              # Data migration scripts
│
├── .env.example                     # Environment variables template
├── .gitignore
├── requirements.txt                 # Python dependencies
├── pyproject.toml                   # Poetry configuration (alternative)
├── Dockerfile                       # Docker container definition
├── docker-compose.yml               # Local development setup
└── README.md                        # This file

---

🎯 Core Functions & Responsibilities

1. Authentication & Authorization

Module: app/api/v1/auth.py, app/core/auth.py, app/core/security.py

Functions:

• User login/logout (JWT token generation)
• Password reset flow
• Role-based access control (RBAC)
• Row-level security (RLS) enforcement
• Multi-tenancy isolation (client/contractor scoping)

Supabase Integration:

• Uses Supabase Auth for user authentication
• Validates JWT tokens from Supabase
• Syncs user roles with custom Users table

---

2. Ticket Management

Module: app/services/ticket_service.py, app/api/v1/tickets.py

Functions:

• Create tickets from sales orders, incidents, or tasks
• Ticket assignment logic (manual, auto-assignment)
• Limit self-assignment to 3 tickets per agent
• Status transitions with validation
• SLA deadline calculation and monitoring
• GPS-based arrival verification
• Ticket completion workflow

Business Rules:

# Example: Ticket Assignment Logic
def assign_ticket(ticket_id, user_id):
    # 1. Check user has < 3 active assignments
    # 2. Verify user is in project team
    # 3. Check user's region matches ticket region
    # 4. Create TicketAssignment record
    # 5. Update ticket status to 'assigned'
    # 6. Send notification to agent
    # 7. Update SLA target date

---

3. Payroll Processing

Module: app/services/payroll_service.py, app/api/v1/payroll.py

Functions:

• Weekly payroll generation (automated via Celery)
• Calculate earnings based on compensation type:
  • Flat rate (fixed weekly amount)
  • Commission (% of package price)
  • Hybrid (base + commission)
  • Hourly (based on timesheet hours)
• Apply deductions (advances, penalties)
• Generate payroll reports
• Mark payroll as paid with payment reference

Business Rules:

# Example: Payroll Calculation
def calculate_payroll(user_id, project_id, period_start, period_end):
    role = get_user_project_role(user_id, project_id)
    tickets_closed = count_tickets_closed(user_id, period_start, period_end)
    hours_worked = sum_timesheet_hours(user_id, period_start, period_end)
    
    if role.compensation_type == 'flat_rate':
        earnings = role.flat_rate_amount
    elif role.compensation_type == 'commission':
        earnings = sum_ticket_values(tickets) * role.commission_percentage
    elif role.compensation_type == 'hybrid':
        earnings = role.base_amount + (sum_ticket_values(tickets) * role.commission_percentage)
    elif role.compensation_type == 'hourly':
        earnings = hours_worked * role.hourly_rate
    
    deductions = get_user_deductions(user_id, period_start, period_end)
    total = earnings - deductions
    
    return create_payroll_record(...)

---

4. Inventory Management

Module: app/services/inventory_service.py, app/api/v1/inventory.py

Functions:

• Allocate inventory to regional hubs
• Issue equipment to field agents
• Track serial numbers and unit identifiers
• Prevent duplicate assignments (same serial issued twice)
• Handle returns (tools) and installations (equipment)
• Optimistic locking for concurrent updates
• Inventory reconciliation reports

Business Rules:

# Example: Inventory Assignment
def assign_inventory_to_agent(distribution_id, user_id, unit_identifier):
    # 1. Check unit_identifier not already assigned
    # 2. Verify quantity_available > 0
    # 3. Check user is in same region as distribution
    # 4. Create InventoryAssignment record
    # 5. Update distribution.quantity_issued
    # 6. Use optimistic locking (version field)

---

5. Financial Management

Module: app/services/finance_service.py, app/api/v1/finance.py

Functions:

• Record project expenses (inflows/outflows)
• Approval workflow (pending → approved → paid)
• Link to entities (tickets, payroll, invoices)
• Payment gateway integration (M-Pesa, bank transfers)
• Transaction reconciliation
• Financial reporting and audit trails

Business Rules:

# Example: Expense Approval Workflow
def approve_expense(expense_id, approver_id):
    # 1. Check approver has 'dispatcher' or 'project_manager' role
    # 2. Verify expense is in 'pending' status
    # 3. Validate location_verified = TRUE (if required)
    # 4. Update status to 'approved'
    # 5. Create ProjectFinance record
    # 6. Send notification to incurred_by_user

---

6. Location Tracking

Module: app/services/location_service.py, app/api/websockets/location.py

Functions:

• Real-time GPS tracking via WebSocket
• Validate agent arrival at customer site
• Calculate distance between agent and customer
• Journey tracking (breadcrumb trail)
• Fraud detection (impossible travel speeds)
• Update user current location

Business Rules:

# Example: Arrival Verification
def verify_arrival(ticket_id, agent_location):
    ticket = get_ticket(ticket_id)
    customer_location = get_customer_location(ticket.customer_id)
    
    distance = calculate_distance(agent_location, customer_location)
    
    if distance <= 100:  # Within 100 meters
        mark_arrived(ticket_id, agent_location, verified=True)
    else:
        mark_arrived(ticket_id, agent_location, verified=False)

---

7. SLA Monitoring

Module: app/services/sla_service.py, app/tasks/sla_tasks.py

Functions:

• Calculate SLA deadlines based on priority
• Monitor tickets approaching deadline
• Send alerts to managers and agents
• Mark tickets as SLA violated
• Generate SLA compliance reports

Background Task (runs every hour):

@celery_app.task
def monitor_sla_violations():
    # 1. Find tickets with sla_target_date < now() and status != 'completed'
    # 2. Mark sla_violated = TRUE
    # 3. Send notifications to project managers
    # 4. Log SLA violation event

---

8. Payment Gateway Integration

Module: app/services/payment_service.py, app/integrations/mpesa.py

Functions:

• Initiate M-Pesa payments (B2C, B2B)
• Handle payment callbacks/webhooks
• Retry failed payments
• Log all payment attempts (PaymentLogs)
• Reconcile payments with transactions

Webhook Handler:

@router.post("/webhooks/mpesa")
async def mpesa_callback(payload: dict):
    # 1. Validate webhook signature
    # 2. Parse payment result
    # 3. Update ProjectFinance status
    # 4. Create PaymentLog entry
    # 5. Send notification to user

---

9. Subscription Management

Module: app/services/subscription_service.py, app/api/v1/subscriptions.py

Functions:

• Activate subscriptions after installation
• Validate activation requirements (dynamic per project)
• Manage subscription lifecycle (active, suspended, cancelled)
• Link subscriptions to tickets and sales orders

Business Rules:

# Example: Subscription Activation
def activate_subscription(subscription_id, activation_data):
    subscription = get_subscription(subscription_id)
    project = get_project(subscription.project_id)
    
    # Validate dynamic activation requirements
    for requirement in project.activation_requirements:
        if requirement['required'] and requirement['field'] not in activation_data:
            raise ValidationError(f"Missing required field: {requirement['label']}")
    
    # Update subscription
    subscription.status = 'active'
    subscription.activation_date = date.today()
    subscription.activation_details = activation_data
    subscription.activated_by_user_id = current_user.id

---

10. Analytics & Reporting

Module: app/services/analytics_service.py, app/api/v1/analytics.py

Functions:

• Dashboard metrics (tickets, revenue, SLA compliance)
• Agent performance reports
• Project financial summaries
• Inventory utilization reports
• Pre-computed aggregations (via Celery)

Background Task (runs daily):

@celery_app.task
def compute_daily_metrics():
    # 1. Calculate tickets closed per agent
    # 2. Compute average completion time
    # 3. Calculate SLA compliance rate
    # 4. Store in cache (Redis)

---

🔗 Supabase Integration Points

1. Authentication

# app/integrations/supabase.py
from supabase import create_client

supabase = create_client(SUPABASE_URL, SUPABASE_KEY)

# Verify JWT token from frontend
def verify_token(token: str):
    user = supabase.auth.get_user(token)
    return user

2. Storage (File Uploads)

# Upload ticket photos, receipts
def upload_document(file, bucket="documents"):
    path = f"{user_id}/{ticket_id}/{filename}"
    supabase.storage.from_(bucket).upload(path, file)
    return supabase.storage.from_(bucket).get_public_url(path)

3. Realtime (WebSocket)

# Subscribe to ticket updates
supabase.realtime.channel('tickets')
    .on('postgres_changes', 
        event='UPDATE', 
        schema='public', 
        table='Tickets',
        callback=handle_ticket_update)
    .subscribe()

4. Database (Direct Queries)

# Use Supabase client for simple queries (optional)
# For complex queries, use SQLAlchemy ORM
tickets = supabase.table('Tickets').select('*').eq('status', 'open').execute()

---

🚀 Getting Started

Prerequisites

• Python 3.11+
• PostgreSQL 15+ (Supabase)
• Redis (for caching and Celery)
• Docker & Docker Compose (optional)

Installation

1. Clone the repository

git clone <repo-url>
cd backend

2. Create virtual environment

python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

3. Install dependencies

pip install -r requirements.txt

4. Set up environment variables

cp .env.example .env
# Edit .env with your Supabase credentials

5. Run database migrations

alembic upgrade head

6. Start the development server

uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

7. Start Celery worker (in separate terminal)

celery -A app.tasks.celery_app worker --loglevel=info

8. Start Celery beat (scheduler, in separate terminal)

celery -A app.tasks.celery_app beat --loglevel=info

---

🐳 Docker Setup

# Start all services (API, Celery, Redis)
docker-compose up -d

# View logs
docker-compose logs -f api

# Stop services
docker-compose down

---

📝 Environment Variables

# .env.example

# Application
APP_NAME=SwiftOps API
APP_VERSION=1.0.0
DEBUG=True
ENVIRONMENT=development

# Database (Supabase PostgreSQL)
DATABASE_URL=postgresql://user:password@host:5432/dbname
SUPABASE_URL=https://your-project.supabase.co
SUPABASE_KEY=your-anon-key
SUPABASE_SERVICE_KEY=your-service-role-key

# JWT
SECRET_KEY=your-secret-key-here
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30

# Redis
REDIS_URL=redis://localhost:6379/0

# Celery
CELERY_BROKER_URL=redis://localhost:6379/0
CELERY_RESULT_BACKEND=redis://localhost:6379/0

# Payment Gateways
MPESA_CONSUMER_KEY=your-mpesa-key
MPESA_CONSUMER_SECRET=your-mpesa-secret
MPESA_SHORTCODE=174379
MPESA_PASSKEY=your-passkey

# Google Maps
GOOGLE_MAPS_API_KEY=your-google-maps-key

# SMS Provider (Africa's Talking)
AFRICASTALKING_USERNAME=your-username
AFRICASTALKING_API_KEY=your-api-key

# Email Provider (SendGrid)
SENDGRID_API_KEY=your-sendgrid-key
FROM_EMAIL=noreply@swiftops.com

# File Storage
STORAGE_PROVIDER=supabase  # or 's3'
AWS_ACCESS_KEY_ID=your-aws-key
AWS_SECRET_ACCESS_KEY=your-aws-secret
AWS_S3_BUCKET=your-bucket-name

---

🧪 Testing

# Run all tests
pytest

# Run with coverage
pytest --cov=app --cov-report=html

# Run specific test file
pytest tests/unit/test_services/test_payroll_service.py

# Run integration tests only
pytest tests/integration/

---

📚 API Documentation

Once the server is running, visit:

• Swagger UI: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc
• OpenAPI JSON: http://localhost:8000/openapi.json

---

🏛️ Design Patterns & Best Practices

1. Repository Pattern

Separates data access logic from business logic.

# app/repositories/ticket_repository.py
class TicketRepository:
    def get_by_id(self, ticket_id: UUID) -> Ticket:
        return db.query(Ticket).filter(Ticket.id == ticket_id).first()
    
    def get_open_tickets(self, project_id: UUID) -> List[Ticket]:
        return db.query(Ticket).filter(
            Ticket.project_id == project_id,
            Ticket.status == 'open',
            Ticket.deleted_at.is_(None)
        ).all()

2. Service Layer Pattern

Encapsulates business logic.

# app/services/ticket_service.py
class TicketService:
    def __init__(self, ticket_repo: TicketRepository):
        self.ticket_repo = ticket_repo
    
    def assign_ticket(self, ticket_id: UUID, user_id: UUID):
        # Business logic here
        ticket = self.ticket_repo.get_by_id(ticket_id)
        self._validate_assignment(ticket, user_id)
        assignment = self._create_assignment(ticket, user_id)
        self._send_notification(user_id, ticket)
        return assignment

3. Dependency Injection

Use FastAPI's dependency injection system.

# app/api/deps.py
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_current_user(token: str = Depends(oauth2_scheme)):
    # Validate token and return user
    return user

# Usage in endpoint
@router.get("/tickets")
def get_tickets(
    db: Session = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    return ticket_service.get_tickets(db, current_user)

4. Optimistic Locking

Prevent concurrent update conflicts.

def update_inventory_distribution(distribution_id, updates, version):
    distribution = db.query(ProjectInventoryDistribution).filter(
        ProjectInventoryDistribution.id == distribution_id,
        ProjectInventoryDistribution.version == version
    ).first()
    
    if not distribution:
        raise ConcurrentUpdateError("Record was modified by another user")
    
    # Apply updates
    distribution.quantity_issued += updates.quantity
    distribution.version += 1
    db.commit()

5. Soft Deletes

Always filter out soft-deleted records.

# Base repository method
def get_active(self, model_class):
    return db.query(model_class).filter(model_class.deleted_at.is_(None))

---

🔐 Security Considerations

1. Row-Level Security (RLS)

   • Enforce tenant isolation at database level
   • Use Supabase RLS policies

2. API Rate Limiting

   • Implement rate limiting per client/user
   • Use Redis for rate limit tracking

3. Input Validation

   • Use Pydantic schemas for all inputs
   • Validate file uploads (type, size)

4. Audit Logging

   • Log all financial transactions
   • Track who created/updated records

5. Secrets Management

   • Never commit secrets to git
   • Use environment variables or secret managers

---

📈 Scalability Considerations

1. Database Connection Pooling

   • Use SQLAlchemy connection pool
   • Configure pool size based on load

2. Caching Strategy

   • Cache frequently accessed data (Redis)
   • Invalidate cache on updates

3. Background Jobs

   • Offload heavy tasks to Celery
   • Use task queues for async processing

4. Horizontal Scaling

   • Stateless API design
   • Load balance across multiple instances

5. Database Optimization

   • Use indexes effectively
   • Monitor slow queries
   • Consider read replicas for reporting

---

🤝 Contributing

1. Create a feature branch
2. Write tests for new features
3. Follow PEP 8 style guide
4. Update documentation
5. Submit pull request

---

📄 License

[Your License Here]

---

📞 Support

For questions or issues, contact: [your-email@example.com]