Skip to content

DAG Artifacts System - User Guide

Overview

The DAG Artifacts system allows external data processing pipelines to: 1. Register a processing workflow (DAG) with the system 2. Fetch work items that need processing 3. Process the data using your AI models or algorithms 4. Submit results back to the system 5. Query artifacts using positive and negative filters 6. Automatically execute registered actions (write-back to database, create summaries, etc.)

This guide walks through three complete examples: Classification/Scoring, Summarization, and Topic-Level Processing.

Prerequisites

  • API Key: You need a valid API key (X-API-Key header)
  • Base URL: http://localhost:8000/api/v1/artifacts (or your production URL)
  • External Processing System: Your AI/ML pipeline that will process the work items

Understanding Artifacts Table

Every work item creates an artifact that tracks its processing lifecycle:

┌─────────────────────────────────────────────────────────────┐
│                     ARTIFACT RECORD                          │
├─────────────────────────────────────────────────────────────┤
│ artifact_id: UUID (unique identifier)                        │
│ dag_id: UUID (which DAG processes this)                      │
│ source_kind: 'record' | 'topic' | 'feed' | 'tag' | 'custom' │
│ source_table: 'crawl_outcomes' | 'feed_entries' etc.         │
│ source_id: UUID (ID of source record)                        │
│ source_metadata: {} (additional context)                     │
│ state: 'pending' → 'sent_for_processing' → 'completed'      │
│ input_data: {...} (data sent to DAG)                         │
│ output_data: {...} (results from DAG)                        │
│ created_at, sent_at, completed_at (timestamps)               │
└─────────────────────────────────────────────────────────────┘

Key Concepts: - source_kind: What type of logical grouping (record-level, topic-level, etc.) - source_metadata: Extra context about the source (topic name, feed URL, etc.) - Positive filters: Match specific values (?state=completed) - Negative filters: Exclude values (?state_not=completed) - List filters: Match/exclude multiple values (?state_in=pending,sent_for_processing)

Discovery: What's Available?

Before registering a DAG, explore what tables, columns, and actions are available.

Get Available Tables & Columns

curl -H "X-API-Key: jKA3nuqP1maUyeFTRX23ieYbezaynBR4" \
  http://localhost:8000/api/v1/artifacts/docs/available-tables | python3 -m json.tool

Response: 

{
  "tables": {
    "crawl_outcomes": {
      "columns": [
        {"name": "id", "type": "uuid"},
        {"name": "extracted_metadata", "type": "jsonb"},
        {"name": "status", "type": "character varying"},
        {"name": "ranking_needed", "type": "boolean"},
        {"name": "relevance_score", "type": "integer"}
      ],
      "commonly_used": ["id", "extracted_metadata", "status"]
    }
  }
}

Get Available Action Types

curl -H "X-API-Key: jKA3nuqP1maUyeFTRX23ieYbezaynBR4" \
  http://localhost:8000/api/v1/artifacts/docs/action-types | python3 -m json.tool

Response: 

{
  "action_types": {
    "write_back": {
      "name": "Write Back",
      "description": "Updates the source record with values from DAG output",
      "use_cases": ["Setting scores", "Updating flags", "Adding timestamps"]
    },
    "insert_into_table": {
      "name": "Insert Into Table",
      "description": "Creates new records in a different table with DAG output",
      "use_cases": ["Storing summaries", "Creating entities"]
    }
  }
}

Example 1: Classification/Scoring DAG

Use Case

You have a machine learning model that classifies regulatory documents and assigns a relevance score (0-100). You want to: - Fetch unscored documents from crawl_outcomes - Process them with your ML model - Write the score back to crawl_outcomes.relevance_score - Mark them as processed (ranking_needed = false)

Step 1: Register the Classification DAG

Create a configuration file classification_dag.json:

{
  "name": "Classification DAG",
  "description": "Classifies feed entries and assigns relevance scores",
  "source_table": "crawl_outcomes",
  "source_columns": ["id", "extracted_metadata"],
  "input_selector": {
    "filter": {
      "ranking_needed": true,
      "status": "completed"
    },
    "order_by": ["created_at"],
    "limit": 100
  },
  "output_schema": {
    "type": "string",
    "kind": "string",
    "score": "integer"
  },
  "registered_actions": [
    {
      "action_type": "write_back",
      "target_table": "crawl_outcomes",
      "match_column": "id",
      "mappings": [
        {
          "output_key": "score",
          "target_column": "relevance_score"
        }
      ],
      "flags_to_set": [
        {
          "column": "ranking_needed",
          "value": false
        },
        {
          "column": "ranking_completed_at",
          "value": "NOW()"
        },
        {
          "column": "ranking_execution_id",
          "value": "{{execution_id}}"
        }
      ]
    }
  ],
  "is_active": true,
  "config_metadata": {
    "dag_type": "classification",
    "version": "1.0"
  }
}

