API reference¶

def app_setting_specs(self):
    return self.settings_manager.app_setting_specs(self.app_id)

def get_app_setting(self, key: str):
    return self.settings_manager.get_app_setting(self.app_id, key)

def update_app_setting(self, key: str, value, *, save_global_too: bool = False) -> None:
    self.settings_manager.update_app_setting(
        self.app_id,
        key,
        value,
        save_global_too=save_global_too,
    )

def get_project_resource_contract(self):
    return self._project_resource_contract

def project_resource_specs(self):
    return self._project_resource_contract.specs

def list_project_resources(self, project_id: str | UUID | None = None) -> dict[str, ProjectResource]:
    project_root = self._resolve_project_root(project_id=project_id)
    if project_root is None:
        return {}
    return self._project_resource_contract.resolve(project_root)

def get_project_resource(self, key: str, project_id: str | UUID | None = None) -> ProjectResource:
    project_root = self._resolve_project_root(project_id=project_id)
    if project_root is None:
        raise RuntimeError(f"There is no active project to resolve resource '{key}'.")
    return self._project_resource_contract.resolve_one(project_root, key)

def get_project_resource_path(
    self,
    key: str,
    *parts: str | Path,
    project_id: str | UUID | None = None,
    create_parent: bool = False,
) -> Path:
    resource = self.get_project_resource(key, project_id=project_id)
    path = resource.path.joinpath(*(str(part) for part in parts))
    if create_parent:
        path.parent.mkdir(parents=True, exist_ok=True)
    return path

def get_project_resource_map(self, project_id: str | UUID | None = None) -> dict[str, dict[str, str]]:
    return {
        key: resource.to_mapping()
        for key, resource in self.list_project_resources(project_id=project_id).items()
    }

def list_projects(self, page: int = 1, items_per_page: int = 20) -> list[Project]:
    return self.project_manager.repository.get_projects_paginated(page, items_per_page)

def open_project(
    self,
    project_id: str | UUID,
    *,
    cancel_running_tasks_on_switch: bool = True,
    touch_master: bool = True,
    extra_dirs: list[str] | None = None,
) -> ProjectContext:
    normalized_id = self._normalize_id(project_id)
    self._validate_project_access(normalized_id)

    if self.active_context:
        if self.active_context.id == normalized_id:
            return self.active_context
        self.close_project(cancel_running_tasks=cancel_running_tasks_on_switch)

    context = self.project_manager.load_project_by_id(
        normalized_id,
        sm=self.settings_manager,
        extra_dirs=self._merge_project_dirs(extra_dirs),
    )
    self._activate_context(context, touch_master=touch_master)
    self.app_logger.info(
        "Project activated: id=%s name=%s",
        context.id,
        context.name,
        extra={"project_id": str(context.id)},
    )
    return context

def create_project(
    self,
    name: str,
    folder: Path | str,
    description: str = "",
    tags: list[str] | None = None,
    base_settings: SettingsProfile = "global",
    scope: str = "full",
    activate: bool = True,
    extra_dirs: list[str] | None = None,
) -> ProjectContext:
    folder = Path(folder).expanduser().resolve()
    normalized_scope = str(scope).strip()
    effective_scope = normalized_scope or "full"
    if effective_scope == "full":
        effective_scope = self.app_id
    context = self.project_manager.create_project(
        name=name,
        folder=folder,
        sm=self.settings_manager,
        base=base_settings,
        description=description,
        scope=effective_scope,
        app_id=self.app_id,
        tags=tags,
        extra_dirs=self._merge_project_dirs(extra_dirs),
    )
    if not activate:
        return context

    if self.active_context and self.active_context.id != context.id:
        self.close_project(cancel_running_tasks=True)
    self._activate_context(context, touch_master=True)
    self.app_logger.info(
        "Project created and activated: id=%s name=%s",
        context.id,
        context.name,
        extra={"project_id": str(context.id)},
    )
    return context

def find_project(
    self,
    *,
    name: str | None = None,
    folder: Path | str | None = None,
) -> Project | None:
    return self.project_manager.find_project(name=name, folder=folder)

