Skip to content

Instantly share code, notes, and snippets.

@discrimy
Created November 30, 2019 12:18
Show Gist options
  • Save discrimy/547cc94a1aa4a170d2e17ea72fcc7fe8 to your computer and use it in GitHub Desktop.
Save discrimy/547cc94a1aa4a170d2e17ea72fcc7fe8 to your computer and use it in GitHub Desktop.
from typing import Set
# Библиотека pyyaml (https://pyyaml.org/wiki/PyYAMLDocumentation)
import yaml
class IllegalServicesDependenciesError(RuntimeError):
pass
class Service:
"""
Класс, описывающий сервис
"""
name: str
depends_on: Set[str]
initialized: bool = False
def __init__(self, name: str, depends_on: Set[str]):
self.name = name
self.depends_on = depends_on
def init(self):
print(f'Инициализация сервиса {self.name}')
self.initialized = True
class ServiceManager:
services: Set[Service] = set()
@classmethod
def from_yaml(cls, file_path: str):
"""
Фабричный метод, создающий объект менеджера из YAML файла конфигурации
:param file_path: путь до файла
:return: Объект ServiceManager
"""
manager = cls()
with open(file_path, 'r') as file:
conf_raw = yaml.load(file, Loader=yaml.SafeLoader)
for service_raw in conf_raw['services']:
name = service_raw['name']
depends_on = set(service_raw['dependsOn']) \
if 'dependsOn' in service_raw and service_raw['dependsOn'] is not None \
else set()
manager.services.add(Service(name, depends_on))
return manager
def init(self):
"""
Инициализация всех сервисов
:return: None
"""
# initialized - множество имён всех инициализированных сервисов
initialized = set()
# Исполнять до тех пор, пока есть хотя бы один неиниц. сервис
while any(filter(lambda service: not service.initialized, self.services)):
# Список сервисов для инициализации (тех, что ещё не иниц.,
# и их зависимости - подмножество инициализированных
to_initialize = list(filter(
lambda service: service.depends_on <= initialized and not service.initialized,
self.services))
# Если таких сервисов нет, а неиниц. остались - значит присутствует цикл. или отсутств. зависимость
if not to_initialize:
raise IllegalServicesDependenciesError("Замечена отсутствующая или циклическая зависимость. "
"Проверьте конфигурацию.")
for service in to_initialize:
service.init()
initialized.add(service.name)
print("Инициализация завершена")
if __name__ == '__main__':
service_manager = ServiceManager.from_yaml('config.yaml')
service_manager.init()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment