92 lines
2.5 KiB
Python
92 lines
2.5 KiB
Python
"""
|
|
api_client.py — HTTP-клиент для stats-приложения.
|
|
Запросы идут к основному API без авторизации.
|
|
"""
|
|
import os
|
|
from typing import Optional
|
|
import httpx
|
|
|
|
API_URL = os.getenv("API_URL", "http://api:8000")
|
|
_TIMEOUT = 15
|
|
|
|
|
|
async def get_summary() -> dict:
|
|
try:
|
|
async with httpx.AsyncClient(base_url=API_URL, timeout=_TIMEOUT) as client:
|
|
resp = await client.get("/stats/summary")
|
|
if resp.status_code == 200:
|
|
return resp.json()
|
|
except Exception:
|
|
pass
|
|
return {"total_responses": 0, "total_polls": 0, "active_polls": 0, "orgs_active": 0}
|
|
|
|
|
|
async def get_years() -> list[int]:
|
|
try:
|
|
async with httpx.AsyncClient(base_url=API_URL, timeout=_TIMEOUT) as client:
|
|
resp = await client.get("/stats/years")
|
|
if resp.status_code == 200:
|
|
return resp.json()
|
|
except Exception:
|
|
pass
|
|
return []
|
|
|
|
|
|
async def get_chart_data(
|
|
group_by: str,
|
|
poll_id: Optional[str] = None,
|
|
org_id: Optional[str] = None,
|
|
group_id: Optional[str] = None,
|
|
year: Optional[int] = None,
|
|
) -> list[dict]:
|
|
params: dict = {"group_by": group_by}
|
|
if poll_id:
|
|
params["poll_id"] = poll_id
|
|
if org_id:
|
|
params["org_id"] = org_id
|
|
if group_id:
|
|
params["group_id"] = group_id
|
|
if year:
|
|
params["year"] = year
|
|
try:
|
|
async with httpx.AsyncClient(base_url=API_URL, timeout=_TIMEOUT) as client:
|
|
resp = await client.get("/stats/responses", params=params)
|
|
if resp.status_code == 200:
|
|
return resp.json()
|
|
except Exception:
|
|
pass
|
|
return []
|
|
|
|
|
|
async def get_polls() -> list[dict]:
|
|
try:
|
|
async with httpx.AsyncClient(base_url=API_URL, timeout=_TIMEOUT) as client:
|
|
resp = await client.get("/polls/")
|
|
if resp.status_code == 200:
|
|
return resp.json()
|
|
except Exception:
|
|
pass
|
|
return []
|
|
|
|
|
|
async def get_organizations() -> list[dict]:
|
|
try:
|
|
async with httpx.AsyncClient(base_url=API_URL, timeout=_TIMEOUT) as client:
|
|
resp = await client.get("/organizations/")
|
|
if resp.status_code == 200:
|
|
return resp.json()
|
|
except Exception:
|
|
pass
|
|
return []
|
|
|
|
|
|
async def get_groups() -> list[dict]:
|
|
try:
|
|
async with httpx.AsyncClient(base_url=API_URL, timeout=_TIMEOUT) as client:
|
|
resp = await client.get("/groups/")
|
|
if resp.status_code == 200:
|
|
return resp.json()
|
|
except Exception:
|
|
pass
|
|
return []
|