Skip to content

mamut_routing_lib.remote

Release archives on GitHub: the manifest contract and the download client.

A MAMUT-routing release ships one zip per (problem type, family) plus a snapshot-manifest.json (ReleaseArchiveManifest) listing every asset with its download URL, sha256 and size. GitHubReleaseClient reads the manifest of a tag (or of the latest release), downloads assets with retries and verifies their checksum; the mamut-routing remote commands are thin wrappers. Configuration: MAMUT_ROUTING_RELEASE_REPO (default ANR-MAMUT/MAMUT-routing) and MAMUT_ROUTING_GITHUB_TOKEN (optional, for rate limits or private releases).

DEFAULT_RELEASE_REPO_ENV module-attribute 

DEFAULT_RELEASE_REPO_ENV = 'MAMUT_ROUTING_RELEASE_REPO'

DEFAULT_GITHUB_TOKEN_ENV module-attribute 

DEFAULT_GITHUB_TOKEN_ENV = 'MAMUT_ROUTING_GITHUB_TOKEN'

DEFAULT_MANIFEST_FILENAME module-attribute 

DEFAULT_MANIFEST_FILENAME = 'snapshot-manifest.json'

MANIFEST_SCHEMA_VERSION module-attribute 

MANIFEST_SCHEMA_VERSION = '1.0.0'

DOWNLOAD_CHUNK_SIZE module-attribute 

DOWNLOAD_CHUNK_SIZE = 1024 * 1024

DEFAULT_REQUEST_TIMEOUT_SECONDS module-attribute 

DEFAULT_REQUEST_TIMEOUT_SECONDS = 30

DEFAULT_RETRY_ATTEMPTS module-attribute 

DEFAULT_RETRY_ATTEMPTS = 3

DEFAULT_RETRY_DELAY_SECONDS module-attribute 

DEFAULT_RETRY_DELAY_SECONDS = 0.5

ProgressCallback module-attribute 

ProgressCallback: TypeAlias = Callable[[int, int | None], None]

ReleaseArchiveScope

Bases: str, Enum

What an archive covers: today always problem_family (one problem type of one family).

PROBLEM_FAMILY class-attribute instance-attribute 

PROBLEM_FAMILY = 'problem_family'

FAMILY_COLLECTION class-attribute instance-attribute 

FAMILY_COLLECTION = 'family_collection'

ReleaseArchiveAsset

Bases: BaseModel

One archive of a release: scope, filename, download_url, family identity, checksum_sha256, size_bytes, archive_root.

scope instance-attribute 

scope: ReleaseArchiveScope

filename instance-attribute 

filename: str

download_url instance-attribute 

download_url: str

problem_type class-attribute instance-attribute 

problem_type: ProblemType | None = None

benchmark_name class-attribute instance-attribute 

benchmark_name: BenchmarkName | None = None

checksum_sha256 class-attribute instance-attribute 

checksum_sha256: str | None = None

size_bytes class-attribute instance-attribute 

size_bytes: int | None = None

archive_root class-attribute instance-attribute 

archive_root: str | None = None

ReleaseArchiveManifest

Bases: BaseModel

The snapshot-manifest.json of a release: snapshot identity (id, date, commit, tag) and its assets.

schema_version class-attribute instance-attribute 

schema_version: str = MANIFEST_SCHEMA_VERSION

snapshot_id instance-attribute 

snapshot_id: str

published_at instance-attribute 

published_at: str

source_commit instance-attribute 

source_commit: str

source_branch class-attribute instance-attribute 

source_branch: str | None = None

release_tag class-attribute instance-attribute 

release_tag: str | None = None

assets class-attribute instance-attribute 

assets: list[ReleaseArchiveAsset] = Field(default_factory=list)

select_assets 

select_assets(*, scope: ReleaseArchiveScope | None = None, problem_type: ProblemType | None = None, benchmark_name: BenchmarkName | None = None) -> list[ReleaseArchiveAsset]

Assets matching every given filter (scope, problem_type, benchmark_name); no filter returns all.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/remote.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def select_assets(
    self,
    *,
    scope: ReleaseArchiveScope | None = None,
    problem_type: ProblemType | None = None,
    benchmark_name: BenchmarkName | None = None,
) -> list[ReleaseArchiveAsset]:
    """Assets matching every given filter (``scope``, ``problem_type``, ``benchmark_name``); no filter returns all."""
    selected = self.assets
    if scope is not None:
        selected = [asset for asset in selected if asset.scope == scope]
    if problem_type is not None:
        selected = [asset for asset in selected if asset.problem_type == problem_type]
    if benchmark_name is not None:
        selected = [asset for asset in selected if asset.benchmark_name == benchmark_name]
    return selected

GitHubReleaseSource dataclass 

GitHubReleaseSource(repo_full_name: str, manifest_filename: str = DEFAULT_MANIFEST_FILENAME, token: str | None = None)

Where releases are read from: owner/name repository, manifest filename, optional API token.

repo_full_name instance-attribute 

repo_full_name: str

manifest_filename class-attribute instance-attribute 

manifest_filename: str = DEFAULT_MANIFEST_FILENAME

token class-attribute instance-attribute 

token: str | None = None

from_env classmethod 

from_env() -> 'GitHubReleaseSource'