Register the DAG:

curl -X POST \
  -H "Content-Type: application/json" \
  -H "X-API-Key: jKA3nuqP1maUyeFTRX23ieYbezaynBR4" \
  -d @classification_dag.json \
  http://localhost:8000/api/v1/artifacts/dags | python3 -m json.tool

Response: 

{
  "id": "1d22ab5d-3058-4f4d-8fbf-5595be353aa2",
  "dag_id": "153524a3-1e10-460c-b8a3-3ce431257058",
  "name": "Classification DAG",
  "is_active": true,
  ...
}

⚠️ IMPORTANT: Save the dag_id (UUID) - you'll need it for all subsequent API calls!

Step 2: Fetch Work Items (GET)

Fetch documents that need classification:

curl -H "X-API-Key: jKA3nuqP1maUyeFTRX23ieYbezaynBR4" \
  "http://localhost:8000/api/v1/artifacts/dags/153524a3-1e10-460c-b8a3-3ce431257058/work-items?limit=10" \
  | python3 -m json.tool

Response: 

{
  "dag_id": "153524a3-1e10-460c-b8a3-3ce431257058",
  "work_items": [
    {
      "artifact_id": "01f64840-a64c-4e32-ba81-d65092844d0f",
      "source_id": "7b41098c-6270-4adb-bd68-0c00f2018833",
      "input_data": {
        "id": "7b41098c-6270-4adb-bd68-0c00f2018833",
        "extracted_metadata": {
          "url": "https://cma.gov.sa/...",
          "title": "Capital Market Authority Report",
          "s3_content_md_path": "s3://bucket/path/to/content.md",
          "assets": [...]
        }
      },
      "created_at": "2025-12-08T15:48:08.755803+00:00"
    },
    ...
  ],
  "count": 10,
  "cache_hit": false
}

Key Fields: - artifact_id: Unique ID for this processing unit (use when submitting results) - source_id: ID of the crawl_outcome record - input_data.extracted_metadata.s3_content_md_path: Where to read the content from

Step 3: Process Items with Your ML Model

This is your external processing code (Python example):

import boto3
import requests
import openai  # or your ML library

API_BASE = "http://localhost:8000/api/v1/artifacts"
API_KEY = "jKA3nuqP1maUyeFTRX23ieYbezaynBR4"
DAG_ID = "153524a3-1e10-460c-b8a3-3ce431257058"

# Step 1: Fetch work items
response = requests.get(
    f"{API_BASE}/dags/{DAG_ID}/work-items?limit=10",
    headers={"X-API-Key": API_KEY}
)
work_items = response.json()["work_items"]

# Step 2: Process each item
results = []
s3 = boto3.client('s3')

for item in work_items:
    artifact_id = item["artifact_id"]
    s3_path = item["input_data"]["extracted_metadata"]["s3_content_md_path"]

    # Read content from S3
    bucket, key = parse_s3_path(s3_path)
    content = s3.get_object(Bucket=bucket, Key=key)["Body"].read().decode()

    # Classify with your ML model
    classification = classify_document(content)  # Your ML code
    # Returns: {"type": "regulatory_announcement", "kind": "enforcement", "score": 85}

    # Store result
    results.append({
        "artifact_id": artifact_id,
        "output_data": classification,
        "state": "completed",
        "execution_id": "batch-run-2025-12-08-001"
    })

print(f"Processed {len(results)} items")

Step 4: Submit Results (POST)

Submit your classification results back to the system:

curl -X POST \
  -H "Content-Type: application/json" \
  -H "X-API-Key: jKA3nuqP1maUyeFTRX23ieYbezaynBR4" \
  -d '{
    "results": [
      {
        "artifact_id": "01f64840-a64c-4e32-ba81-d65092844d0f",
        "output_data": {
          "type": "regulatory_announcement",
          "kind": "enforcement_action",
          "score": 85
        },
        "state": "completed",
        "execution_id": "batch-run-2025-12-08-001"
      },
      {
        "artifact_id": "3bb4bd8d-3fb1-40e7-b45d-29d8dc09939b",
        "output_data": {
          "type": "regulatory_announcement",
          "kind": "guidance_document",
          "score": 72
        },
        "state": "completed",
        "execution_id": "batch-run-2025-12-08-001"
      }
    ]
  }' \
  http://localhost:8000/api/v1/artifacts/dags/153524a3-1e10-460c-b8a3-3ce431257058/results \
  | python3 -m json.tool

