144 lines
5.5 KiB
Python
144 lines
5.5 KiB
Python
"""
|
||
Порты (интерфейсы) для hexagonal architecture
|
||
"""
|
||
from abc import ABC, abstractmethod
|
||
from typing import Optional, Dict, List, Any, Callable
|
||
import pandas as pd
|
||
|
||
|
||
class ParserPort(ABC):
|
||
"""Интерфейс для парсеров с поддержкой множественных геттеров"""
|
||
|
||
def __init__(self):
|
||
"""Инициализация с пустым словарем геттеров"""
|
||
self.getters: Dict[str, Dict[str, Any]] = {}
|
||
self._register_default_getters()
|
||
|
||
def _register_default_getters(self):
|
||
"""Регистрация геттеров по умолчанию - переопределяется в наследниках"""
|
||
pass
|
||
|
||
def register_getter(self, name: str, method: Callable, required_params: List[str],
|
||
optional_params: List[str] = None, description: str = ""):
|
||
"""
|
||
Регистрация нового геттера
|
||
|
||
Args:
|
||
name: Имя геттера
|
||
method: Метод для выполнения
|
||
required_params: Список обязательных параметров
|
||
optional_params: Список необязательных параметров
|
||
description: Описание геттера
|
||
"""
|
||
if optional_params is None:
|
||
optional_params = []
|
||
|
||
self.getters[name] = {
|
||
"method": method,
|
||
"required_params": required_params,
|
||
"optional_params": optional_params,
|
||
"description": description
|
||
}
|
||
|
||
def get_available_getters(self) -> Dict[str, Dict[str, Any]]:
|
||
"""Получение списка доступных геттеров с их описанием"""
|
||
return {
|
||
name: {
|
||
"required_params": info["required_params"],
|
||
"optional_params": info["optional_params"],
|
||
"description": info["description"]
|
||
}
|
||
for name, info in self.getters.items()
|
||
}
|
||
|
||
# Добавить схему
|
||
def get_value(self, getter_name: str, params: Dict[str, Any]):
|
||
"""
|
||
Получение значения через указанный геттер
|
||
|
||
Args:
|
||
getter_name: Имя геттера
|
||
params: Параметры для геттера
|
||
|
||
Returns:
|
||
Результат выполнения геттера
|
||
|
||
Raises:
|
||
ValueError: Если геттер не найден или параметры неверны
|
||
"""
|
||
if getter_name not in self.getters:
|
||
available = list(self.getters.keys())
|
||
raise ValueError(f"Геттер '{getter_name}' не найден. Доступные: {available}")
|
||
|
||
getter_info = self.getters[getter_name]
|
||
required = getter_info["required_params"]
|
||
|
||
# Проверка обязательных параметров
|
||
missing = [p for p in required if p not in params]
|
||
if missing:
|
||
raise ValueError(f"Отсутствуют обязательные параметры для геттера '{getter_name}': {missing}")
|
||
|
||
# Вызов метода геттера
|
||
try:
|
||
return getter_info["method"](params)
|
||
except Exception as e:
|
||
raise ValueError(f"Ошибка выполнения геттера '{getter_name}': {str(e)}")
|
||
|
||
def determine_getter(self, get_params: Dict[str, Any]) -> str:
|
||
"""
|
||
Определение имени геттера на основе параметров запроса
|
||
|
||
Args:
|
||
get_params: Параметры запроса
|
||
|
||
Returns:
|
||
Имя геттера для выполнения
|
||
|
||
Raises:
|
||
ValueError: Если не удается определить геттер
|
||
"""
|
||
# По умолчанию используем первый доступный геттер
|
||
available_getters = list(self.getters.keys())
|
||
if not available_getters:
|
||
raise ValueError("Парсер не имеет доступных геттеров")
|
||
|
||
# Если указан режим, используем его
|
||
if 'mode' in get_params:
|
||
mode = get_params['mode']
|
||
if mode in self.getters:
|
||
return mode
|
||
else:
|
||
raise ValueError(f"Режим '{mode}' не найден. Доступные: {available_getters}")
|
||
|
||
# Иначе используем первый доступный
|
||
return available_getters[0]
|
||
|
||
@abstractmethod
|
||
def parse(self, file_path: str, params: dict) -> pd.DataFrame:
|
||
"""Парсинг файла и возврат DataFrame"""
|
||
pass
|
||
|
||
|
||
class StoragePort(ABC):
|
||
"""Интерфейс для хранилища данных"""
|
||
|
||
@abstractmethod
|
||
def save_dataframe(self, df: pd.DataFrame, object_id: str) -> bool:
|
||
"""Сохранение DataFrame"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def load_dataframe(self, object_id: str) -> Optional[pd.DataFrame]:
|
||
"""Загрузка DataFrame"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def delete_object(self, object_id: str) -> bool:
|
||
"""Удаление объекта"""
|
||
pass
|
||
|
||
@abstractmethod
|
||
def object_exists(self, object_id: str) -> bool:
|
||
"""Проверка существования объекта"""
|
||
pass
|