141 lines
5.8 KiB
Python
141 lines
5.8 KiB
Python
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.core.config import settings
|
|
|
|
|
|
class ApiClient:
|
|
def __init__(self) -> None:
|
|
self.base_url = settings.api_base_url.rstrip("/")
|
|
|
|
def headers(self, telegram_id: int | None = None) -> dict[str, str]:
|
|
headers = {"X-Internal-API-Token": settings.internal_api_token}
|
|
if telegram_id is not None:
|
|
headers["X-Telegram-User-Id"] = str(telegram_id)
|
|
return headers
|
|
|
|
async def request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
telegram_id: int | None = None,
|
|
json: dict[str, Any] | None = None,
|
|
params: dict[str, Any] | None = None,
|
|
) -> Any:
|
|
async with httpx.AsyncClient(base_url=self.base_url, timeout=15) as client:
|
|
response = await client.request(
|
|
method,
|
|
path,
|
|
json=json,
|
|
params=params,
|
|
headers=self.headers(telegram_id),
|
|
)
|
|
response.raise_for_status()
|
|
if response.status_code == 204:
|
|
return None
|
|
return response.json()
|
|
|
|
async def upsert_user(self, telegram_user: Any) -> dict[str, Any]:
|
|
payload = {
|
|
"telegram_id": telegram_user.id,
|
|
"username": telegram_user.username,
|
|
"first_name": telegram_user.first_name,
|
|
"last_name": telegram_user.last_name,
|
|
}
|
|
async with httpx.AsyncClient(base_url=self.base_url, timeout=10) as client:
|
|
response = await client.post("/api/users", json=payload, headers=self.headers())
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def list_cars(self, owner_id: int, telegram_id: int) -> list[dict[str, Any]]:
|
|
async with httpx.AsyncClient(base_url=self.base_url, timeout=10) as client:
|
|
response = await client.get(
|
|
"/api/cars", params={"owner_id": owner_id}, headers=self.headers(telegram_id)
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def create_car(self, owner_id: int, name: str, telegram_id: int) -> dict[str, Any]:
|
|
async with httpx.AsyncClient(base_url=self.base_url, timeout=10) as client:
|
|
response = await client.post(
|
|
"/api/cars",
|
|
json={"owner_id": owner_id, "name": name},
|
|
headers=self.headers(telegram_id),
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def stats(self, car_id: int, telegram_id: int) -> dict[str, Any]:
|
|
async with httpx.AsyncClient(base_url=self.base_url, timeout=10) as client:
|
|
response = await client.get(f"/api/cars/{car_id}/stats", headers=self.headers(telegram_id))
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
async def create_fuel(self, telegram_id: int, payload: dict[str, Any]) -> dict[str, Any]:
|
|
return await self.request("POST", "/api/fuel", telegram_id=telegram_id, json=payload)
|
|
|
|
async def create_service(self, telegram_id: int, payload: dict[str, Any]) -> dict[str, Any]:
|
|
return await self.request("POST", "/api/service", telegram_id=telegram_id, json=payload)
|
|
|
|
async def create_expense(self, telegram_id: int, payload: dict[str, Any]) -> dict[str, Any]:
|
|
return await self.request("POST", "/api/expenses", telegram_id=telegram_id, json=payload)
|
|
|
|
async def parse_record(self, telegram_id: int, text: str) -> dict[str, Any]:
|
|
return await self.request("POST", "/api/parse/record", telegram_id=telegram_id, json={"text": text})
|
|
|
|
async def public_service_centers(self, telegram_id: int) -> list[dict[str, Any]]:
|
|
return await self.request("GET", "/api/service-centers/public", telegram_id=telegram_id)
|
|
|
|
async def sto_catalog(self, telegram_id: int) -> list[dict[str, Any]]:
|
|
return await self.request("GET", "/api/sto/catalog", telegram_id=telegram_id)
|
|
|
|
async def my_appointments(self, telegram_id: int) -> list[dict[str, Any]]:
|
|
return await self.request("GET", "/api/appointments/my", telegram_id=telegram_id)
|
|
|
|
async def sto_dashboard(self, telegram_id: int, service_center_id: int) -> dict[str, Any]:
|
|
return await self.request(
|
|
"GET",
|
|
"/api/sto/dashboard",
|
|
telegram_id=telegram_id,
|
|
params={"service_center_id": service_center_id},
|
|
)
|
|
|
|
async def sto_appointments(self, telegram_id: int, service_center_id: int) -> list[dict[str, Any]]:
|
|
return await self.request(
|
|
"GET",
|
|
"/api/sto/appointments",
|
|
telegram_id=telegram_id,
|
|
params={"service_center_id": service_center_id, "status": "requested"},
|
|
)
|
|
|
|
async def my_service_centers(self, telegram_id: int) -> list[dict[str, Any]]:
|
|
return await self.request("GET", "/api/service-centers/my", telegram_id=telegram_id)
|
|
|
|
async def register_service_center(self, telegram_id: int, payload: dict[str, Any]) -> dict[str, Any]:
|
|
return await self.request("POST", "/api/service-centers", telegram_id=telegram_id, json=payload)
|
|
|
|
async def pending_service_centers(self, telegram_id: int) -> list[dict[str, Any]]:
|
|
return await self.request("GET", "/api/admin/service-centers/pending", telegram_id=telegram_id)
|
|
|
|
async def moderate_service_center(
|
|
self,
|
|
telegram_id: int,
|
|
service_center_id: int,
|
|
action: str,
|
|
payload: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
endpoint = {
|
|
"approve": "verify",
|
|
"reject": "reject",
|
|
"suspend": "suspend",
|
|
"changes": "request-changes",
|
|
}[action]
|
|
return await self.request(
|
|
"POST",
|
|
f"/api/admin/service-centers/{service_center_id}/{endpoint}",
|
|
telegram_id=telegram_id,
|
|
json=payload or {},
|
|
)
|