Macumba Travel Backend - FastAPI AI Travel Recommendation Engine¶

HTTP Request → FastAPI Router → Dependency Injection → Service Layer → AI/External APIs → Database/Cache → Response

┌─────────────────┐
│   API Layer     │  FastAPI routers with automatic docs
│                 │  /api/v1/travel/recommendations
└─────────────────┘
         │
         ▼
┌─────────────────┐
│  Dependencies   │  Dependency injection for services
│                 │  Rate limiting, authentication
└─────────────────┘
         │
         ▼
┌─────────────────┐
│ Service Layer   │  Business logic and AI integration
│                 │  TravelService, AIService, etc.
└─────────────────┘
         │
         ▼
┌─────────────────┐
│   Data Layer    │  Models, schemas, database
│                 │  SQLAlchemy ORM with PostgreSQL
└─────────────────┘

fastapi-backend/
├── app/
│   ├── main.py                    # FastAPI application entry point
│   │
│   ├── api/                       # API layer - HTTP endpoints
│   │   ├── deps.py               # Dependency injection (auth, rate limiting)
│   │   ├── deps_services.py      # Service dependency injection
│   │   └── v1/
│   │       ├── api.py            # Main API router configuration
│   │       └── endpoints/        # Individual endpoint modules
│   │           ├── recommendations.py  # Travel recommendations API
│   │           ├── auth.py             # Authentication endpoints
│   │           ├── users.py            # User management
│   │           └── saved_trips.py      # Trip saving/retrieval
│   │
│   ├── core/                      # Core configuration and setup
│   │   ├── config.py             # Environment configuration (Pydantic)
│   │   ├── db.py                 # Database connection setup
│   │   ├── security.py           # JWT token handling
│   │   └── enums.py              # Application enumerations
│   │
│   ├── models/                    # Database models (SQLAlchemy)
│   │   ├── user.py               # User account model
│   │   ├── saved_trip.py         # Saved trip model with JSONB data
│   │   ├── travel.py             # Travel-related models
│   │   └── associations.py       # Many-to-many relationships
│   │
│   ├── schemas/                   # API request/response schemas (Pydantic)
│   │   ├── user.py               # User registration/profile schemas
│   │   ├── token.py              # Authentication token schemas
│   │   ├── destination.py        # Destination and recommendation schemas
│   │   └── saved_trip.py         # Trip saving schemas
│   │
│   ├── services/                  # Business logic layer
│   │   ├── travel_services.py    # Main travel orchestration service
│   │   ├── recommendations.py    # Legacy recommendation service
│   │   │
│   │   ├── ai/                   # AI service abstraction layer
│   │   │   ├── service.py        # Main AI service with provider switching
│   │   │   ├── base.py           # Base AI provider interface
│   │   │   └── providers/        # Individual AI provider implementations
│   │   │       ├── claude.py     # Anthropic Claude integration
│   │   │       └── gemini.py     # Google Gemini integration
│   │   │
│   │   ├── cache/                # Caching infrastructure
│   │   │   ├── cache_factory.py  # Cache abstraction with multiple backends
│   │   │   ├── postgres_cache.py # PostgreSQL-based cache implementation
│   │   │   ├── file_cache.py     # File-based cache for development
│   │   │   ├── redis_manager.py  # Redis cache (legacy, optional)
│   │   │   ├── decorators.py     # @cached decorator for functions
│   │   │   ├── interface.py      # Cache interface definition
│   │   │   └── key_generator.py  # Consistent cache key generation
│   │   │
│   │   ├── external/             # External API integrations
│   │   │   ├── google_maps.py    # Google Maps/Places API
│   │   │   ├── amadeus.py        # Flight data (Amadeus API)
│   │   │   ├── weather.py        # Weather API integration
│   │   │   └── exchange_rate_service.py  # Currency conversion
│   │   │
│   │   ├── image_service.py      # Destination image fetching
│   │   ├── weather_service.py    # Weather data service
│   │   └── exceptions.py         # Custom exception classes
│   │
│   └── utils/                     # Utility functions
│       ├── logging.py            # Structured logging setup
│       ├── middleware.py         # Custom FastAPI middleware
│       └── exceptions.py         # Global exception handlers
│
├── migrations/                    # Alembic database migrations
│   ├── versions/                 # Individual migration files
│   ├── env.py                    # Alembic environment configuration
│   └── script.py.mako           # Migration template
│
├── tests/                        # Test suite
│   ├── api/                      # API endpoint tests
│   ├── services/                 # Service layer tests
│   ├── integration/              # Integration tests
│   └── conftest.py              # Test configuration and fixtures
│
├── scripts/                      # Utility scripts
│   ├── setup_db.py              # Database initialization
│   └── populate_airports.py     # Seed airport data
│
├── docker-compose.yml           # Local development environment
├── Dockerfile                   # Container configuration
├── pyproject.toml              # Poetry dependencies and config
├── requirements.txt            # Pip dependencies
└── alembic.ini                 # Alembic configuration

