feat(api): FastAPI — auth, RFC 7807 и роуты фазы 1
Префикс /api/v1, operationId = "<tag>_<name>", чтобы Dart-клиент получил методы вроде accountsList, а не list_accounts_api_v1_accounts_get. Ошибки — application/problem+json. Auth: access-JWT на час + refresh на 30 дней, который хранится хэшем и ротируется, логин ограничен по частоте в памяти. Деньги в JSON — всегда позиционные строки (api/schemas/common.py): float в проводе потерял бы копейки, которые NUMERIC(24,10) бережёт. Тестовый harness поднимает свой Postgres через pg_ctl (pytest-postgresql), мигрирует его один раз на сессию и усекает таблицы после каждого теста.
This commit is contained in:
@@ -0,0 +1,31 @@
|
||||
"""Tiny in-memory sliding-window limiter for the login endpoint.
|
||||
|
||||
Single-process API, single user: no need for Redis. Caddy adds nothing here by
|
||||
default, so this is the only brute-force protection — keep it.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from collections import defaultdict, deque
|
||||
|
||||
|
||||
class SlidingWindowLimiter:
|
||||
def __init__(self, limit: int, window_seconds: float = 60.0) -> None:
|
||||
self.limit = limit
|
||||
self.window = window_seconds
|
||||
self._hits: dict[str, deque[float]] = defaultdict(deque)
|
||||
|
||||
def hit(self, key: str) -> float | None:
|
||||
"""Register an attempt. Returns seconds to wait if over the limit, else None."""
|
||||
now = time.monotonic()
|
||||
q = self._hits[key]
|
||||
while q and now - q[0] > self.window:
|
||||
q.popleft()
|
||||
if len(q) >= self.limit:
|
||||
return self.window - (now - q[0])
|
||||
q.append(now)
|
||||
return None
|
||||
|
||||
def reset(self, key: str) -> None:
|
||||
self._hits.pop(key, None)
|
||||
Reference in New Issue
Block a user