LEO Customer 360 — Documentation

c360 Promotions API

16 min read

c360 Promotions API

High-performance, multi-tenant marketing promotions API built on FastAPI + SQLAlchemy + PostgreSQL 16.

Designed for:

  • Standard digital ad banner delivery, affiliate marketing link tracking, and sponsored content in native ads
  • Real-time content and product recommendations for websites, mobile applications, and AI agents through the MCP protocol
  • Multi-tenant isolation with strict tenant_id filtering on all operations
  • Production-grade architecture with clean separation of concerns across repositories, services, and controllers
  • Scalable promotion content spanning display, native, video, carousel, affiliate, and sponsored formats

Overview

The c360 Promotions API exposes:

  1. Health endpoints for monitoring service and database availability
  2. Promotion and campaign endpoints for retrieving sponsored content and delivery candidates by placement
  3. Placement endpoints for inventory configuration lookups
  4. Affiliate link tracking for measurable partner and campaign conversions
  5. Recommendation endpoints for real-time website, mobile, and MCP-based AI-agent experiences
  6. JavaScript widget (ads.loader.js) for client-side banner and sponsored-content rendering

Key characteristics:

  • All data scoped by tenant_id (multi-tenant isolation)
  • Composable ad structure: Ad = Campaign + Creative + Placement
  • Extensible JSONB payloads for provider-specific configuration
  • Ready for Redis caching on hot paths (placement lookups, promotion candidates, and recommendations)
  • Comprehensive test coverage (unit, integration, repository)

Quick Start

Prerequisites

  • Python 3.9+
  • PostgreSQL 16+
  • Redis (optional, for caching; API functions without it)
  • Docker Compose (recommended for local dev)

Installation

cd customer360-promotions
 
# Create virtual environment
python3 -m venv .venv
source .venv/bin/activate
 
# Install dependencies
pip install -r requirements.txt
 
# Copy .env template and edit
cp .env.example .env
# Edit .env:
#   c360_PROMOTION_API_HOST=localhost
#   c360_PROMOTION_API_PORT=9009
#   c360_PROMOTION_DB_HOST=localhost
#   c360_PROMOTION_DB_PORT=5432
#   c360_PROMOTION_DB_USER=postgres
#   c360_PROMOTION_DB_PASSWORD=your_password
#   c360_PROMOTION_DB_NAME=customer360

Running the Server

Basic startup:

./start.sh

With database initialization + demo data:

./start.sh --seed-demo-customer360-promotions

Stop/restart:

./stop.sh      # Shutdown gracefully
./restart.sh   # Stop and start

Development Mode

# With auto-reload on file changes
UVICORN_RELOAD=true ./start.sh
 
# Or run directly with uvicorn
uvicorn app:app --reload --port 9009

API Documentation

API Endpoints

Health & Monitoring

GET / Service information and links.

{
    "service": "customer360-promotions",
  "status": "ok",
  "version": "1.0.0",
  "schema": "leo_ads",
  "docs": "/docs"
}

GET /health Lightweight health check (no database access).

{
  "status": "ok",
    "service": "customer360-promotions"
}

GET /health/database Verify PostgreSQL connectivity.

{
  "status": "ok",
  "database": "reachable",
  "schema": "leo_ads"
}

Ads

GET /ads/{ad_id} Retrieve a single ad by its primary key.

Parameters:

  • ad_id (int, path): Internal ad ID

Response:

{
  "ad_id": 12345,
  "tenant_id": 1,
  "ad_key": "ad_1001",
  "campaign_id": 100,
  "creative_id": 500,
  "placement_id": 10,
  "status": "active",
  "score_weight": 1.5,
  "frequency_cap": 3,
  "metadata": {},
  "created_at": "2024-01-01T10:00:00Z",
  "updated_at": "2024-01-15T14:30:00Z"
}

Note: Use ad_key for external/user-facing references, not ad_id.

Placements

GET /placements/{placement_key} Retrieve an active placement by its external key.

Parameters:

  • placement_key (str, path): External identifier (e.g., “homepage_top”)

Response:

{
  "placement_id": 10,
  "tenant_id": 1,
  "placement_key": "homepage_top",
  "name": "Homepage Top Banner",
  "status": "active",
  "min_width_px": 728,
  "max_width_px": 728,
  "min_height_px": 90,
  "max_height_px": 90,
  "responsive": false,
  "metadata": {
    "section": "header",
    "priority": "high"
  },
  "created_at": "2024-01-01T10:00:00Z",
  "updated_at": "2024-01-15T14:30:00Z"
}