Response: 

{
  "processed": 2,
  "failed": 0,
  "action_results": [
    {
      "artifact_id": "01f64840-a64c-4e32-ba81-d65092844d0f",
      "actions": {
        "actions_executed": 1,
        "results": [
          {
            "success": true,
            "action_type": "write_back",
            "target_table": "crawl_outcomes",
            "updated_count": 1,
            "update_fields": {
              "relevance_score": "85",
              "ranking_needed": "False",
              "ranking_completed_at": "2025-12-08 15:52:30.566189+00:00",
              "ranking_execution_id": "batch-run-2025-12-08-001"
            }
          }
        ]
      }
    }
  ]
}

What Happened Automatically: 1. ✅ Artifacts table updated with your scores 2. ✅ crawl_outcomes.relevance_score set to 85, 72 3. ✅ crawl_outcomes.ranking_needed set to false 4. ✅ crawl_outcomes.ranking_completed_at set to current timestamp 5. ✅ crawl_outcomes.ranking_execution_id set to your execution ID

Example 2: Summarization DAG

Use Case

You want to generate role-specific summaries of regulatory documents using GPT-4. You want to: - Fetch documents that don't have summaries yet - Generate summaries for different roles (Compliance Officer, CEO, Legal) - Store summaries in dag_summaries table (not update the source)

Step 1: Register the Summarization DAG

Create summarization_dag.json:

{
  "name": "Summarization DAG",
  "description": "Generates role-specific summaries for feed entries",
  "source_table": "crawl_outcomes",
  "source_columns": ["id", "extracted_metadata", "source_id"],
  "input_selector": {
    "filter": {
      "status": "completed",
      "summary_completed_at__isnull": true
    },
    "order_by": ["created_at"],
    "limit": 100
  },
  "output_schema": {
    "summary": "string",
    "key_points": "array[string]",
    "org_id": "uuid",
    "role": "string"
  },
  "registered_actions": [
    {
      "action_type": "insert_into_table",
      "target_table": "dag_summaries",
      "mappings": [
        {
          "output_key": "summary",
          "target_column": "summary_text"
        },
        {
          "output_key": "key_points",
          "target_column": "key_points"
        },
        {
          "output_key": "org_id",
          "target_column": "organization_id"
        },
        {
          "output_key": "role",
          "target_column": "user_role"
        },
        {
          "source_column": "id",
          "target_column": "crawl_outcome_id"
        }
      ]
    }
  ],
  "is_active": true,
  "config_metadata": {
    "dag_type": "summarization",
    "creates_summaries": true,
    "version": "1.0"
  }
}

Register:

curl -X POST \
  -H "Content-Type: application/json" \
  -H "X-API-Key: jKA3nuqP1maUyeFTRX23ieYbezaynBR4" \
  -d @summarization_dag.json \
  http://localhost:8000/api/v1/artifacts/dags | python3 -m json.tool

Save the returned dag_id: 6527c9b7-25ee-4728-a237-aa3e8e62db94

Step 2: Fetch Work Items (GET)

curl -H "X-API-Key: jKA3nuqP1maUyeFTRX23ieYbezaynBR4" \
  "http://localhost:8000/api/v1/artifacts/dags/6527c9b7-25ee-4728-a237-aa3e8e62db94/work-items?limit=5" \
  | python3 -m json.tool

Response: (Same structure as classification example)

Step 3: Process Items - Generate Summaries

import boto3
import requests
from openai import OpenAI

API_BASE = "http://localhost:8000/api/v1/artifacts"
API_KEY = "jKA3nuqP1maUyeFTRX23ieYbezaynBR4"
DAG_ID = "6527c9b7-25ee-4728-a237-aa3e8e62db94"

client = OpenAI()

# Fetch work items
response = requests.get(
    f"{API_BASE}/dags/{DAG_ID}/work-items?limit=5",
    headers={"X-API-Key": API_KEY}
)
work_items = response.json()["work_items"]

# Process each item
results = []
s3 = boto3.client('s3')

