06 - Architecture Documentation¶
Technical documentation covering the architecture, code organization, and internal workings of the API v2 system.
🏗️ Architecture Overview¶
The API v2 follows Atomic Design principles for a modular, maintainable, and scalable codebase:
API v2 Architecture
├── Atoms (Basic Building Blocks)
│ ├── Configuration (settings.py)
│ ├── Models (requests.py, responses.py)
│ └── Utilities (file_utils.py)
├── Molecules (Single-Purpose Components)
│ ├── TextSummarizer
│ ├── TTSService
│ ├── VideoService
│ ├── YouTubeService
│ └── ShortlinkService
├── Organisms (Complex Components)
│ └── ProductProcessingService
└── Templates/Pages (Application Layer)
└── FastAPI Application (main.py)
📁 Directory Structure¶
Complete Project Layout¶
apiv2/
├── main.py # FastAPI application entry point
├── requirements.txt # Python dependencies
├── .env # Environment variables
├── README.md # Project documentation
├── handover/ # Documentation folder
│ ├── README.md
│ ├── 01-installation.md
│ ├── 02-environment-setup.md
│ ├── 03-running-the-api.md
│ ├── 04-api-endpoints.md
│ ├── 05-troubleshooting.md
│ └── 06-architecture.md
├── internal/ # Core application code
│ ├── __init__.py
│ ├── config/ # Configuration atoms
│ │ ├── __init__.py
│ │ └── settings.py # Environment settings
│ ├── models/ # Data models atoms
│ │ ├── __init__.py
│ │ ├── requests.py # Pydantic request models
│ │ └── responses.py # Pydantic response models
│ ├── services/ # Business logic molecules
│ │ ├── __init__.py
│ │ ├── text_summarizer.py # AI text summarization
│ │ ├── tts_service.py # Text-to-speech conversion
│ │ ├── video_service.py # Video generation
│ │ ├── youtube_service.py # YouTube integration
│ │ ├── shortlink_service.py # URL shortening
│ │ └── product_service.py # Product processing orchestrator
│ └── utils/ # Utility atoms
│ ├── __init__.py
│ └── file_utils.py # File operations
├── sa/ # Service account credentials
│ ├── supradmin-0c4280905b78.json # Google Cloud service account
│ ├── oauth.json # YouTube OAuth2 configuration
│ └── youtube_token.pickle # YouTube authentication tokens
├── audio/ # Audio files storage
├── images/ # Image files storage
├── videos_output/ # Generated video files
├── tts/ # Generated TTS files
└── __pycache__/ # Python bytecode cache
🧩 Atomic Design Components¶
Atoms (Basic Building Blocks)¶
1. Configuration (internal/config/settings.py)¶
class Settings:
"""Centralized configuration management"""
# API Keys
GEMINI_API_KEY: str
ELEVENLABS_API_KEY: str
YOUTUBE_CLIENT_ID: str
YOUTUBE_CLIENT_SECRET: str
# Google Cloud Storage
GOOGLE_APPLICATION_CREDENTIALS: str
DEFAULT_BUCKET_NAME: str
# Service defaults
DEFAULT_VOICE_ID: str
DEFAULT_MODEL: str
# File paths
AUDIO_FOLDER: str
IMAGES_FOLDER: str
TTS_FOLDER: str
VIDEO_OUTPUT_FOLDER: str
# Methods
validate_required_settings()
setup_google_credentials()
create_directories()
2. Data Models (internal/models/)¶
Request Models (requests.py):
class SummarizeRequest(BaseModel):
text: str = Field(min_length=10, max_length=10000)
max_length: int = Field(default=100, ge=20, le=500)
min_length: int = Field(default=20, ge=10)
temperature: float = Field(default=0.7, ge=0.1, le=1.0)
class VideoGenerationRequest(BaseModel):
title: str = Field(min_length=1)
price: Optional[str] = None
video_format: str = Field(default="regular")
upload_to_youtube: bool = False
youtube_public: bool = False
Response Models (responses.py):
class SummaryResponse(BaseModel):
summary: str
class VideoGenerationResponse(BaseModel):
success: bool
message: str
video_path: Optional[str]
youtube_video_id: Optional[str]
# ... additional fields
Molecules (Single-Purpose Services)¶
1. TextSummarizer (internal/services/text_summarizer.py)¶
class TextSummarizer:
"""AI-powered text summarization using Google Gemini"""
def __init__(self):
self.model = genai.GenerativeModel(settings.DEFAULT_MODEL)
def summarize(self, text: str, max_length: int = 100, **kwargs) -> str:
"""Generate concise summary"""
def bullet_summary(self, text: str, num_points: int = 5) -> str:
"""Generate bullet point summary"""
def marketing_product_summary(self, product_data: dict, tone: str = "enthusiastic") -> str:
"""Generate marketing copy for products"""
2. TTSService (internal/services/tts_service.py)¶
class TTSService:
"""Text-to-speech conversion using ElevenLabs"""
def __init__(self):
self.api_key = settings.ELEVENLABS_API_KEY
self.voice_id = settings.DEFAULT_VOICE_ID
def generate_audio(self, text: str, filename: str) -> str:
"""Convert text to speech"""
def upload_to_gcs(self, local_path: str, bucket_name: str) -> str:
"""Upload audio to Google Cloud Storage"""
def get_signed_url(self, gcs_path: str, expiration: int = 3600) -> str:
"""Generate signed URL for audio access"""
3. VideoService (internal/services/video_service.py)¶
class VideoService:
"""Video generation from images and audio"""
def create_video_ffmpeg(self, image_path: str, audio_path: str, output_path: str, format_type: str = "regular") -> bool:
"""Create video using FFmpeg"""
def create_slideshow_ffmpeg(self, image_paths: list, audio_path: str, output_path: str, format_type: str = "regular") -> bool:
"""Create slideshow video from multiple images"""
def upload_video_to_gcs(self, local_path: str, bucket_name: str) -> str:
"""Upload video to Google Cloud Storage"""
4. YouTubeService (internal/services/youtube_service.py)¶
class YouTubeService:
"""YouTube API integration for video uploads"""
def get_auth_url(self) -> str:
"""Get OAuth2 authorization URL"""
def complete_authorization(self, auth_code: str) -> bool:
"""Complete OAuth2 flow with authorization code"""
def upload_video(self, video_path: str, product_data: dict, privacy_status: str = "private") -> dict:
"""Upload video to YouTube with metadata"""
5. ShortlinkService (internal/services/shortlink_service.py)¶
class ShortlinkService:
"""URL shortening using multiple providers"""
def create_shortlink(self, url: str) -> str:
"""Create shortened URL with fallback providers"""
def create_shortlink_ulvis(self, url: str) -> str:
"""Use ulvis.net provider"""
def create_shortlink_tinyurl(self, url: str) -> str:
"""Use tinyurl.com provider"""
Organisms (Complex Components)¶
ProductProcessingService (internal/services/product_service.py)¶
class ProductProcessingService:
"""Orchestrates multiple services for complete product processing"""
def __init__(self):
self.text_summarizer = TextSummarizer()
self.tts_service = TTSService()
self.video_service = VideoService()
self.youtube_service = YouTubeService()
async def process_batch_products(self, products: list, tone: str = "enthusiastic", max_length: int = 150) -> list:
"""Process multiple products with rate limiting"""
def generate_video_from_files(self, title: str, **kwargs) -> dict:
"""Complete video generation workflow"""
🔄 Data Flow Architecture¶
Request Processing Flow¶
1. FastAPI Request → 2. Pydantic Validation → 3. Service Layer → 4. External APIs → 5. Response Formation
┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Client │───▶│ FastAPI │───▶│ Services │───▶│ External │───▶│ Response │
│ Request │ │ Validation │ │ Logic │ │ APIs │ │ Formation │
└─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘ └─────────────┘
Service Interaction Diagram¶
┌─────────────────┐
│ FastAPI App │
└─────┬───────────┘
│
▼
┌─────────────────┐ ┌──────────────┐ ┌─────────────┐
│ ProductService │───▶│TextSummarizer│───▶│ Gemini API │
│ (Orchestrator) │ └──────────────┘ └─────────────┘
└─────┬───────────┘
│
├─────┬─────────────┬─────────────┬─────────────┐
▼ ▼ ▼ ▼ ▼
┌──────────┐ ┌─────────┐ ┌───────────┐ ┌───────────┐ ┌─────────────┐
│TTSService│ │VideoSvc │ │YouTubeSvc │ │ShortlinkSvc│ │ GCS │
└────┬─────┘ └────┬────┘ └─────┬─────┘ └─────┬─────┘ └─────────────┘
│ │ │ │
▼ ▼ ▼ ▼
┌──────────┐ ┌─────────┐ ┌───────────┐ ┌───────────┐
│ElevenLabs│ │ FFmpeg │ │YouTube API│ │ Ulvis.net │
│ API │ │ Process │ │ │ │ API │
└──────────┘ └─────────┘ └───────────┘ └───────────┘
🛠️ Technology Stack¶
Backend Framework¶
- FastAPI: Modern, fast web framework with automatic API documentation
- Uvicorn: ASGI server for production deployment
- Pydantic: Data validation and serialization
AI/ML Services¶
- Google Gemini: Large language model for text summarization
- ElevenLabs: Text-to-speech conversion
- Hugging Face: Alternative AI model access
Cloud Services¶
- Google Cloud Storage: File storage and hosting
- YouTube Data API v3: Video upload and management
- Google OAuth2: Authentication for YouTube
Media Processing¶
- FFmpeg: Video/audio processing and conversion
- MoviePy: Python video editing (fallback)
- PIL/Pillow: Image processing
Utilities¶
- python-dotenv: Environment variable management
- httpx: Async HTTP client
- aiofiles: Async file operations
🔐 Security Architecture¶
API Key Management¶
# Environment-based configuration
class Settings:
GEMINI_API_KEY: str = os.getenv("GEMINI_API")
ELEVENLABS_API_KEY: str = os.getenv("ELEVENLABS_API_KEY")
@classmethod
def validate_required_settings(cls):
"""Validate all required keys are present"""
File Security¶
- Service account JSON files stored in
sa/directory - File permissions:
chmod 600for sensitive files - No hardcoded credentials in source code
- Environment variables for all secrets
Authentication Flow¶
1. YouTube OAuth2 Flow:
Client → Auth URL → Google → Auth Code → API → Credentials
2. Service Account Flow:
API → Service Account JSON → Google Cloud → Authenticated Access
📊 Performance Considerations¶
Async Operations¶
# Async batch processing
async def process_batch_products(self, products: list) -> list:
tasks = []
for product in products:
task = asyncio.create_task(self.process_single_product(product))
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
Rate Limiting¶
- Batch processing limited to 10 products per request
- 1-hour cooldown between large batch operations
- API-specific rate limits respected (ElevenLabs, YouTube)
Resource Management¶
- Temporary file cleanup after processing
- Memory-efficient streaming for large files
- Connection pooling for HTTP requests
Caching Strategy¶
- Generated audio files cached in
audio/directory - Video files stored locally and in GCS
- Service account credentials cached in memory
🧪 Testing Architecture¶
Unit Testing Structure¶
tests/
├── test_services/
│ ├── test_text_summarizer.py
│ ├── test_tts_service.py
│ ├── test_video_service.py
│ └── test_youtube_service.py
├── test_models/
│ ├── test_requests.py
│ └── test_responses.py
└── test_config/
└── test_settings.py
Testing Commands¶
# Run all tests
python -m pytest
# Test specific service
python -m pytest tests/test_services/test_text_summarizer.py
# Test with coverage
python -m pytest --cov=internal
🔄 Deployment Architecture¶
Development Deployment¶
Production Deployment¶
# Multiple workers with Gunicorn
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000
# Or with Uvicorn directly
uvicorn main:app --workers 4 --host 0.0.0.0 --port 8000
Container Deployment (Docker)¶
FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
📈 Scalability Considerations¶
Horizontal Scaling¶
- Stateless service design allows multiple instances
- Shared file storage via Google Cloud Storage
- Database-free architecture (configuration-based)
Vertical Scaling¶
- FFmpeg processing benefits from more CPU cores
- Memory usage scales with concurrent requests
- Storage requirements grow with generated content
Load Balancing¶
upstream api_servers {
server 127.0.0.1:8000;
server 127.0.0.1:8001;
server 127.0.0.1:8002;
server 127.0.0.1:8003;
}
server {
location / {
proxy_pass http://api_servers;
}
}
🔧 Configuration Management¶
Environment-Based Configuration¶
# Development
ENVIRONMENT=development
DEBUG=true
LOG_LEVEL=DEBUG
# Production
ENVIRONMENT=production
DEBUG=false
LOG_LEVEL=INFO
WORKERS=4
Feature Flags¶
class Settings:
ENABLE_YOUTUBE_UPLOAD: bool = os.getenv("ENABLE_YOUTUBE_UPLOAD", "true").lower() == "true"
ENABLE_VIDEO_GENERATION: bool = os.getenv("ENABLE_VIDEO_GENERATION", "true").lower() == "true"
ENABLE_BATCH_PROCESSING: bool = os.getenv("ENABLE_BATCH_PROCESSING", "true").lower() == "true"
📝 Code Quality Standards¶
Type Hints¶
from typing import Optional, List, Dict, Any, Union
def summarize(self, text: str, max_length: int = 100) -> str:
"""Function with proper type hints"""
Error Handling¶
try:
result = external_api_call()
return {"success": True, "data": result}
except APIException as e:
logger.error(f"API error: {e}")
return {"success": False, "error": str(e)}
except Exception as e:
logger.exception("Unexpected error")
return {"success": False, "error": "Internal server error"}
Logging Strategy¶
import logging
logger = logging.getLogger(__name__)
class TextSummarizer:
def summarize(self, text: str) -> str:
logger.info(f"Summarizing text of length {len(text)}")
try:
result = self._process(text)
logger.info("Summarization completed successfully")
return result
except Exception as e:
logger.error(f"Summarization failed: {e}")
raise
🔮 Future Enhancements¶
Planned Features¶
- Database Integration: PostgreSQL for request logging and analytics
- Redis Caching: Response caching for repeated requests
- WebSocket Support: Real-time progress updates for long operations
- Admin Dashboard: Web interface for monitoring and management
- API Versioning: Support for multiple API versions
Scalability Improvements¶
- Queue System: Celery/Redis for background processing
- Microservices: Split into separate services for each function
- Container Orchestration: Kubernetes deployment
- CDN Integration: CloudFlare for static asset delivery
Security Enhancements¶
- API Authentication: JWT tokens for API access
- Rate Limiting: Redis-based rate limiting per client
- Input Sanitization: Enhanced validation and sanitization
- Audit Logging: Comprehensive request/response logging
📚 Documentation Standards¶
Code Documentation¶
- Docstrings for all public methods
- Type hints for all function parameters
- Inline comments for complex logic
- README files for each module
API Documentation¶
- OpenAPI/Swagger automatic generation
- Example requests and responses
- Error code documentation
- Rate limit information
🎯 Best Practices¶
Service Design¶
- Single responsibility principle
- Dependency injection
- Interface segregation
- Error handling consistency
File Organization¶
- Logical grouping by functionality
- Clear naming conventions
- Minimal circular dependencies
- Proper import structure
Performance¶
- Lazy loading of heavy dependencies
- Connection pooling
- Async operations where beneficial
- Resource cleanup
This architecture documentation provides the technical foundation for understanding, maintaining, and extending the API v2 system.