Ad Serving

GET /serve/{placement_ref} Assemble a full ad-serving payload for a placement or ad.

Parameters:

  • placement_ref (str, path): Placement key or ad key
  • tenant (str, query): Tenant key (default: “demo”)
  • limit (int, query): Max ads to return (default: 5)

Response:

{
  "ads": [
    {
      "ad_id": 1,
      "ad_key": "ad_1001",
      "creative_id": 500,
      "campaign_id": 100,
      "placement_id": 10,
      "status": "active",
      "score_weight": 2.0,
      "frequency_cap": 3,
      "metadata": {}
    }
  ]
}

Note: This is a development/test endpoint; production usage should implement proper tenant authentication via JWT/OAuth.

Data Models

The promotions API is built on four core entities: Placement, Campaign, Creative, and Ad.

Placement (inventory)
    ↓ (1:M)
Ad (delivery config)
    ├─ → Campaign (buying/business logic)
    ├─ → Creative (content/assets)
    └─ → Tenant (multi-tenancy)

Placement

Publisher inventory slot. Represents a stable, named location on a website/app.

Schema: leo_ads.placement

Key fields:

  • placement_id (BigInteger, PK): Internal surrogate key
  • placement_key (String): External identifier (e.g., “homepage_top”) — use this for APIs
  • tenant_id (BigInteger, FK): Multi-tenant isolation
  • name (String): Display name
  • status (String): 'active' | 'paused' | 'archived'
  • min_width_px, max_width_px, min_height_px, max_height_px (Integer, nullable): Dimension constraints
  • responsive (Boolean): Whether placement supports responsive sizing
  • metadata (JSONB): Extensible placement-specific attributes
  • created_at, updated_at (DateTime): Audit timestamps

Example placements:

homepage_top        → 728×90 banner
article_inline      → 300×250 / 336×280
sidebar             → 300×600 / 320×600
native_feed         → responsive, flexible

Campaign

Represents the business/buying configuration. One campaign may fund multiple ads across different placements.

Schema: leo_ads.campaign

Key fields:

  • campaign_id (BigInteger, PK): Internal ID
  • campaign_key (String): External identifier — use this for APIs
  • tenant_id (BigInteger, FK): Multi-tenant isolation
  • advertiser_id (BigInteger, FK, nullable): Advertiser owner
  • source_account_id (BigInteger, FK, nullable): Source/provider account (Google Ads, Shopee, etc.)
  • name (String): Campaign display name
  • objective (String): Campaign goal ('awareness' | 'traffic' | 'conversions' | 'retention' | ...)
  • buying_model (String): Pricing model ('CPM' | 'CPC' | 'CPA' | 'vCPM' | ...)
  • budget_amount (Decimal): Total budget
  • currency (String): ISO 4217 code (e.g., ‘USD’)
  • daily_budget_amount (Decimal, nullable): Daily cap
  • status (String): 'draft' | 'approved' | 'running' | 'paused' | 'ended' | 'archived'
  • starts_at, ends_at (DateTime, nullable): Campaign date range
  • metadata (JSONB): Campaign-specific config (targeting rules, bid strategy, etc.)

Creative

Represents reusable ad content/assets. Multiple ads can reference the same creative.

Schema: leo_ads.creative

Key fields:

  • creative_id (BigInteger, PK): Internal ID
  • creative_key (String): External identifier — use this for APIs
  • tenant_id (BigInteger, FK): Multi-tenant isolation
  • ad_type (String): 'display' | 'native' | 'video' | 'carousel' | 'dynamic' | ...
  • format_code (String): Dimensions/layout ('300x250' | '728x90' | '1200x628' | 'responsive' | ...)
  • status (String): 'active' | 'paused' | 'archived'
  • version_no (Integer): Content version tracking
  • headline, subheadline, body, cta (Text): Common rendering fields
  • image_url, video_url, logo_url (Text): Asset URLs
  • content_payload (JSONB): Provider-specific payload (e.g., Google Ad Manager JSON, Shopee widget config)
  • created_at, updated_at (DateTime): Audit timestamps

