Da

✏️ Editing: session_subscriptions_status.py

import threading

from parallels.core import messages
from parallels.core.logging import get_logger
from parallels.core.utils.common import default
from parallels.core.utils.common_constants import LOCK_SUBSCRIPTIONS_STATUS
from parallels.core.utils.json_utils import read_json, write_json
from parallels.core.utils.locks.session_file_lock import SessionFileLock
from parallels.core.utils.migration_progress import SubscriptionMigrationStatus
from parallels.core.utils.subscription_operations import SubscriptionOperation

logger = get_logger(__name__)

class SessionSubscriptionsStatus(object):
    """File with per-session status of each subscription: whether migration was finished, and for which operations"""

def __init__(self, filename):
        self._filename = filename
        self._update_thread_lock = threading.Lock()
        self._update_process_lock = SessionFileLock(LOCK_SUBSCRIPTIONS_STATUS)

def set_operation_status(self, subscription_name, operation, status):
        with self._update_thread_lock:
            self._update_process_lock.acquire_block()
            try:
                self._load()
                self._init_subscription_dict(subscription_name)
                self._data[subscription_name]['operations'][operation] = status
                self._data[subscription_name]['queued'] = False
                self._data[subscription_name]['last_operation'] = operation
                self._save()
            finally:
                self._update_process_lock.release()

def get_operation_status(self, subscription_name, operation):
        with self._update_thread_lock:
            self._update_process_lock.acquire_block()
            try:
                self._load()
                self._init_subscription_dict(subscription_name)
                subscription_operations = self._data.get(subscription_name, {}).get('operations', {})
                return subscription_operations.get(operation, SubscriptionMigrationStatus.NOT_STARTED)
            finally:
                self._update_process_lock.release()

def init_subscription_status(self, subscription_name):
        with self._update_thread_lock:
            self._update_process_lock.acquire_block()
            try:
                self._load()
                self._init_subscription_dict(subscription_name)
                self._save()
            finally:
                self._update_process_lock.release()

def reset_operation_status(self, subscription_name):
        """Reset operation status of subscription for all operations - like subscription was never migrated

:type subscription_name: str | unicode
        """
        with self._update_thread_lock:
            self._update_process_lock.acquire_block()
            try:
                self._load()
                if subscription_name in self._data:
                    del self._data[subscription_name]
                self._init_subscription_dict(subscription_name)
                self._save()
            finally:
                self._update_process_lock.release()

def set_queued(self, subscription_name):
        with self._update_thread_lock:
            self._update_process_lock.acquire_block()
            try:
                self._load()
                self._init_subscription_dict(subscription_name)
                self._data[subscription_name]['queued'] = True
                self._save()
            finally:
                self._update_process_lock.release()

def remove_queued(self, subscription_name):
        with self._update_thread_lock:
            self._update_process_lock.acquire_block()
            try:
                self._load()
                self._init_subscription_dict(subscription_name)
                self._data[subscription_name]['queued'] = False
                self._save()
            finally:
                self._update_process_lock.release()

def set_stop(self, subscription_name):
        with self._update_thread_lock:
            self._update_process_lock.acquire_block()
            try:
                self._load()
                self._init_subscription_dict(subscription_name)
                subscription_dict = self._data[subscription_name]
                subscription_dict['stop'] = True
                for operation_name in subscription_dict.get('operations', {}).keys():
                    if subscription_dict['operations'][operation_name] == SubscriptionMigrationStatus.ON_HOLD:
                        subscription_dict['operations'][operation_name] = SubscriptionMigrationStatus.CANCELLED
                self._save()
            finally:
                self._update_process_lock.release()

def is_stop(self, subscription_name):
        with self._update_thread_lock:
            self._update_process_lock.acquire_block()
            try:
                self._load()
                return self._data.get(subscription_name, {}).get('stop', False)
            finally:
                self._update_process_lock.release()

def remove_stop(self, subscription_name):
        with self._update_thread_lock:
            self._update_process_lock.acquire_block()
            try:
                self._load()
                self._init_subscription_dict(subscription_name)
                self._data[subscription_name]['stop'] = False
                self._save()
            finally:
                self._update_process_lock.release()

def set_all_not_queued(self):
        with self._update_thread_lock:
            self._update_process_lock.acquire_block()
            try:
                self._load()
                for subscription_dict in self._data.values():
                    subscription_dict['queued'] = False
                    for operation_name in subscription_dict.get('operations', {}).keys():
                        if subscription_dict['operations'][operation_name] == SubscriptionMigrationStatus.ON_HOLD:
                            subscription_dict['operations'][operation_name] = SubscriptionMigrationStatus.CANCELLED
                self._save()
            finally:
                self._update_process_lock.release()

def _load(self):
        try:
            self._data = default(read_json(self._filename), {})
        except Exception as e:
            logger.exception()
            logger.fwarn(messages.ERROR_READING_SUBSCRIPTION_STATUS, filename=self._filename, reason=e)
            self._data = {}

def _save(self):
        write_json(self._filename, self._data, pretty_print=True)

def _init_subscription_dict(self, subscription_name):
        if subscription_name not in self._data:
            self._data[subscription_name] = {
                'operations': {
                    SubscriptionOperation.OPERATION_MIGRATED: SubscriptionMigrationStatus.NOT_STARTED,
                    SubscriptionOperation.OPERATION_POST_MIGRATION_CHECKS: SubscriptionMigrationStatus.NOT_STARTED,
                    SubscriptionOperation.OPERATION_CONTENT_SYNCED: SubscriptionMigrationStatus.NOT_STARTED,
                    SubscriptionOperation.OPERATION_DNS_SWITCHED: SubscriptionMigrationStatus.NOT_STARTED,
                },
                'queued': False,
                'stop': False,
                'last_operation': None,
            }

📁 Sid Gifari File Manager