How to Query Tracked Metadata Programmatically Using the CmfQuery API

Use the CmfQuery class in cmflib/cmfquery.py to connect to an MLMD backend and retrieve pipeline, execution, and artifact metadata as pandas DataFrames.

The Hewlett Packard Enterprise CMF (Component Management Framework) stores all machine learning pipeline information—stages, executions, and artifacts—in an ML Metadata (MLMD) database. The CmfQuery API provides the primary Python interface for programmatically querying this tracked metadata, abstracting away protobuf complexity and returning results as native pandas objects.

CmfQuery Architecture and Data Flow

The CmfQuery implementation in cmflib/cmfquery.py follows a layered architecture that separates storage concerns from data transformation.

Storage Backend Abstraction

The CmfQuery.__init__ method initializes either an SQLite file connection (SqlliteStore) or a PostgreSQL server connection (PostgresStore) based on the is_server parameter. When is_server=False, the class instantiates SqlliteStore pointing to the local MLMD file; when is_server=True, it connects to PostgreSQL using environment variables read via get_postgres_config from cmflib/utils/helper_functions.py.

Data Transformation Layer

Raw MLMD protobuf objects retrieved from the store undergo transformation through private helper methods. The _transform_to_dataframe function (line 173 in cmfquery.py) flattens properties and custom_properties into dictionary rows using _copy and _KeyMapper subclasses. This ensures collision-free column names when converting to pandas DataFrames.

High-Level Query Methods

Public methods such as get_all_executions_in_stage, get_all_artifacts_for_execution, and get_all_executions_for_artifact combine primitive store calls with the transformation layer. These methods iterate over pipeline contexts, stage contexts, and execution events, concatenating rows via pd.concat to return complete DataFrames.

Connecting to the MLMD Backend

Instantiate CmfQuery with the appropriate backend parameters to establish a connection to your tracked metadata store.

Local SQLite File:

from cmflib.cmfquery import CmfQuery

# Connect to a local SQLite MLMD file (default name is "mlmd")

query = CmfQuery(filepath="mlmd", is_server=False)

PostgreSQL Server:

# Connect to PostgreSQL (requires CMF_POSTGRES_*, CMF_DB_* environment variables)

query = CmfQuery(is_server=True)

The connect() method internally creates self.store, which exposes low-level retrieval methods like get_contexts_by_type and get_events_by_artifact_ids.

Querying Pipeline and Stage Metadata

Retrieve high-level pipeline structures and drill down into stage-specific execution history.

List All Pipelines

The get_pipeline_names() method iterates over pipeline contexts and returns a list of pipeline identifiers:

pipeline_names = query.get_pipeline_names()
print("Available pipelines:", pipeline_names)

Implementation reference: This method calls _get_pipelines() and extracts the name attribute from each context object at line 74 in cmfquery.py.

Retrieve Stage Executions

Fetch all execution records for a specific stage using get_all_executions_in_stage:

stage_name = "pipeline1.stage_preprocess"
executions_df = query.get_all_executions_in_stage(stage_name)
print(executions_df[['execution_id', 'execution_name', 'start_time']])

Implementation reference: This method traverses the hierarchy from pipeline to stage to execution, transforming each protobuf into a DataFrame row via _transform_to_dataframe.

Retrieving Artifact Information

Access detailed metadata for individual artifacts and their relationships to executions.

Fetch Single Artifact Metadata

The get_artifact() method retrieves comprehensive metadata including URI, type, timestamps, and custom properties:

artifact_name = "pipeline1.stage_preprocess.output_artifact_12345"
artifact_df = query.get_artifact(artifact_name)
print(artifact_df[['id', 'type', 'uri', 'create_time']])

Implementation reference: The private _get_artifact helper locates the protobuf object, while get_artifact_df constructs the row representation.

List Artifacts by Execution

Retrieve all input and output artifacts associated with a specific execution ID:

execution_id = 42
artifacts_df = query.get_all_artifacts_for_execution(execution_id)
print(artifacts_df[['artifact_name', 'type', 'event_type']])  # event_type shows INPUT or OUTPUT

Implementation reference: This method queries events linked to the execution ID, distinguishes between input and output types, and builds rows using get_artifact_df.

Tracing Execution Lineage

Trace the provenance of artifacts by querying which executions consumed or produced specific data objects.

The get_all_executions_for_artifact() method walks the event graph to build a lineage table:

artifact_name = "pipeline1.stage_train.model"
lineage_df = query.get_all_executions_for_artifact(artifact_name)
print(lineage_df[['execution_id', 'execution_name', 'stage', 'pipeline', 'Type']])

Implementation reference: This function traverses from artifact events to parent executions, then resolves the associated stage and pipeline contexts to provide complete lineage context at line 76 in cmfquery.py.

Server-Side Deployment with FastAPI

The same CmfQuery methods power the CMF web UI through FastAPI endpoints. Instantiate CmfQuery within route handlers to expose metadata via HTTP:

from fastapi import FastAPI
from cmflib.cmfquery import CmfQuery

app = FastAPI()
cmf = CmfQuery(is_server=True)  # PostgreSQL backend

@app.get("/pipelines")
def list_pipelines():
    return cmf.get_pipeline_names()

@app.get("/lineage/artifact/{artifact_name}")
def get_artifact_lineage(artifact_name: str):
    return cmf.get_all_executions_for_artifact(artifact_name).to_dict('records')

Implementation reference: The server endpoints in server/app/query_artifact_lineage_d3tree.py and server/app/query_execution_lineage_d3tree.py follow this pattern, wrapping CmfQuery results in JSON responses.

Summary

• CmfQuery in cmflib/cmfquery.py serves as the primary Python interface for querying CMF-tracked metadata stored in MLMD.
• Storage flexibility: Supports both SQLite files (filepath parameter) and PostgreSQL servers (is_server=True).
• Pandas-native: All query methods return pandas DataFrames via the _transform_to_dataframe helper, enabling immediate data manipulation.
• Key methods: get_pipeline_names(), get_all_executions_in_stage(), get_artifact(), get_all_artifacts_for_execution(), and get_all_executions_for_artifact() cover the core query patterns.
• Production-ready: The same API powers both interactive notebooks and FastAPI server deployments as seen in server/app/query_*.py.

Frequently Asked Questions

What storage backends does CmfQuery support?

CmfQuery supports SQLite for local file-based storage and PostgreSQL for server deployments. The backend is selected via the is_server boolean parameter during initialization, which determines whether SqlliteStore or PostgresStore is instantiated in cmflib/cmfquery.py.

How do I filter executions by specific custom properties?

Filter the pandas DataFrame returned by get_all_executions_in_stage() using standard pandas boolean indexing. The DataFrame includes columns for both properties and custom_properties (flattened by _transform_to_dataframe), allowing you to filter like: df[df['custom_properties.batch_size'] == '128'].

Can I use CmfQuery in a Jupyter notebook for interactive analysis?

Yes. Import CmfQuery from cmflib.cmfquery, instantiate it with filepath="mlmd" pointing to your local metadata file, and call any query method. The pandas DataFrame output integrates directly with Jupyter's display system and visualization libraries like matplotlib or plotly.

What is the performance impact of querying large MLMD databases?

Query performance depends on the storage backend and database size. SQLite performs well for local development with thousands of executions, while PostgreSQL (via is_server=True) is recommended for production workloads. The CmfQuery methods use efficient MLMD store lookups, but lineage queries involving get_all_executions_for_artifact may traverse many event records; consider caching results for frequently accessed artifacts.

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