To implement this reliably, package the model, compute checksum, upload artifact, create registry entry (via API), and trigger CI/CD—while ensuring idempotency, retries, error handling, and secure credentials. Approach: small modular functions, exponential backoff retries, deterministic artifact naming (version + git sha), and transactional semantics (write metadata after successful upload).python
import hashlib, json, os, subprocess, time
from pathlib import Path
import boto3, requests
from botocore.exceptions import ClientError
MAX_RETRIES = 5
def compute_checksum(path):
h = hashlib.sha256()
with open(path,'rb') as f:
for chunk in iter(lambda: f.read(8192), b''):
h.update(chunk)
return h.hexdigest()
def retry(fn, *args, **kwargs):
for attempt in range(1, MAX_RETRIES+1):
try:
return fn(*args, **kwargs)
except Exception as e:
if attempt==MAX_RETRIES: raise
time.sleep(2**attempt)
def package_model(model_dir, out_dir, version, git_sha):
out = Path(out_dir)/f"model-{version}-{git_sha}.tar.gz"
if out.exists(): return out # idempotent
subprocess.check_call(["tar","-czf",str(out),"-C",model_dir,"."])
return out
def upload_s3(file_path, bucket, key):
s3 = boto3.client('s3') # credentials via env/role
retry(lambda: s3.upload_file(str(file_path), bucket, key))
def create_registry_entry(api_url, token, payload):
headers = {'Authorization':f'Bearer {token}','Content-Type':'application/json'}
# idempotency via client-provided id (e.g., checksum)
retry(lambda: requests.post(api_url, headers=headers, data=json.dumps(payload)).raise_for_status())
def trigger_pipeline(ci_api, token, body):
headers = {'Authorization':f'Bearer {token}','Content-Type':'application/json'}
retry(lambda: requests.post(ci_api, headers=headers, json=body).raise_for_status())
# Example orchestration (omitting argument parsing)
# 1) package -> 2) checksum -> 3) upload -> 4) registry entry -> 5) trigger pipeline