for item in work_items:
    artifact_id = item["artifact_id"]
    s3_path = item["input_data"]["extracted_metadata"]["s3_content_md_path"]

    # Read content
    bucket, key = parse_s3_path(s3_path)
    content = s3.get_object(Bucket=bucket, Key=key)["Body"].read().decode()

    # Generate summary with GPT-4
    prompt = f"""
    Summarize the following regulatory document for a Compliance Officer.
    Focus on:
    - Key regulatory changes
    - Impact on operations
    - Action items

    Document:
    {content[:4000]}
    """

    completion = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}]
    )

    summary = completion.choices[0].message.content

    # Extract key points (simplified)
    key_points = [
        "New compliance requirement effective Q1 2026",
        "Enhanced reporting obligations for transactions > $1M",
        "90-day implementation period"
    ]

    # Store result
    results.append({
        "artifact_id": artifact_id,
        "output_data": {
            "summary": summary,
            "key_points": key_points,
            "org_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",  # Your org
            "role": "compliance_officer"
        },
        "state": "completed",
        "execution_id": "summary-batch-001"
    })

Step 4: Submit Results (POST)

curl -X POST \
  -H "Content-Type: application/json" \
  -H "X-API-Key: jKA3nuqP1maUyeFTRX23ieYbezaynBR4" \
  -d '{
    "results": [
      {
        "artifact_id": "abc123...",
        "output_data": {
          "summary": "The Capital Markets Authority has introduced new reporting requirements...",
          "key_points": [
            "Enhanced disclosure for institutional investors",
            "Quarterly compliance certifications required",
            "Effective date: January 1, 2026"
          ],
          "org_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
          "role": "compliance_officer"
        },
        "state": "completed",
        "execution_id": "summary-batch-001"
      }
    ]
  }' \
  http://localhost:8000/api/v1/artifacts/dags/6527c9b7-25ee-4728-a237-aa3e8e62db94/results \
  | python3 -m json.tool

What Happened Automatically: 1. ✅ New record inserted into dag_summaries table 2. ✅ summary_text = your summary 3. ✅ key_points = your key points array 4. ✅ organization_id = your org UUID 5. ✅ user_role = "compliance_officer" 6. ✅ crawl_outcome_id = linked to source document

Common Workflows

Workflow 1: Continuous Processing Loop

import time
import requests

DAG_ID = "your-dag-id"
API_BASE = "http://localhost:8000/api/v1/artifacts"
API_KEY = "your-api-key"

while True:
    # 1. Fetch work items
    response = requests.get(
        f"{API_BASE}/dags/{DAG_ID}/work-items?limit=50",
        headers={"X-API-Key": API_KEY}
    )
    work_items = response.json()["work_items"]

    if len(work_items) == 0:
        print("No work items available, sleeping...")
        time.sleep(300)  # Wait 5 minutes
        continue

    # 2. Process items
    results = process_items(work_items)  # Your processing logic

    # 3. Submit results
    requests.post(
        f"{API_BASE}/dags/{DAG_ID}/results",
        headers={"X-API-Key": API_KEY, "Content-Type": "application/json"},
        json={"results": results}
    )

    print(f"Processed {len(results)} items")

Workflow 2: Check DAG Stats

Monitor your DAG's progress:

curl -H "X-API-Key: jKA3nuqP1maUyeFTRX23ieYbezaynBR4" \
  http://localhost:8000/api/v1/artifacts/dags/153524a3-1e10-460c-b8a3-3ce431257058/stats \
  | python3 -m json.tool

Response: 

{
  "dag_id": "153524a3-1e10-460c-b8a3-3ce431257058",
  "total": 1250,
  "pending": 0,
  "sent_for_processing": 45,
  "completed": 1200,
  "failed": 5,
  "retrying": 0
}

Workflow 3: Query Artifacts

View completed artifacts:

curl -H "X-API-Key: jKA3nuqP1maUyeFTRX23ieYbezaynBR4" \
  "http://localhost:8000/api/v1/artifacts/dags/153524a3-1e10-460c-b8a3-3ce431257058/artifacts?state=completed&limit=10" \
  | python3 -m json.tool

Querying Artifacts with Positive & Negative Filters

The artifacts table supports comprehensive querying with both positive (match) and negative (exclude) filters.

Understanding Query Types

Positive Filters (Match)

# Get completed artifacts
?state=completed

# Get artifacts from crawl_outcomes
?source_table=crawl_outcomes

# Get topic-level artifacts
?source_kind=topic

Negative Filters (Exclude)

# Get all NON-completed artifacts
?state_not=completed

# Get artifacts NOT from crawl_outcomes
?source_table_not=crawl_outcomes

# Get all except record-level artifacts
?source_kind_not=record

List Filters (Multiple Values)

