Data Platform Health

# Data Platform Health — ingestion pipeline
import asyncio
from neo4j import AsyncGraphDatabase
from pydantic import BaseModel

class ModelRecord(BaseModel):
    entity_id: str
    structured_score: float
    metadata: dict[str, str]

async def run_pipeline(
    records: list[ModelRecord],
    uri: str,
    auth: tuple[str, str],
) -> None:
    driver = AsyncGraphDatabase.driver(uri, auth=auth)
    async with driver.session(database="neo4j") as session:
        await session.execute_write(
            _merge_entities,
            [r.model_dump() for r in records],
            technique="ML",
        )
    await driver.close()

async def _merge_entities(tx, batch, technique):
    await tx.run("""
        UNWIND $batch AS row
        MERGE (e:Entity {id: row.entity_id})
          ON CREATE SET e.created   = datetime(),
                        e.technique = $technique
          ON MATCH  SET e.updated   = datetime(),
                        e.score     = row.structured_score
        """, batch=batch, technique=technique)