from dataclasses import dataclass
from enum import Enum
from importlib import import_module
from q2_sdk.core import configuration
from q2_sdk.core.q2_logging.logger import Q2LoggerType
SETTINGS = configuration.get_settings()
[docs]
@dataclass
class LifeCycleHook:
"""Base class for all LifeCycle Hooks"""
logger: Q2LoggerType
debug: bool
local_dev: bool
[docs]
@dataclass
class InitializeData(LifeCycleHook):
"""Used with lifecycle.initialize"""
[docs]
@dataclass
class StartupData(LifeCycleHook):
"""Used with lifecycle.startup"""
[docs]
@dataclass
class ShutdownData(LifeCycleHook):
"""Used with lifecycle.shutdown"""
[docs]
class LifeCycle(Enum):
Startup = "lifecycle.startup"
Shutdown = "lifecycle.shutdown"
Initialize = "lifecycle.initialize"
async def execute_lifecycle_hook(logger: Q2LoggerType, hook: LifeCycle) -> None:
try:
lifecycle_module = import_module(hook.value)
except ModuleNotFoundError:
lifecycle_module = None
try:
lifecycle_hook = lifecycle_module.handle
except AttributeError:
lifecycle_hook = None
if lifecycle_hook:
hook_data = _get_lifecycle_hook(logger, hook)
logger.debug("Executing LifeCycle hook: %s", hook.name)
await lifecycle_hook(hook_data)
def _get_lifecycle_hook(logger: Q2LoggerType, hook: LifeCycle) -> LifeCycleHook:
match hook:
case LifeCycle.Initialize:
return InitializeData(
logger, debug=SETTINGS.DEBUG, local_dev=SETTINGS.LOCAL_DEV
)
case LifeCycle.Startup:
return StartupData(
logger, debug=SETTINGS.DEBUG, local_dev=SETTINGS.LOCAL_DEV
)
case LifeCycle.Shutdown:
return ShutdownData(
logger, debug=SETTINGS.DEBUG, local_dev=SETTINGS.LOCAL_DEV
)