# app/services/ai/service.py
class AIService:
    def __init__(self):
        self.primary_provider = ClaudeProvider()    # Primary: Anthropic Claude
        self.fallback_provider = GeminiProvider()   # Fallback: Google Gemini

    async def get_recommendations(self, **params):
        try:
            return await self.primary_provider.get_recommendations(**params)
        except Exception:
            # Automatic failover to Gemini if Claude fails
            return await self.fallback_provider.get_recommendations(**params)

# app/services/travel_services.py
class TravelService:
    async def get_initial_recommendations(self, budget, duration, departure_city, **params):
        # 1. Calculate traveler-specific budget
        total_travelers = num_adults + num_children
        per_person_budget = budget / total_travelers

        # 2. Get AI recommendations with traveler count
        raw_recs = await self.ai_service.get_recommendations(
            budget=budget,
            travelers=total_travelers,  # Critical for proper budget calculation!
            **params
        )

        # 3. Filter recommendations within budget
        filtered_recs = self._filter_by_budget(raw_recs, budget, total_travelers)

        # 4. Enrich with images and external data
        await self._fetch_and_attach_images(filtered_recs)

        return filtered_recs

# PostgreSQL caching decorator (default)
@cached(prefix="travel_recs", ttl=timedelta(hours=1), key_params=["budget", "departure_city"])
async def get_travel_recommendations(self, budget, departure_city, **params):
    # Expensive AI call only happens if not cached
    return await self.ai_service.get_recommendations(**params)

# User model with authentication
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, index=True)
    username = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    query_count = Column(Integer, default=0)  # Track API usage

    # Relationship to saved trips
    saved_trips = relationship("SavedTrip", back_populates="user")

# Saved trip with flexible JSONB storage
class SavedTrip(Base):
    __tablename__ = "saved_trips"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    user_id = Column(Integer, ForeignKey("users.id"))
    name = Column(String)

    # JSONB columns for flexible data storage
    search_params = Column(JSONB)      # Original search criteria
    results = Column(JSONB)            # Simple recommendations
    enriched_results = Column(JSONB)   # Full enriched data

    user = relationship("User", back_populates="saved_trips")

# app/api/deps.py
def check_query_limit(request: Request, current_user: Optional[User] = None):
    if current_user:
        return  # Authenticated users have unlimited access

    client_id = get_client_identifier(request)

    if client_id in ip_query_tracker:
        if ip_query_tracker[client_id]["count"] >= settings.RATE_LIMIT_UNAUTHENTICATED:
            raise HTTPException(status_code=429, detail="Rate limit exceeded")

# Start all services (PostgreSQL, FastAPI)
cd fastapi-backend
docker compose up

# The backend will be available at:
# - API: http://localhost:8000
# - Docs: http://localhost:8000/docs
# - Database: localhost:5432 (includes cache_entries table)

# Run database migrations
docker compose exec backend alembic upgrade head

# Initialize database with seed data
docker compose exec backend python scripts/setup_db.py
docker compose exec backend python scripts/populate_airports.py

# Run tests
docker compose exec backend pytest

# Access Python shell
docker compose exec backend python

# Check installed package versions
docker compose exec backend python -c "import fastapi; print(f'FastAPI: {fastapi.__version__}')"

# View logs
docker compose logs backend -f

# Create virtual environment
python -m venv venv
source venv/bin/activate  # Windows: .\venv\Scripts\activate

# Install dependencies
poetry install
# OR
pip install -r requirements.txt

# Set up environment variables
cp .env.example .env
# Edit .env with your API keys

# Run database migrations
alembic upgrade head

# Initialize database with seed data
python scripts/setup_db.py
python scripts/populate_airports.py

# Start development server
uvicorn app.main:app --reload

# Database
DATABASE_URL=postgresql://postgres:password@localhost:5432/traveldb
POSTGRES_USER=postgres
POSTGRES_PASSWORD=password
POSTGRES_DB=traveldb

# Cache Configuration
CACHE_TYPE=postgres  # Options: postgres, file, redis
CACHE_ENABLED=true

# AI Services
AI_PROVIDER=gemini  # or "claude"
ANTHROPIC_API_KEY=sk-ant-api03-...
GOOGLE_AI_API_KEY=AIzaSy...
GOOGLE_AI_MODEL=models/gemini-2.0-flash

# External APIs
GOOGLE_MAPS_API_KEY=AIzaSy...
WEATHER_API_KEY=b626a390f...
AMADEUS_API_KEY=3Bo4eVi40W...
AMADEUS_API_SECRET=Y48KIATb7n...

