ips_etl/src/state.py

65 lines
2.2 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import abc
from typing import Any, Dict
import json
import os
class BaseStorage(abc.ABC):
"""Абстрактное хранилище состояния.
Позволяет сохранять и получать состояние.
"""
@abc.abstractmethod
def save_state(self, state: Dict[str, Any]) -> None:
"""Сохранить состояние в хранилище."""
raise NotImplementedError
@abc.abstractmethod
def retrieve_state(self) -> Dict[str, Any]:
"""Получить состояние из хранилища."""
raise NotImplementedError
class JsonFileStorage(BaseStorage):
"""Реализация хранилища, использующего локальный файл.
Формат хранения: JSON
"""
def __init__(self, file_path: str = 'state.json') -> None:
self.file_path = file_path
try:
with open(file_path, 'r', encoding='UTF-8') as f:
self.local_storage = json.load(f)
except FileNotFoundError:
self.local_storage = {}
def save_state(self, state: Dict[str, Any]) -> None:
"""Сохранить состояние в хранилище."""
with open(self.file_path, 'w+', encoding='UTF-8') as f:
json.dump(state, f, ensure_ascii=False)
def retrieve_state(self) -> Dict[str, Any]:
"""Получить состояние из хранилища."""
if not os.path.exists(self.file_path):
return {}
with open(self.file_path, 'r') as file:
return json.load(file)
class State:
"""Класс для работы с состояниями."""
def __init__(self, storage: BaseStorage) -> None:
self.storage = storage
self.local_state = storage.retrieve_state()
def set_state(self, key: str, value: Any) -> None:
"""Установить состояние для определённого ключа."""
self.local_state[key] = value
def get_state(self, key: str) -> Any:
"""Получить состояние по определённому ключу."""
return self.local_state.get(key)