def create_or_open_project(
    self,
    *,
    name: str,
    folder: Path | str,
    description: str = "",
    tags: list[str] | None = None,
    base_settings: SettingsProfile = "global",
    scope: str = "full",
    cancel_running_tasks_on_switch: bool = True,
    activate: bool = True,
    extra_dirs: list[str] | None = None,
) -> ProjectContext:
    folder_path = Path(folder).expanduser().resolve()
    existing_at_path = self.project_manager.find_project_global(folder=folder_path)
    if existing_at_path is not None:
        existing_app_id = (existing_at_path.app_id or "").strip()
        if existing_app_id != self.app_id:
            raise ValueError(
                f"Path '{folder_path}' already belongs to project '{existing_at_path.name}' "
                f"of app_id='{existing_app_id or '<empty>'}'."
            )
        if activate:
            return self.open_project(
                existing_at_path.id,
                cancel_running_tasks_on_switch=cancel_running_tasks_on_switch,
                touch_master=True,
                extra_dirs=self._merge_project_dirs(extra_dirs),
            )
        return self.project_manager.load_project(
            project=existing_at_path,
            sm=self.settings_manager,
            extra_dirs=self._merge_project_dirs(extra_dirs),
        )
    existing = self.project_manager.find_project(name=name)
    if existing is not None:
        if activate:
            return self.open_project(
                existing.id,
                cancel_running_tasks_on_switch=cancel_running_tasks_on_switch,
                touch_master=True,
                extra_dirs=self._merge_project_dirs(extra_dirs),
            )
        return self.project_manager.load_project(
            project=existing,
            sm=self.settings_manager,
            extra_dirs=self._merge_project_dirs(extra_dirs),
        )
    return self.create_project(
        name=name,
        folder=folder_path,
        description=description,
        tags=tags,
        base_settings=base_settings,
        scope=scope,
        activate=activate,
        extra_dirs=self._merge_project_dirs(extra_dirs),
    )

def register_task_canceller(self, project_id: str | UUID, canceller: Callable[[], None]):
    normalized_id = self._normalize_id(project_id)
    runtime = self._require_active_project_runtime()
    if runtime.context.id != normalized_id:
        raise RuntimeError("Cancellers can only be registered for the active project.")
    runtime.task_cancellers.append(canceller)

def get_app_logger(self, name: str = "core"):
    return self.logging_manager.get_app_logger(name)

def get_executor_logger(self, name: str = "core"):
    return self.logging_manager.get_executor_logger(name)

def get_project_logger(self, name: str = "core"):
    return self.logging_manager.get_project_logger(name)

def get_entity_loader_context(self) -> ProjectDataContext:
    return self.advanced.entity_loader_context()

def get_project_data_context(self) -> ProjectDataContext:
    return self.advanced.project_data_context()

def refresh_master_db(self):
    self.master_db.reconnect()
    self.app_logger.debug("Master DB reconnected")

def close_project(self, cancel_running_tasks: bool = True):
    """Close the active project. Its jobs that have not ended are interrupted, not lost:
    submitted again under the same key in a later session, each continues where it stopped.

    With `cancel_running_tasks=False`, raises instead when one is still going on.
    """
    if not self.active_context:
        self.settings_manager.clear_project()
        self.logging_manager.clear_project_logging()
        return

    context = self.active_context
    if cancel_running_tasks:
        self._cancel_project_tasks(context.id)
    elif self.get_project_activity()["jobs_active"]:
        raise RuntimeError("The project cannot be closed while active jobs remain.")

    self.project_manager.close_project(context)
    self._close_project_runtime()
    self.settings_manager.clear_project()
    self.app_logger.info(
        "Project closed: id=%s name=%s",
        context.id,
        context.name,
        extra={"project_id": str(context.id)},
    )

def get_project_activity(self, project_id: str | UUID | None = None) -> dict[str, Any]:
    """What is still going on in a project: its jobs that have not ended, and the external
    tasks registered with `register_task_canceller`. Only the active project has any.
    """
    context = self.active_context
    mine = context is not None and (project_id is None or context.id == self._normalize_id(project_id))
    # ponytail: every recorded job of the engine counts, as the project's journal is the
    # only one this facade attaches.
    jobs = [
        job
        for job in (self._engine.jobs() if mine and self._flow is not None else ())
        if job.key is not None and job.status in ("waiting", "running")
    ]
    cancellers = len(self._runtime_state.active_project.task_cancellers) if mine else 0
    active = bool(jobs or cancellers)
    return {
        "project_id": str(project_id or context.id) if project_id or context else None,
        "active": active,
        "can_switch_project": not active,
        "jobs_active": len(jobs),
        "job_ids": [job.id for job in jobs],
        "statuses": sorted({job.status for job in jobs}),
        "external_cancellers": cancellers,
    }

