gitea_admin/sientia-dataops-laborious_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add scripts

Browse Source
This commit is contained in:
vitor-aignosi
2026-09-01 13:28:13 -03:00
parent 82b4a53271
commit 10c8723db6
3 changed files with 615 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

586
scripts/import_model_manual_run.py Normal file
View File

@@ -0,0 +1,586 @@
"""Run one `.sientia` model import by hand, standing in for the frontend that does not exist yet.
**What this is.** The Angular screen plus the Spring Boot BFF will, between them, pick a `.sientia`
file, hash it, wrap the bundle password for the worker, upload the object and insert one `PENDING`
row into `public.experiment_run` before starting `import_model`. None of that exists yet. This
script does exactly those four things, against whatever environment `.env` points at, so the import
path can be exercised for real and so whoever writes the frontend has a working reference for the
wire shapes (the digest, the envelope, the object key, the row).
**What it writes.** Two things, and it can undo both: one object under
`{IMPORT_BUNDLE_BUCKET}/{IMPORT_BUNDLE_PREFIX}` in MinIO, and one row in `public.experiment_run` in
the BFF Postgres database. Everything else it does is a read. If the workflow runs, *it* creates the
MLflow experiment, run and registered version and the MongoDB model document — cell 10 can remove
those too, behind its own flag.
**The safety rule.** Every target comes from the environment, and this file contains no host, user,
password or key literal. A `.env` inherited from a cluster deployment can perfectly well name a
production database and a production bucket, so cell 5 and cell 6 — the only cells that write — are
gated on cell 2, where the operator has to restate the Postgres host that cell 1 printed. This is a
development tool. It is not for production, and the guard is a mitigation, not a guarantee.
**Cells.** Run them one at a time in the VS Code interactive window, or run the module top to
bottom. Cell order is the contract:
1 resolve every target and print it 6 insert the PENDING row -> import_run_id
2 confirmation guard (first write is after) 7 print the workflow input
3 pick the bundle, compute expected_digest 8 optionally start the workflow and wait
4 build the password envelope 9 read the outcome back
5 upload the object, verify the stored bytes 10 undo
**Boundaries.** This script imports nothing from `laborious.activities` or `laborious.workflows` —
it stands in for the frontend, so it must not borrow the consumer's code. The one exception is
`laborious.utils.bundle.format`, which is pure `struct`/`hashlib` over bytes and lets cell 3 print
the header of the file it is about to upload. The envelope is built here in four visible lines
rather than imported, because those four lines are what the frontend has to transliterate to
WebCrypto; `laborious/utils/import_password.py` is the decrypt counterpart.
"""
# %%
# Cell 1 — resolve every target from the environment and print what we are pointed at.
# Reads only. No secret is printed here or anywhere else in this script.
from __future__ import annotations
import base64
import hashlib
import io
import json
import os
import secrets
import sys
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from urllib.parse import urlparse
import mlflow
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from dotenv import load_dotenv
from minio import Minio
from pymongo import MongoClient
from sqlalchemy import create_engine, text
from sqlalchemy.engine import URL
REPO_ROOT = Path(__file__).resolve().parent.parent
if str(REPO_ROOT) not in sys.path:
sys.path.insert(0, str(REPO_ROOT))
load_dotenv(REPO_ROOT / '.env')
# Below the `sys.path` insert on purpose: `laborious` is only importable once the repository root is
# on the path, which is what those three lines above just arranged. Hoisting this to the top of the
# cell is a `ModuleNotFoundError` when the script runs from anywhere but the repository root.
from laborious.utils.bundle.format import DIGEST_OFFSET, HEADER_SIZE, parse_header # noqa: E402
def env(name: str, default: str = '') -> str:
"""The environment variable, stripped, or `default`."""
return os.getenv(name, default).strip()
def env_required(name: str) -> str:
"""The environment variable, or a `RuntimeError` naming it."""
value = env(name)
if not value:
raise RuntimeError(f'{name} is not set — see .env.example')
return value
def env_secret_required(name: str) -> str:
"""The environment variable *verbatim*, or a `RuntimeError` naming it.
Unlike `env_required` this does not strip, and the difference is not cosmetic. A key and a
password are byte strings whose surrounding whitespace is part of the secret. The worker reads
the same variables with a plain `os.getenv`, so stripping here would seal the envelope under
bytes the worker never derives, and the import would fail at a gate with nothing in the message
to say why.
"""
value = os.getenv(name, '')
if not value:
raise RuntimeError(f'{name} is not set — see .env.example')
return value
def _line(label: str, value: object) -> str:
"""One aligned `label value` line, for output that stays readable in a terminal."""
return f' {label:<24} {value}'
# Every connection target below is *required* rather than defaulted. The worker's own config falls
# back to localhost, which is right for a service that must boot; it is wrong for a script that
# writes, because a silent fallback is a write to somewhere nobody read out loud.
#
# The import log database: the POSTGRES_* server from .env, with only the database name of its own,
# because the log lives in the Spring Boot BFF's database rather than in `sientia`. The schema and
# table are constants of the worker's code, never configuration.
#
# `build_import_status_config` also honours IMPORT_STATUS_DB_{HOST,PORT,USER,PASSWORD} as overrides,
# for a deployment whose log sits on another server. This script does not: it is a development tool
# pointed at a real environment, and the override exists for the e2e suite's two containers.
STATUS_DB_HOST = env_required('POSTGRES_HOST')
STATUS_DB_PORT = int(env_required('POSTGRES_PORT'))
STATUS_DB_NAME = env_required('IMPORT_STATUS_DB_NAME')
STATUS_DB_USER = env_required('POSTGRES_USER')
STATUS_DB_PASSWORD = os.getenv('POSTGRES_PASSWORD', '')
STATUS_TABLE = 'public.experiment_run'
# Where the frontend uploads the bundle. The prefix keeps the worker's own default, which is a
# format decision rather than a target.
MINIO_ENDPOINT_URL = env_required('MINIO_ENDPOINT_URL')
MINIO_ACCESS_KEY = env_required('MINIO_ACCESS_KEY')
MINIO_SECRET_KEY = env_required('MINIO_SECRET_KEY')
BUNDLE_BUCKET = env_required('IMPORT_BUNDLE_BUCKET')
_raw_prefix = env('IMPORT_BUNDLE_PREFIX', 'imported_models/')
BUNDLE_PREFIX = _raw_prefix if _raw_prefix.endswith('/') else f'{_raw_prefix}/'
# Read-back targets. This script never writes to these; the workflow does.
MONGODB_URL = env_required('MONGODB_URL')
MONGODB_USERNAME = env('MONGODB_USERNAME')
MONGODB_PASSWORD = os.getenv('MONGODB_PASSWORD', '')
MONGODB_DATABASE = env_required('MONGODB_DATABASE')
MODELS_COLLECTION = env('IMPORT_MODELS_COLLECTION', 'models')
MLFLOW_HOST = env_required('MLFLOW_HOST')
MLFLOW_PORT = env_required('MLFLOW_PORT')
MLFLOW_TRACKING_URI = f'{MLFLOW_HOST}:{MLFLOW_PORT}'
TEMPORAL_HOST = env_required('TEMPORAL_HOST')
TEMPORAL_NAMESPACE = env_required('TEMPORAL_NAMESPACE')
RUNTIME = env_required('RUNTIME')
IMPORT_TASK_QUEUE = 'import_model-legacy-queue'
# This script's own inputs.
BUNDLE_PATH = Path(
env('IMPORT_SCRIPT_BUNDLE_PATH', 'tests/fixtures/bundle/golden_model_v3.sientia')
)
if not BUNDLE_PATH.is_absolute():
BUNDLE_PATH = REPO_ROOT / BUNDLE_PATH
IMPORT_USERNAME = env_required('IMPORT_SCRIPT_USERNAME')
print('resolved targets — nothing has been written yet')
print(_line('status postgres', f'{STATUS_DB_HOST}:{STATUS_DB_PORT}/{STATUS_DB_NAME}'))
print(_line('status user', STATUS_DB_USER))
print(_line('status table', STATUS_TABLE))
print(_line('minio endpoint', MINIO_ENDPOINT_URL))
print(_line('minio bucket/prefix', f'{BUNDLE_BUCKET}/{BUNDLE_PREFIX}'))
print(_line('mongo', f'{MONGODB_URL}/{MONGODB_DATABASE}.{MODELS_COLLECTION}'))
print(_line('mlflow', MLFLOW_TRACKING_URI))
print(_line('temporal', f'{TEMPORAL_HOST} ns={TEMPORAL_NAMESPACE}'))
print(_line('task queue', IMPORT_TASK_QUEUE))
print(_line('bundle', BUNDLE_PATH))
print(_line('username', IMPORT_USERNAME))
print()
print(f'to continue, set the cell 2 confirmation to: {STATUS_DB_HOST}')
# %%
# Cell 2 — the confirmation guard. Cells 5 and 6 write; this is the last cell before them.
#
# Restate the Postgres host cell 1 printed. Either set IMPORT_SCRIPT_CONFIRM_TARGET in .env, or
# replace the right-hand side below with the value cell 1 told you. It is a read-then-restate on
# purpose: a boolean flag can be flipped without ever looking at what it is pointed at.
CONFIRM_TARGET = 'localhost'
if CONFIRM_TARGET != STATUS_DB_HOST:
raise RuntimeError(
'refusing to write: the confirmation does not match the resolved status database host. '
f'confirmation={CONFIRM_TARGET!r}, resolved POSTGRES_HOST={STATUS_DB_HOST!r}. '
'Read the targets cell 1 printed, then restate the host here.'
)
print(f'target confirmed: {STATUS_DB_HOST} — the write cells may run')
# %%
# Cell 3 — pick the bundle, compute the digest the frontend must send, print its header.
#
# `expected_digest` is the SHA-256 of the exact bytes that get uploaded, computed the way the
# frontend computes it: over the file, not over anything derived from it. The header fields printed
# here are plaintext at the front of the file — reading them decrypts nothing.
bundle_bytes = BUNDLE_PATH.read_bytes()
expected_digest = hashlib.sha256(bundle_bytes).hexdigest()
object_name = BUNDLE_PATH.name
object_key = f'{BUNDLE_PREFIX}{object_name}'
header = parse_header(bundle_bytes[:HEADER_SIZE])
print(_line('file', BUNDLE_PATH))
print(_line('size', f'{len(bundle_bytes)} bytes'))
print(_line('expected_digest', expected_digest))
print(_line('object key', f'{BUNDLE_BUCKET}/{object_key}'))
print(_line('format version', header.format_version))
print(_line('aead / kdf id', f'{header.aead_id} / {header.kdf_id}'))
print(_line('kdf memlimit', f'{header.kdf_memlimit_bytes} bytes'))
print(_line('kdf opslimit', header.kdf_opslimit))
print(_line('kdf parallelism', header.kdf_parallelism))
print(_line('salt', header.salt.hex()))
print(_line('ciphertext digest', header.ciphertext_digest.hex()))
print(_line('chunk stream from', f'offset {DIGEST_OFFSET}'))
# %%
# Cell 4 — build the password envelope, exactly as the frontend must build it.
#
# Wire shape: base64( nonce(12) || ciphertext || GCM tag(16) ) under AES-256-GCM with the 32-byte
# key that `IMPORT_PASSWORD_KEY` carries base64-encoded. The four lines below are the whole of it,
# written out rather than imported so the Angular side can transliterate them to WebCrypto:
#
# const k = await crypto.subtle.importKey('raw', keyBytes, 'AES-GCM', false, ['encrypt']);
# const nonce = crypto.getRandomValues(new Uint8Array(12));
# const sealed = await crypto.subtle.encrypt({name: 'AES-GCM', iv: nonce}, k, passwordBytes);
# const envelope = base64(concat(nonce, new Uint8Array(sealed)));
#
# WebCrypto appends the GCM tag to the ciphertext, and so does `cryptography`'s AESGCM, so the two
# produce byte-identical payloads. The decrypt counterpart the worker runs is
# `laborious/utils/import_password.py` (`decrypt_password_envelope`).
#
# `IMPORT_PASSWORD_KEY` must be base64 of exactly 32 bytes — `openssl rand -base64 32`, or the
# value the deployment's Kubernetes Secret carries. Anything else fails here, which is the point:
# the worker resolves the same variable the same way, so a value this cell cannot decode is a value
# the worker could not have decoded either.
#
# The bundle password is read verbatim — `env_secret_required`, not `env_required` — because a
# trailing space is part of the password, and stripping it here would seal an envelope whose
# plaintext the bundle reader rejects at gate 3.
#
# Neither the key nor the password nor the plaintext is printed — only the envelope's length.
envelope_key = base64.b64decode(env_required('IMPORT_PASSWORD_KEY'), validate=True)
nonce = secrets.token_bytes(12)
sealed = AESGCM(envelope_key).encrypt(
nonce, env_secret_required('IMPORT_SCRIPT_BUNDLE_PASSWORD').encode('utf-8'), None
)
password_envelope = base64.b64encode(nonce + sealed).decode('ascii')
print(_line('envelope key', f'{len(envelope_key)} bytes (not printed)'))
print(_line('envelope', f'{len(password_envelope)} base64 characters (not printed)'))
# %%
# Cell 5 — FIRST WRITE. Upload the object, then prove the stored bytes are the ones we hashed.
#
# The re-read is the point: `expected_digest` is a promise about what is in the bucket, and the
# worker's gate 1 holds the upload to it. If the digest of the stored object differs, the frontend
# contract is already broken and there is no reason to insert a row.
_endpoint = urlparse(MINIO_ENDPOINT_URL)
minio_client = Minio(
_endpoint.netloc or MINIO_ENDPOINT_URL,
access_key=MINIO_ACCESS_KEY,
secret_key=MINIO_SECRET_KEY,
secure=_endpoint.scheme == 'https',
)
minio_client.put_object(
BUNDLE_BUCKET,
object_key,
io.BytesIO(bundle_bytes),
length=len(bundle_bytes),
content_type='application/octet-stream',
)
stat = minio_client.stat_object(BUNDLE_BUCKET, object_key)
_response = minio_client.get_object(BUNDLE_BUCKET, object_key)
try:
stored_bytes = _response.read()
finally:
_response.close()
_response.release_conn()
stored_digest = hashlib.sha256(stored_bytes).hexdigest()
if stored_digest != expected_digest:
raise RuntimeError(
f'the stored object does not hash to the digest we computed: stored {stored_digest}, '
f'expected {expected_digest} — do not insert a row for this upload'
)
print(_line('uploaded', f'{BUNDLE_BUCKET}/{object_key}'))
print(_line('reported size', f'{stat.size} bytes'))
print(_line('etag', stat.etag))
print(_line('re-read digest', f'{stored_digest} (matches expected_digest)'))
# %%
# Cell 6 — SECOND WRITE. Insert the import log row: the BFF's half of the contract.
#
# `URL.create` rather than a URL string, so a password containing `@ : /` survives. `experiment_name`
# is a placeholder: the column is NOT NULL and the frontend cannot see inside an encrypted bundle,
# so the worker overwrites it with the bundle's own experiment name as soon as the bundle is open
# (`record_import_names`). `file_name` is the object key, because the worker compares the two and
# refuses the import at the `received` step when they disagree.
status_engine = create_engine(
URL.create(
'postgresql',
username=STATUS_DB_USER,
password=STATUS_DB_PASSWORD,
host=STATUS_DB_HOST,
port=STATUS_DB_PORT,
database=STATUS_DB_NAME,
),
pool_pre_ping=True,
)
request_data = {
'expected_digest': expected_digest,
'bucket': BUNDLE_BUCKET,
'object_key': object_key,
'file_name': object_name,
'source': 'scripts/import_model_manual_run.py',
}
_now = datetime.now(UTC).replace(tzinfo=None)
_placeholder_experiment = f'import-{Path(object_name).stem}'[:50]
INSERT_ROW_SQL = text(
'INSERT INTO public.experiment_run '
'(experiment_name, username, status, created_at, updated_at, '
' bucket_name, file_name, request_data, run_type) '
'VALUES (:experiment_name, :username, :status, :created_at, :updated_at, '
' :bucket_name, :file_name, CAST(:request_data AS JSONB), :run_type) '
'RETURNING id'
)
with status_engine.begin() as connection:
import_run_id = connection.execute(
INSERT_ROW_SQL,
{
'experiment_name': _placeholder_experiment,
'username': IMPORT_USERNAME,
'status': 'PENDING',
'created_at': _now,
'updated_at': _now,
'bucket_name': BUNDLE_BUCKET,
'file_name': object_key,
'request_data': json.dumps(request_data),
'run_type': 'IMPORT',
},
).scalar_one()
print(_line('import_run_id', import_run_id))
print(_line('status / run_type', 'PENDING / IMPORT'))
print(_line('experiment_name', f'{_placeholder_experiment} (the worker overwrites this)'))
print(_line('file_name', object_key))
# %%
# Cell 7 — the workflow input, ready to paste into the Temporal UI.
#
# These five keys are the whole contract. `bucket` is optional — the worker falls back to
# IMPORT_BUNDLE_BUCKET — and is sent explicitly so the printed input is self-describing. No model
# name is sent: the worker takes it from the bundle's own metadata, verbatim.
workflow_input = {
'import_run_id': import_run_id,
'bucket': BUNDLE_BUCKET,
'object_key': object_key,
'expected_digest': expected_digest,
'password_envelope': password_envelope,
}
print(f'workflow: import_model task queue: {IMPORT_TASK_QUEUE}')
print(json.dumps(workflow_input, indent=2))
# %%
# Cell 8 — optionally start the workflow and wait for it.
#
# Skipping this cell is normal: the operator may prefer to start the run from the Temporal UI with
# the JSON cell 7 printed. Cell 9 works either way, because it looks the outcome up by row id.
#
# The coroutine runs on a loop of its own in a worker thread rather than through `asyncio.run`, so
# this cell behaves the same in the interactive window (which already has a running loop) as it
# does when the module is executed top to bottom (which does not).
START_THE_WORKFLOW = True
if START_THE_WORKFLOW:
import asyncio
from concurrent.futures import ThreadPoolExecutor
from temporalio.client import Client
async def start_and_wait() -> Any:
"""Start `import_model` on the import queue and return its result."""
client = await Client.connect(TEMPORAL_HOST, namespace=TEMPORAL_NAMESPACE)
handle = await client.start_workflow(
'import_model',
workflow_input,
id=f'manual-import-{import_run_id}',
task_queue=IMPORT_TASK_QUEUE,
)
print(_line('workflow id', handle.id))
print(_line('run id', handle.result_run_id))
return await handle.result()
with ThreadPoolExecutor(max_workers=1) as _pool:
workflow_result = _pool.submit(asyncio.run, start_and_wait()).result()
print(json.dumps(workflow_result, indent=2, default=str))
else:
print('not started — paste the input from cell 7 into the Temporal UI, or set')
print(f'START_THE_WORKFLOW = True and re-run this cell. Task queue: {IMPORT_TASK_QUEUE}')
# %%
# Cell 9 — read the outcome back from all three places it lands.
#
# The row is authoritative for the verdict; MLflow and MongoDB are what the verdict is about. Each
# block is labelled, so the output can be pasted into a ticket as it is.
print('--- import log row ---')
READ_ROW_SQL = text(
'SELECT id, experiment_name, run_name, status, error_message, run_type, '
' bucket_name, file_name, created_at, updated_at, '
' request_data, orchestrator_response_data '
'FROM public.experiment_run WHERE id = :id'
)
with status_engine.begin() as connection:
_result = connection.execute(READ_ROW_SQL, {'id': import_run_id})
row = _result.mappings().one_or_none()
if row is None:
print(f' no row with id {import_run_id}')
else:
for _column, _value in row.items():
print(_line(_column, _value))
print()
print('--- mlflow ---')
os.environ['MLFLOW_TRACKING_USERNAME'] = env('MLFLOW_USERNAME')
os.environ['MLFLOW_TRACKING_PASSWORD'] = os.getenv('MLFLOW_PASSWORD', '')
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
mlflow_client = mlflow.tracking.MlflowClient()
_experiment_name = row['experiment_name'] if row is not None else ''
_experiment = mlflow_client.get_experiment_by_name(_experiment_name) if _experiment_name else None
# The experiment is get-or-create, so it can perfectly well be the model's own experiment with
# months of training runs in it. `_run_ids` is everything in it — printed, and used to decide
# whether the experiment is this import's alone — while `_import_run_ids` is only the run this
# import created, matched on the run name the record carries. The undo acts on the second list.
_row_run_name = (row['run_name'] if row is not None else '') or ''
_run_ids: list[str] = []
_import_run_ids: list[str] = []
if _experiment is None:
print(f' no experiment named {_experiment_name!r}')
else:
print(_line('experiment', f'{_experiment.name} ({_experiment.experiment_id})'))
for _run in mlflow_client.search_runs([_experiment.experiment_id]):
_mine = bool(_row_run_name) and _run.info.run_name == _row_run_name
print(
_line(
'run',
f'{_run.info.run_name} ({_run.info.run_id}) {_run.info.status}'
f'{" <- this import" if _mine else ""}',
)
)
_run_ids.append(_run.info.run_id)
if _mine:
_import_run_ids.append(_run.info.run_id)
for _artifact in mlflow_client.list_artifacts(_run.info.run_id):
print(_line(' artifact', _artifact.path))
# The registered model is looked up by `run_id`, not by a name reconstructed from `run_name`. The
# run name is `import-{model_name}-v{model_version}` truncated to 50 characters (`derive_run_name`),
# and a 32-character `model_version` leaves ten characters for the name — so `Quartz-demo` comes
# back as `Quartz-dem` and the lookup finds nothing. The version's `run_id` is exact, and it is what
# ties the registered model to the run this import created whatever the name was truncated to.
_registered_names: set[str] = set()
_registered_versions: list[tuple[str, str]] = []
for _run_id in _import_run_ids:
try:
for _version in mlflow_client.search_model_versions(f"run_id='{_run_id}'"):
_registered_names.add(_version.name)
_registered_versions.append((_version.name, str(_version.version)))
print(
_line(
'version',
f'{_version.name} v{_version.version} stage={_version.current_stage} '
f'run={_version.run_id}',
)
)
print(_line(' source', _version.source))
except Exception as error:
# A read-back cell reports what it could not read; it does not fail the inspection.
print(f' registry read failed: {type(error).__name__}: {error}')
if not _import_run_ids:
print(' no run of this import in the experiment yet, so no registered model to look up')
elif not _registered_names:
print(" no registered version points at this import's run")
print()
print('--- mongodb model document ---')
_credentials = f'{MONGODB_USERNAME}:{MONGODB_PASSWORD}@' if MONGODB_USERNAME else ''
mongo_client: MongoClient = MongoClient(f'mongodb://{_credentials}{MONGODB_URL}')
_collection = mongo_client[MONGODB_DATABASE][MODELS_COLLECTION]
for _document in _collection.find({}, {'_id': False}).sort('id', 1):
print(_line('document', _document))
# %%
# Cell 10 — undo.
#
# The object and the row always: they are what this script created, and leaving them behind leaves a
# PENDING import nobody will ever run. What MLflow and MongoDB hold is left alone unless
# DELETE_PROVISIONING is set, because after a successful run that is the thing you wanted to look at.
#
# When it *is* set, the undo removes only what this import added. An import of a model the platform
# already had lands as a new *version* of that registered model and writes no document at all, so
# deleting the registered model by name would take versions trained months ago with it. The version
# this import created is deleted by number; the registered model and the document go only when
# nothing else is left pointing at that name — which is exactly the case where this import was the
# first one.
DELETE_PROVISIONING = False
try:
minio_client.remove_object(BUNDLE_BUCKET, object_key)
print(_line('deleted object', f'{BUNDLE_BUCKET}/{object_key}'))
except Exception as error:
# Undo reports what it could not remove and carries on to the row.
print(f' could not delete the object: {type(error).__name__}: {error}')
with status_engine.begin() as connection:
_deleted = connection.execute(
text('DELETE FROM public.experiment_run WHERE id = :id'),
{'id': import_run_id},
).rowcount
print(_line('deleted rows', _deleted))
if DELETE_PROVISIONING:
for _name, _number in _registered_versions:
try:
mlflow_client.delete_model_version(_name, _number)
print(_line('deleted version', f'{_name} v{_number}'))
except Exception as error:
# Same rule as above: report and carry on to the next one.
print(f' could not delete the model version: {type(error).__name__}: {error}')
for _registered in sorted(_registered_names):
try:
_remaining = list(mlflow_client.search_model_versions(f"name='{_registered}'"))
except Exception as error:
print(f' could not count the remaining versions: {type(error).__name__}: {error}')
continue
if _remaining:
# Versions this import did not create: the model was on the platform before it ran, so
# the registered model and its document are somebody else's and stay exactly as they are.
print(_line('kept model', f'{_registered} ({len(_remaining)} other versions)'))
continue
try:
mlflow_client.delete_registered_model(_registered)
print(_line('deleted model', _registered))
except Exception as error:
print(f' could not delete the registered model: {type(error).__name__}: {error}')
_removed = _collection.delete_many({'name': _registered}).deleted_count
print(_line('deleted documents', f'{_removed} named {_registered}'))
# Last, and only when the experiment holds nothing but this import's own run: the experiment is
# get-or-create, so an import into a model the platform already had reuses the experiment its
# training runs live in, and deleting it would take them all.
if _experiment is not None:
_others = len(_run_ids) - len(_import_run_ids)
if _others:
print(_line('kept experiment', f'{_experiment.name} ({_others} other runs)'))
else:
mlflow_client.delete_experiment(_experiment.experiment_id)
print(_line('deleted experiment', _experiment.experiment_id))
else:
print(' provisioning kept — set DELETE_PROVISIONING = True to remove it too')
status_engine.dispose()
mongo_client.close()

11
scripts/run_coverage.sh Executable file
View File

@@ -0,0 +1,11 @@
#!/bin/bash
# Exit on any error
set -e
echo "Activating virtual environment..."
source ./venv/bin/activate
pytest --cov=laborious --cov-report=html
xdg-open htmlcov/index.html

18
scripts/run_local.sh Executable file
View File

@@ -0,0 +1,18 @@
#!/bin/bash
# Exit on any error
set -e
echo "Activating virtual environment..."
source ./venv/bin/activate
echo "Loading environment variables from .env..."
if [ -f .env ]; then
export $(cat .env | grep -v '^#' | xargs)
echo "Environment variables loaded from .env"
else
echo "Warning: .env file not found. Continuing without environment variables."
fi
echo "Starting ingestor application..."
python -m laborious.worker.worker