- Исправлен импорт core.celery в тестах health-check вместо устаревшего config.celery - Добавлен fallback для Checko datasets при отсутствии JSON-файлов в CI - Обновлен BaseDataset: загрузка встроенных данных при missing JSON - Прокинуты TG_BOT_KEY/TG_CHANNEL из secrets в jobs lint/test/build_push
188 lines
5.5 KiB
Python
188 lines
5.5 KiB
Python
"""
|
||
Базовые классы для работы со справочниками Checko.
|
||
|
||
Справочники загружаются лениво (lazy loading) при первом обращении.
|
||
Данные кэшируются на уровне класса (singleton pattern).
|
||
"""
|
||
|
||
import json
|
||
from dataclasses import dataclass
|
||
from pathlib import Path
|
||
from typing import ClassVar, Generic, TypeVar
|
||
|
||
T = TypeVar("T")
|
||
|
||
# Путь к директории с JSON файлами
|
||
DATA_DIR = Path(__file__).parent / "data"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class DatasetItem:
|
||
"""Базовый элемент справочника."""
|
||
|
||
code: str
|
||
name: str
|
||
|
||
|
||
class BaseDataset(Generic[T]):
|
||
"""
|
||
Базовый класс для работы со справочниками.
|
||
|
||
Реализует:
|
||
- Lazy loading данных при первом обращении
|
||
- Кэширование на уровне класса
|
||
- Поиск по коду и названию
|
||
|
||
Наследники должны определить:
|
||
- _json_filename: имя JSON файла
|
||
- _parse_item: метод парсинга элемента из JSON
|
||
"""
|
||
|
||
_data: ClassVar[dict[str, T] | None] = None
|
||
_json_filename: ClassVar[str] = ""
|
||
_builtin_raw_data: ClassVar[list[dict]] = []
|
||
|
||
@classmethod
|
||
def _get_json_path(cls) -> Path:
|
||
"""Получить путь к JSON файлу."""
|
||
return DATA_DIR / cls._json_filename
|
||
|
||
@classmethod
|
||
def _parse_item(cls, raw: dict) -> T:
|
||
"""Распарсить элемент из JSON. Переопределяется в наследниках."""
|
||
raise NotImplementedError
|
||
|
||
@classmethod
|
||
def _get_item_code(cls, raw: dict) -> str:
|
||
"""Получить код элемента из JSON."""
|
||
return raw.get("code", "")
|
||
|
||
@classmethod
|
||
def _ensure_loaded(cls) -> None:
|
||
"""Загрузить данные если ещё не загружены."""
|
||
if cls._data is not None:
|
||
return
|
||
|
||
json_path = cls._get_json_path()
|
||
if not json_path.exists():
|
||
if cls._builtin_raw_data:
|
||
cls._data = {}
|
||
for raw in cls._builtin_raw_data:
|
||
item = cls._parse_item(raw)
|
||
code = cls._get_item_code(raw)
|
||
cls._data[code] = item
|
||
return
|
||
cls._data = {}
|
||
return
|
||
|
||
with open(json_path, encoding="utf-8") as f:
|
||
raw_data = json.load(f)
|
||
|
||
cls._data = {}
|
||
for raw in raw_data:
|
||
item = cls._parse_item(raw)
|
||
code = cls._get_item_code(raw)
|
||
cls._data[code] = item
|
||
|
||
@classmethod
|
||
def get(cls, code: str) -> T | None:
|
||
"""
|
||
Получить элемент по коду.
|
||
|
||
Args:
|
||
code: Код элемента справочника.
|
||
|
||
Returns:
|
||
Элемент справочника или None если не найден.
|
||
"""
|
||
cls._ensure_loaded()
|
||
return cls._data.get(code)
|
||
|
||
@classmethod
|
||
def get_name(cls, code: str) -> str | None:
|
||
"""
|
||
Получить название по коду.
|
||
|
||
Args:
|
||
code: Код элемента справочника.
|
||
|
||
Returns:
|
||
Название элемента или None если не найден.
|
||
"""
|
||
item = cls.get(code)
|
||
if item is None:
|
||
return None
|
||
return getattr(item, "name", None) or getattr(item, "full_name", None)
|
||
|
||
@classmethod
|
||
def all(cls) -> list[T]:
|
||
"""
|
||
Получить все элементы справочника.
|
||
|
||
Returns:
|
||
Список всех элементов.
|
||
"""
|
||
cls._ensure_loaded()
|
||
return list(cls._data.values())
|
||
|
||
@classmethod
|
||
def codes(cls) -> list[str]:
|
||
"""
|
||
Получить все коды справочника.
|
||
|
||
Returns:
|
||
Список всех кодов.
|
||
"""
|
||
cls._ensure_loaded()
|
||
return list(cls._data.keys())
|
||
|
||
@classmethod
|
||
def search(cls, query: str) -> list[T]:
|
||
"""
|
||
Поиск по названию (регистронезависимый).
|
||
|
||
Args:
|
||
query: Поисковый запрос.
|
||
|
||
Returns:
|
||
Список найденных элементов.
|
||
"""
|
||
cls._ensure_loaded()
|
||
q = query.lower()
|
||
results = []
|
||
for item in cls._data.values():
|
||
name = getattr(item, "name", "") or getattr(item, "full_name", "")
|
||
if name and q in name.lower():
|
||
results.append(item)
|
||
return results
|
||
|
||
@classmethod
|
||
def exists(cls, code: str) -> bool:
|
||
"""
|
||
Проверить существование кода.
|
||
|
||
Args:
|
||
code: Код элемента справочника.
|
||
|
||
Returns:
|
||
True если код существует.
|
||
"""
|
||
return cls.get(code) is not None
|
||
|
||
@classmethod
|
||
def count(cls) -> int:
|
||
"""
|
||
Получить количество элементов в справочнике.
|
||
|
||
Returns:
|
||
Количество элементов.
|
||
"""
|
||
cls._ensure_loaded()
|
||
return len(cls._data)
|
||
|
||
@classmethod
|
||
def reload(cls) -> None:
|
||
"""Перезагрузить данные из JSON файла."""
|
||
cls._data = None
|
||
cls._ensure_loaded()
|