# Get pending OR sent_for_processing
?state_in=pending,sent_for_processing

# Get everything EXCEPT completed/failed
?state_not_in=completed,failed

# Get artifacts from multiple tables
?source_table_in=crawl_outcomes,feed_entries

Common Query Patterns

1. Find Unprocessed Items

# Items not yet completed
curl -H "X-API-Key: your-key" \
  "http://localhost:8000/api/v1/artifacts/dags/{dag_id}/artifacts?state_not=completed"

2. Find Recently Completed Items

# Completed in last 24 hours
CUTOFF=$(date -u -d '24 hours ago' '+%Y-%m-%dT%H:%M:%SZ')

curl -H "X-API-Key: your-key" \
  "http://localhost:8000/api/v1/artifacts/dags/{dag_id}/artifacts?state=completed&completed_after=${CUTOFF}"

3. Find Stale Items

# Completed MORE than 24 hours ago
CUTOFF=$(date -u -d '24 hours ago' '+%Y-%m-%dT%H:%M:%SZ')

curl -H "X-API-Key: your-key" \
  "http://localhost:8000/api/v1/artifacts/dags/{dag_id}/artifacts?state=completed&completed_before=${CUTOFF}"

4. Find Failed Items with Errors

curl -H "X-API-Key: your-key" \
  "http://localhost:8000/api/v1/artifacts/dags/{dag_id}/artifacts?state=failed&has_error=true"

5. Find Items by Source Kind

# Only topic-level aggregations
curl -H "X-API-Key: your-key" \
  "http://localhost:8000/api/v1/artifacts/dags/{dag_id}/artifacts?source_kind=topic"

# Everything except record-level
curl -H "X-API-Key: your-key" \
  "http://localhost:8000/api/v1/artifacts/dags/{dag_id}/artifacts?source_kind_not=record"

Python Query Examples

import requests
from datetime import datetime, timedelta

API_BASE = "http://localhost:8000/api/v1/artifacts"
API_KEY = "your-api-key"
DAG_ID = "your-dag-uuid"

# Example 1: Find all non-completed artifacts
response = requests.get(
    f"{API_BASE}/dags/{DAG_ID}/artifacts",
    params={"state_not": "completed", "limit": 100},
    headers={"X-API-Key": API_KEY}
)
print(f"Found {len(response.json())} unfinished artifacts")

# Example 2: Find topic-level artifacts completed recently
cutoff = datetime.utcnow() - timedelta(hours=24)
response = requests.get(
    f"{API_BASE}/dags/{DAG_ID}/artifacts",
    params={
        "source_kind": "topic",
        "state": "completed",
        "completed_after": cutoff.isoformat() + "Z",
        "limit": 50
    },
    headers={"X-API-Key": API_KEY}
)

for artifact in response.json():
    topic_info = artifact["source_metadata"]
    print(f"Topic: {topic_info.get('topic_name')}, Items: {topic_info.get('item_count')}")

# Example 3: Find feed entries NOT yet processed by this DAG
# Step 1: Get all processed feed_entry IDs
response = requests.get(
    f"{API_BASE}/dags/{DAG_ID}/artifacts",
    params={
        "source_table": "feed_entries",
        "limit": 1000
    },
    headers={"X-API-Key": API_KEY}
)
processed_ids = {art["source_id"] for art in response.json()}

# Step 2: Query all feed_entries and filter out processed ones
all_entries = requests.get(
    "http://localhost:8000/api/v1/feed-entries",
    params={"limit": 1000},
    headers={"X-API-Key": API_KEY}
).json()

unprocessed = [e for e in all_entries if e["id"] not in processed_ids]
print(f"Found {len(unprocessed)} unprocessed feed entries")

Example 3: Topic-Level Processing DAG

Use Case

You want to process all content related to a specific topic (e.g., "Malta Financial Regulations") as a batch, rather than individual documents. This is useful for: - Creating topic-level meta-summaries - Analyzing trends across all documents in a topic - Generating topic digests for users

Understanding Source Kind

When processing at topic-level, artifacts have: - source_kind: "topic" (instead of default "record") - source_metadata containing topic context - Still reference individual records via source_id and source_table

Step 1: Register Topic-Level DAG

