Файловый менеджер - Редактировать - /opt/cpmigrate/environments/base.py
Ðазад
""" Constructor for base Environment class. """ from functools import wraps import logging import sys from typing import Union, TYPE_CHECKING if TYPE_CHECKING: from transfer import Transfer from journal import Journal def environment_run(func): """Wrapper for the Environment run() function.""" @wraps(func) def wrapper(self: 'Environment', *args): self.xfer = args[0] self.xfer.journal.save() if len(self.actions) > 0: self.start_message() func(self, *args) self.end_message() self.completed = True self.xfer.journal.save() return wrapper def environment_check(func): """Wrapper for the Environment check() function.""" @wraps(func) def wrapper(self: 'Environment', *args): self.xfer = args[0] self.xfer.journal.save() func(self, *args) self.xfer.journal.save() return wrapper class MetaEnvironment(type): """Attaches the wrapper to all run() and check() functions.""" def __new__(mcs, name, bases, classes): if 'run' in classes: classes['run'] = environment_run(classes['run']) if 'check' in classes: classes['check'] = environment_check(classes['check']) return super().__new__(mcs, name, bases, classes) class Environment(metaclass=MetaEnvironment): """Base class responsible for Environment modules.""" def __init__(self): self.xfer: Union["Transfer", None] = None self.journal: Union["Journal", None] = None self.name: str = self.__class__.__name__.lower() self.actions: list[str] = [] self.priority: int = 1 self.skipped: bool = False self.completed: bool = False self.resync_relevant: bool = False def debug(self, msg: str): """Logs debug for the Environment, with the name.""" logging.debug("[Environment %s] %s", self.name, msg) def info(self, msg: str): """Logs information for the Environment, with the name.""" logging.info("[Environment %s] %s", self.name, msg) def warning(self, msg: str, note: bool = False): """Logs a warning for the Environment, with the name.""" logging.warning("[Environment %s] %s", self.name, msg) if note: self.xfer.notes.append(f"[Environment {self.name}] {msg}") def error(self, msg: str, note: bool = False): """Logs an error for the Environment, with the name.""" logging.error("[Environment %s] %s", self.name, msg) if note: self.xfer.notes.append(f"[Environment {self.name}] {msg}") def critical(self, msg: str, note: bool = False): """Logs a critical error for the Environment, with the name.""" logging.critical("[Environment %s] %s", self.name, msg) if note: self.xfer.notes.append(f"[Environment {self.name}] {msg}") def start_message(self): """Logs that it's starting a run().""" self.info("Starting run.") def end_message(self): """Logs that it's finishing the run().""" self.info("Finished run") def check(self, _): """ Required to be overriden per Environment, performs checks neccesary. """ logging.debug( "Environment %s check() does not exist, skipping.", self.name ) def run(self, _): """ Required to be overriden per Environment, performs the run. """ logging.debug( "Environment %s run() does not exist, skipping.", self.name ) def post_transfer(self): """ Called after the migration completes (user pkg+rsync) and before domain validation. This is meant to be overriden when needed. """ def capture_state(self, p_state=None): """ Returns current state of Environment. Can be overriden to include custom states. Remember to also override load_state. """ env_state = { 'name': self.name, 'skipped': self.skipped, 'completed': self.completed, 'actions': self.actions, } if p_state: env_state.update(p_state) return env_state def load_state(self, loadstate: dict): """ Loads state of Environment from provided journal. Override this if you override capture_state to ensure custom vars are loaded. """ self.skipped = loadstate.get('skipped') self.completed = loadstate.get('completed') self.actions = loadstate.get('actions') def __repr__(self): return f"<Environment {self.name}>" class FatalMigrationFailure(Exception): """ This exception is to be used on fatal failures that require ending the migration. Example would be a connection failure. """ def __init__(self, message): self.message = message super().__init__(self.message) logging.critical(message) sys.exit(1) class APIFailure(Exception): """ This exception is to be used when a module/function does not exist or fails for an unknown reason. This will does require ending the migration. """ def __init__(self, message, suppress_errors=False): self.message = message super().__init__(self.message) if not suppress_errors: logging.error(message)
| ver. 1.1 | |
.
| PHP 8.3.30 | Ð“ÐµÐ½ÐµÑ€Ð°Ñ†Ð¸Ñ Ñтраницы: 0 |
proxy
|
phpinfo
|
ÐаÑтройка