Design note: Creative separates content (text, images, media) from delivery config (Ad) and business config (Campaign). This enables:

  • Multiple Ads to reference one Creative (reuse)
  • Multiple Campaigns to share Creative variants
  • Flexible provider-specific data in content_payload without schema migrations
  • Easy A/B testing by swapping creatives

Represents the serving configuration linking Campaign, Creative, and Placement.

Schema: leo_ads.ad

Key fields:

  • ad_id (BigInteger, PK): Internal surrogate key
  • ad_key (String): External identifier — use this for APIs, never expose ad_id
  • tenant_id (BigInteger, FK): Multi-tenant isolation
  • campaign_id (BigInteger, FK, nullable): Link to Campaign (optional; ads may exist without campaigns)
  • creative_id (BigInteger, FK): Link to Creative content
  • placement_id (BigInteger, FK): Link to Placement inventory
  • status (String): 'active' | 'paused' | 'archived' — only active ads serve
  • score_weight (Float): Ranking weight for candidate selection (higher = preferred)
  • frequency_cap (Integer, nullable): Max impressions per user per session/day
  • metadata (JSONB): Ad-tech-specific configuration (targeting rules, bid adjustments, etc.)
  • created_at, updated_at (DateTime): Audit timestamps

Multi-tenancy: ⚠️ CRITICAL: Every query must filter by tenant_id to prevent cross-tenant data exposure.

Example query (from AdRepository):

statement = (
    select(Ad)
    .where(
        Ad.tenant_id == tenant_id,      # ← MANDATORY
        Ad.placement_id == placement_id,
        Ad.status == "active",
    )
    .order_by(Ad.score_weight.desc(), Ad.ad_id.asc())
    .limit(20)
)

Supporting Models

Tenant

  • Represents an isolated tenant/customer
  • All tables reference tenant_id
  • Enforce tenant isolation on all queries

Advertiser

  • Represents a brand/advertiser within a tenant
  • Campaigns link to advertisers (optional)

SourceAccount

  • Represents an external provider account (Google Ads, Shopee, Lazada, etc.)
  • Links to advertiser and tenant
  • Stores provider-specific credentials/config

SourceAsset

  • External assets managed by providers (Google campaigns, Shopee offers, etc.)
  • Maps external IDs to local campaign/creative/ad IDs

Project Structure

customer360-promotions/
├── app.py                      # FastAPI entrypoint & local dev runner
├── core/
│   ├── __init__.py
│   ├── application.py          # Application composition root, routing, lifecycle
│   ├── config.py               # Pydantic settings, database engine, connection pooling
│   └── database.py             # Database utilities (future)
├── model/
│   ├── __init__.py
│   ├── base.py                 # SQLAlchemy DeclarativeBase
│   ├── ad.py                   # Ad ORM model
│   ├── campaign.py             # Campaign ORM model
│   ├── creative.py             # Creative ORM model
│   ├── placement.py            # Placement ORM model
│   └── tenant.py               # Tenant ORM model
├── repository/
│   ├── __init__.py
│   ├── ad_repository.py        # Ad query operations
│   ├── placement_repository.py # Placement query operations
│   └── ad_cache_utils.py       # Redis caching utilities (future)
├── static/
│   ├── ads.loader.js           # Client-side ad widget & loader
│   ├── leo.ads.css             # Ad widget styling (light/dark theme)
│   ├── ads-banner.html         # HTML test page for ad rendering
│   └── ads.data.json           # Static test data fixture
├── sql-scripts/
│   ├── db-schema-init.sql      # PostgreSQL schema creation
│   └── sample-data-init.sql    # Demo data seeding
├── tests/
│   ├── __init__.py
│   ├── conftest.py             # Pytest fixtures (test DB, sessions, etc.)
│   ├── _pg.py                  # PostgreSQL availability check
│   ├── test_api.py             # Integration tests for endpoints
│   ├── test_models.py          # ORM field mapping tests
│   ├── test_repositories.py    # Repository query logic tests
│   └── test_model_metadata.py  # Model introspection tests
├── logs/                        # Application logs (created at runtime)
├── .env.example                 # Environment variable template
├── .env                         # Actual env config (git-ignored)
├── .gitignore
├── .dockerignore
├── Dockerfile                   # Container image definition
├── requirements.txt             # Python dependencies
├── start.sh                     # Startup script (with optional --seed-demo-customer360-promotions)
├── stop.sh                      # Shutdown script
├── restart.sh                   # Stop + start convenience script
├── run_unit_tests.sh            # Test runner
└──  README.md                    # This file