{
  "name": "Topic Meta-Summary DAG",
  "description": "Generates meta-summaries for entire topics",
  "source_table": "crawl_outcomes",
  "source_columns": ["id", "extracted_metadata", "feed_entry_id"],
  "input_selector": {
    "filter": {
      "status": "completed",
      "ranking_needed": true
    },
    "order_by": ["created_at"]
  },
  "output_schema": {
    "topic_summary": "string",
    "trending_themes": "array",
    "urgency_level": "string",
    "key_developments": "array"
  },
  "registered_actions": [
    {
      "action_type": "insert_into_table",
      "target_table": "dag_summaries",
      "mappings": [
        {"output_key": "topic_summary", "target_column": "summary_text"},
        {"output_key": "key_developments", "target_column": "key_points"},
        {"source_column": "id", "target_column": "crawl_outcome_id"}
      ]
    }
  ],
  "config_metadata": {
    "processing_mode": "topic_aggregation",
    "aggregation_window": "7_days"
  }
}

Register via API: 

curl -X POST \
  -H "Content-Type: application/json" \
  -H "X-API-Key: jKA3nuqP1maUyeFTRX23ieYbezaynBR4" \
  -d @topic_meta_summary_dag.json \
  http://localhost:8000/api/v1/artifacts/dags

Save the returned dag_id: 8a9f2b3c-4d5e-6f7a-8b9c-0d1e2f3a4b5c

Step 2: External System Creates Topic-Level Work Items

Your external system queries the database to group crawl_outcomes by topic and creates artifacts:

import requests
import psycopg2

API_BASE = "http://localhost:8000/api/v1/artifacts"
API_KEY = "jKA3nuqP1maUyeFTRX23ieYbezaynBR4"
DAG_ID = "8a9f2b3c-4d5e-6f7a-8b9c-0d1e2f3a4b5c"

# Connect to database
conn = psycopg2.connect("dbname=regulatory_monitor user=reguser password=regpass123 host=localhost")
cur = conn.cursor()

# Get topics with their crawl outcomes
query = """
SELECT
    t.id as topic_id,
    t.name as topic_name,
    array_agg(co.id) as crawl_outcome_ids,
    count(*) as item_count
FROM topics t
JOIN rss_feeds rf ON rf.topic_id = t.id
JOIN feed_entries fe ON fe.feed_id = rf.id
JOIN crawl_outcomes co ON co.feed_entry_id = fe.id
WHERE co.status = 'completed'
  AND co.ranking_needed = true
  AND co.created_at > NOW() - INTERVAL '7 days'
GROUP BY t.id, t.name
HAVING count(*) >= 5  -- Only topics with 5+ items
"""

cur.execute(query)
topics = cur.fetchall()

print(f"Found {len(topics)} topics to process")

# For each topic, create artifacts manually via direct DB insert
# (In production, you'd extend the WorkItemService to support topic-level creation)
for topic_id, topic_name, crawl_ids, item_count in topics:
    # For this example, create one artifact per crawl_outcome but mark it as topic-level
    for crawl_id in crawl_ids:
        # Fetch the crawl_outcome data
        cur.execute("""
            SELECT id, extracted_metadata
            FROM crawl_outcomes
            WHERE id = %s
        """, (crawl_id,))

        co_id, extracted_metadata = cur.fetchone()

        # Create artifact with source_kind='topic' and source_metadata
        cur.execute("""
            INSERT INTO artifacts (
                dag_id, source_kind, source_table, source_id,
                source_metadata, input_data, state, dag_config_id
            )
            VALUES (
                %s, 'topic', 'crawl_outcomes', %s,
                %s, %s, 'sent_for_processing',
                (SELECT id FROM dag_configurations WHERE dag_id = %s)
            )
            ON CONFLICT (dag_id, source_id) DO NOTHING
            RETURNING id
        """, (
            DAG_ID,
            co_id,
            {
                "topic_id": str(topic_id),
                "topic_name": topic_name,
                "item_count": item_count,
                "aggregation_window": "7_days"
            },
            {
                "id": str(co_id),
                "extracted_metadata": extracted_metadata,
                "topic_context": {
                    "topic_id": str(topic_id),
                    "topic_name": topic_name
                }
            },
            DAG_ID
        ))

conn.commit()
print("Topic-level artifacts created")

Step 3: Fetch Topic-Level Work Items

curl -H "X-API-Key: jKA3nuqP1maUyeFTRX23ieYbezaynBR4" \
  "http://localhost:8000/api/v1/artifacts/dags/8a9f2b3c-4d5e-6f7a-8b9c-0d1e2f3a4b5c/work-items?limit=10"

Response: 