def delete_projects(self, project_ids: list[str | UUID], delete_files: bool = True):
    normalized_ids = []
    for project_id in project_ids:
        try:
            normalized_ids.append(self._normalize_id(project_id))
        except ValueError:
            continue

    if not normalized_ids:
        return

    if self.active_context and self.active_context.id in normalized_ids:
        self.close_project()

    self.project_manager.delete_projects(normalized_ids, delete_files=delete_files)
    self.app_logger.info("Projects deleted: count=%s delete_files=%s", len(normalized_ids), delete_files)

def shutdown(self):
    try:
        self.close_project(cancel_running_tasks=True)
    except Exception:
        pass
    self._shutdown_runtime_stack()
    try:
        self.logging_manager.stop()
    except Exception:
        pass

def executors(self) -> list[dict]:
    """One row for each executor kept: `name`, `kind`, `hosts`, `default`, and `state`:
    `on`, `starting`, `off`, or `blocked` when it is off and not to be started now, with
    `why`. Safe to poll."""
    settings, active = self.settings_manager.settings, self._active()
    rows = []
    for name, config in settings.executors.items():
        why = "" if name in active or name in self._starting else blocked(name, settings, active)
        state = "starting" if name in self._starting else "on" if name in active else "blocked" if why else "off"
        rows.append({"name": name, "kind": config.kind, "hosts": list(config.hosts), "state": state, "why": why,
                     "default": name == settings.general.default_executor})
    return rows

def activate(self, name: str, on_line=None) -> None:
    """Start the executor kept as `name`, which takes jobs under that name from then on.

    Refused with `ValueError` while another active one has one of its machines or is a Ray
    too (see `blocked`). A Ray that takes this machine puts the pool of it aside until it
    is deactivated, so the pool must have no job that has not ended. A host that is not
    reached or not prepared raises, and leaves everything as it was. `on_line` hears what
    is done on a host that is being prepared. It waits for its hosts: not from a thread
    that cannot.
    """
    settings, engine = self.settings_manager.settings, self.engine
    if name not in settings.executors:
        raise ValueError(f"no executor is kept as `{name}`")
    self._starting.add(name)
    try:
        with self._activation:
            if name in engine._executors:
                return
            why = blocked(name, settings, self._active())
            if why:
                raise ValueError(f"`{name}` is not started: {why}")
            config = settings.executors[name]
            aside = config.kind == "ray" and "local" in config.hosts and "local" in engine._executors
            if aside:
                try:
                    engine.unregister("local")
                except ValueError:
                    raise ValueError(f"`{name}` takes this machine, and its pool has jobs that have not ended") from None
            try:
                host_env = self.host_env and (lambda address, busy: self.host_env(address, busy, on_line))
                executor = build_executor(name, settings, host_env)
                if executor is None:
                    raise RuntimeError(_LOGIN_NODE)
                engine.register(name, executor)
            except BaseException:
                if aside:
                    self._local_back()
                raise
            if aside:
                self._aside = name
    finally:
        self._starting.discard(name)

def deactivate(self, name: str) -> None:
    """Stop the executor active as `name`. Refused with `ValueError` while a job submitted
    to it has not ended. The pool of this machine comes back if it was aside for this one."""
    if self._engine is None or name not in self._active():
        return
    with self._activation:
        self._engine.unregister(name)
        if self._aside == name:
            self._aside = None
            self._local_back()

def save_host(self, name: str, **fields) -> None:
    """`SettingsManager.save_host`, refused while an active executor works on that machine."""
    self._change("hosts", name, lambda: self.settings_manager.save_host(name, **fields))

def remove_host(self, name: str) -> None:
    self._change("hosts", name, lambda: self.settings_manager.remove_host(name))

def save_executor(self, name: str, **fields) -> None:
    """`SettingsManager.save_executor`, refused while that executor is active."""
    self._change("executors", name, lambda: self.settings_manager.save_executor(name, **fields))

def remove_executor(self, name: str) -> None:
    self._change("executors", name, lambda: self.settings_manager.remove_executor(name))

def submit(self, job, params=None, *, executor: str | None = None, **options) -> int:
    """Run a `ms_flow.engine.job` in the active project and return its id, for `engine.status`
    and `engine.wait`.

    Its results go to the project's store, and the job is recorded in the project's
    `executor.db`: closing the project interrupts it, and submitting it again with the same
    `key=` continues it. `executor` is `thread` or the name of an active one (see
    `activate`), by default the one the settings give (`general.default_executor`);
    `options` are those of `ms_flow.engine.Flow.submit`.
    """
    flow = self._project_flow()
    name = executor or self.settings_manager.settings.general.default_executor
    self._ready(name)
    return flow.submit(job, params, journal="project", executor=name, **options)