Key files explained:

  • app.py: Entry point. Bootstraps PromotionsApplication and exposes app object for uvicorn
  • core/application.py: Composition root; creates FastAPI instance, registers routes, manages lifecycle
  • core/config.py: Loads .env, creates SQLAlchemy engine with connection pooling
  • model/*.py: SQLAlchemy ORM definitions (declarative syntax)
  • repository/*.py: Persistence layer; all database queries live here
  • static/: Frontend assets (JS widget, CSS, HTML test pages)
  • sql-scripts/: DDL for schema initialization and DML for seed data
  • tests/: Full test suite with fixtures for test database

Repositories

Repositories encapsulate all database queries and serialization. Business logic belongs in Services (future).

AdRepository

Located in: repository/ad_repository.py

Responsibilities:

  • Retrieve ads by ID
  • Retrieve active ads for a placement (hot path)
  • Serialize ORM objects to API-safe dicts

Methods:

get_by_id(ad_id: int) -> dict | None

Retrieve a single ad by primary key.

from repository.ad_repository import AdRepository
 
repo = AdRepository(engine=db_engine)
ad = repo.get_by_id(ad_id=12345)
# Returns: {"ad_id": 12345, "tenant_id": 1, "ad_key": "ad_1001", ...} or None

get_active_by_placement(tenant_id: int, placement_id: int, limit: int = 20) -> list[dict]

Retrieve active ads for a placement (production hot path).

repo = AdRepository(engine=db_engine)
ads = repo.get_active_by_placement(
    tenant_id=1,
    placement_id=10,
    limit=20
)
# Returns: [{"ad_id": 1, ...}, {"ad_id": 2, ...}, ...]

Performance notes:

  • Uses indexed fields: (tenant_id, placement_id, status)
  • Results ordered by score_weight DESC, ad_id ASC for deterministic ranking
  • Limits to max 100 results to prevent over-fetching
  • Should be cached in Redis with TTL=300 seconds
  • Every query includes tenant_id filter (multi-tenancy isolation)

get_serving_ads(tenant_key: str, placement_ref: str, limit: int = 5) -> list[dict]

High-level serving endpoint combining placement + ad queries.

repo = AdRepository(engine=db_engine)
ads = repo.get_serving_ads(
    tenant_key="demo",
    placement_ref="homepage_top",  # Can also be an ad_key
    limit=5
)
# Returns: [{"ad_id": 1, ...}, {"ad_id": 2, ...}, ...]

PlacementRepository

Located in: repository/placement_repository.py

Responsibilities:

  • Retrieve active placements by key
  • Optional tenant filtering
  • Serialize ORM objects to API-safe dicts

Not responsible for:

  • Validating placement constraints (width/height)
  • Fetching associated formats (PlacementFormat table)
  • Fetching candidate ads (use AdRepository)
  • Caching (should migrate to RedisRepository)

Methods:

get_active_by_key(placement_key: str, tenant_id: int | None = None) -> dict | None

Retrieve an active placement by its external key.

from repository.placement_repository import PlacementRepository
 
repo = PlacementRepository(engine=db_engine)
placement = repo.get_active_by_key(
    placement_key="homepage_top",
    tenant_id=1
)
# Returns: {"placement_id": 10, "placement_key": "homepage_top", ...} or None

Performance notes:

  • Placement lookups should be cached in Redis (TTL=3600 seconds)
  • No joins; plain SELECT
  • Tenant filtering is currently optional but should become mandatory
  • Future versions will enforce tenant_id on all queries

Testing

Running Tests

# Run all tests
./run_unit_tests.sh
 
# Run specific test file
.venv/bin/python -m pytest tests/test_models.py -v
 
# Run with coverage report
.venv/bin/python -m pytest tests/ --cov=. --cov-report=html
 
# Run only integration tests (requires PostgreSQL)
.venv/bin/python -m pytest tests/test_api.py -v
 
# Run specific test function
.venv/bin/python -m pytest tests/test_api.py::test_get_placement -v

Test Structure

Unit Tests: tests/test_models.py

  • Test ORM field definitions and mappings
  • Verify model constraints and defaults
  • Test SQLAlchemy type system
  • No database required (SQLAlchemy type checks only)

Unit Tests: tests/test_repositories.py

  • Test repository query logic
  • Test serialization (_to_dict methods)
  • Test edge cases (NULL values, empty results, multi-tenancy)
  • Verifies queries return correct structure

Integration Tests: tests/test_api.py

  • Test HTTP endpoints via TestClient
  • Test request/response serialization
  • Verify multi-tenancy isolation
  • Test health checks and error handling
  • Requires PostgreSQL — creates a transactional test database

Model Metadata Tests: tests/test_model_metadata.py

  • Test ORM introspection
  • Verify table names, column names, relationships

Test Database

Tests use transactional fixtures (from conftest.py) that:

  1. Create a test database connection
  2. Start a SAVEPOINT before each test
  3. Roll back the SAVEPOINT after each test (no data persists)

Requirement: PostgreSQL must be running; SQLite is not compatible due to:

  • Identity columns (GENERATED ALWAYS AS IDENTITY)
  • Foreign key cascades
  • JSONB columns
  • Type-specific features

Pytest Fixtures

From tests/conftest.py:

  • test_engine: SQLAlchemy Engine connected to test database
  • test_session: Transactional session for queries
  • seed: Fixture providing sample tenant_id, campaign_id, creative_id
  • test_app: FastAPI test application
  • client: TestClient for making HTTP requests
  • sample_placement: Pre-seeded Placement object

Database Schema

Overview

All tables are in the leo_ads schema. The schema supports:

  • Multi-tenancy via tenant_id on every table
  • Flexible provider integration (Google Ads, Shopee, Lazada, etc.)
  • Extensible metadata (JSONB) on most tables
  • Historical tracking (created_at, updated_at)

Initialization

Initialize the database with:

./start.sh --seed-demo-customer360-promotions

This:

  1. Creates the leo_ads schema
  2. Creates all tables (from sql-scripts/db-schema-init.sql)
  3. Seeds demo data (from sql-scripts/sample-data-init.sql)
  4. Starts the API server

Connection Configuration

From .env:

c360_PROMOTION_DB_HOST=localhost
c360_PROMOTION_DB_PORT=5432
c360_PROMOTION_DB_USER=postgres
c360_PROMOTION_DB_PASSWORD=your_password
c360_PROMOTION_DB_NAME=customer360
c360_PROMOTION_DB_SCHEMA=leo_ads
c360_PROMOTION_DB_POOL_SIZE=10
c360_PROMOTION_DB_MAX_OVERFLOW=20
c360_PROMOTION_DB_ECHO_SQL=false  # Set to true for SQL debugging

Connection pooling:

  • Pool size: 10 connections (default)
  • Max overflow: 20 (temporary connections during spikes)
  • Recycle: 1800 seconds (30 min) — prevents stale connections
  • Pre-ping: Enabled — verifies connection health before reuse

Core Tables

leo_ads.tenant

  • Represents an isolated customer/tenant
  • All other tables reference tenant_id

leo_ads.advertiser

  • Brand/advertiser within a tenant
  • Campaigns optionally link to advertisers

leo_ads.source_account

  • External provider account (Google Ads, Shopee, etc.)
  • Stores provider credentials and configuration

leo_ads.source_asset

  • External assets from providers
  • Maps external IDs (Google campaign ID, Shopee offer ID) to local IDs

leo_ads.placement

  • Publisher inventory slots
  • Dimensions, responsiveness, and constraints
  • (tenant_id, placement_key) unique

leo_ads.placement_format

  • Supported formats for a placement
  • Stores width, height, responsive, and format-specific constraints

leo_ads.campaign

  • Business/buying configuration
  • Budget, objective, date range, status
  • (tenant_id, campaign_key) unique

leo_ads.creative

  • Reusable ad content and assets
  • Text, images, videos, metadata
  • Provider-specific payload in content_payload
  • (tenant_id, creative_key) unique

leo_ads.ad

  • Serving configuration: Campaign + Creative + Placement
  • Status, ranking weight, frequency cap
  • (tenant_id, ad_key) unique

Key Constraints

  1. Multi-tenancy: Every query must filter by tenant_id to prevent cross-tenant leakage
  2. Foreign keys:
    • Ad → Campaign (ON DELETE SET NULL, nullable)
    • Ad → Creative (ON DELETE RESTRICT, protected)
    • Ad → Placement (ON DELETE RESTRICT, protected)
  3. Unique keys: All external identifiers (*_key columns) are unique per tenant
  4. Indexes: Optimized for hot paths:
    • (tenant_id, placement_id, status) for ad serving
    • (tenant_id, placement_key) for placement lookups
    • (tenant_id, campaign_key), (tenant_id, creative_key), etc.

Important Notes

  • No soft deletes — use status = 'archived' instead
  • JSONB columns store provider-specific, extensible data
  • Type safety — use PostgreSQL types (BIGINT, DECIMAL, etc.) not application types
  • Concurrency — connection pooling handles multiple concurrent requests

Development Guidelines

Code Architecture

The promotions API follows clean layering to keep code maintainable and testable:

HTTP Layer (FastAPI endpoints) → Requests & Response serialization
    ↓
Repository Layer (persistence) → Database queries & ORM mapping
    ↓
SQLAlchemy ORM (models) → Object-relational mapping
    ↓
PostgreSQL (durability) → Source of truth

Separation of concerns:

  • Models (model/): ORM definitions only; no business logic
  • Repositories (repository/): Database queries only; no HTTP or business logic
  • Application (core/application.py): Routing, lifecycle, dependency injection
  • Handlers (in core/application.py): HTTP request/response; minimal logic
  • Services (future): Business logic (targeting, ranking, caching, etc.)

Adding a New Endpoint

  1. Add repository method for database access:

    # repository/ad_repository.py
def get_by_campaign(self, tenant_id: int, campaign_id: int) -> list[dict]:
    """Get all active ads for a campaign."""
    # Query logic...

  2. Add route handler in core/application.py:

    def get_campaign_ads(self, campaign_id: int):
    """GET /campaigns/{campaign_id}/ads"""
    return self.ad_repository.get_by_campaign(
        tenant_id=1,  # TODO: Extract from JWT/auth context
        campaign_id=campaign_id,
    )

  3. Register route:

    # In _register_routes()
application.add_api_route(
    "/campaigns/{campaign_id}/ads",
    self.get_campaign_ads,
    methods=["GET"],
    tags=["Ads"],
)

  4. Add tests:

    • Unit test in tests/test_repositories.py
    • Integration test in tests/test_api.py

  5. Update README with endpoint documentation

Code Quality Standards

  1. Always filter by tenant_id in queries to prevent cross-tenant data leakage

    # ✅ Correct
select(Ad).where(Ad.tenant_id == tenant_id, Ad.status == "active")
 
# ❌ Wrong
select(Ad).where(Ad.status == "active")  # Missing tenant_id!

  2. Use repositories for all database access (never write SQL in handlers)

    # ✅ Correct
def get_ad(self, ad_id: int):
    return self.ad_repository.get_by_id(ad_id)
 
# ❌ Wrong
def get_ad(self, ad_id: int):
    session = Session(bind=self.engine)
    return session.query(Ad).filter(Ad.ad_id == ad_id).one()

  3. Add docstrings to all public methods:

    def get_active_by_placement(
    self,
    tenant_id: int,
    placement_id: int,
    limit: int = 20,
) -> list[dict]:
    """
    Retrieve active ads for a placement.
    
    Args:
        tenant_id: Tenant to filter by (required for multi-tenancy)
        placement_id: Placement to fetch ads for
        limit: Max results (default: 20, max: 100)
    
    Returns:
        List of ad dicts with serialized fields
    """

  4. Use type hints on all function signatures:

    # ✅ Correct
def get_by_id(self, ad_id: int) -> dict | None:
    pass
 
# ❌ Wrong
def get_by_id(self, ad_id):
    pass

  5. Validate inputs in repositories and services:

    def get_by_id(self, ad_id: int) -> dict | None:
    if ad_id <= 0:
        raise ValueError("ad_id must be positive")
    # Query...

  6. Handle errors explicitly:

    # ✅ Correct - re-raise or handle
try:
    with self.engine.connect() as conn:
        conn.execute(text("SELECT 1"))
except Exception as e:
    logger.exception("Database connection failed")
    raise
 
# ❌ Wrong - silent failure
try:
    # ...
except Exception:
    pass

Multi-Tenancy Best Practices

  • Always validate tenant context before returning data
  • Use (tenant_id, *_key) composite unique constraints
  • Include tenant_id in all WHERE clauses
  • Never assume tenant context from user input alone (use JWT/auth)
  • Test multi-tenancy isolation in integration tests

Environment Variables

VariableDefaultPurpose
c360_PROMOTION_API_HOSTlocalhostAPI listen address
c360_PROMOTION_API_PORT9009API listen port
c360_PROMOTION_DB_HOSTlocalhostPostgreSQL host
c360_PROMOTION_DB_PORT5432PostgreSQL port
c360_PROMOTION_DB_USERpostgresPostgreSQL user
c360_PROMOTION_DB_PASSWORDpasswordPostgreSQL password
c360_PROMOTION_DB_NAMEcustomer360Database name
c360_PROMOTION_DB_SCHEMAleo_adsSchema name
c360_PROMOTION_DB_POOL_SIZE10Connection pool size
c360_PROMOTION_DB_MAX_OVERFLOW20Max overflow connections
c360_PROMOTION_DB_POOL_RECYCLE_SECONDS1800Connection recycle interval
c360_PROMOTION_DB_ECHO_SQLfalseEnable SQL logging (debug)
c360_PROMOTION_UVICORN_RELOADfalseEnable auto-reload (dev only)
c360_PROMOTION_CACHE_ENABLEDtrueEnable Redis caching
c360_PROMOTION_REDIS_HOSTlocalhostRedis host
c360_PROMOTION_REDIS_PORT6580Redis port
c360_PROMOTION_REDIS_DB0Redis database number
c360_PROMOTION_REDIS_PASSWORDunsetRedis password
c360_PROMOTION_CACHE_TTL_SECONDS60Cache TTL

Performance Tips

  1. Cache placements: Placement lookups have 90% read ratio → cache in Redis
  2. Cache ad candidates: Pre-compute placement_id → [ad_ids] in Redis
  3. Batch ad queries: Use get_active_by_placement instead of loop of get_by_id
  4. Connection pooling: SQLAlchemy uses configured pool (default: 10 connections, recycle every 30 min)
  5. Query optimization: All hot queries use indexed fields: (tenant_id, placement_id, status)

Production Deployment Checklist

  • Database is PostgreSQL 16+
  • Implement JWT/OAuth authentication middleware
  • Enforce tenant_id extraction from auth context
  • Add request logging and distributed tracing (OpenTelemetry)
  • Set up Redis caching for placement_key → Placement lookups
  • Implement ad ranking service (not just score_weight ordering)
  • Add pagination to list endpoints
  • Implement input validation with Pydantic models
  • Add rate limiting and DDoS protection
  • Set up monitoring, alerting, and health checks
  • Configure database connection pooling
  • Enable SSL/TLS for database and API
  • Rotate secrets (DB password, API keys)
  • Document SLAs and performance targets
  • Load test with realistic placement and ad volumes

Monitoring & Observability

Health Checks

API provides three health endpoints for monitoring:

# Service is running
curl http://localhost:9009/health
 
# Database is reachable
curl http://localhost:9009/health/database
 
# Basic info
curl http://localhost:9009/

Logging

Logs are written to logs/app.log with timestamps.

Troubleshooting

PostgreSQL connection error

PostgreSQL connection failed

Solution:

  • Ensure PostgreSQL is running: docker ps | grep postgres
  • Check credentials in .env
  • Verify c360_PROMOTION_DB_HOST is reachable: ping localhost

Schema not found

ERROR:  schema "leo_ads" does not exist

Solution:

./start.sh --seed-demo-customer360-promotions

This creates the schema and seeds demo data.

c360 Promotions already running

[API] Already running | PID 12345
[API] Stop first | ./stop.sh

Solution:

./stop.sh
./start.sh

Contributing

Code Style

  • Follow PEP 8
  • Use type hints
  • Document public APIs with docstrings
  • Add tests for new features

Pull Request Checklist

  • Tests pass: ./run_unit_tests.sh
  • Code is documented
  • Multi-tenancy is preserved (tenant_id filtering)
  • No SQL injection vectors (use parameterized queries)
  • README is updated if needed

License

See LICENSE file in repository root.

Graph View

Backlinks