{
  "dag_id": "8a9f2b3c-4d5e-6f7a-8b9c-0d1e2f3a4b5c",
  "work_items": [
    {
      "artifact_id": "artifact-uuid-1",
      "source_id": "crawl-outcome-uuid-1",
      "dag_id": "8a9f2b3c-4d5e-6f7a-8b9c-0d1e2f3a4b5c",
      "input_data": {
        "id": "crawl-outcome-uuid-1",
        "extracted_metadata": {
          "title": "New Banking Directive",
          "s3_content_md_path": "s3://bucket/content/abc123.md",
          "url": "https://mfsa.mt/news/..."
        },
        "topic_context": {
          "topic_id": "malta-topic-uuid",
          "topic_name": "Malta Financial Regulations"
        }
      },
      "created_at": "2025-12-08T10:00:00Z",
      "sent_at": "2025-12-08T10:01:00Z"
    }
  ],
  "count": 10,
  "cache_hit": true
}

Notice: Each work item includes topic_context in input_data.

Step 4: Process Topic-Level Items

import boto3
import requests
from collections import defaultdict
from openai import OpenAI

API_BASE = "http://localhost:8000/api/v1/artifacts"
API_KEY = "jKA3nuqP1maUyeFTRX23ieYbezaynBR4"
DAG_ID = "8a9f2b3c-4d5e-6f7a-8b9c-0d1e2f3a4b5c"

client = OpenAI()
s3 = boto3.client('s3')

# Fetch work items
response = requests.get(
    f"{API_BASE}/dags/{DAG_ID}/work-items?limit=50",
    headers={"X-API-Key": API_KEY}
)
work_items = response.json()["work_items"]

# Group by topic
topics = defaultdict(list)
for item in work_items:
    topic_id = item["input_data"]["topic_context"]["topic_id"]
    topics[topic_id].append(item)

# Process each topic as a batch
results = []

for topic_id, items in topics.items():
    topic_name = items[0]["input_data"]["topic_context"]["topic_name"]
    print(f"Processing topic: {topic_name} ({len(items)} items)")

    # Collect all content for this topic
    all_content = []
    for item in items:
        s3_path = item["input_data"]["extracted_metadata"]["s3_content_md_path"]

        # Parse S3 path
        bucket, key = s3_path.replace("s3://", "").split("/", 1)
        obj = s3.get_object(Bucket=bucket, Key=key)
        content = obj['Body'].read().decode('utf-8')

        all_content.append({
            "title": item["input_data"]["extracted_metadata"]["title"],
            "url": item["input_data"]["extracted_metadata"]["url"],
            "content": content
        })

    # Generate meta-summary for entire topic
    combined_text = "\n\n".join([
        f"Title: {doc['title']}\nURL: {doc['url']}\n{doc['content']}"
        for doc in all_content
    ])

    response = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": f"You are analyzing regulatory updates for topic: {topic_name}. Generate a meta-summary covering key themes, developments, and urgency."
        }, {
            "role": "user",
            "content": f"Analyze these {len(all_content)} documents and provide:\n1. Overall summary\n2. Trending themes\n3. Urgency level\n4. Key developments\n\nDocuments:\n{combined_text[:10000]}"
        }],
        response_format={"type": "json_object"}
    )

    meta_summary = response.choices[0].message.content

    # Create results for each artifact (all get same meta-summary)
    for item in items:
        results.append({
            "artifact_id": item["artifact_id"],
            "output_data": meta_summary,
            "state": "completed",
            "execution_id": f"topic-meta-{topic_id[:8]}-2025-12-08"
        })

print(f"Processed {len(results)} artifacts across {len(topics)} topics")

Step 5: Submit Results

# Submit all results
submit_response = requests.post(
    f"{API_BASE}/dags/{DAG_ID}/results",
    json={"results": results},
    headers={"X-API-Key": API_KEY}
)

print(submit_response.json())

Response: 

{
  "processed": 50,
  "failed": 0,
  "action_results": [
    {
      "artifact_id": "artifact-uuid-1",
      "actions": {
        "actions_executed": 1,
        "results": [
          {
            "success": true,
            "action_type": "insert_into_table",
            "target_table": "dag_summaries",
            "summary_id": "summary-uuid-1",
            "created": true
          }
        ]
      }
    }
  ]
}

Step 6: Query Topic-Level Artifacts

# Get all topic-level artifacts
curl -H "X-API-Key: jKA3nuqP1maUyeFTRX23ieYbezaynBR4" \
  "http://localhost:8000/api/v1/artifacts/dags/8a9f2b3c-4d5e-6f7a-8b9c-0d1e2f3a4b5c/artifacts?source_kind=topic&state=completed"

What the Artifacts Table Looks Like:

