80 lines
3.0 KiB
Python
80 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
import re
|
|
|
|
|
|
_PROFILE_MONTHLY_RE = re.compile(r"^monthly-(\d{4}-\d{2})$")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ResolvedWorkspace:
|
|
branch: str
|
|
workspace_name: str
|
|
workspace_path: Path
|
|
qmd_collection: str
|
|
qmd_index: str
|
|
|
|
|
|
class WorkspaceManager:
|
|
def __init__(self, workspaces_root: Path, default_branch: str = "main", qmd_index_prefix: str = "ws") -> None:
|
|
self.workspaces_root = workspaces_root
|
|
self.default_branch = default_branch
|
|
self.qmd_index_prefix = qmd_index_prefix
|
|
self.workspaces_root.mkdir(parents=True, exist_ok=True)
|
|
|
|
def resolve_branch(self, branch: str | None, memory_profile: str | None) -> str:
|
|
if branch and branch.strip():
|
|
return branch.strip()
|
|
if memory_profile and memory_profile.strip():
|
|
return self._profile_to_branch(memory_profile.strip())
|
|
return self.default_branch
|
|
|
|
def resolve_workspace(self, branch: str | None, memory_profile: str | None) -> ResolvedWorkspace:
|
|
resolved_branch = self.resolve_branch(branch=branch, memory_profile=memory_profile)
|
|
workspace_name = self.workspace_name_for_branch(resolved_branch)
|
|
workspace_path = self.workspaces_root / workspace_name
|
|
qmd_collection = self.collection_name_for_workspace(workspace_name)
|
|
qmd_index = self.index_name_for_workspace(workspace_name)
|
|
return ResolvedWorkspace(
|
|
branch=resolved_branch,
|
|
workspace_name=workspace_name,
|
|
workspace_path=workspace_path,
|
|
qmd_collection=qmd_collection,
|
|
qmd_index=qmd_index,
|
|
)
|
|
|
|
def _profile_to_branch(self, memory_profile: str) -> str:
|
|
if memory_profile == "stable":
|
|
return "main"
|
|
monthly = _PROFILE_MONTHLY_RE.match(memory_profile)
|
|
if monthly:
|
|
return f"memory/{monthly.group(1)}"
|
|
if memory_profile.startswith("task-") and len(memory_profile) > len("task-"):
|
|
return f"task/{memory_profile[len('task-') :]}"
|
|
raise ValueError(
|
|
"unsupported memory_profile, expected stable | monthly-YYYY-MM | task-<task-id>"
|
|
)
|
|
|
|
def workspace_name_for_branch(self, branch: str) -> str:
|
|
if branch in {"main", "stable"}:
|
|
return "main"
|
|
if branch.startswith("memory/"):
|
|
return f"memory-{branch.split('/', 1)[1].replace('/', '-') }"
|
|
if branch.startswith("task/"):
|
|
return f"task-{branch.split('/', 1)[1].replace('/', '-') }"
|
|
|
|
return "branch-" + self._slugify(branch)
|
|
|
|
def collection_name_for_workspace(self, workspace_name: str) -> str:
|
|
return self._slugify(workspace_name)
|
|
|
|
def index_name_for_workspace(self, workspace_name: str) -> str:
|
|
return f"{self.qmd_index_prefix}_{self._slugify(workspace_name)}"
|
|
|
|
def _slugify(self, value: str) -> str:
|
|
cleaned = re.sub(r"[^A-Za-z0-9_.-]+", "-", value)
|
|
cleaned = cleaned.strip("-")
|
|
return cleaned or "default"
|