Source from MAMUT_ROUTING_RELEASE_REPO and MAMUT_ROUTING_GITHUB_TOKEN.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/remote.py 
115
116
117
118
119
120
@classmethod
def from_env(cls) -> "GitHubReleaseSource":
    """Source from ``MAMUT_ROUTING_RELEASE_REPO`` and ``MAMUT_ROUTING_GITHUB_TOKEN``."""
    repo_full_name = os.getenv(DEFAULT_RELEASE_REPO_ENV, "ANR-MAMUT/MAMUT-routing")
    token = os.getenv(DEFAULT_GITHUB_TOKEN_ENV)
    return cls(repo_full_name=repo_full_name, token=token)

GitHubReleaseClient 

GitHubReleaseClient(source: GitHubReleaseSource | None = None, *, retry_attempts: int = DEFAULT_RETRY_ATTEMPTS, retry_delay_seconds: float = DEFAULT_RETRY_DELAY_SECONDS, request_timeout_seconds: float = DEFAULT_REQUEST_TIMEOUT_SECONDS)

Reads manifests and downloads assets of GitHub releases with retries and a request timeout.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/remote.py 
146
147
148
149
150
151
152
153
154
155
156
157
def __init__(
    self,
    source: GitHubReleaseSource | None = None,
    *,
    retry_attempts: int = DEFAULT_RETRY_ATTEMPTS,
    retry_delay_seconds: float = DEFAULT_RETRY_DELAY_SECONDS,
    request_timeout_seconds: float = DEFAULT_REQUEST_TIMEOUT_SECONDS,
) -> None:
    self.source = source or GitHubReleaseSource.from_env()
    self.retry_attempts = max(1, retry_attempts)
    self.retry_delay_seconds = max(0.0, retry_delay_seconds)
    self.request_timeout_seconds = request_timeout_seconds

source instance-attribute 

source = source or GitHubReleaseSource.from_env()

retry_attempts instance-attribute 

retry_attempts = max(1, retry_attempts)

retry_delay_seconds instance-attribute 

retry_delay_seconds = max(0.0, retry_delay_seconds)

request_timeout_seconds instance-attribute 

request_timeout_seconds = request_timeout_seconds

fetch_manifest 

fetch_manifest(tag: str | None = None) -> ReleaseArchiveManifest

The manifest of tag, or of the latest release when tag is None.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/remote.py 
159
160
161
162
163
164
def fetch_manifest(self, tag: str | None = None) -> ReleaseArchiveManifest:
    """The manifest of ``tag``, or of the latest release when ``tag`` is ``None``."""
    resolved_tag = tag if tag is not None else self._resolve_latest_tag()
    manifest_url = self._build_release_asset_url(resolved_tag, self.source.manifest_filename)
    payload = self._download_json(manifest_url)
    return ReleaseArchiveManifest(**payload)

download_asset 

download_asset(asset: ReleaseArchiveAsset, destination_dir: str | Path, *, extract: bool = False, progress_callback: ProgressCallback | None = None) -> Path

Download asset into destination_dir, verify its checksum, optionally extract.

Returns the archive path, or the extraction directory (<destination>/<stem>, replaced if it exists) when extract is true. progress_callback receives (bytes_so_far, total_or_None).

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/remote.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def download_asset(
    self,
    asset: ReleaseArchiveAsset,
    destination_dir: str | Path,
    *,
    extract: bool = False,
    progress_callback: ProgressCallback | None = None,
) -> Path:
    """Download ``asset`` into ``destination_dir``, verify its checksum, optionally extract.

    Returns the archive path, or the extraction directory (``<destination>/<stem>``,
    replaced if it exists) when ``extract`` is true. ``progress_callback``
    receives ``(bytes_so_far, total_or_None)``.
    """
    destination_root = Path(destination_dir)
    destination_root.mkdir(parents=True, exist_ok=True)
    destination_path = destination_root / asset.filename
    self._download_file(
        asset.download_url,
        destination_path,
        expected_total_bytes=asset.size_bytes,
        progress_callback=progress_callback,
    )

    if asset.checksum_sha256 is not None:
        verify_sha256(destination_path, asset.checksum_sha256)

    if not extract:
        return destination_path

    extract_dir = destination_root / destination_path.stem
    if extract_dir.exists():
        shutil.rmtree(extract_dir)
    extract_dir.mkdir(parents=True, exist_ok=True)
    with zipfile.ZipFile(destination_path, "r") as archive:
        archive.extractall(extract_dir)
    return extract_dir

load_release_manifest 

load_release_manifest(path: str | Path) -> ReleaseArchiveManifest

Load and validate a manifest file.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/remote.py 
123
124
125
def load_release_manifest(path: str | Path) -> ReleaseArchiveManifest:
    """Load and validate a manifest file."""
    return ReleaseArchiveManifest(**load_json_from_file(path))

compute_sha256 

compute_sha256(filepath: str | Path) -> str

Hex sha256 of a file, streamed.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/remote.py 
128
129
130
131
132
133
134
def compute_sha256(filepath: str | Path) -> str:
    """Hex sha256 of a file, streamed."""
    digest = hashlib.sha256()
    with Path(filepath).open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()

verify_sha256 

verify_sha256(filepath: str | Path, expected_sha256: str) -> None

Raise ValueError unless the file's sha256 equals expected_sha256.

Source code in MAMUT-routing-tools/MAMUT-routing-lib/src/mamut_routing_lib/remote.py 
137
138
139
140
141
def verify_sha256(filepath: str | Path, expected_sha256: str) -> None:
    """Raise ``ValueError`` unless the file's sha256 equals ``expected_sha256``."""
    actual_sha256 = compute_sha256(filepath)
    if actual_sha256 != expected_sha256:
        raise ValueError(f"SHA256 mismatch for {filepath}: expected {expected_sha256}, got {actual_sha256}")