trevor/marriage_frontend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

main features

Browse Source
This commit is contained in:
Andrey K. Choi
2025-08-10 17:28:38 +09:00
parent 3e4a21d5b1
commit 95bef94c53
30 changed files with 4246 additions and 1066 deletions
Download Patch File Download Diff File Expand all files Collapse all files

636
ui/api.py
View File

@@ -1,10 +1,7 @@
import logging
import os
import os.path
import json
import time
import uuid
from typing import Any, Dict, Optional, Tuple, List, Union
from typing import Any, Dict, Optional, Tuple, List, Union, IO
import requests
from django.conf import settings
@@ -12,189 +9,111 @@ from django.core.exceptions import PermissionDenied
logger = logging.getLogger(__name__)
# ===== Логирование / флаги =====
API_DEBUG = os.environ.get('API_DEBUG', '1') == '1'
API_LOG_BODY_MAX = int(os.environ.get('API_LOG_BODY_MAX', '2000'))
API_LOG_HEADERS = os.environ.get('API_LOG_HEADERS', '1') == '1'
API_LOG_CURL = os.environ.get('API_LOG_CURL', '0') == '1'
# === Конфиг лога (можно подкрутить через .env): ===============================
API_DEBUG = bool(int(os.environ.get("API_DEBUG", getattr(settings, "API_DEBUG", 1))))
API_LOG_BODY_MAX = int(os.environ.get("API_LOG_BODY_MAX", getattr(settings, "API_LOG_BODY_MAX", 2000)))
API_LOG_HEADERS = bool(int(os.environ.get("API_LOG_HEADERS", getattr(settings, "API_LOG_HEADERS", 1))))
API_LOG_CURL = bool(int(os.environ.get("API_LOG_CURL", getattr(settings, "API_LOG_CURL", 1))))
API_FALLBACK_OPENAPI_ON_404 = bool(int(os.environ.get("API_FALLBACK_OPENAPI_ON_404", getattr(settings, "API_FALLBACK_OPENAPI_ON_404", 1))))
# Переключение базы при 404: сначала servers[0].url из openapi.json, потом жёстко http://localhost:8080
API_FALLBACK_OPENAPI_ON_404 = os.environ.get('API_FALLBACK_OPENAPI_ON_404', '1') == '1'
# === База API и пути по умолчанию (OpenAPI v1): ===============================
def _from_settings_or_env(name: str, default: Optional[str] = None) -> Optional[str]:
return getattr(settings, name, None) or os.environ.get(name, None) or default
SENSITIVE_KEYS = {'password', 'refresh_token', 'access_token', 'authorization', 'token', 'api_key'}
_API_BASE_PRIMARY = (_from_settings_or_env("API_BASE_URL", None) or "http://localhost:8080").rstrip("/")
_API_BASE_CACHE = None # текущая выбранная база
_API_LAST_SELECT_SRC = "ENV/SETTINGS"
def _sanitize(obj: Any) -> Any:
try:
if isinstance(obj, dict):
return {k: ('***' if isinstance(k, str) and k.lower() in SENSITIVE_KEYS else _sanitize(v))
for k, v in obj.items()}
if isinstance(obj, list):
return [_sanitize(x) for x in obj]
return obj
except Exception:
return obj
def _shorten(s: str, limit: int) -> str:
if s is None:
return ''
return s if len(s) <= limit else (s[:limit] + f'... <truncated {len(s)-limit} chars>')
def _build_curl(method: str, url: str, headers: Dict[str, str],
params: Optional[dict], json_body: Optional[dict], data: Optional[dict]) -> str:
parts = [f"curl -X '{method.upper()}' \\\n '{url}'"]
for k, v in headers.items():
if k.lower() == 'authorization':
v = 'Bearer ***'
parts.append(f" -H '{k}: {v}'")
if json_body is not None:
parts.append(" -H 'Content-Type: application/json'")
try:
body_str = json.dumps(_sanitize(json_body), ensure_ascii=False)
except Exception:
body_str = str(_sanitize(json_body))
parts.append(f" -d '{body_str}'")
elif data is not None:
try:
body_str = json.dumps(_sanitize(data), ensure_ascii=False)
except Exception:
body_str = str(_sanitize(data))
parts.append(f" --data-raw '{body_str}'")
return " \\\n".join(parts)
# ===== Пути эндпоинтов (как в swagger) =====
EP_DEFAULTS: Dict[str, str] = {
# Auth
'AUTH_REGISTER_PATH': '/auth/v1/register',
'AUTH_TOKEN_PATH': '/auth/v1/token',
'AUTH_REFRESH_PATH': '/auth/v1/refresh',
'ME_PATH': '/auth/v1/me',
'USERS_LIST_PATH': '/auth/v1/users',
'USER_DETAIL_PATH': '/auth/v1/users/{user_id}',
# Profiles
'PROFILE_ME_PATH': '/profiles/v1/profiles/me',
'PROFILES_CREATE_PATH': '/profiles/v1/profiles',
'PROFILE_PHOTO_UPLOAD_PATH': '/profiles/v1/profiles/me/photo',
'PROFILE_PHOTO_DELETE_PATH': '/profiles/v1/profiles/me/photo',
# Pairs
'PAIRS_PATH': '/match/v1/pairs',
'PAIR_DETAIL_PATH': '/match/v1/pairs/{pair_id}',
'PAIR_ACCEPT_PATH': '/match/v1/pairs/{pair_id}/accept',
'PAIR_REJECT_PATH': '/match/v1/pairs/{pair_id}/reject',
# Chat
'ROOMS_PATH': '/chat/v1/rooms',
'ROOM_DETAIL_PATH': '/chat/v1/rooms/{room_id}',
'ROOM_MESSAGES_PATH': '/chat/v1/rooms/{room_id}/messages',
# Payments
'INVOICES_PATH': '/payments/v1/invoices',
'INVOICE_DETAIL_PATH': '/payments/v1/invoices/{inv_id}',
'INVOICE_MARK_PAID_PATH': '/payments/v1/invoices/{inv_id}/mark-paid',
# AUTH
"AUTH_REGISTER_PATH": "/auth/v1/register",
"AUTH_TOKEN_PATH": "/auth/v1/token",
"AUTH_REFRESH_PATH": "/auth/v1/refresh",
"AUTH_ME_PATH": "/auth/v1/me",
# USERS (admin and/or owner)
"USERS_LIST_PATH": "/auth/v1/users",
"USER_DETAIL_PATH": "/auth/v1/users/{user_id}",
# PROFILES
"PROFILE_ME_PATH": "/profiles/v1/profiles/me",
"PROFILE_CREATE_PATH": "/profiles/v1/profiles",
# Optional, если сервер будет поддерживать:
"PROFILE_ME_PATCH_PATH": "", # пусто → нет на бэкенде
"PROFILE_PHOTO_UPLOAD_PATH": "", # "/profiles/v1/profiles/me/photo"
"PROFILE_PHOTO_DELETE_PATH": "", # "/profiles/v1/profiles/me/photo"
}
def EP(key: str) -> str:
return os.environ.get(f'API_{key}', EP_DEFAULTS[key])
def EP(key: str, **fmt) -> str:
# при желании можно переопределить путями в settings или ENV: API_<KEY>
override = _from_settings_or_env(f"API_{key}", None)
path = (override or EP_DEFAULTS[key]).format(**fmt)
if not path.startswith("/"):
path = "/" + path
return path
# ===== База API с авто‑детектом =====
_API_BASE_CACHE: Optional[str] = None
_API_LAST_SELECT_SRC = 'DEFAULT' # для логов
def _detect_api_base_from_openapi() -> Optional[str]:
"""
Берём servers[0].url из openapi.json (по схеме — http://localhost:8080).
"""
candidates = [
os.environ.get('API_SPEC_PATH'),
getattr(settings, 'API_SPEC_PATH', None),
os.path.join(getattr(settings, 'BASE_DIR', ''), 'openapi.json') if getattr(settings, 'BASE_DIR', None) else None,
os.path.join(getattr(settings, 'BASE_DIR', ''), 'agency', 'openapi.json') if getattr(settings, 'BASE_DIR', None) else None,
'/mnt/data/openapi.json',
]
for p in candidates:
if p and os.path.isfile(p):
try:
with open(p, 'r', encoding='utf-8') as f:
spec = json.load(f)
servers = spec.get('servers') or []
if servers and isinstance(servers[0], dict):
url = servers[0].get('url')
if url:
return url
except Exception as e:
if API_DEBUG:
logger.debug('API: cannot read OpenAPI from %s: %s', p, e)
return None
def _get_api_base_url() -> str:
"""
Источники (по приоритету):
1) ENV/Settings: API_BASE_URL, затем BASE_URL
2) servers[0].url из openapi.json
3) 'http://localhost:8080'
"""
def _get_api_base() -> str:
global _API_BASE_CACHE, _API_LAST_SELECT_SRC
if _API_BASE_CACHE:
return _API_BASE_CACHE
base = (os.environ.get('API_BASE_URL') or getattr(settings, 'API_BASE_URL', '')
or os.environ.get('BASE_URL') or getattr(settings, 'BASE_URL', ''))
if base:
_API_BASE_CACHE = base.rstrip('/')
_API_LAST_SELECT_SRC = 'ENV/SETTINGS'
else:
detected = _detect_api_base_from_openapi()
if detected:
_API_BASE_CACHE = detected.rstrip('/')
_API_LAST_SELECT_SRC = 'OPENAPI'
else:
_API_BASE_CACHE = 'http://localhost:8080'
_API_LAST_SELECT_SRC = 'HARDCODED'
# пробуем взять из ENV/SETTINGS
_API_BASE_CACHE = _API_BASE_PRIMARY
_API_LAST_SELECT_SRC = "ENV/SETTINGS"
if API_DEBUG:
logger.info("API base selected [%s]: %s", _API_LAST_SELECT_SRC, _API_BASE_CACHE)
return _API_BASE_CACHE
def _set_api_base(new_base: str):
global _API_BASE_CACHE, _API_LAST_SELECT_SRC
if new_base and new_base.rstrip("/") != _API_BASE_CACHE:
old = _API_BASE_CACHE
_API_BASE_CACHE = new_base.rstrip("/")
_API_LAST_SELECT_SRC = "FALLBACK"
if API_DEBUG:
logger.warning("API BASE SWITCHED: %s → %s", old, _API_BASE_CACHE)
# === Исключение API: ==========================================================
class ApiError(Exception):
def __init__(self, status: int, message: str = 'API error', payload: Optional[dict] = None, req_id: Optional[str] = None):
if req_id and message and 'req_id=' not in message:
message = f"{message} (req_id={req_id})"
def __init__(self, status: int, message: str = "API error", payload: Optional[dict] = None):
super().__init__(message)
self.status = status
self.payload = payload or {}
self.req_id = req_id
def _base_headers(request, extra: Optional[Dict[str, str]] = None) -> Dict[str, str]:
"""
Достаём токен сначала из сессии, затем из куки — чтобы «куки‑режим» работал без доп. настроек.
"""
headers: Dict[str, str] = {'Accept': 'application/json'}
token = request.session.get('access_token') or request.COOKIES.get('access_token')
# === Вспомогательные: =========================================================
def _headers(request, *, extra: Optional[Dict[str, str]] = None, json_mode: bool = True) -> Dict[str, str]:
h = {"Accept": "application/json"}
if json_mode:
h["Content-Type"] = "application/json"
# Токен: из сессии или из cookie (HTTPCookie)
token = request.session.get("access_token") or request.COOKIES.get("access_token")
if token:
headers['Authorization'] = f'Bearer {token}'
api_key = getattr(settings, 'API_KEY', '') or os.environ.get('API_KEY', '')
h["Authorization"] = f"Bearer {token}"
api_key = getattr(settings, "API_KEY", "") or os.environ.get("API_KEY", "")
if api_key:
headers['X-API-Key'] = api_key
h["X-API-Key"] = api_key
if extra:
headers.update(extra)
return headers
h.update(extra)
return h
def _url(path: str) -> str:
base = _get_api_base_url()
path = path if path.startswith('/') else '/' + path
base = _get_api_base()
path = path if path.startswith("/") else "/" + path
return base + path
def _log_curl(method: str, url: str, headers: Dict[str, str], body: Optional[Union[dict, str]]):
if not API_LOG_CURL:
return
parts = ["curl", "-X", method.upper(), f"'{url}'"]
if headers:
for k, v in headers.items():
if k.lower() == "authorization":
v = "*****"
parts += ["-H", f"'{k}: {v}'"]
if body is not None and body != {}:
import json as _json
b = body if isinstance(body, str) else _json.dumps(body, ensure_ascii=False)
parts += ["-d", f"'{b}'"]
logger.debug("CURL> %s", " ".join(parts))
# === Базовый HTTP вызов с ретраями/refresh/fallback: ==========================
def request_api(
request,
method: str,
@@ -203,325 +122,184 @@ def request_api(
params: Optional[dict] = None,
json: Optional[dict] = None,
files: Optional[dict] = None,
data: Optional[dict] = None,
) -> Tuple[int, Any]:
"""
Универсальный HTTP-вызов (упрощённая версия):
- токен/ключ в заголовках (токен берём из сессии или куки),
- auto-refresh при 401 (refresh берём тоже из сессии или куки),
- 404 → переключаем базу: servers[0].url из openapi.json → 'http://localhost:8080',
- подробные логи; БЕЗ «retry со слэшем».
"""
global _API_BASE_CACHE, _API_LAST_SELECT_SRC
req_id = uuid.uuid4().hex[:8]
base_before = _get_api_base_url()
rid = uuid.uuid4().hex[:8]
url = _url(path)
json_mode = files is None
headers = _headers(request, json_mode=json_mode)
def _do(_url: str):
headers = _base_headers(request)
if json is not None and files is None and data is None:
headers['Content-Type'] = 'application/json'
if API_DEBUG:
b_preview = ("***" if json and "password" in (json or {}) else json)
if isinstance(b_preview, dict) and "password" in (b_preview or {}):
b_preview = dict(b_preview); b_preview["password"] = "***"
logger.info("API[req_id=%s] REQUEST %s %s params=%s headers=%s body=%s",
rid, method.upper(), url, params, (headers if API_LOG_HEADERS else "{…}"),
(str(b_preview)[:API_LOG_BODY_MAX] if b_preview else None))
_log_curl(method, url, headers, b_preview)
if API_DEBUG:
log_headers = _sanitize(headers) if API_LOG_HEADERS else {}
log_body = _sanitize(json if json is not None else data)
if API_LOG_CURL:
try:
curl = _build_curl(method, _url, headers, params, json, data)
logger.debug("API[req_id=%s] cURL:\n%s", req_id, curl)
except Exception:
pass
logger.info(
"API[req_id=%s] REQUEST %s %s params=%s headers=%s body=%s",
req_id, method.upper(), _url, _sanitize(params), log_headers, log_body
)
t0 = time.time()
try:
resp = requests.request(
method=method.upper(),
url=_url,
url=url,
headers=headers,
params=params,
json=json,
data=data,
files=files,
timeout=float(getattr(settings, 'API_TIMEOUT', 8.0)),
json=json if json_mode else None,
files=files if not json_mode else None,
timeout=float(getattr(settings, "API_TIMEOUT", os.environ.get("API_TIMEOUT", 6.0))),
)
dt = int((time.time() - t0) * 1000)
content_type = resp.headers.get('Content-Type', '')
try:
payload = resp.json() if 'application/json' in content_type else {}
except ValueError:
payload = {}
if API_DEBUG:
body_str = ""
try:
body_str = json.dumps(_sanitize(payload), ensure_ascii=False)
except Exception:
body_str = str(_sanitize(payload))
headers_out = _sanitize(dict(resp.headers)) if API_LOG_HEADERS else {}
logger.info(
"API[req_id=%s] RESPONSE %s %sms ct=%s headers=%s body=%s",
req_id, resp.status_code, dt, content_type, headers_out, _shorten(body_str, API_LOG_BODY_MAX)
)
return resp, payload
# 1) Первый запрос
try:
resp, payload = _do(url)
except requests.RequestException as e:
logger.exception('API[req_id=%s] network error: %s', req_id, e)
raise ApiError(0, f'Network unavailable or timeout when accessing API ({e})', req_id=req_id)
logger.exception("API network error: %s", e)
raise ApiError(0, f"Network unavailable or timeout when accessing API ({e})")
# 2) 404 → переключаем базу (openapi → 8080) и повторяем
if resp.status_code == 404:
candidates: List[tuple[str, str]] = []
if API_FALLBACK_OPENAPI_ON_404:
detected = _detect_api_base_from_openapi()
if detected:
candidates.append((detected.rstrip('/'), 'OPENAPI(FAILOVER)'))
candidates.append(('http://localhost:8080', 'DEFAULT(FAILOVER)'))
content_type = resp.headers.get("Content-Type", "")
try:
data = resp.json() if "application/json" in content_type else {}
except ValueError:
data = {}
for cand_base, label in candidates:
if not cand_base or cand_base == _API_BASE_CACHE:
continue
if API_DEBUG:
logger.warning("API[req_id=%s] 404 on base %s → switch API base to %s and retry",
req_id, _API_BASE_CACHE, cand_base)
_API_BASE_CACHE = cand_base
_API_LAST_SELECT_SRC = label
try:
resp, payload = _do(_url(path))
if resp.status_code != 404:
break
except requests.RequestException:
continue
if API_DEBUG:
b_preview = data
logger.info("API[req_id=%s] RESPONSE %s %sms ct=%s headers=%s body=%s",
rid, resp.status_code, int(resp.elapsed.total_seconds()*1000),
content_type, (dict(resp.headers) if API_LOG_HEADERS else "{…}"),
(str(b_preview)[:API_LOG_BODY_MAX] if b_preview else None))
# 3) 401 → refresh и повтор
refresh_token = request.session.get('refresh_token') or request.COOKIES.get('refresh_token')
if resp.status_code == 401 and refresh_token:
# авто-ретрай на 401: пробуем refresh
if resp.status_code == 401 and (request.session.get("refresh_token") or request.COOKIES.get("refresh_token")):
if API_DEBUG:
logger.info("API[req_id=%s] 401 → try refresh token", req_id)
logger.info("API[req_id=%s] 401 → try refresh token", rid)
try:
refresh_url = _url(EP('AUTH_REFRESH_PATH'))
refresh_body = {'refresh_token': refresh_token}
logger.info("API[req_id=%s] REFRESH POST %s body=%s", req_id, refresh_url, _sanitize(refresh_body))
refresh_resp = requests.post(refresh_url, json=refresh_body, timeout=float(getattr(settings, 'API_TIMEOUT', 8.0)))
if refresh_resp.status_code == 200:
try:
rj = refresh_resp.json()
except ValueError:
rj = {}
if rj.get('access_token'):
request.session['access_token'] = rj['access_token']
if rj.get('refresh_token'):
request.session['refresh_token'] = rj['refresh_token']
refresh_json = {"refresh_token": request.session.get("refresh_token") or request.COOKIES.get("refresh_token")}
r = requests.post(_url(EP("AUTH_REFRESH_PATH")), json=refresh_json, timeout=float(getattr(settings, "API_TIMEOUT", 6.0)))
if r.status_code == 200:
tokens = r.json()
request.session["access_token"] = tokens.get("access_token")
request.session["refresh_token"] = tokens.get("refresh_token")
request.session.modified = True
if API_DEBUG:
logger.info("API[req_id=%s] REFRESH OK → retry original request", req_id)
resp, payload = _do(_url(path))
logger.info("API[req_id=%s] REFRESH OK → retry original request", rid)
# повторяем исходный
headers = _headers(request, json_mode=json_mode)
resp = requests.request(
method=method.upper(), url=url, headers=headers,
params=params, json=json if json_mode else None, files=files if not json_mode else None,
timeout=float(getattr(settings, "API_TIMEOUT", 6.0)),
)
content_type = resp.headers.get("Content-Type", "")
data = resp.json() if "application/json" in content_type else {}
else:
logger.warning("API[req_id=%s] REFRESH failed: %s", req_id, refresh_resp.status_code)
if API_DEBUG:
logger.warning("API[req_id=%s] REFRESH FAIL %s body=%s", rid, r.status_code, r.text[:200])
except requests.RequestException as e:
logger.exception('API[req_id=%s] Refresh token network error: %s', req_id, e)
raise ApiError(401, f'Token refresh failed ({e})', req_id=req_id)
logger.exception("Refresh token error: %s", e)
raise ApiError(401, f"Token refresh failed ({e})")
# 4) Ошибки
# fallback базы на 404 (например, локальный gateway vs 8080)
if resp.status_code == 404 and API_FALLBACK_OPENAPI_ON_404:
# пробуем со слешем
if not url.endswith("/"):
url_slash = url + "/"
if API_DEBUG:
logger.info("API[req_id=%s] 404 → retry with trailing slash", rid)
resp2 = requests.request(method=method.upper(), url=url_slash, headers=headers,
params=params, json=json if json_mode else None, files=files if not json_mode else None,
timeout=float(getattr(settings, "API_TIMEOUT", 6.0)))
content_type2 = resp2.headers.get("Content-Type", "")
data2 = resp2.json() if "application/json" in content_type2 else {}
if resp2.status_code != 404:
resp, data, content_type = resp2, data2, content_type2
# если всё ещё 404 и текущая база не 8080 — переключаемся и повторяем
if resp.status_code == 404 and _get_api_base() != "http://localhost:8080":
if API_DEBUG:
logger.warning("API[req_id=%s] 404 on base %s → switch API base to http://localhost:8080 and retry",
rid, _get_api_base())
_set_api_base("http://localhost:8080")
url = _url(path)
headers = _headers(request, json_mode=json_mode)
resp = requests.request(method=method.upper(), url=url, headers=headers,
params=params, json=json if json_mode else None, files=files if not json_mode else None,
timeout=float(getattr(settings, "API_TIMEOUT", 6.0)))
content_type = resp.headers.get("Content-Type", "")
try:
data = resp.json() if "application/json" in content_type else {}
except ValueError:
data = {}
# финальная проверка кода
if not (200 <= resp.status_code < 300):
msg = None
if isinstance(payload, dict):
msg = payload.get('detail') or payload.get('message')
msg = msg or f'API error: {resp.status_code}'
msg = data.get("detail") or data.get("message") or f"API error: {resp.status_code}"
if resp.status_code in (401, 403):
# PermissionDenied обрабатываем во view (не всегда это «выйти и войти заново»)
raise PermissionDenied(f"{msg} (req_id={req_id})")
raise ApiError(resp.status_code, msg, payload if isinstance(payload, dict) else {}, req_id=req_id)
raise PermissionDenied(msg)
raise ApiError(resp.status_code, msg, data)
# 5) База сменилась — отметим
base_after = _get_api_base_url()
if API_DEBUG and base_before != base_after:
logger.warning("API[req_id=%s] BASE SWITCHED: %s → %s", req_id, base_before, base_after)
return resp.status_code, data
return resp.status_code, payload
# === High-level helpers в соответствии с OpenAPI v1 ==========================
# ==========================
# AUTH
# ==========================
def register_user(request, email: str, password: str, full_name: Optional[str] = None, role: str = 'CLIENT') -> Dict[str, Any]:
body = {'email': email, 'password': password, 'full_name': full_name, 'role': role}
_, data = request_api(request, 'POST', EP('AUTH_REGISTER_PATH'), json=body)
def register_user(request, email: str, password: str, full_name: Optional[str] = None, role: str = "CLIENT") -> Dict[str, Any]:
body = {"email": email, "password": password, "full_name": full_name, "role": role}
_, data = request_api(request, "POST", EP("AUTH_REGISTER_PATH"), json=body)
return data # UserRead
def login(request, email: str, password: str) -> Dict[str, Any]:
body = {'email': email, 'password': password}
_, data = request_api(request, 'POST', EP('AUTH_TOKEN_PATH'), json=body)
body = {"email": email, "password": password}
_, data = request_api(request, "POST", EP("AUTH_TOKEN_PATH"), json=body)
return data # TokenPair
def get_current_user(request) -> Dict[str, Any]:
_, data = request_api(request, 'GET', EP('ME_PATH'))
def get_me(request) -> Dict[str, Any]:
_, data = request_api(request, "GET", EP("AUTH_ME_PATH"))
return data # UserRead
def list_users(request, offset: int = 0, limit: int = 50) -> Union[List[Dict[str, Any]], Dict[str, Any]]:
params = {'offset': offset, 'limit': limit}
_, data = request_api(request, 'GET', EP('USERS_LIST_PATH'), params=params)
return data
def list_users(request, offset: int = 0, limit: int = 50) -> List[Dict[str, Any]] | Dict[str, Any]:
params = {"offset": offset, "limit": min(max(limit, 1), 200)}
_, data = request_api(request, "GET", EP("USERS_LIST_PATH"), params=params)
return data # array[UserRead] (в схеме — массив) или {"items": [...]} если backend так возвращает
def get_user(request, user_id: str) -> Dict[str, Any]:
path = EP('USER_DETAIL_PATH').format(user_id=user_id)
_, data = request_api(request, 'GET', path)
return data
_, data = request_api(request, "GET", EP("USER_DETAIL_PATH", user_id=user_id))
return data # UserRead
def update_user(request, user_id: str, **fields) -> Dict[str, Any]:
path = EP('USER_DETAIL_PATH').format(user_id=user_id)
_, data = request_api(request, 'PATCH', path, json=fields)
return data
def delete_user(request, user_id: str) -> None:
path = EP('USER_DETAIL_PATH').format(user_id=user_id)
request_api(request, 'DELETE', path)
# ==========================
# PROFILES
# ==========================
def update_user_me(request, user_id: str, *, full_name: Optional[str] = None, password: Optional[str] = None) -> Dict[str, Any]:
body: Dict[str, Any] = {}
if full_name is not None:
body["full_name"] = full_name
if password is not None:
body["password"] = password
_, data = request_api(request, "PATCH", EP("USER_DETAIL_PATH", user_id=user_id), json=body)
return data # UserRead
def get_my_profile(request) -> Dict[str, Any]:
_, data = request_api(request, 'GET', EP('PROFILE_ME_PATH'))
_, data = request_api(request, "GET", EP("PROFILE_ME_PATH"))
return data # ProfileOut
def create_my_profile(request, gender: str, city: str, languages: List[str], interests: List[str]) -> Dict[str, Any]:
body = {'gender': gender, 'city': city, 'languages': languages, 'interests': interests}
_, data = request_api(request, 'POST', EP('PROFILES_CREATE_PATH'), json=body)
body = {"gender": gender, "city": city, "languages": languages, "interests": interests}
_, data = request_api(request, "POST", EP("PROFILE_CREATE_PATH"), json=body)
return data # ProfileOut
# ==========================
# PAIRS
# ==========================
def create_pair(request, user_id_a: str, user_id_b: str, score: Optional[float] = None, notes: Optional[str] = None) -> Dict[str, Any]:
body = {'user_id_a': user_id_a, 'user_id_b': user_id_b, 'score': score, 'notes': notes}
_, data = request_api(request, 'POST', EP('PAIRS_PATH'), json=body)
return data # PairRead
def list_pairs(request, for_user_id: Optional[str] = None, status: Optional[str] = None, offset: int = 0, limit: int = 50) -> Union[List[Dict[str, Any]], Dict[str, Any]]:
params = {'for_user_id': for_user_id, 'status': status, 'offset': offset, 'limit': limit}
params = {k: v for k, v in params.items() if v is not None}
_, data = request_api(request, 'GET', EP('PAIRS_PATH'), params=params)
# Опционально: если сервер добавит PATCH /profiles/v1/profiles/me
def patch_my_profile(request, **fields) -> Dict[str, Any]:
patch_path = EP("PROFILE_ME_PATCH_PATH")
if not patch_path or patch_path.strip("/") == "":
raise ApiError(405, "Profile update endpoint is not available on backend")
_, data = request_api(request, "PATCH", patch_path, json=fields)
return data
def get_pair(request, pair_id: str) -> Dict[str, Any]:
path = EP('PAIR_DETAIL_PATH').format(pair_id=pair_id)
_, data = request_api(request, 'GET', path)
return data
def update_pair(request, pair_id: str, **fields) -> Dict[str, Any]:
path = EP('PAIR_DETAIL_PATH').format(pair_id=pair_id)
_, data = request_api(request, 'PATCH', path, json=fields)
return data
def delete_pair(request, pair_id: str) -> None:
path = EP('PAIR_DETAIL_PATH').format(pair_id=pair_id)
request_api(request, 'DELETE', path)
def accept_pair(request, pair_id: str) -> Dict[str, Any]:
path = EP('PAIR_ACCEPT_PATH').format(pair_id=pair_id)
_, data = request_api(request, 'POST', path)
return data
def reject_pair(request, pair_id: str) -> Dict[str, Any]:
path = EP('PAIR_REJECT_PATH').format(pair_id=pair_id)
_, data = request_api(request, 'POST', path)
return data
# ==========================
# CHAT
# ==========================
def my_rooms(request) -> Union[List[Dict[str, Any]], Dict[str, Any]]:
_, data = request_api(request, 'GET', EP('ROOMS_PATH'))
return data
def get_room(request, room_id: str) -> Dict[str, Any]:
path = EP('ROOM_DETAIL_PATH').format(room_id=room_id)
_, data = request_api(request, 'GET', path)
return data
def create_room(request, title: Optional[str] = None, participants: List[str] = []) -> Dict[str, Any]:
body = {'title': title, 'participants': participants}
_, data = request_api(request, 'POST', EP('ROOMS_PATH'), json=body)
return data
def send_message(request, room_id: str, content: str) -> Dict[str, Any]:
body = {'content': content}
path = EP('ROOM_MESSAGES_PATH').format(room_id=room_id)
_, data = request_api(request, 'POST', path, json=body)
return data
def list_messages(request, room_id: str, offset: int = 0, limit: int = 100) -> Union[List[Dict[str, Any]], Dict[str, Any]]:
params = {'offset': offset, 'limit': limit}
path = EP('ROOM_MESSAGES_PATH').format(room_id=room_id)
_, data = request_api(request, 'GET', path, params=params)
return data
# ==========================
# PAYMENTS
# ==========================
def create_invoice(request, client_id: str, amount: float, currency: str, description: Optional[str] = None) -> Dict[str, Any]:
body = {'client_id': client_id, 'amount': amount, 'currency': currency, 'description': description}
_, data = request_api(request, 'POST', EP('INVOICES_PATH'), json=body)
return data
def list_invoices(request, client_id: Optional[str] = None, status: Optional[str] = None, offset: int = 0, limit: int = 50) -> Union[List[Dict[str, Any]], Dict[str, Any]]:
params = {'client_id': client_id, 'status': status, 'offset': offset, 'limit': limit}
params = {k: v for k, v in params.items() if v is not None}
_, data = request_api(request, 'GET', EP('INVOICES_PATH'), params=params)
return data
def get_invoice(request, inv_id: str) -> Dict[str, Any]:
path = EP('INVOICE_DETAIL_PATH').format(inv_id=inv_id)
_, data = request_api(request, 'GET', path)
return data
def update_invoice(request, inv_id: str, **fields) -> Dict[str, Any]:
path = EP('INVOICE_DETAIL_PATH').format(inv_id=inv_id)
_, data = request_api(request, 'PATCH', path, json=fields)
return data
def delete_invoice(request, inv_id: str) -> None:
path = EP('INVOICE_DETAIL_PATH').format(inv_id=inv_id)
request_api(request, 'DELETE', path)
def mark_invoice_paid(request, inv_id: str) -> Dict[str, Any]:
path = EP('INVOICE_MARK_PAID_PATH').format(inv_id=inv_id)
_, data = request_api(request, 'POST', path)
return data
def upload_my_photo(request, file_obj) -> Dict[str, Any]:
def upload_my_profile_photo(request, file_obj: IO[bytes]) -> Dict[str, Any]:
"""
Отправляет multipart/form-data на бекенд для загрузки фото профиля.
Ожидаем, что сервер примет поле 'file' и вернёт обновлённый профиль или {photo_url: "..."}.
POST /profiles/v1/profiles/me/photo (multipart/form-data, field: file)
Возвращает ProfileOut с photo_url. (см. OpenAPI) :contentReference[oaicite:1]{index=1}
"""
path = EP('PROFILE_PHOTO_UPLOAD_PATH')
filename = getattr(file_obj, 'name', 'photo.jpg')
content_type = getattr(file_obj, 'content_type', 'application/octet-stream')
files = {'file': (filename, file_obj, content_type)}
_, data = request_api(request, 'POST', path, files=files)
_, data = request_api(request, 'POST', '/profiles/v1/profiles/me/photo', files=files)
return data
def delete_my_photo(request) -> Dict[str, Any]:
"""
Удаляет фото профиля (если сервер поддерживает DELETE на том же пути).
"""
path = EP('PROFILE_PHOTO_DELETE_PATH')
_, data = request_api(request, 'DELETE', path)
return data
path = EP("PROFILE_PHOTO_DELETE_PATH")
if not path or path.strip("/") == "":
raise ApiError(405, "Photo delete endpoint is not available on backend")
_, data = request_api(request, "DELETE", path)
return data