Source code for

# coding=utf-8
from __future__ import absolute_import, print_function, division

import logging
from functools import wraps

from flask import current_app
from abilian.core.util import fqcn

[docs]class ServiceNotRegistered(Exception): pass
[docs]class ServiceState(object): """ Service state stored in Application.extensions """ #: reference to :class:`Service` instance service = None running = False def __init__(self, service, running=False): self.service = service self.running = running self.logger = logging.getLogger(fqcn(self.__class__))
[docs]class Service(object): """ Base class for services. """ #: State class to use for this Service AppStateClass = ServiceState #: service name in Application.extensions / name = None def __init__(self, app=None): if is None: raise ValueError( 'Service must have a name ({})'.format(fqcn(self.__class__))) self.logger = logging.getLogger(fqcn(self.__class__)) if app: self.init_app(app)
[docs] def init_app(self, app): app.extensions[] = self.AppStateClass(self)[] = self
[docs] def start(self, ignore_state=False): """ Starts the service. """ self.logger.debug('Start service') self._toggle_running(True, ignore_state)
[docs] def stop(self, ignore_state=False): """ Stops the service. """ self.logger.debug('Stop service') self._toggle_running(False, ignore_state)
def _toggle_running(self, run_state, ignore_state=False): state = self.app_state run_state = bool(run_state) if not ignore_state: assert run_state ^ state.running state.running = run_state @property def app_state(self): """ Current service state in current application. :raise:RuntimeError if working outside application context. """ try: return current_app.extensions[] except KeyError: raise ServiceNotRegistered( @property def running(self): """ :returns: `False` if working outside application context, if service is not registered on current application, or if service is halted for current application. """ try: return self.app_state.running except (RuntimeError, ServiceNotRegistered): # RuntimeError: happens when current_app is None: working outside # application context return False @staticmethod
[docs] def if_running(meth): """ Decorator for service methods that must be ran only if service is in running state. """ @wraps(meth) def check_running(self, *args, **kwargs): if not self.running: return return meth(self, *args, **kwargs) return check_running