sema-api / app /api /v1 /endpoints.py
kamau1's picture
update: Fastapi codebase structure with api endpoints
a7d24e3
raw
history blame
18.2 kB
"""
API v1 endpoints
"""
import time
from fastapi import APIRouter, HTTPException, Request
from slowapi import Limiter
from slowapi.util import get_remote_address
from prometheus_client import generate_latest, CONTENT_TYPE_LATEST
from fastapi.responses import Response
from ...models.schemas import (
TranslationRequest,
TranslationResponse,
HealthResponse,
LanguagesResponse,
LanguageStatsResponse,
LanguageInfo
)
from ...services.translation import (
translate_with_detection,
translate_with_source,
models_loaded
)
from ...services.languages import (
get_all_languages,
get_languages_by_region,
get_language_info,
is_language_supported,
get_popular_languages,
get_african_languages,
search_languages,
get_language_statistics
)
from ...core.config import settings
from ...core.logging import get_logger
from ...core.metrics import TRANSLATION_COUNT, CHARACTER_COUNT, ERROR_COUNT
from ...utils.helpers import get_nairobi_time
logger = get_logger()
limiter = Limiter(key_func=get_remote_address)
# Application start time for uptime calculation
app_start_time = time.time()
# Create router
router = APIRouter()
@router.get(
"/",
response_model=HealthResponse,
tags=["Health & Monitoring"],
summary="Basic Health Check",
description="Quick health check endpoint that returns basic API status information."
)
async def root():
"""
## Basic Health Check
Returns essential API status information including:
- βœ… API operational status
- πŸ“¦ Model loading status
- ⏱️ System uptime
- 🏷️ API version
**Use this endpoint for:**
- Load balancer health checks
- Basic monitoring
- API availability verification
"""
uptime = time.time() - app_start_time
full_date, _ = get_nairobi_time()
return HealthResponse(
status="healthy" if models_loaded() else "degraded",
version=settings.app_version,
models_loaded=models_loaded(),
uptime=uptime,
timestamp=full_date
)
@router.get(
"/health",
response_model=HealthResponse,
tags=["Health & Monitoring"],
summary="Detailed Health Check",
description="Comprehensive health check with detailed system status for monitoring systems.",
responses={
200: {"description": "System is healthy"},
503: {"description": "System is unhealthy - models not loaded"}
}
)
async def health_check():
"""
## Detailed Health Check
Comprehensive health check endpoint designed for monitoring systems like:
- πŸ“Š Prometheus/Grafana
- 🚨 Alerting systems
- πŸ” APM tools
- πŸ₯ Health monitoring dashboards
**Returns detailed information about:**
- System health status
- Model loading status
- API uptime
- Timestamp information
**HTTP Status Codes:**
- `200`: All systems operational
- `503`: Service unavailable (models not loaded)
"""
uptime = time.time() - app_start_time
full_date, _ = get_nairobi_time()
# Perform additional health checks here
models_healthy = models_loaded()
return HealthResponse(
status="healthy" if models_healthy else "unhealthy",
version=settings.app_version,
models_loaded=models_healthy,
uptime=uptime,
timestamp=full_date
)
@router.get(
"/metrics",
tags=["Health & Monitoring"],
summary="Prometheus Metrics",
description="Prometheus-compatible metrics endpoint for monitoring and alerting.",
responses={
200: {"description": "Metrics in Prometheus format", "content": {"text/plain": {}}},
404: {"description": "Metrics disabled"}
}
)
async def get_metrics():
"""
## Prometheus Metrics
Returns metrics in Prometheus format for monitoring and alerting.
**Available Metrics:**
- πŸ“Š `sema_requests_total` - Total API requests by endpoint and status
- ⏱️ `sema_request_duration_seconds` - Request duration histogram
- 🌍 `sema_translations_total` - Translation count by language pair
- πŸ“ `sema_characters_translated_total` - Total characters translated
- ❌ `sema_errors_total` - Error count by type
**Integration Examples:**
```yaml
# Prometheus scrape config
scrape_configs:
- job_name: 'sema-api'
static_configs:
- targets: ['your-api-url:port']
metrics_path: '/metrics'
```
"""
if not settings.enable_metrics:
raise HTTPException(status_code=404, detail="Metrics disabled")
return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)
@router.post(
"/translate",
response_model=TranslationResponse,
tags=["Translation"],
summary="Translate Text",
description="Translate text between 200+ languages with automatic language detection.",
responses={
200: {"description": "Translation successful"},
400: {"description": "Invalid request - empty text or invalid language code"},
413: {"description": "Text too long - exceeds character limit"},
429: {"description": "Rate limit exceeded"},
500: {"description": "Translation service error"}
}
)
@limiter.limit(f"{settings.max_requests_per_minute}/minute")
async def translate_endpoint(
request: TranslationRequest,
http_request: Request
):
"""
## 🌍 Translate Text
Translate text between 200+ languages using state-of-the-art neural machine translation.
### ✨ Features
- **Automatic Language Detection**: Leave `source_language` empty for auto-detection
- **200+ Languages**: Full FLORES-200 language support
- **High Performance**: Optimized CTranslate2 inference engine
- **Usage Tracking**: Character count and request metrics
- **Request Tracking**: Unique request IDs for debugging
### πŸ”’ Limits & Constraints
- **Rate Limit**: 60 requests per minute per IP address
- **Character Limit**: Maximum 5000 characters per request
- **Language Codes**: Must use FLORES-200 format (e.g., `eng_Latn`, `swh_Latn`)
### πŸ“ Language Code Examples
| Language | Code | Example |
|----------|------|---------|
| English | `eng_Latn` | "Hello world" |
| Swahili | `swh_Latn` | "Habari ya dunia" |
| French | `fra_Latn` | "Bonjour le monde" |
| Kikuyu | `kik_Latn` | "WΔ© mwega?" |
| Spanish | `spa_Latn` | "Hola mundo" |
### πŸš€ Usage Examples
**Auto-detect source language:**
```json
{
"text": "Habari ya asubuhi",
"target_language": "eng_Latn"
}
```
**Specify source language:**
```json
{
"text": "Good morning",
"source_language": "eng_Latn",
"target_language": "swh_Latn"
}
```
### πŸ“Š Response Information
The response includes:
- Translated text
- Detected/provided source language
- Character count for usage tracking
- Inference time for performance monitoring
- Unique request ID for debugging
- Timestamp in Nairobi timezone
"""
request_id = http_request.state.request_id
# Validate text length
if len(request.text) > settings.max_text_length:
raise HTTPException(
status_code=413,
detail=f"Text too long. Maximum {settings.max_text_length} characters allowed."
)
full_date, _ = get_nairobi_time()
character_count = len(request.text)
# Log translation request
logger.info(
"translation_started",
request_id=request_id,
source_language=request.source_language,
target_language=request.target_language,
character_count=character_count
)
try:
if request.source_language:
# Use provided source language
translated_text, inference_time = translate_with_source(
request.text,
request.source_language,
request.target_language
)
source_lang = request.source_language
else:
# Auto-detect source language
source_lang, translated_text, inference_time = translate_with_detection(
request.text,
request.target_language
)
# Update metrics
TRANSLATION_COUNT.labels(
source_lang=source_lang,
target_lang=request.target_language
).inc()
CHARACTER_COUNT.inc(character_count)
# Log successful translation
logger.info(
"translation_completed",
request_id=request_id,
source_language=source_lang,
target_language=request.target_language,
character_count=character_count,
inference_time=inference_time
)
return TranslationResponse(
translated_text=translated_text,
source_language=source_lang,
target_language=request.target_language,
inference_time=inference_time,
character_count=character_count,
timestamp=full_date,
request_id=request_id
)
except Exception as e:
# Log translation error
logger.error(
"translation_failed",
request_id=request_id,
error=str(e),
error_type=type(e).__name__,
source_language=request.source_language,
target_language=request.target_language
)
# Update error metrics
ERROR_COUNT.labels(error_type="translation_error").inc()
raise HTTPException(
status_code=500,
detail="Translation service temporarily unavailable. Please try again later."
)
@router.get(
"/languages",
response_model=LanguagesResponse,
tags=["Languages"],
summary="Get All Supported Languages",
description="Retrieve a complete list of all supported languages with metadata."
)
async def get_languages():
"""
## 🌍 Get All Supported Languages
Returns a comprehensive list of all 200+ supported languages with detailed metadata.
### πŸ“‹ Response Information
Each language includes:
- **English Name**: Standard English name
- **Native Name**: Name in the language's native script
- **Region**: Geographic region (Africa, Europe, Asia, etc.)
- **Script**: Writing system (Latin, Arabic, Cyrillic, etc.)
### 🎯 Use Cases
- **Frontend Language Selectors**: Populate dropdown menus
- **API Integration**: Validate language codes before translation
- **Documentation**: Generate language support documentation
- **Analytics**: Track language usage patterns
### πŸ“Š Language Coverage
- **African Languages**: 25+ languages including Swahili, Hausa, Yoruba
- **European Languages**: 40+ languages including major EU languages
- **Asian Languages**: 80+ languages including Chinese, Japanese, Hindi
- **Middle Eastern**: 15+ languages including Arabic, Hebrew, Persian
- **Americas**: 30+ languages including indigenous languages
"""
languages = get_all_languages()
return LanguagesResponse(
languages={code: LanguageInfo(**info) for code, info in languages.items()},
total_count=len(languages)
)
@router.get(
"/languages/popular",
response_model=LanguagesResponse,
tags=["Languages"],
summary="Get Popular Languages",
description="Get the most commonly used languages for quick access."
)
async def get_popular_languages_endpoint():
"""
## ⭐ Get Popular Languages
Returns the most commonly requested languages for quick access and better UX.
### πŸ”₯ Included Languages
- **Global**: English, Spanish, French, German, Portuguese, Russian
- **Asian**: Chinese, Japanese, Korean, Hindi, Arabic
- **African**: Swahili, Hausa, Yoruba, Amharic, Somali, Kikuyu
### πŸ’‘ Perfect For
- **Quick Selection**: Show popular options first
- **Mobile Apps**: Reduced list for smaller screens
- **Default Options**: Pre-populate common language pairs
"""
languages = get_popular_languages()
return LanguagesResponse(
languages={code: LanguageInfo(**info) for code, info in languages.items()},
total_count=len(languages)
)
@router.get(
"/languages/african",
response_model=LanguagesResponse,
tags=["Languages"],
summary="Get African Languages",
description="Get all supported African languages."
)
async def get_african_languages_endpoint():
"""
## 🌍 Get African Languages
Returns all supported African languages - our specialty!
### 🎯 Featured African Languages
- **East Africa**: Swahili, Kikuyu, Luo, Amharic, Somali, Tigrinya
- **West Africa**: Hausa, Yoruba, Igbo, Wolof, Lingala
- **Southern Africa**: Zulu, Xhosa, Afrikaans, Tswana, Sotho, Shona
- **Central Africa**: Lingala, Umbundu
### ✨ Special Features
- High-quality translations for African languages
- Cultural context preservation
- Support for various scripts (Latin, Ethiopic)
"""
languages = get_african_languages()
return LanguagesResponse(
languages={code: LanguageInfo(**info) for code, info in languages.items()},
total_count=len(languages)
)
@router.get(
"/languages/region/{region}",
response_model=LanguagesResponse,
tags=["Languages"],
summary="Get Languages by Region",
description="Get all languages from a specific geographic region."
)
async def get_languages_by_region_endpoint(region: str):
"""
## πŸ—ΊοΈ Get Languages by Region
Filter languages by geographic region for targeted language support.
### 🌍 Available Regions
- **Africa**: African languages (Swahili, Hausa, Yoruba, etc.)
- **Europe**: European languages (English, French, German, etc.)
- **Asia**: Asian languages (Chinese, Japanese, Hindi, etc.)
- **Middle East**: Middle Eastern languages (Arabic, Hebrew, Persian, etc.)
- **Americas**: Languages from the Americas
### πŸ“ Usage Examples
```
GET /languages/region/Africa
GET /languages/region/Europe
GET /languages/region/Asia
```
"""
languages = get_languages_by_region(region)
if not languages:
raise HTTPException(
status_code=404,
detail=f"No languages found for region: {region}. Available regions: Africa, Europe, Asia, Middle East, Americas"
)
return LanguagesResponse(
languages={code: LanguageInfo(**info) for code, info in languages.items()},
total_count=len(languages)
)
@router.get(
"/languages/search",
response_model=LanguagesResponse,
tags=["Languages"],
summary="Search Languages",
description="Search for languages by name, native name, or language code."
)
async def search_languages_endpoint(q: str):
"""
## πŸ” Search Languages
Search for languages using flexible text matching.
### 🎯 Search Capabilities
- **English Names**: "Swahili", "French", "Chinese"
- **Native Names**: "Kiswahili", "FranΓ§ais", "δΈ­ζ–‡"
- **Language Codes**: "swh_Latn", "fra_Latn", "cmn_Hans"
- **Partial Matches**: "Span" matches "Spanish"
### πŸ’‘ Perfect For
- **Autocomplete**: Real-time language search
- **User Input**: Find languages by any name variation
- **Validation**: Check if a language exists
### πŸ“ Query Examples
```
GET /languages/search?q=Swahili
GET /languages/search?q=δΈ­ζ–‡
GET /languages/search?q=ara
```
"""
if not q or len(q.strip()) < 2:
raise HTTPException(
status_code=400,
detail="Search query must be at least 2 characters long"
)
languages = search_languages(q.strip())
return LanguagesResponse(
languages={code: LanguageInfo(**info) for code, info in languages.items()},
total_count=len(languages)
)
@router.get(
"/languages/stats",
response_model=LanguageStatsResponse,
tags=["Languages"],
summary="Get Language Statistics",
description="Get comprehensive statistics about supported languages."
)
async def get_language_stats():
"""
## πŸ“Š Language Statistics
Get comprehensive statistics about our language support coverage.
### πŸ“ˆ Statistics Include
- **Total Languages**: Complete count of supported languages
- **Regional Distribution**: Languages per geographic region
- **Script Coverage**: Number of writing systems supported
- **Detailed Breakdown**: Languages by region with counts
### 🎯 Use Cases
- **Analytics Dashboards**: Display language coverage metrics
- **Marketing Materials**: Showcase translation capabilities
- **API Documentation**: Provide coverage statistics
- **Business Intelligence**: Track language support growth
"""
stats = get_language_statistics()
return LanguageStatsResponse(**stats)
@router.get(
"/languages/{language_code}",
response_model=LanguageInfo,
tags=["Languages"],
summary="Get Language Information",
description="Get detailed information about a specific language."
)
async def get_language_info_endpoint(language_code: str):
"""
## πŸ” Get Language Information
Get detailed metadata about a specific language using its FLORES-200 code.
### πŸ“‹ Information Provided
- **English Name**: Standard English name
- **Native Name**: Name in native script
- **Region**: Geographic region
- **Script**: Writing system used
### 🎯 Use Cases
- **Language Validation**: Check if a code is supported
- **UI Display**: Show language names in interfaces
- **Documentation**: Generate language-specific docs
### πŸ“ Example Codes
```
GET /languages/swh_Latn # Swahili
GET /languages/eng_Latn # English
GET /languages/cmn_Hans # Chinese (Simplified)
```
"""
language_info = get_language_info(language_code)
if not language_info:
raise HTTPException(
status_code=404,
detail=f"Language code '{language_code}' not supported. Use /languages to see all supported languages."
)
return LanguageInfo(**language_info)