You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
81 lines
3.3 KiB
81 lines
3.3 KiB
import httpx |
|
from typing import Optional, Dict, List, Any |
|
import asyncio |
|
from datetime import datetime, date |
|
|
|
class APIClient: |
|
def __init__(self, base_url: str, auth_token: Optional[str] = None): |
|
self.base_url = base_url.rstrip('/') |
|
self.auth_token = auth_token |
|
self.client = httpx.AsyncClient(timeout=30.0, follow_redirects=True) |
|
|
|
def set_auth_token(self, token: str): |
|
"""Set authentication token""" |
|
self.auth_token = token |
|
|
|
@property |
|
def headers(self) -> Dict[str, str]: |
|
"""Get headers with authentication""" |
|
headers = {"Content-Type": "application/json"} |
|
if self.auth_token: |
|
headers["Authorization"] = f"Bearer {self.auth_token}" |
|
return headers |
|
|
|
async def _request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]: |
|
"""Make HTTP request""" |
|
url = f"{self.base_url}{endpoint}" |
|
kwargs.setdefault('headers', {}).update(self.headers) |
|
|
|
try: |
|
response = await self.client.request(method, url, **kwargs) |
|
response.raise_for_status() |
|
return response.json() |
|
except httpx.HTTPError as e: |
|
raise Exception(f"API request failed: {str(e)}") |
|
|
|
# Pietanze endpoints |
|
async def get_pietanze(self, skip: int = 0, limit: int = 100, search: str = None, allergeni: List[str] = None) -> Dict[str, Any]: |
|
"""Get pietanze with optional filters""" |
|
params = {"skip": skip, "limit": limit} |
|
if search: |
|
params["search"] = search |
|
if allergeni: |
|
params["allergeni"] = ",".join(allergeni) |
|
return await self._request("GET", "/api/v1/pietanze", params=params) |
|
|
|
async def get_pietanza(self, pietanza_id: int) -> Dict[str, Any]: |
|
"""Get single pietanza by ID""" |
|
return await self._request("GET", f"/api/v1/pietanze/{pietanza_id}") |
|
|
|
async def create_pietanza(self, pietanza_data: Dict[str, Any]) -> Dict[str, Any]: |
|
"""Create new pietanza""" |
|
return await self._request("POST", "/api/v1/pietanze", json=pietanza_data) |
|
|
|
async def update_pietanza(self, pietanza_id: int, pietanza_data: Dict[str, Any]) -> Dict[str, Any]: |
|
"""Update existing pietanza""" |
|
return await self._request("PUT", f"/api/v1/pietanze/{pietanza_id}", json=pietanza_data) |
|
|
|
async def delete_pietanza(self, pietanza_id: int) -> None: |
|
"""Delete pietanza""" |
|
await self._request("DELETE", f"/api/v1/pietanze/{pietanza_id}") |
|
|
|
# Dashboard/stats endpoints |
|
async def get_dashboard_stats(self) -> Dict[str, Any]: |
|
"""Get dashboard statistics""" |
|
return await self._request("GET", "/api/v1/stats/dashboard") |
|
|
|
async def get_prenotazioni_today(self) -> List[Dict[str, Any]]: |
|
"""Get today's prenotazioni""" |
|
today = date.today().isoformat() |
|
return await self._request("GET", f"/api/v1/prenotazioni/by-date/{today}") |
|
|
|
async def get_pasti_disponibili(self) -> List[Dict[str, Any]]: |
|
"""Get available pasti""" |
|
return await self._request("GET", "/api/v1/pasti", params={"disponibile": True}) |
|
|
|
def __del__(self): |
|
"""Cleanup client""" |
|
try: |
|
asyncio.create_task(self.client.aclose()) |
|
except: |
|
pass
|
|
|