63 lines
2.0 KiB
Python
63 lines
2.0 KiB
Python
import inspect
|
|
import logging
|
|
from abc import ABC
|
|
from datetime import datetime
|
|
from types import FrameType
|
|
from typing import Any
|
|
|
|
from flask import Blueprint, Flask, request
|
|
from flask.blueprints import BlueprintSetupState
|
|
from flask_assets import Environment
|
|
|
|
from starfall.types import SnapshotQueue
|
|
|
|
|
|
class BaseBlueprint(ABC):
|
|
def __init__(self, blueprint: Blueprint, assets: Environment, app: Flask):
|
|
self.queue: SnapshotQueue | None = None
|
|
self.assets: Environment = assets
|
|
self.date: datetime = datetime.now()
|
|
self.app: Flask = app
|
|
self.data: dict[str, Any] = {}
|
|
blueprint.record(self.on_blueprint_setup)
|
|
|
|
if type(self) is not BaseBlueprint:
|
|
self._debug("Attaching blueprint of type %r to web service", type(self))
|
|
|
|
def on_blueprint_setup(self, state: BlueprintSetupState):
|
|
self.queue = state.options["options"]["queue"]
|
|
|
|
if type(self) is not BaseBlueprint:
|
|
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.
|
|
frame: FrameType | None = inspect.currentframe()
|
|
if not isinstance(frame, FrameType):
|
|
return
|
|
|
|
f_back: FrameType | None = frame.f_back
|
|
if not isinstance(f_back, FrameType):
|
|
return
|
|
|
|
self._info("Route access: %r", 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)
|
|
|
|
def _route(self):
|
|
rule = request.url_rule
|
|
if rule is None:
|
|
return ""
|
|
return rule.rule
|