How to Implement Custom Storage Backends for CMF Artifacts: A Complete Guide

Implementing a custom storage backend for CMF artifacts requires creating a Python class with download_file and download_directory methods in cmflib/storage_backends/, then registering it in cmflib/commands/artifact/pull.py to handle the cmf artifact pull command.

CMF (Continuous Machine-learning Framework) stores pipeline artifacts in a pluggable storage-backend layer that supports multiple remote protocols. Each backend implements a standardized Python interface invoked by the cmf artifact pull command, allowing seamless integration with local filesystems, cloud providers, or custom storage solutions.

Understanding the CMF Storage Backend Interface

CMF backends follow a strict contract defined by two public methods. These methods handle the retrieval of single files and entire directories from remote storage locations.

The Required Method Contract

Every custom backend must implement the following signatures exactly as found in the existing implementations:

def download_file(
    self,
    current_directory: str,
    object_name: str,
    download_loc: str,
) -> tuple[str, str, bool]:
    ...

def download_directory(
    self,
    current_directory: str,
    object_name: str,
    download_loc: str,
) -> tuple[int, int, bool]:
    ...

The download_file method returns a tuple containing the object name, download location, and a boolean success flag. The download_directory method returns the total files expected, files successfully downloaded, and a success flag indicating complete retrieval.

Built-in Storage Backend Reference

The CMF repository includes five reference implementations in cmflib/storage_backends/ that demonstrate different storage patterns:

• LocalArtifacts (local_artifacts.py): Uses DVC's DVCFileSystem for local repository access
• MinioArtifacts (minio_artifacts.py): Wraps the MinIO Python client for S3-compatible storage
• AmazonS3Artifacts (amazonS3_artifacts.py): Implements AWS S3 retrieval using boto3
• SSHremoteArtifacts (sshremote_artifacts.py): Employs paramiko for SFTP-based transfers
• OSDFremoteArtifacts (osdf_artifacts.py): Handles HTTP GET requests with token-based authentication and cache fallback

Each backend initializes its client in __init__ using configuration from dvc_config_op, then implements the download logic according to the storage protocol.

Step-by-Step Implementation Guide

Step 1: Create the Backend Class

Create a new Python file in cmflib/storage_backends/ (e.g., azure_blob_artifacts.py). Your class must implement the two required methods with exact signatures, handling client initialization, directory creation, and error logging:

import os
import logging
from azure.storage.blob import BlobServiceClient

logger = logging.getLogger(__name__)

class AzureBlobArtifacts:
    def __init__(self, dvc_config_op: dict):
        account_url = dvc_config_op["remote.azure-blob.account_url"]
        credential = dvc_config_op["remote.azure-blob.credential"]
        self.client = BlobServiceClient(
            account_url=account_url, 
            credential=credential
        )

The __init__ method receives dvc_config_op, a dictionary containing parsed DVC remote configuration including credentials and endpoint URLs.

Step 2: Integrate with the Pull Command

Register your backend in cmflib/commands/artifact/pull.py by adding an import and conditional branch. The dispatcher reads dvc_config_op["core.remote"] to select the appropriate backend:

# Add near other imports

from cmflib.storage_backends import azure_blob_artifacts

# Inside the main execution logic

if dvc_config_op["core.remote"] == "minio":
    backend = minio_artifacts.MinioArtifacts(dvc_config_op)
elif dvc_config_op["core.remote"] == "azure-blob":
    backend = azure_blob_artifacts.AzureBlobArtifacts(dvc_config_op)

Step 3: Add URL Parsing Logic

Extend the extract_repo_args method in pull.py to handle your backend's URL format. This method parses artifact URLs and extracts components needed for downloading:

elif type == "azure-blob":
    token = url.split("/")
    container_name = token[2]
    object_name = "/".join(token[3:])
    download_loc = current_directory + "/" + name
    return container_name, object_name, download_loc

The method returns backend-specific arguments that get passed to your download_file or download_directory methods.

Step 4: Configure DVC Remote Settings

Register the remote with DVC so CMF can discover it in the configuration:

dvc remote add -d azure-blob azure://myaccount.blob.core.windows.net/mycontainer
dvc remote modify azure-blob account_name myaccount
dvc remote modify azure-blob credential <SAS-token-or-key>

These settings populate dvc_config_op with keys like remote.azure-blob.account_url and remote.azure-blob.credential, accessible in your backend's constructor.

Step 5: Document Your Backend

Add documentation in docs/ui/artifacts.md describing the required DVC configuration, expected URL formats, and any environment variables or CLI flags specific to your implementation.

Complete Example: Azure Blob Storage Backend

Below is a production-ready implementation following CMF's patterns for directory metadata handling:

# File: cmflib/storage_backends/azure_blob_artifacts.py

import os
import logging
from azure.storage.blob import BlobServiceClient

logger = logging.getLogger(__name__)