def origin(self, key: str) -> dict:
    """What the job recorded under `key` in the active project was submitted with: `job` (its
    `module:qualname`), `params` and `options`. Raises for one `resume` cannot submit again."""
    return self._project_flow().origin(key, journal="project")

def resume(self, key: str, **options) -> int:
    """Submit again the job recorded under `key` (`JobSnapshot.key`) in the active project,
    with the params and options it was submitted with: one that was interrupted (the project
    or the application was closed) or canceled continues where it was. `options` say
    otherwise for this run, and bring what the record cannot keep (`on_terminal`)."""
    from ms_flow.engine.flow import named

    origin = self.origin(key)
    return self.submit(named(origin["job"]), origin["params"], key=key, **{**origin["options"], **options})

def move(self, key: str, executor: str, **options) -> int:
    """Continue on `executor` the job recorded under `key`: `resume`, for one that may be running.
    A job this session runs is canceled first and waited for, so not from a thread that cannot
    wait: what its tasks in flight had done is done again there."""
    self.origin(key)  # nothing is canceled for a job that cannot be submitted again,
    self._ready(executor)  # or that has nowhere to go
    for job in self._engine.jobs():
        if job.key == key and job.finished_at is None:
            self._engine.cancel(job.id)
            self._engine.wait(job.id)
    return self.resume(key, executor=executor, **options)

def remote_jobs(self, executor: str) -> list[dict]:
    """What the executor of that name has on another machine, whichever session or project
    sent it and with or without a project open: see `RemoteExecutor.jobs_there`. Nothing
    when it runs its jobs here. It asks that machine, so not from a thread that cannot wait."""
    slot = self.engine._executors.get(executor)
    return slot.impl.jobs_there() if hasattr(getattr(slot, "impl", None), "jobs_there") else []

def remove_remote_job(self, executor: str, dir: str) -> None:
    """Take one of `remote_jobs` out of the queue, with what it has there. One that a job of
    this session waits for is refused: cancel the job."""
    self.engine.executor(executor).remove_there(dir)

def jobs(self, limit: int | None = None) -> list:
    """The jobs of the active project, oldest first; with `limit`, the last ones.

    What a monitor shows: every job on record in the project's journal, the ones of earlier
    sessions too, each as the engine has it now when it ran it in this session. Reads the
    journal without owning it and never builds the engine, so it is safe to poll.
    """
    project = self._runtime_state.active_project
    if project is None:
        return []
    from ms_flow.engine import JobSnapshot
    from ms_flow.engine.journal import recorded

    live = {(job.key, job.started_at): job for job in (self._engine.jobs() if self._flow is not None else ())}
    # ponytail: a job of a process that died reads as running until this session's first
    # submit opens the journal, which is what marks it interrupted.
    listed = [live.pop((row["job_id"], row["started_at"]), None)
              or JobSnapshot(None, row["name"], row["status"], row["total"], row["done"], row["failed"], 0,
                             row["error"], 0, row["started_at"], row["finished_at"], row["job_id"], 0,
                             min(1.0, (row["done"] + row["failed"]) / row["total"]) if row["total"] else None)
              for row in recorded(project.context.path / "executor.db", limit)]
    # One just submitted is not in the file until the journal's next write. Only the active
    # project has jobs that have not ended: closing a project interrupts its own.
    return listed + [job for job in live.values() if job.finished_at is None]

def job_tasks(self, job_id: int | None) -> list:
    """The tasks in flight of a job this session runs, by its `JobSnapshot.id`: see
    `Engine.tasks`. Nothing for a job of another session. Never builds the engine."""
    return self._engine.tasks(job_id) if self._engine is not None and job_id is not None else []

def job_failures(self, key: str, limit: int = 200) -> list[dict]:
    """`attempt`, `error` (the traceback) and `input` (a short repr of the element) of the
    chunks that failed in the job recorded under `key` (`JobSnapshot.key`) in the active
    project, at most `limit`."""
    from ms_flow.engine.journal import failures

    return failures(self._require_active_project_runtime().context.path / "executor.db", key, limit)

def get_project_store(self):
    return self.advanced.project_store