Model Deployment API Demo

This notebook demonstrates how to use the Datamint API to deploy machine learning models as Docker images.

IMPORTANT: Registered Model it not the same as a deployed model!

A Registered Model is a model that has been uploaded to the platform, that might or might not be deployed. Is a requirement to have a Registered Model before deploying it. In this demo we will show how to deploy a registered model.

Features Covered:

• Starting deployment jobs

• Monitoring deployment status

• Cancelling jobs

• Managing deployed images

• Checking image existence

Setup and Initialization

First, import the necessary libraries and initialize the API client.

import time

from datamint import Api

# Initialize the API client
# The API key can be set via environment variable DATAMINT_API_KEY
# or passed directly to the Api constructor
api = Api()

print("API client initialized successfully!")

List current registered models

# get all versions of all registered models
from datetime import datetime

from mlflow import MlflowClient

client = MlflowClient()
all_registered_models = client.search_registered_models()
for rm in all_registered_models:
    print(f"Registered Model: {rm.name}")
    model_versions = client.search_model_versions(f"name='{rm.name}'")
    # get all all attributes of each model version
    for mv in model_versions:
        created_datetime = datetime.fromtimestamp(mv.creation_timestamp / 1000.0) # converting from milliseconds
        created_datetime = created_datetime.strftime("%Y-%m-%d %H:%M:%S") # formatting
        print(f"  Version: {mv.version} | Created at: {created_datetime}")
        # INFO: do try to access ``mv.aliases``, since mlflow does not populate this field with the ``search_model_versions``.
        # If you need aliases, use ``get_model_version(mv.name, mv.version)`` method instead.

Loading a Registered Model Locally

Before deploying, you can load a model version directly in Python via ModelVersion.load_model() (a shortcut over datamint.mlflow.flavors.load_model).

# Load a model version locally, without deploying it
model = api.models.get_by_name(all_registered_models[1].name)
latest = model.get_latest_version()

loaded_model = latest.load_model()
print(loaded_model)

Starting a Deployment Job

Deploy a model by specifying the model name and optionally a version or alias.

# Start a deployment job
# You can specify either model_version (int) or model_alias (str), but not both
job = api.deploy.start(
    model_name=all_registered_models[1].name,
    model_alias="latest",
    # or use model_version=1
    with_gpu=False,
)

print("Deployment job started!")
print(f"Job ID: {job.id}")
print(f"Status: {job.status}")
print(f"Model: {job.model_name}")

Monitoring Job Status

Check the status of a deployment job and monitor its progress.

# Get job status by ID
job_status = api.deploy.get_by_id(job.id)

print(f"Job Status: {job_status.status}")
print(f"Progress: {job_status.progress_percentage}%")
print(f"Current Step: {job_status.current_step}")
print(f"Image Name: {job_status.image_name}")
print(f"Image Tag: {job_status.image_tag}")

if job_status.error_message:
    print(f"Error: {job_status.error_message}")

if job_status.recent_logs:
    print("\nRecent Logs:")
    for log in job_status.recent_logs[-5:]:  # Show last 5 logs
        print(f"  {log}")

Polling for Job Completion

Monitor a job until it completes (or fails).

def wait_for_job_completion(job_id, max_wait_seconds=600, poll_interval=2):
    """
    Wait for a deployment job to complete.

    Args:
        job_id: The job ID to monitor
        max_wait_seconds: Maximum time to wait (default 10 minutes)
        poll_interval: How often to check status (default 2 seconds)
    """
    start_time = time.time()
    last_msg = ""
    while time.time() - start_time < max_wait_seconds:
        job_status = api.deploy.get_by_id(job_id)

        msg = f"Status: {job_status.status} | Progress: {job_status.progress_percentage}% | Step: {job_status.current_step}"
        if msg != last_msg:
            print(msg)
        last_msg = msg

        if job_status.status in ['completed', 'failed', 'cancelled']:
            return job_status

        time.sleep(poll_interval)

    print("Timeout waiting for job completion")
    return None

