Skip to content

Macrostrat application frame

This is a CLI toolset for managing data and services for a Dockerized application. It was originally developed for the Sparrow application, but it has been generalized for use with other projects including Macrostrat, Mapboard and other systems.

It is designed to work well with projects managed using Docker Compose, but may eventually acquire features for managing Kubernetes-based applications.

Basic usage

app = Application(
    "Mapboard",
    restart_commands={"gateway": "caddy reload --config /etc/caddy/Caddyfile"},
    log_modules=["mapboard.server"],
    compose_files=[MAPBOARD_ROOT / "system" / "docker-compose.yaml"],
)
cli = app.control_command()

The cli is a Typer application that can be used to control the system, with abilities to start, stop, and restart services, with extension points to add new functionality.

Application

Bases: ApplicationBase

Source code in app-frame/macrostrat/app_frame/core.py
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class Application(ApplicationBase):
    console: Console
    _dotenv_cfg: bool | Path | list[Path]
    _log_modules: list[str] = []

    def __init__(
        self,
        name: str,
        *,
        command_name: Optional[str] = None,
        project_prefix: Optional[str] = None,
        log_modules: Optional[str | list[str]] = None,
        env: EnvironmentDependency = {},
        load_dotenv: DotenvDependency = None,
        env_prefix: Optional[str] = None,
    ):
        self.name = name
        self.command_name = command_name or name.lower()
        self.project_prefix = project_prefix or name.lower().replace(" ", "_")

        _env_prefix = env_prefix or self.project_prefix.upper() + "_"

        self.envvar_prefix = _env_prefix
        self.console = Console()

        if isinstance(log_modules, str):
            log_modules = [log_modules]
        if log_modules is None:
            log_modules = []
        # Always log this module
        log_modules.append("macrostrat.app_frame")
        if log_modules is not None:
            self._log_modules = log_modules

        self._dotenv_cfg = load_dotenv

        # Environment setup should possibly be postponed until within a command context.
        self.setup_environment(env)

    def replace_names(self, text: str) -> str:
        text = text.replace(":app_name:", self.name)
        return text.replace(":command_name:", self.name.lower())

    def info(self, text, style=None):
        self.console.print(self.replace_names(text), style=style)

    def load_dotenv(self, dependency: DotenvDependency = None):
        if dependency is None:
            dependency = self._dotenv_cfg

        if isinstance(dependency, list):
            for path in dependency:
                self.load_dotenv(path)
        elif isinstance(dependency, Path):
            load_dotenv(dependency)
        elif dependency is True:
            load_dotenv()

    def setup_environment(self, env: EnvironmentDependency):
        environ["DOCKER_SCAN_SUGGEST"] = "false"
        # environ["DOCKER_BUILDKIT"] = "1"

        # Additional user-specified environment variables
        if callable(env):
            env = env(self)
        for k, v in env.items():
            environ[k] = v

        # Load other environment variables
        self.load_dotenv()

    def setup_logs(self, verbose: bool = False):
        if len(self._log_modules) == 0:
            log.warning("No modules specified, not setting up logs")
            return
        if verbose:
            setup_stderr_logs(*self._log_modules)
        else:
            # Disable all logging
            # TODO: This is a hack, we shouldn't have to explicitly disable
            # logging in the CLI. Perhaps there's somewhere that it's being
            # enabled that we haven't chased down?
            setup_stderr_logs("", level=logging.CRITICAL)

    def control_command(self, *args, **kwargs):
        from .control_command import ControlCommand

        return ControlCommand(self, *args, **kwargs)