Adriano
dbdbb77daf
Struttura monorepo completa con server FastAPI e client Flask: - Server: FastAPI + SQLAlchemy 2.0 async + Alembic migrations - Client: Flask + blueprints (auth, measure, maker, statistics) - Database: docker-compose MySQL 8.0 + Alembic async config - Config: pydantic-settings, TailwindCSS, Flask-Babel i18n - Piano implementazione completo (18 sezioni, 1600 righe) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
81 lines
2.4 KiB
Python
81 lines
2.4 KiB
Python
"""API Client - wrapper for HTTP requests to FastAPI server."""
|
|
from typing import Any
|
|
|
|
import requests
|
|
from flask import session
|
|
|
|
from config import Config
|
|
|
|
|
|
class APIClient:
|
|
"""HTTP client for TieMeasureFlow API server."""
|
|
|
|
def __init__(self):
|
|
self.base_url = Config.API_SERVER_URL.rstrip("/")
|
|
self.timeout = 30
|
|
|
|
@property
|
|
def _headers(self) -> dict[str, str]:
|
|
"""Build request headers with API key from session."""
|
|
headers = {"Content-Type": "application/json"}
|
|
api_key = session.get("api_key")
|
|
if api_key:
|
|
headers["X-API-Key"] = api_key
|
|
return headers
|
|
|
|
def get(self, endpoint: str, params: dict | None = None) -> dict[str, Any]:
|
|
"""GET request to API server."""
|
|
response = requests.get(
|
|
f"{self.base_url}{endpoint}",
|
|
headers=self._headers,
|
|
params=params,
|
|
timeout=self.timeout,
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def post(self, endpoint: str, data: dict | None = None, files: dict | None = None) -> dict[str, Any]:
|
|
"""POST request to API server."""
|
|
if files:
|
|
headers = {"X-API-Key": session.get("api_key", "")}
|
|
response = requests.post(
|
|
f"{self.base_url}{endpoint}",
|
|
headers=headers,
|
|
data=data,
|
|
files=files,
|
|
timeout=self.timeout,
|
|
)
|
|
else:
|
|
response = requests.post(
|
|
f"{self.base_url}{endpoint}",
|
|
headers=self._headers,
|
|
json=data,
|
|
timeout=self.timeout,
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def put(self, endpoint: str, data: dict | None = None) -> dict[str, Any]:
|
|
"""PUT request to API server."""
|
|
response = requests.put(
|
|
f"{self.base_url}{endpoint}",
|
|
headers=self._headers,
|
|
json=data,
|
|
timeout=self.timeout,
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def delete(self, endpoint: str) -> dict[str, Any]:
|
|
"""DELETE request to API server."""
|
|
response = requests.delete(
|
|
f"{self.base_url}{endpoint}",
|
|
headers=self._headers,
|
|
timeout=self.timeout,
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
api_client = APIClient()
|