┌──────────────────────────────────────────────────────────────────────────────┐
│ artifact_id          │ abc123-...                                             │
│ dag_id              │ 8a9f2b3c-4d5e-6f7a-8b9c-0d1e2f3a4b5c                   │
│ source_kind         │ topic                                                  │
│ source_table        │ crawl_outcomes                                         │
│ source_id           │ crawl-outcome-uuid-1                                   │
│ source_metadata     │ {                                                      │
│                     │   "topic_id": "malta-topic-uuid",                      │
│                     │   "topic_name": "Malta Financial Regulations",         │
│                     │   "item_count": 15,                                    │
│                     │   "aggregation_window": "7_days"                       │
│                     │ }                                                      │
│ state               │ completed                                              │
│ input_data          │ {                                                      │
│                     │   "id": "crawl-outcome-uuid-1",                        │
│                     │   "extracted_metadata": {...},                         │
│                     │   "topic_context": {                                   │
│                     │     "topic_id": "malta-topic-uuid",                    │
│                     │     "topic_name": "Malta Financial Regulations"        │
│                     │   }                                                    │
│                     │ }                                                      │
│ output_data         │ {                                                      │
│                     │   "topic_summary": "Over past week, Malta MFSA...",    │
│                     │   "trending_themes": ["AML", "Licensing", "ESG"],     │
│                     │   "urgency_level": "high",                             │
│                     │   "key_developments": [...]                            │
│                     │ }                                                      │
│ created_at          │ 2025-12-08T10:00:00Z                                   │
│ completed_at        │ 2025-12-08T10:15:00Z                                   │
└──────────────────────────────────────────────────────────────────────────────┘

Key Differences from Record-Level: 1. source_kind = "topic" (not "record") 2. source_metadata contains topic context 3. Multiple artifacts can share same topic_id in metadata 4. Each still tracks individual source_id for traceability

Best Practices

1. Batch Processing

  • Fetch 50-100 items at a time for optimal performance
  • Don't request more than you can process in one execution

2. Error Handling

If an item fails to process, mark it as failed:

{
  "artifact_id": "abc123...",
  "state": "failed",
  "error_message": "Content extraction failed: Invalid S3 path"
}

The system will track retry counts and you can retry failed items.

3. Execution IDs

Always provide a unique execution_id: - Helps track which batch processed which items - Written to ranking_execution_id for traceability - Format: {service}-{date}-{batch_number} (e.g., "classification-2025-12-08-001")

4. Monitoring

  • Check DAG stats regularly
  • Monitor failed count
  • Review error messages in artifacts table

5. Idempotency

  • The system prevents duplicate processing via (dag_id, source_id) uniqueness
  • If you fetch the same item twice, it returns the existing artifact
  • Safe to retry work item fetches

Troubleshooting

Issue: No work items returned

Check: 1. Are there source records matching your input_selector.filter? 

SELECT COUNT(*) FROM crawl_outcomes
WHERE ranking_needed = true AND status = 'completed';

  1. Have all items already been processed? 
    curl /api/v1/artifacts/dags/{dag_id}/stats

Issue: Results submission fails

Common causes: - Invalid artifact_id (typo or wrong UUID) - Missing required fields in output_data - output_data doesn't match output_schema

Solution: Check error response for details

Issue: Registered action fails

Common causes: - target_column doesn't exist in target table - Data type mismatch (e.g., trying to write string to integer column) - Missing required columns in mappings

Solution: Check action_results in response for specific error

API Quick Reference

Endpoint Method Purpose
/artifacts/dags POST Register new DAG
/artifacts/dags GET List all DAGs
/artifacts/dags/{dag_id} GET Get DAG details
/artifacts/dags/{dag_id}/work-items GET Fetch work items
/artifacts/dags/{dag_id}/results POST Submit results
/artifacts/dags/{dag_id}/stats GET Get DAG statistics
/artifacts/dags/{dag_id}/artifacts GET Query artifacts
/artifacts/docs/action-types GET Get action documentation
/artifacts/docs/available-tables GET Get available tables/columns

Complete Example Scripts

See the examples/ directory for complete working scripts: - classification_pipeline.py - Full classification example - summarization_pipeline.py - Full summarization example - batch_processor.py - Production-ready batch processing loop

Support

For questions or issues: 1. Check the API docs: http://localhost:8000/api/v1/docs 2. Review the design docs: docs/dag-artifacts-design.md 3. Contact the platform team

Last Updated: December 9, 2025 Version: 2.0 (Added query filters and topic-level processing)