fix: consolidate coordinator-api entrypoints
Some checks failed
Deploy to Testnet / deploy-testnet (push) Waiting to run
Integration Tests / test-service-integration (push) Waiting to run
Multi-Node Stress Testing / stress-test (push) Waiting to run
Node Failover Simulation / failover-test (push) Waiting to run
API Endpoint Tests / test-api-endpoints (push) Successful in 22s
Cross-Node Transaction Testing / transaction-test (push) Successful in 15s
Python Tests / test-python (push) Failing after 15s
Security Scanning / security-scan (push) Successful in 18s
Deploy to Testnet / notify-deployment (push) Has been cancelled

- Fix app.py to properly import FastAPI app from main.py
- Remove unused python_13_optimized.py variant
- Ensure single clean entrypoint for coordinator-api service
- Systemd wrapper uses app.main:app which now properly resolves
This commit is contained in:
aitbc
2026-04-30 09:11:48 +02:00
parent 750f81a098
commit 7596ef1c1a
2 changed files with 2 additions and 259 deletions

View File

@@ -3,3 +3,5 @@ import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from .main import app

View File

@@ -1,259 +0,0 @@
"""
Python 3.13.5 Optimized FastAPI Application
This demonstrates how to leverage Python 3.13.5 features
in the AITBC Coordinator API for improved performance and maintainability.
"""
import time
from contextlib import asynccontextmanager
from typing import TypeVar, override
from fastapi import FastAPI, Request
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from .config import settings
from .storage import init_db
# ============================================================================
# Python 13.5 Type Parameter Defaults for Generic Middleware
# ============================================================================
T = TypeVar("T")
class GenericMiddleware[T]:
"""Generic middleware base class using Python 3.13 type parameter defaults"""
def __init__(self, app: FastAPI) -> None:
self.app = app
self.metrics: list[T] = []
async def record_metric(self, metric: T) -> None:
"""Record performance metric"""
self.metrics.append(metric)
@override
async def __call__(self, scope: dict, receive, send) -> None:
"""Generic middleware call method"""
start_time = time.time()
# Process request
await self.app(scope, receive, send)
# Record performance metric
end_time = time.time()
processing_time = end_time - start_time
await self.record_metric(processing_time)
# ============================================================================
# Performance Monitoring Middleware
# ============================================================================
class PerformanceMiddleware:
"""Performance monitoring middleware using Python 3.13 features"""
def __init__(self, app: FastAPI) -> None:
self.app = app
self.request_times: list[float] = []
self.error_count = 0
self.total_requests = 0
async def __call__(self, scope: dict, receive, send) -> None:
start_time = time.time()
# Track request
self.total_requests += 1
try:
await self.app(scope, receive, send)
except Exception:
self.error_count += 1
raise
finally:
# Record performance
end_time = time.time()
processing_time = end_time - start_time
self.request_times.append(processing_time)
# Keep only last 1000 requests to prevent memory issues
if len(self.request_times) > 1000:
self.request_times = self.request_times[-1000:]
def get_stats(self) -> dict:
"""Get performance statistics"""
if not self.request_times:
return {"total_requests": self.total_requests, "error_rate": 0.0, "avg_response_time": 0.0}
avg_time = sum(self.request_times) / len(self.request_times)
error_rate = (self.error_count / self.total_requests) * 100
return {
"total_requests": self.total_requests,
"error_rate": error_rate,
"avg_response_time": avg_time,
"max_response_time": max(self.request_times),
"min_response_time": min(self.request_times),
}
# ============================================================================
# Enhanced Error Handler with Python 3.13 Features
# ============================================================================
class EnhancedErrorHandler:
"""Enhanced error handler using Python 3.13 improved error messages"""
def __init__(self, app: FastAPI) -> None:
self.app = app
self.error_log: list[dict] = []
async def __call__(self, request: Request, call_next):
try:
return await call_next(request)
except RequestValidationError as exc:
# Python 3.13 provides better error messages
error_detail = {
"type": "validation_error",
"message": "Validation failed",
"errors": exc.errors() if hasattr(exc, "errors") else [],
"timestamp": time.time(),
"path": request.url.path,
"method": request.method,
}
self.error_log.append(error_detail)
return JSONResponse(status_code=422, content={"detail": error_detail})
except Exception as exc:
# Enhanced error logging
error_detail = {
"type": "internal_error",
"message": "Internal error occurred",
"timestamp": time.time(),
"path": request.url.path,
"method": request.method,
}
self.error_log.append(error_detail)
return JSONResponse(status_code=500, content={"detail": "Internal server error"})
# ============================================================================
# Optimized Application Factory
# ============================================================================
def create_optimized_app() -> FastAPI:
"""Create FastAPI app with Python 3.13.5 optimizations"""
# Initialize database
init_db()
# Create FastAPI app
app = FastAPI(
title="AITBC Coordinator API",
description="Python 3.13.5 Optimized AITBC Coordinator API",
version="1.0.0",
python_version="3.13.5+",
)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=settings.allow_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Add performance monitoring
performance_middleware = PerformanceMiddleware(app)
app.middleware("http")(performance_middleware)
# Add enhanced error handling
error_handler = EnhancedErrorHandler(app)
app.middleware("http")(error_handler)
# Add performance monitoring endpoint
@app.get("/v1/performance")
async def get_performance_stats():
"""Get performance statistics"""
return performance_middleware.get_stats()
# Add health check with enhanced features
@app.get("/v1/health")
async def health_check():
"""Enhanced health check with Python 3.13 features"""
return {
"status": "ok",
"env": settings.app_env,
"python_version": "3.13.5+",
"database": "connected",
"performance": performance_middleware.get_stats(),
"timestamp": time.time(),
}
# Add error log endpoint for debugging
@app.get("/v1/errors")
async def get_error_log():
"""Get recent error logs for debugging"""
error_handler = error_handler
return {"recent_errors": error_handler.error_log[-10:], "total_errors": len(error_handler.error_log)} # Last 10 errors
return app
# ============================================================================
# Async Context Manager for Database Operations
# ============================================================================
@asynccontextmanager
async def get_db_session():
"""Async context manager for database sessions using Python 3.13 features"""
from .storage.db import get_session
async with get_session() as session:
try:
yield session
finally:
# Session is automatically closed by context manager
pass
# ============================================================================
# Example Usage
# ============================================================================
async def demonstrate_optimized_features():
"""Demonstrate Python 3.13.5 optimized features"""
create_optimized_app()
print("🚀 Python 3.13.5 Optimized FastAPI Features:")
print("=" * 50)
print("✅ Enhanced error messages for debugging")
print("✅ Performance monitoring middleware")
print("✅ Generic middleware with type safety")
print("✅ Async context managers")
print("✅ @override decorators for method safety")
print("✅ 5-10% performance improvements")
print("✅ Enhanced security features")
print("✅ Better memory management")
if __name__ == "__main__":
import uvicorn
# Create and run optimized app
app = create_optimized_app()
print("🚀 Starting Python 3.13.5 optimized AITBC Coordinator API...")
uvicorn.run(app, host="127.0.0.1", port=8000, log_level="info")