# Wait for the job to complete
final_status = wait_for_job_completion(job.id)

if final_status:
    print(f"\nFinal Status: {final_status.status}")
    if final_status.status == 'completed':
        print(f"Image built successfully: {final_status.image_name}:{final_status.image_tag}")
    elif final_status.status == 'failed':
        print(f"Error: {final_status.error_message}")

Cancelling a Job

Cancel a running deployment job.

# Cancel a job
cancelled = api.deploy.cancel(job.id)

if cancelled:
    print(f"Job {job.id} cancelled successfully")
else:
    print(f"Job {job.id} could not be cancelled (may have already completed)")

Listing Active Jobs

Get information about active deployment jobs.

# List active jobs
active_jobs = api.deploy.list_active_jobs()

print(f"Active jobs count: {active_jobs['active_jobs_count']}")
active_jobs

Managing Deployed Images

List and manage Docker images that have been deployed.

# List all deployed images
all_images = api.deploy.list_images()

print(f"Total deployed images: {len(all_images)}")
for img in all_images[:5]:  # Show first 5
    print(f"  - {img['full_name']} ({img['size_mb']:.2f} MB)")
    print(f"    Created: {img['created']}")

# List images for a specific model
model_images = api.deploy.list_images(model_name="FracAtlas_adapted")

print(f"Images for 'FracAtlas_adapted': {len(model_images)}")
for img in model_images:
    print(f"  - {img['name']}:{img['tag']}")

Checking if an Image Exists

Verify if a specific model image exists.

# Check if an image exists
exists = api.deploy.image_exists(
    model_name="FracAtlas_adapted",
    tag="champion"
)

if exists:
    print("Image exists!")
else:
    print("Image not found")

Removing Deployed Images

Remove Docker images that are no longer needed.

# Remove a specific image tag
result = api.deploy.remove_image(
    model_name="my_model",
    tag="v1.0"
)

print(f"Removal {'successful' if result['success'] else 'failed'}")
print(f"Message: {result['message']}")
if result.get('removed_tags'):
    print(f"Removed tags: {result['removed_tags']}")

# Remove all images for a model
result = api.deploy.remove_image(
    model_name="my_model"
    # No tag specified = remove all tags
)

print(f"Removal {'successful' if result['success'] else 'failed'}")
print(f"Message: {result['message']}")
if result.get('removed_tags'):
    print(f"Removed tags: {', '.join(result['removed_tags'])}")

Complete Workflow Example

Here’s a complete workflow from deployment to cleanup.

def deploy_and_monitor_model(model_name, model_alias="champion"):
    """
    Complete workflow: deploy, monitor, and verify a model.
    """
    print(f"1. Starting deployment for {model_name}@{model_alias}")
    job = api.deploy.start(
        model_name=model_name,
        model_alias=model_alias,
        with_gpu=False
    )
    print(f"   Job ID: {job.id}")

    print("\n2. Monitoring deployment progress...")
    final_status = wait_for_job_completion(job.id, poll_interval=5)

    if final_status and final_status.status == 'completed':
        print("\n3. Deployment completed successfully!")
        print(f"   Image: {final_status.image_name}:{final_status.image_tag}")

        # Verify image exists
        exists = api.deploy.image_exists(
            model_name=model_name,
            tag=final_status.image_tag
        )
        print(f"   Image verification: {'✓' if exists else '✗'}")

        return final_status
    else:
        print("\n3. Deployment failed or timed out")
        if final_status and final_status.error_message:
            print(f"   Error: {final_status.error_message}")
        return None

# Run the complete workflow
result = deploy_and_monitor_model("my_production_model", "champion")

Summary

This notebook covered:

• ✓ Starting deployment jobs with various configurations

• ✓ Monitoring job progress and status

• ✓ Cancelling running jobs

• ✓ Listing and managing deployed images

• ✓ Checking image existence

• ✓ Removing images

• ✓ Complete workflow examples