Query pipeline run metadata from database

SELECT
    *
FROM pipeline_run
WHERE pipeline_uuid = 'example_pipeline' -- Filter by pipeline_uuid
AND status = 'COMPLETED' -- Filter by status

from mage_ai.orchestration.db import db_connection
from mage_ai.orchestration.db.models.schedules import PipelineRun, BlockRun

db_connection.start_session()

# Fetch running pipeline runs of a pipeline
pipeline_runs = PipelineRun.query.filter(
    # Filter by pipeline uuid
    PipelineRun.pipeline_uuid == 'example_pipeline',
    # Filter by status
    PipelineRun.status == PipelineRun.PipelineRunStatus.RUNNING,
    # Filter by execution_date
    PipelineRun.execution_date >= '2024-01-01',
    PipelineRun.execution_date < '2024-02-01',
)

# Get the block runs of a pipeline run
block_runs = pipeline_runs.all()[0].block_runs
print(f'Block runs: {[b.to_dict() for b in block_runs]}')

# Get pipeline run count
print(f'Number of pipeline runs {pipeline_runs.count()}')

# Batch update pipeline run status
PipelineRun.batch_update_status([p.id for p in pipeline_runs], PipelineRun.PipelineRunStatus.CANCELLED)