40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
import inspect
|
|
import logging
|
|
from abc import ABC
|
|
|
|
from flask import Blueprint
|
|
from flask.blueprints import BlueprintSetupState
|
|
|
|
from starfall.types import SnapshotQueue
|
|
|
|
|
|
class BaseBlueprint(ABC):
|
|
def __init__(self, blueprint: Blueprint):
|
|
blueprint.record(self.on_blueprint_setup)
|
|
|
|
if type(self) != BaseBlueprint: # noqa: E721
|
|
self._debug("Attaching blueprint of type %r to web service", type(self))
|
|
|
|
def on_blueprint_setup(self, state: BlueprintSetupState):
|
|
self.queue: SnapshotQueue = state.options["options"]["queue"]
|
|
|
|
if type(self) != BaseBlueprint: # noqa: E721
|
|
self._info("Blueprint of type %r successfully set up", type(self))
|
|
|
|
def _log_access(self):
|
|
# There is always a previous frame - This function is called when a
|
|
# route is accessed by a user through a method unknown to this one.
|
|
self._info("Route access: %r", inspect.currentframe().f_back.f_code.co_name)
|
|
|
|
def _debug(self, msg: str, *args: object):
|
|
logging.getLogger("web").debug(msg, *args)
|
|
|
|
def _info(self, msg: str, *args: object):
|
|
logging.getLogger("web").info(msg, *args)
|
|
|
|
def _warn(self, msg: str, *args: object):
|
|
logging.getLogger("web").warning(msg, *args)
|
|
|
|
def _crit(self, msg: str, *args: object):
|
|
logging.getLogger("web").critical(msg, *args)
|