Source code for ewoksserver.app.config

from __future__ import annotations
import os
import sys
import importlib.util
import logging
from typing_extensions import Annotated
from typing import Optional
import warnings

from fastapi import Depends


from .models import AppSettings, EwoksSettings

try:
    from ewoksweb.serverutils import get_test_config
except ImportError:
    get_test_config = None

logger = logging.getLogger(__name__)

_APP_SETTINGS = None

_EWOKS_SETTINGS = None


def _resolve_ewoks_execution_settings(
    ewoks_execution: dict | None, ewoks: dict | None
) -> dict:
    if ewoks is None:
        return ewoks_execution if ewoks_execution else dict()

    if ewoks_execution is None:
        warnings.warn(
            "EWOKS configuration field has been renamed EWOKS_EXECUTION",
            DeprecationWarning,
        )
        return ewoks

    logger.warning(
        "Both EWOKS_EXECUTION and EWOKS fields were specified but EWOKS field is deprecated. EWOKS field will be ignored."
    )
    return ewoks_execution


def _resolve_ewoks_discovery_settings(
    ewoks_discovery: dict | None, discover_timeout: float | None
) -> dict:

    if discover_timeout is None:
        return ewoks_discovery if ewoks_discovery else dict()

    if ewoks_discovery is None:
        warnings.warn(
            "DISCOVER_TIMEOUT is deprecated. The timeout should be specified via the `timeout` field of EWOKS_DISCOVERY",
            DeprecationWarning,
        )
        return {"timeout": discover_timeout}

    logger.warning(
        "Both EWOKS_DISCOVERY and DISCOVER_TIMEOUT fields were specified but DISCOVER_TIMEOUT field is deprecated. DISCOVER_TIMEOUT field will be ignored."
    )
    return ewoks_discovery


def _resolve_ewoks_scheduling_settings(celery: dict | None) -> dict:
    if celery is None:
        return {"type": "local"}

    return {"type": "celery", "configuration": celery}


[docs] def create_ewoks_settings( config: Optional[str] = None, dir: Optional[str] = None, without_events: bool = False, frontend_tests: bool = False, no_discovery_at_launch: bool = False, ) -> EwoksSettings: global _EWOKS_SETTINGS # Get configuration file filename = os.environ.get("EWOKSSERVER_SETTINGS") if config: filename = config if frontend_tests: if get_test_config is None: raise RuntimeError("ewoksweb is not installed") filename = get_test_config() # Extract settings from configuration file resource_directory = None ewoks = None ewoks_execution = None ewoks_discovery = None celery = None discover_timeout = None if filename: spec = importlib.util.spec_from_file_location("ewoksserverconfig", filename) mod = importlib.util.module_from_spec(spec) sys.modules["ewoksserverconfig"] = mod spec.loader.exec_module(mod) resource_directory = getattr(mod, "RESOURCE_DIRECTORY", resource_directory) celery = getattr(mod, "CELERY", celery) ewoks_execution = getattr(mod, "EWOKS_EXECUTION", None) ewoks_discovery = getattr(mod, "EWOKS_DISCOVERY", None) # DEPRECATED ewoks = getattr(mod, "EWOKS", ewoks) discover_timeout = getattr(mod, "DISCOVER_TIMEOUT", discover_timeout) # Overwrite resource directory if dir: resource_directory = dir if not resource_directory: resource_directory = "." ewoks_discovery = _resolve_ewoks_discovery_settings( ewoks_discovery, discover_timeout ) ewoks_execution = _resolve_ewoks_execution_settings(ewoks_execution, ewoks) ewoks_scheduling = _resolve_ewoks_scheduling_settings(celery) # Overwrite on_start_up if asked if no_discovery_at_launch: ewoks_discovery["on_start_up"] = False configured = bool(filename) or bool(dir) _EWOKS_SETTINGS = EwoksSettings( configured=configured, resource_directory=resource_directory, without_events=without_events, ewoks_execution=ewoks_execution, ewoks_discovery=ewoks_discovery, ewoks_scheduling=ewoks_scheduling, ) return _EWOKS_SETTINGS
[docs] def create_app_settings(no_older_versions: bool = False) -> AppSettings: global _APP_SETTINGS _APP_SETTINGS = AppSettings(no_older_versions=no_older_versions) return _APP_SETTINGS
[docs] def get_ewoks_settings(): global _EWOKS_SETTINGS if _EWOKS_SETTINGS is not None: return _EWOKS_SETTINGS return create_ewoks_settings()
[docs] def get_app_settings(): global _APP_SETTINGS if _APP_SETTINGS is not None: return _APP_SETTINGS return create_app_settings()
EwoksSettingsType = Annotated[EwoksSettings, Depends(get_ewoks_settings)]