class AzureBlobArtifacts:
    """
    Storage backend that pulls artifacts from Azure Blob Storage.
    Expected DVC remote config keys:
        remote.azure-blob.account_url
        remote.azure-blob.credential
    """

    def __init__(self, dvc_config_op: dict):
        account_url = dvc_config_op["remote.azure-blob.account_url"]
        credential = dvc_config_op["remote.azure-blob.credential"]
        self.client = BlobServiceClient(
            account_url=account_url, 
            credential=credential
        )

    def download_file(
        self,
        current_directory: str,
        container_name: str,
        object_name: str,
        download_loc: str,
    ) -> tuple[str, str, bool]:
        """Pull a single blob to download_loc."""
        os.makedirs(os.path.dirname(download_loc), mode=0o777, exist_ok=True)
        
        try:
            container = self.client.get_container_client(container_name)
            blob = container.get_blob_client(object_name)
            
            with open(download_loc, "wb") as f:
                stream = blob.download_blob()
                f.write(stream.readall())
            return object_name, download_loc, True
        except Exception as exc:
            logger.error(f"[AzureBlob] download_file failed: {exc}")
            return object_name, download_loc, False

    def download_directory(
        self,
        current_directory: str,
        container_name: str,
        dir_metadata_name: str,
        download_loc: str,
    ) -> tuple[int, int, bool]:
        """Pull a directory described by a *.dir metadata blob."""
        os.makedirs(download_loc, mode=0o777, exist_ok=True)
        
        # Download metadata file

        temp_meta = os.path.join(download_loc, "temp.dir")
        success, _ = self.download_file(
            current_directory, container_name, 
            dir_metadata_name, temp_meta
        )
        if not success:
            return 1, 0, False

        with open(temp_meta, "r") as f:
            tracked = eval(f.read())
        os.remove(temp_meta)

        total, downloaded = 0, 0
        repo_path = "/".join(dir_metadata_name.split("/")[:-2])

        for info in tracked:
            total += 1
            rel = info["relpath"]
            md5 = info["md5"]
            formatted = f"{md5[:2]}/{md5[2:]}"
            blob_name = f"{repo_path}/{formatted}"
            local_path = os.path.join(download_loc, rel)

            os.makedirs(os.path.dirname(local_path), mode=0o777, exist_ok=True)
            
            ok = self.download_file(
                current_directory, container_name, 
                blob_name, local_path
            )[2]
            if ok:
                downloaded += 1
            else:
                logger.error(f"[AzureBlob] failed to download {blob_name}")

        return total, downloaded, (total == downloaded)

This implementation handles both single-file artifacts and DVC-tracked directories (represented by .dir metadata files), following the same parsing logic used in minio_artifacts.py and amazonS3_artifacts.py.

Key Source Files and References

When implementing custom storage backends, reference these existing implementations in the hewlettpackard/cmf repository:

• cmflib/storage_backends/local_artifacts.py: Reference for local filesystem access using DVCFileSystem
• cmflib/storage_backends/minio_artifacts.py: Cloud-native pattern for S3-compatible APIs
• cmflib/storage_backends/amazonS3_artifacts.py: AWS S3 implementation using boto3 client initialization
• cmflib/storage_backends/sshremote_artifacts.py: Paramiko-based remote file handling
• cmflib/storage_backends/osdf_artifacts.py: HTTP-based retrieval with cache fallback logic
• cmflib/commands/artifact/pull.py: Central dispatcher that selects backends based on dvc_config_op["core.remote"]
• cmflib/utils/helper_functions.py: Configuration loading utilities (fetch_cmf_config_path)
• cmflib/cmflib_config.py: .cmfconfig file parsing for credential management

Summary

Implementing a custom storage backend for CMF artifacts involves:

• Creating a Python class in cmflib/storage_backends/ that implements the download_file and download_directory interface with exact method signatures
• Registering the backend in cmflib/commands/artifact/pull.py by adding an import and conditional branch based on dvc_config_op["core.remote"]
• Extending URL parsing in pull.py's extract_repo_args method to handle your storage backend's specific URL format
• Configuring DVC remotes to populate the dvc_config_op dictionary with credentials and endpoint information required by your backend constructor

Frequently Asked Questions

What Python methods must a custom CMF storage backend implement?

A custom backend must implement download_file(self, current_directory, object_name, download_loc) and download_directory(self, current_directory, object_name, download_loc). The file method returns a tuple of (object_name, download_loc, success_flag), while the directory method returns (total_files, files_downloaded, success_flag).

How does CMF discover which storage backend to use?

The cmf artifact pull command reads the core.remote setting from .cmfconfig via dvc_config_op["core.remote"]. It matches this value against hardcoded strings in cmflib/commands/artifact/pull.py (e.g., "minio", "local-storage", "azure-blob") to instantiate the appropriate backend class.

Can custom backends access DVC configuration credentials?

Yes. The dvc_config_op dictionary passed to your backend's __init__ method contains all DVC remote configuration values. For a remote named azure-blob, settings like account_url and credential are accessible via dvc_config_op["remote.azure-blob.account_url"] and dvc_config_op["remote.azure-blob.credential"].

Where should custom storage backend files be placed?

All storage backend implementations belong in the cmflib/storage_backends/ directory of the CMF repository. Follow the naming convention <backend>_artifacts.py (e.g., azure_blob_artifacts.py) and ensure your class name uses CamelCase (e.g., AzureBlobArtifacts).

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}