import logging from typing import Any import httpx from bot.config import settings logger = logging.getLogger("bot.api") class ApiError(Exception): """Backend returned an error. `code` is the machine-readable message key.""" def __init__(self, code: str, status: int = 400, errors: list[str] | None = None) -> None: super().__init__(code) self.code = code self.status = status self.errors = errors or [] class ApiClient: """Thin HTTP wrapper. The bot never touches the database directly.""" def __init__(self) -> None: self._client = httpx.AsyncClient( base_url=settings.api_base_url.rstrip("/"), timeout=settings.api_timeout, headers={"X-Bot-Token": settings.bot_api_token}, ) async def close(self) -> None: await self._client.aclose() 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, files: dict[str, Any] | None = None, ) -> Any: headers = {"X-Telegram-Id": str(telegram_id)} if telegram_id else {} try: response = await self._client.request( method, path, json=json, params=params, files=files, headers=headers ) except httpx.HTTPError as exc: logger.error("api request failed %s %s: %s", method, path, exc) raise ApiError("api_unavailable", 503) from exc try: payload = response.json() except ValueError: logger.error("invalid api response %s: %s", response.status_code, response.text[:300]) raise ApiError("invalid_response", response.status_code) from None if response.status_code >= 400 or not payload.get("success", False): code = str(payload.get("message") or "error_generic") logger.warning("api error %s %s -> %s", method, path, code) raise ApiError(code, response.status_code, payload.get("errors") or []) return payload.get("data") async def get(self, path: str, **kwargs: Any) -> Any: return await self.request("GET", path, **kwargs) async def post(self, path: str, **kwargs: Any) -> Any: return await self.request("POST", path, **kwargs) async def put(self, path: str, **kwargs: Any) -> Any: return await self.request("PUT", path, **kwargs) async def delete(self, path: str, **kwargs: Any) -> Any: return await self.request("DELETE", path, **kwargs) # -- endpoints used by the handlers ------------------------------ async def start(self, payload: dict[str, Any]) -> dict[str, Any]: return await self.post("/telegram/start", json=payload) async def config(self) -> dict[str, Any]: return await self.get("/bot/config") async def categories(self) -> list[dict[str, Any]]: return await self.get("/categories") async def products(self, category_id: str) -> list[dict[str, Any]]: return await self.get("/products", params={"category_id": category_id, "enabled": True}) async def product(self, product_id: str) -> dict[str, Any]: return await self.get(f"/products/{product_id}") async def flow(self, name: str, product_id: str | None = None) -> dict[str, Any]: params = {"product_id": product_id} if product_id else None return await self.get(f"/flows/{name}", params=params) async def me(self, telegram_id: int) -> dict[str, Any]: return await self.get("/users/me", telegram_id=telegram_id) async def wallet(self, telegram_id: int) -> dict[str, Any]: return await self.get("/wallet", telegram_id=telegram_id) async def transactions(self, telegram_id: int, size: int = 10) -> dict[str, Any]: return await self.get("/wallet/transactions", telegram_id=telegram_id, params={"size": size}) async def max_affordable( self, telegram_id: int, product_id: str | None, flow: str | None ) -> dict[str, Any]: params = {key: value for key, value in {"product_id": product_id, "flow": flow}.items() if value} return await self.get("/orders/max-affordable", telegram_id=telegram_id, params=params) async def create_order(self, telegram_id: int, payload: dict[str, Any]) -> dict[str, Any]: return await self.post("/orders", telegram_id=telegram_id, json=payload) async def order(self, telegram_id: int, order_id: str) -> dict[str, Any]: return await self.get(f"/orders/{order_id}", telegram_id=telegram_id) async def orders(self, telegram_id: int, size: int = 10) -> dict[str, Any]: return await self.get("/orders", telegram_id=telegram_id, params={"size": size}) async def track(self, telegram_id: int, order_number: str) -> dict[str, Any]: return await self.get(f"/orders/track/{order_number}", telegram_id=telegram_id) async def update_inputs( self, telegram_id: int, order_id: str, input_data: dict[str, Any] ) -> dict[str, Any]: return await self.put( f"/orders/{order_id}/inputs", telegram_id=telegram_id, json={"input_data": input_data} ) async def apply_coupon(self, telegram_id: int, order_id: str, code: str) -> dict[str, Any]: return await self.post( f"/orders/{order_id}/coupon", telegram_id=telegram_id, json={"code": code} ) async def confirm_order(self, telegram_id: int, order_id: str) -> dict[str, Any]: return await self.put(f"/orders/{order_id}/confirm", telegram_id=telegram_id) async def cancel_order(self, telegram_id: int, order_id: str) -> dict[str, Any]: return await self.put(f"/orders/{order_id}/cancel", telegram_id=telegram_id) async def create_payment(self, telegram_id: int, order_id: str, method: str) -> dict[str, Any]: return await self.post( "/payments", telegram_id=telegram_id, json={"order_id": order_id, "method": method} ) async def check_payment(self, telegram_id: int, payment_id: str) -> dict[str, Any]: return await self.post(f"/payments/{payment_id}/check", telegram_id=telegram_id) async def upload_receipt( self, telegram_id: int, payment_id: str, content: bytes, filename: str ) -> dict[str, Any]: return await self.post( f"/payments/{payment_id}/receipt", telegram_id=telegram_id, files={"file": (filename, content, "image/jpeg")}, ) async def referral(self, telegram_id: int) -> dict[str, Any]: return await self.get("/referral", telegram_id=telegram_id) async def create_ticket(self, telegram_id: int, message: str) -> dict[str, Any]: return await self.post("/support", telegram_id=telegram_id, json={"message": message}) api = ApiClient()