644 lines
22 KiB
Python
644 lines
22 KiB
Python
"""Однофайловый клиент для SheetScheduleParser API.
|
||
|
||
Файл можно просто скопировать в другой проект. Внешние зависимости не нужны.
|
||
|
||
Пример::
|
||
|
||
import schedule_client as schedule
|
||
|
||
schedule.configure("http://schedule.example.com", cache_ttl=300)
|
||
db = schedule.ScheduleDB()
|
||
|
||
lessons = (
|
||
db.query(schedule.ScheduleEntry)
|
||
.filter_by(schedule_type="default", group_ids="АТ-501", weekday=1)
|
||
.order_by(schedule.Schedule.time_start)
|
||
.all()
|
||
)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import threading
|
||
import time
|
||
from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
|
||
from dataclasses import dataclass, field
|
||
from typing import Any, Generic, TypeVar
|
||
from urllib.error import HTTPError, URLError
|
||
from urllib.parse import quote
|
||
from urllib.request import Request, urlopen
|
||
|
||
|
||
# Эти значения можно изменить прямо в скопированном файле или через configure().
|
||
API_URL = os.getenv("SCHEDULE_API_URL", "http://127.0.0.1:8000")
|
||
# Должно в точности совпадать с app.version.API_VERSION на сервере.
|
||
API_VERSION = "1"
|
||
CACHE_TTL = float(os.getenv("SCHEDULE_CACHE_TTL", "300"))
|
||
REQUEST_TIMEOUT = float(os.getenv("SCHEDULE_REQUEST_TIMEOUT", "15"))
|
||
FAILURE_RETRY_DELAY = float(os.getenv("SCHEDULE_FAILURE_RETRY_DELAY", "15"))
|
||
|
||
|
||
class ScheduleClientError(RuntimeError):
|
||
"""Базовая ошибка клиента."""
|
||
|
||
|
||
class APIError(ScheduleClientError):
|
||
"""API недоступно или вернуло некорректный ответ."""
|
||
|
||
|
||
class IncompatibleAPIVersionError(ScheduleClientError):
|
||
"""Версия контракта API не поддерживается этим файлом клиента."""
|
||
|
||
|
||
class NoResultFound(ScheduleClientError):
|
||
"""Query.one() не нашёл ни одной записи."""
|
||
|
||
|
||
class MultipleResultsFound(ScheduleClientError):
|
||
"""Query.one() нашёл больше одной записи."""
|
||
|
||
|
||
class Expression:
|
||
def __init__(self, predicate: Callable[["Model"], bool], label: str = "") -> None:
|
||
self._predicate = predicate
|
||
self._label = label
|
||
|
||
def __call__(self, item: "Model") -> bool:
|
||
return self._predicate(item)
|
||
|
||
def __and__(self, other: "Expression") -> "Expression":
|
||
return Expression(lambda item: self(item) and other(item))
|
||
|
||
def __or__(self, other: "Expression") -> "Expression":
|
||
return Expression(lambda item: self(item) or other(item))
|
||
|
||
def __invert__(self) -> "Expression":
|
||
return Expression(lambda item: not self(item))
|
||
|
||
def __repr__(self) -> str:
|
||
return self._label or "Expression(...)"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SortExpression:
|
||
name: str
|
||
reverse: bool = False
|
||
|
||
|
||
class Field:
|
||
"""Дескриптор поля модели и строитель выражений для Query.filter()."""
|
||
|
||
def __init__(self) -> None:
|
||
self.name = ""
|
||
|
||
def __set_name__(self, owner: type["Model"], name: str) -> None:
|
||
self.name = name
|
||
|
||
def __get__(self, instance: "Model | None", owner: type["Model"]) -> Any:
|
||
if instance is None:
|
||
return self
|
||
return instance._data.get(self.name)
|
||
|
||
def _compare(self, operation: Callable[[Any, Any], bool], value: Any, symbol: str) -> Expression:
|
||
return Expression(
|
||
lambda item: operation(item._data.get(self.name), value),
|
||
f"{self.name} {symbol} {value!r}",
|
||
)
|
||
|
||
def __eq__(self, value: Any) -> Expression: # type: ignore[override]
|
||
return self._compare(lambda actual, expected: actual == expected, value, "==")
|
||
|
||
def __ne__(self, value: Any) -> Expression: # type: ignore[override]
|
||
return self._compare(lambda actual, expected: actual != expected, value, "!=")
|
||
|
||
def __lt__(self, value: Any) -> Expression:
|
||
return self._compare(_safe_lt, value, "<")
|
||
|
||
def __le__(self, value: Any) -> Expression:
|
||
return self._compare(_safe_le, value, "<=")
|
||
|
||
def __gt__(self, value: Any) -> Expression:
|
||
return self._compare(_safe_gt, value, ">")
|
||
|
||
def __ge__(self, value: Any) -> Expression:
|
||
return self._compare(_safe_ge, value, ">=")
|
||
|
||
def contains(self, value: Any) -> Expression:
|
||
return Expression(
|
||
lambda item: _contains(item._data.get(self.name), value),
|
||
f"{self.name}.contains({value!r})",
|
||
)
|
||
|
||
def in_(self, values: Iterable[Any]) -> Expression:
|
||
choices = tuple(values)
|
||
return Expression(
|
||
lambda item: item._data.get(self.name) in choices,
|
||
f"{self.name}.in_({choices!r})",
|
||
)
|
||
|
||
def is_(self, value: Any) -> Expression:
|
||
return Expression(lambda item: item._data.get(self.name) is value)
|
||
|
||
def startswith(self, value: str) -> Expression:
|
||
return Expression(
|
||
lambda item: str(item._data.get(self.name) or "").startswith(value)
|
||
)
|
||
|
||
def asc(self) -> SortExpression:
|
||
return SortExpression(self.name)
|
||
|
||
def desc(self) -> SortExpression:
|
||
return SortExpression(self.name, reverse=True)
|
||
|
||
|
||
class Model:
|
||
"""Неизменяемая по соглашению запись из API с доступом через атрибуты."""
|
||
|
||
def __init__(self, data: Mapping[str, Any]) -> None:
|
||
self._data = dict(data)
|
||
|
||
def as_dict(self) -> dict[str, Any]:
|
||
return dict(self._data)
|
||
|
||
def __getitem__(self, key: str) -> Any:
|
||
return self._data[key]
|
||
|
||
def __repr__(self) -> str:
|
||
identity = self._data.get("id", self._data.get("name", "?"))
|
||
return f"{type(self).__name__}(id={identity!r})"
|
||
|
||
|
||
class Schedule(Model):
|
||
id = Field()
|
||
schedule_type = Field()
|
||
name = Field()
|
||
type = Field()
|
||
weekday = Field()
|
||
is_odd_week = Field()
|
||
is_even_week = Field()
|
||
time_start = Field()
|
||
time_end = Field()
|
||
datetime = Field()
|
||
group_ids = Field()
|
||
room_ids = Field()
|
||
teacher_ids = Field()
|
||
teacher_assignments = Field()
|
||
created_at = Field()
|
||
parsed_at = Field()
|
||
|
||
|
||
# Более явное имя для импорта; Schedule оставлено как короткий вариант.
|
||
ScheduleEntry = Schedule
|
||
|
||
|
||
class Group(Model):
|
||
id = Field()
|
||
name = Field()
|
||
faculty = Field()
|
||
course = Field()
|
||
parent_group_id = Field()
|
||
created_at = Field()
|
||
updated_at = Field()
|
||
last_seen_at = Field()
|
||
|
||
|
||
class Teacher(Model):
|
||
id = Field()
|
||
name = Field()
|
||
positions = Field()
|
||
created_at = Field()
|
||
updated_at = Field()
|
||
last_seen_at = Field()
|
||
|
||
|
||
class Room(Model):
|
||
id = Field()
|
||
name = Field()
|
||
created_at = Field()
|
||
updated_at = Field()
|
||
last_seen_at = Field()
|
||
|
||
|
||
class GroupHistory(Model):
|
||
faculty = Field()
|
||
course = Field()
|
||
parent_group_id = Field()
|
||
valid_from = Field()
|
||
valid_to = Field()
|
||
|
||
|
||
M = TypeVar("M", bound=Model)
|
||
|
||
|
||
class Query(Generic[M]):
|
||
"""Небольшое неизменяемое подмножество привычного SQLAlchemy Query API."""
|
||
|
||
def __init__(
|
||
self,
|
||
client: "ScheduleDB",
|
||
dataset: str,
|
||
model: type[M],
|
||
predicates: tuple[Expression, ...] = (),
|
||
sort_expressions: tuple[SortExpression, ...] = (),
|
||
offset_value: int = 0,
|
||
limit_value: int | None = None,
|
||
) -> None:
|
||
self._client = client
|
||
self._dataset = dataset
|
||
self._model = model
|
||
self._predicates = predicates
|
||
self._sort_expressions = sort_expressions
|
||
self._offset = offset_value
|
||
self._limit = limit_value
|
||
|
||
def _clone(self, **changes: Any) -> "Query[M]":
|
||
values = {
|
||
"client": self._client,
|
||
"dataset": self._dataset,
|
||
"model": self._model,
|
||
"predicates": self._predicates,
|
||
"sort_expressions": self._sort_expressions,
|
||
"offset_value": self._offset,
|
||
"limit_value": self._limit,
|
||
}
|
||
values.update(changes)
|
||
return Query(**values)
|
||
|
||
def filter(self, *criteria: Expression | Callable[[M], bool]) -> "Query[M]":
|
||
expressions: list[Expression] = []
|
||
for criterion in criteria:
|
||
if isinstance(criterion, Expression):
|
||
expressions.append(criterion)
|
||
elif callable(criterion):
|
||
expressions.append(Expression(criterion))
|
||
else:
|
||
raise TypeError("filter() принимает выражения полей или callable")
|
||
return self._clone(predicates=self._predicates + tuple(expressions))
|
||
|
||
def filter_by(self, **values: Any) -> "Query[M]":
|
||
expressions: list[Expression] = []
|
||
for name, expected in values.items():
|
||
if not hasattr(self._model, name):
|
||
raise AttributeError(f"У {self._model.__name__} нет поля {name!r}")
|
||
expressions.append(
|
||
Expression(
|
||
lambda item, key=name, value=expected: _filter_by_match(
|
||
item._data.get(key), value
|
||
),
|
||
f"{name}={expected!r}",
|
||
)
|
||
)
|
||
return self._clone(predicates=self._predicates + tuple(expressions))
|
||
|
||
def order_by(self, *fields: Field | SortExpression | str) -> "Query[M]":
|
||
expressions: list[SortExpression] = []
|
||
for value in fields:
|
||
if isinstance(value, Field):
|
||
expressions.append(value.asc())
|
||
elif isinstance(value, SortExpression):
|
||
expressions.append(value)
|
||
elif isinstance(value, str):
|
||
expressions.append(SortExpression(value))
|
||
else:
|
||
raise TypeError("order_by() принимает поле, field.asc()/desc() или имя поля")
|
||
return self._clone(sort_expressions=tuple(expressions))
|
||
|
||
def limit(self, value: int) -> "Query[M]":
|
||
if value < 0:
|
||
raise ValueError("limit не может быть отрицательным")
|
||
return self._clone(limit_value=value)
|
||
|
||
def offset(self, value: int) -> "Query[M]":
|
||
if value < 0:
|
||
raise ValueError("offset не может быть отрицательным")
|
||
return self._clone(offset_value=value)
|
||
|
||
def all(self) -> list[M]:
|
||
records = list(self._client._records(self._dataset))
|
||
for predicate in self._predicates:
|
||
records = [item for item in records if predicate(item)]
|
||
for expression in reversed(self._sort_expressions):
|
||
records.sort(
|
||
key=lambda item, name=expression.name: _sort_key(item._data.get(name)),
|
||
reverse=expression.reverse,
|
||
)
|
||
end = None if self._limit is None else self._offset + self._limit
|
||
return records[self._offset : end]
|
||
|
||
def first(self) -> M | None:
|
||
records = self.limit(1).all()
|
||
return records[0] if records else None
|
||
|
||
def one_or_none(self) -> M | None:
|
||
records = self.limit(2).all()
|
||
if len(records) > 1:
|
||
raise MultipleResultsFound("Ожидалась максимум одна запись")
|
||
return records[0] if records else None
|
||
|
||
def one(self) -> M:
|
||
result = self.one_or_none()
|
||
if result is None:
|
||
raise NoResultFound("Запись не найдена")
|
||
return result
|
||
|
||
def count(self) -> int:
|
||
return len(self.all())
|
||
|
||
def get(self, entity_id: str) -> M | None:
|
||
return self.filter_by(id=entity_id).first()
|
||
|
||
def __iter__(self) -> Iterator[M]:
|
||
return iter(self.all())
|
||
|
||
|
||
@dataclass
|
||
class _CacheState:
|
||
lock: threading.RLock = field(default_factory=threading.RLock)
|
||
datasets: dict[str, tuple[Model, ...]] = field(default_factory=dict)
|
||
refreshed_monotonic: float = 0.0
|
||
refreshed_at: str | None = None
|
||
retry_after: float = 0.0
|
||
last_error: str | None = None
|
||
|
||
|
||
_CACHE_LOCK = threading.Lock()
|
||
_GLOBAL_CACHE: dict[str, _CacheState] = {}
|
||
|
||
|
||
def configure(
|
||
api_url: str,
|
||
*,
|
||
cache_ttl: float | None = None,
|
||
request_timeout: float | None = None,
|
||
) -> None:
|
||
"""Задаёт глобальные настройки для последующих ScheduleDB()."""
|
||
|
||
global API_URL, CACHE_TTL, REQUEST_TIMEOUT
|
||
API_URL = api_url.rstrip("/")
|
||
if cache_ttl is not None:
|
||
CACHE_TTL = float(cache_ttl)
|
||
if request_timeout is not None:
|
||
REQUEST_TIMEOUT = float(request_timeout)
|
||
|
||
|
||
def clear_cache(api_url: str | None = None) -> None:
|
||
"""Очищает один глобальный снимок или весь кеш процесса."""
|
||
|
||
with _CACHE_LOCK:
|
||
if api_url is None:
|
||
states = list(_GLOBAL_CACHE.values())
|
||
_GLOBAL_CACHE.clear()
|
||
else:
|
||
state = _GLOBAL_CACHE.pop(api_url.rstrip("/"), None)
|
||
states = [state] if state is not None else []
|
||
# Сбрасываем и уже выданные клиентам state, а не только запись в словаре.
|
||
for state in states:
|
||
with state.lock:
|
||
state.datasets.clear()
|
||
state.refreshed_monotonic = 0.0
|
||
state.refreshed_at = None
|
||
state.retry_after = 0.0
|
||
state.last_error = None
|
||
|
||
|
||
class ScheduleDB:
|
||
"""Точка входа к кешированной копии всех основных справочников API."""
|
||
|
||
def __init__(
|
||
self,
|
||
api_url: str | None = None,
|
||
*,
|
||
cache_ttl: float | None = None,
|
||
request_timeout: float | None = None,
|
||
eager: bool = True,
|
||
) -> None:
|
||
self.api_url = (api_url or API_URL).rstrip("/")
|
||
self.cache_ttl = CACHE_TTL if cache_ttl is None else float(cache_ttl)
|
||
self.request_timeout = (
|
||
REQUEST_TIMEOUT if request_timeout is None else float(request_timeout)
|
||
)
|
||
with _CACHE_LOCK:
|
||
self._state = _GLOBAL_CACHE.setdefault(self.api_url, _CacheState())
|
||
# Проверяется при каждом создании клиента, даже если общий кеш уже заполнен.
|
||
self._check_compatibility()
|
||
if eager:
|
||
self._ensure_fresh(require_data=True, compatibility_checked=True)
|
||
|
||
@property
|
||
def schedule(self) -> Query[Schedule]:
|
||
return Query(self, "schedule", Schedule)
|
||
|
||
@property
|
||
def default(self) -> Query[Schedule]:
|
||
return Query(self, "default", Schedule)
|
||
|
||
@property
|
||
def exam(self) -> Query[Schedule]:
|
||
return Query(self, "exam", Schedule)
|
||
|
||
@property
|
||
def groups(self) -> Query[Group]:
|
||
return Query(self, "groups", Group)
|
||
|
||
@property
|
||
def teachers(self) -> Query[Teacher]:
|
||
return Query(self, "teachers", Teacher)
|
||
|
||
@property
|
||
def rooms(self) -> Query[Room]:
|
||
return Query(self, "rooms", Room)
|
||
|
||
def query(self, model: type[M]) -> Query[M]:
|
||
"""SQLAlchemy-подобная форма: db.query(ScheduleEntry).filter(...)."""
|
||
|
||
datasets: dict[type[Model], str] = {
|
||
Schedule: "schedule",
|
||
Group: "groups",
|
||
Teacher: "teachers",
|
||
Room: "rooms",
|
||
}
|
||
dataset = datasets.get(model)
|
||
if dataset is None:
|
||
raise TypeError(
|
||
"query() поддерживает ScheduleEntry, Group, Teacher и Room; "
|
||
"для истории используйте group_history(group_id)"
|
||
)
|
||
return Query(self, dataset, model)
|
||
|
||
def group_history(self, group_id: str) -> Query[GroupHistory]:
|
||
"""Загружает историю группы лениво и дальше хранит в общем кеше."""
|
||
|
||
dataset = f"group_history:{group_id}"
|
||
return Query(self, dataset, GroupHistory)
|
||
|
||
def refresh(self) -> None:
|
||
"""Принудительно и атомарно обновляет все основные наборы или бросает APIError."""
|
||
|
||
with self._state.lock:
|
||
try:
|
||
self._refresh_locked()
|
||
except APIError as error:
|
||
self._state.last_error = str(error)
|
||
self._state.retry_after = time.monotonic() + FAILURE_RETRY_DELAY
|
||
raise
|
||
|
||
def cache_info(self) -> dict[str, Any]:
|
||
with self._state.lock:
|
||
return {
|
||
"api_url": self.api_url,
|
||
"refreshed_at": self._state.refreshed_at,
|
||
"last_error": self._state.last_error,
|
||
"counts": {
|
||
name: len(items)
|
||
for name, items in self._state.datasets.items()
|
||
if not name.startswith("group_history:")
|
||
},
|
||
}
|
||
|
||
def _records(self, dataset: str) -> Sequence[Model]:
|
||
self._ensure_fresh(require_data=True)
|
||
with self._state.lock:
|
||
if dataset.startswith("group_history:") and dataset not in self._state.datasets:
|
||
group_id = dataset.partition(":")[2]
|
||
response = self._fetch(f"/groups/{quote(group_id, safe='')}/history")
|
||
self._state.datasets[dataset] = tuple(
|
||
GroupHistory(item) for item in response["result"]
|
||
)
|
||
return self._state.datasets.get(dataset, ())
|
||
|
||
def _ensure_fresh(
|
||
self,
|
||
*,
|
||
require_data: bool,
|
||
compatibility_checked: bool = False,
|
||
) -> None:
|
||
with self._state.lock:
|
||
now = time.monotonic()
|
||
has_data = bool(self._state.datasets)
|
||
stale = not has_data or now - self._state.refreshed_monotonic >= self.cache_ttl
|
||
if not stale or (has_data and now < self._state.retry_after):
|
||
return
|
||
try:
|
||
self._refresh_locked(check_compatibility=not compatibility_checked)
|
||
except APIError as error:
|
||
self._state.last_error = str(error)
|
||
self._state.retry_after = time.monotonic() + FAILURE_RETRY_DELAY
|
||
if require_data and not has_data:
|
||
raise
|
||
|
||
def _refresh_locked(self, *, check_compatibility: bool = True) -> None:
|
||
if check_compatibility:
|
||
self._check_compatibility()
|
||
endpoints = {
|
||
"default": ("/schedule/default", Schedule),
|
||
"exam": ("/schedule/exam", Schedule),
|
||
"groups": ("/groups", Group),
|
||
"teachers": ("/teachers", Teacher),
|
||
"rooms": ("/rooms", Room),
|
||
}
|
||
fresh: dict[str, tuple[Model, ...]] = {}
|
||
for name, (path, model) in endpoints.items():
|
||
response = self._fetch(path)
|
||
records: list[Model] = []
|
||
for raw_item in response["result"]:
|
||
item = dict(raw_item)
|
||
if name in {"default", "exam"}:
|
||
item.setdefault("schedule_type", name)
|
||
records.append(model(item))
|
||
fresh[name] = tuple(records)
|
||
fresh["schedule"] = fresh["default"] + fresh["exam"]
|
||
|
||
# Истории групп инвалидируются: после обновления основной базы они могли измениться.
|
||
self._state.datasets = fresh
|
||
self._state.refreshed_monotonic = time.monotonic()
|
||
self._state.refreshed_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
||
self._state.retry_after = 0.0
|
||
self._state.last_error = None
|
||
|
||
def _fetch(self, path: str) -> dict[str, Any]:
|
||
return _request_json(f"{self.api_url}{path}", self.request_timeout)
|
||
|
||
def _check_compatibility(self) -> None:
|
||
response = self._fetch("/version")
|
||
actual = response.get("api_version")
|
||
if actual != API_VERSION:
|
||
raise IncompatibleAPIVersionError(
|
||
f"Несовместимая версия API: сервер={actual!r}, "
|
||
f"клиент ожидает {API_VERSION!r}"
|
||
)
|
||
|
||
|
||
def _request_json(url: str, timeout: float) -> dict[str, Any]:
|
||
request = Request(url, headers={"Accept": "application/json"})
|
||
try:
|
||
with urlopen(request, timeout=timeout) as response:
|
||
payload = json.load(response)
|
||
except HTTPError as error:
|
||
raise APIError(f"API вернуло HTTP {error.code} для {url}") from error
|
||
except (URLError, TimeoutError, OSError) as error:
|
||
raise APIError(f"Не удалось обратиться к {url}: {error}") from error
|
||
except (UnicodeDecodeError, json.JSONDecodeError) as error:
|
||
raise APIError(f"API вернуло невалидный JSON для {url}") from error
|
||
if not isinstance(payload, dict) or payload.get("ok") is not True:
|
||
raise APIError(f"Некорректный ответ API для {url}")
|
||
return payload
|
||
|
||
|
||
def _filter_by_match(actual: Any, expected: Any) -> bool:
|
||
# Самый частый запрос — filter_by(group_ids="АТ-501"). Для списковых полей
|
||
# одиночное значение означает contains, а коллекция — точное сравнение.
|
||
if isinstance(actual, (list, tuple, set)) and not isinstance(
|
||
expected, (list, tuple, set)
|
||
):
|
||
return expected in actual
|
||
return actual == expected
|
||
|
||
|
||
def _contains(container: Any, value: Any) -> bool:
|
||
try:
|
||
return value in container
|
||
except (TypeError, AttributeError):
|
||
return False
|
||
|
||
|
||
def _sort_key(value: Any) -> tuple[bool, Any]:
|
||
return value is None, value
|
||
|
||
|
||
def _safe_lt(left: Any, right: Any) -> bool:
|
||
return left is not None and right is not None and left < right
|
||
|
||
|
||
def _safe_le(left: Any, right: Any) -> bool:
|
||
return left is not None and right is not None and left <= right
|
||
|
||
|
||
def _safe_gt(left: Any, right: Any) -> bool:
|
||
return left is not None and right is not None and left > right
|
||
|
||
|
||
def _safe_ge(left: Any, right: Any) -> bool:
|
||
return left is not None and right is not None and left >= right
|
||
|
||
|
||
__all__ = [
|
||
"API_URL",
|
||
"API_VERSION",
|
||
"CACHE_TTL",
|
||
"APIError",
|
||
"IncompatibleAPIVersionError",
|
||
"NoResultFound",
|
||
"MultipleResultsFound",
|
||
"ScheduleDB",
|
||
"Schedule",
|
||
"ScheduleEntry",
|
||
"Group",
|
||
"Teacher",
|
||
"Room",
|
||
"GroupHistory",
|
||
"Query",
|
||
"configure",
|
||
"clear_cache",
|
||
]
|