# JWT Authentication
JWT_SECRET=your-secret-key
ACCESS_TOKEN_EXPIRE_MINUTES=30

# Application
PROJECT_NAME=TravelSage API
API_V1_STR=/api/v1
APP_ENV=development

# Run all tests
pytest

# Run with coverage
pytest --cov=app tests/

# Run specific test file
pytest tests/api/test_recommendations.py

# Run specific test
pytest tests/api/test_recommendations.py::test_get_recommendations_success

tests/
├── conftest.py              # Test configuration and fixtures
├── api/                     # API endpoint tests
│   ├── test_recommendations.py
│   ├── test_auth.py
│   └── test_users.py
├── services/                # Service layer tests
│   ├── test_ai_service.py
│   ├── test_travel_service.py
│   └── test_cache.py
└── integration/             # End-to-end tests
    └── test_user_journey.py

# API endpoint test
def test_get_recommendations_success(client, sample_request):
    response = client.post("/api/v1/travel/recommendations", json=sample_request)
    assert response.status_code == 200
    data = response.json()
    assert len(data) > 0
    assert "destination" in data[0]

# Service layer test with mocking
@pytest.mark.asyncio
async def test_ai_service_with_fallback(mock_claude_error, mock_gemini_success):
    ai_service = AIService()
    result = await ai_service.get_recommendations(budget=1000, travelers=2)

    # Should use Gemini after Claude fails
    assert result is not None
    mock_gemini_success.assert_called_once()

# Auto-generate migration from model changes
alembic revision --autogenerate -m "Add new field to user model"

# Create empty migration for custom SQL
alembic revision -m "Add custom index"

# Apply migrations
alembic upgrade head

# Rollback one migration
alembic downgrade -1

# Production database (Cloud SQL)
DATABASE_URL=postgresql://user:pass@host:5432/prod_db

# Cache Configuration (PostgreSQL cache saves $110/month vs Redis)
CACHE_TYPE=postgres

# Production AI keys
ANTHROPIC_API_KEY=sk-ant-prod-...
GOOGLE_AI_API_KEY=AIzaSy-prod-...

# Security
JWT_SECRET=production-secret-key
APP_ENV=production

# External APIs (production keys)
GOOGLE_MAPS_API_KEY=AIzaSy-prod-maps-...

# .github/workflows/deploy.yml
- name: Deploy to Cloud Run
  run: |
    gcloud run deploy macumba-backend \
      --image gcr.io/$PROJECT_ID/macumba-backend:$GITHUB_SHA \
      --platform managed \
      --region us-central1 \
      --set-env-vars APP_ENV=production

# app/utils/logging.py
logger = get_logger("app.services.travel")

# Usage in services
logger.info("Processing recommendation request", 
           departure_city=departure_city, 
           budget=budget, 
           travelers=total_travelers)

logger.error("AI service failed", 
            provider="claude",
            error=str(e),
            request_id=request.headers.get("x-request-id"))

# Application health
curl http://localhost:8000/health

# AI service health
curl http://localhost:8000/api/v1/travel/rate-limit

# Database health (via logs)
docker-compose logs backend | grep "Database"

# Check logs
docker-compose logs backend

# Connect to database
docker-compose exec db psql -U postgres -d traveldb

# Check PostgreSQL cache
docker-compose exec backend python -c "
from app.services.cache.cache_factory import CacheFactory
import asyncio
cache = CacheFactory.create_cache()
keys = asyncio.run(cache.keys('travel_recs:*'))
print('Cache keys:', keys)
"

# Test AI service directly
docker-compose exec backend python -c "
from app.services.ai.service import AIService
import asyncio
ai = AIService()
result = asyncio.run(ai.get_recommendations(budget=1000, travelers=2, departure_city='Sydney'))
print(result)
"

# Get travel recommendations
POST /api/v1/travel/recommendations
Content-Type: application/json
{
  "budget": 1500,
  "from_date": "2025-07-15",
  "to_date": "2025-07-22",
  "departure_city": "Sydney, Australia",
  "preferences": ["beach", "culture"],
  "num_adults": 2,
  "num_children": 1,
  "with_pets": false
}

# Enrich a recommendation with detailed data
GET /api/v1/travel/{recommendation_id}/enrich

# Save a recommendation as a trip (requires auth)
POST /api/v1/travel/{recommendation_id}/save
Authorization: Bearer <token>
{
  "name": "My Gold Coast Trip",
  "notes": "Family vacation",
  "is_public": false
}

# User authentication
POST /api/v1/auth/login
POST /api/v1/auth/register

# Check rate limit status
GET /api/v1/travel/rate-limit