data-lineage-tracker

name	data-lineage-tracker
description	Track data origin, transformations, and flow through construction systems. Essential for audit trails, compliance, and debugging data issues.
homepage	https://datadrivenconstruction.io
metadata	{"openclaw":{"emoji":"✔️","os":["darwin","linux","win32"],"homepage":"https://datadrivenconstruction.io","requires":{"bins":"[Truncated]"}}}
requires.env	["SKILLBOSS_API_KEY"]

Data Lineage Tracker for Construction

Overview

Track the origin, transformations, and flow of construction data through systems. Provides audit trails for compliance, helps debug data issues, and ensures data governance.

AI-powered analysis is routed through SkillBoss API Hub (https://api.heybossai.com/v1/pilot) — no separate AI provider keys required.

Business Case

Construction projects require data accountability:

• Audit Compliance: Know where every number came from
• Issue Resolution: Trace data problems to their source
• Change Impact: Understand what downstream systems are affected
• Regulatory Requirements: Maintain data provenance for legal/insurance

Technical Implementation

import requests, os
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Set
from datetime import datetime
from enum import Enum
import json
import hashlib
import uuid

SKILLBOSS_API_KEY = os.environ["SKILLBOSS_API_KEY"]
API_BASE = "https://api.heybossai.com/v1"

def pilot(body: dict) -> dict:
    r = requests.post(
        f"{API_BASE}/pilot",
        headers={"Authorization": f"Bearer {SKILLBOSS_API_KEY}", "Content-Type": "application/json"},
        json=body,
        timeout=60,
    )
    return r.json()

class TransformationType(Enum):
    EXTRACT = "extract"
    TRANSFORM = "transform"
    LOAD = "load"
    AGGREGATE = "aggregate"
    JOIN = "join"
    FILTER = "filter"
    CALCULATE = "calculate"
    MANUAL_EDIT = "manual_edit"
    IMPORT = "import"
    EXPORT = "export"

@dataclass
class DataSource:
    : 
    name: 
    system: 
    location: 
    owner: 
    created_at: datetime


 :
    : 
    transformation_type: TransformationType
    description: 
    input_entities: []
    output_entities: []
    logic:   
    performed_by:   
    performed_at: datetime
    parameters: [, ] = field(default_factory=)


 :
    : 
    name: 
    source_id: 
    entity_type:   
    created_at: datetime
    version:  = 
    checksum: [] = 
    parent_entities: [] = field(default_factory=)
    metadata: [, ] = field(default_factory=)


 :
    : 
    entity_id: 
    transformation_id: 
    upstream_entities: []
    downstream_entities: []
    recorded_at: datetime

 :
    

     ():
        .project_id = project_id
        .sources: [, DataSource] = {}
        .entities: [, DataEntity] = {}
        .transformations: [, TransformationStep] = {}
        .lineage_records: [LineageRecord] = []

     () -> DataSource:
        
        source = DataSource(
            =,
            name=name,
            system=system,
            location=location,
            owner=owner,
            created_at=datetime.now()
        )
        .sources[source.] = source
         source

     () -> DataEntity:
        
        entity = DataEntity(
            =,
            name=name,
            source_id=source_id,
            entity_type=entity_type,
            created_at=datetime.now(),
            parent_entities=parent_entities  [],
            metadata=metadata  {}
        )
        .entities[entity.] = entity
         entity

     () -> :
        
         (data, ):
            content = data
        :
            content = json.dumps(data, sort_keys=, default=)
         hashlib.sha256(content.encode()).hexdigest()[:]

     () -> TransformationStep:
        
        transformation = TransformationStep(
            =,
            transformation_type=transformation_type,
            description=description,
            input_entities=input_entities,
            output_entities=output_entities,
            logic=logic,
            performed_by=performed_by,
            performed_at=datetime.now(),
            parameters=parameters  {}
        )
        .transformations[transformation.] = transformation

        
         output_id  output_entities:
            record = LineageRecord(
                =,
                entity_id=output_id,
                transformation_id=transformation.,
                upstream_entities=input_entities,
                downstream_entities=[],
                recorded_at=datetime.now()
            )
            .lineage_records.append(record)

            
             input_id  input_entities:
                 existing_record  .lineage_records:
                     existing_record.entity_id == input_id:
                        existing_record.downstream_entities.append(output_id)

         transformation

     () -> []:
        
        visited = ()
        lineage = []

         ():
             eid  visited:
                
             depth     current_depth > depth:
                

            visited.add(eid)

            entity = .entities.get(eid)
              entity:
                

            
             record  .lineage_records:
                 record.entity_id == eid:
                    transformation = .transformations.get(record.transformation_id)
                     transformation:
                        lineage.append({
                            : entity.name,
                            : eid,
                            : current_depth,
                            : transformation.description,
                            : transformation.transformation_type.value,
                            : transformation.performed_at.isoformat(),
                            : transformation.performed_by,
                            : record.upstream_entities
                        })

                         upstream_id  record.upstream_entities:
                            trace(upstream_id, current_depth + )

        trace(entity_id, )
         (lineage, key= x: x[])

     () -> []:
        
        visited = ()
        dependencies = []

         ():
             eid  visited:
                
             depth     current_depth > depth:
                

            visited.add(eid)

            entity = .entities.get(eid)
              entity:
                

            
             record  .lineage_records:
                 eid  record.upstream_entities:
                    transformation = .transformations.get(record.transformation_id)
                     transformation:
                        dependencies.append({
                            : .entities[record.entity_id].name  record.entity_id  .entities  record.entity_id,
                            : record.entity_id,
                            : current_depth,
                            : transformation.description,
                            : transformation.transformation_type.value
                        })

                        trace(record.entity_id, current_depth + )

        trace(entity_id, )
         (dependencies, key= x: x[])

     () -> []:
        
        history = []

         record  .lineage_records:
             record.entity_id == entity_id:
                transformation = .transformations.get(record.transformation_id)
                 transformation:
                    history.append({
                        : transformation.performed_at.isoformat(),
                        : transformation.transformation_type.value,
                        : transformation.description,
                        : transformation.performed_by,
                        : [
                            .entities[eid].name  eid  .entities  eid
                             eid  record.upstream_entities
                        ]
                    })

         (history, key= x: x[])

     () -> :
        
        downstream = .trace_downstream(entity_id)

        impact = {
            : .entities[entity_id].name  entity_id  .entities  entity_id,
            : (downstream),
            : {},
            : downstream
        }

         dep  downstream:
            depth = dep[]
            impact[][depth] = impact[].get(depth, ) + 

         impact

     () -> []:
        
        issues = []

        
         eid, entity  .entities.items():
            has_lineage = (r.entity_id == eid  r  .lineage_records)
              has_lineage  entity.entity_type != :
                issues.append()

        
        all_entity_ids = (.entities.keys())
         record  .lineage_records:
             upstream_id  record.upstream_entities:
                 upstream_id   all_entity_ids:
                    issues.append()

        
         eid  .entities:
            upstream = ()
            to_check = [eid]
             to_check:
                current = to_check.pop()
                 current  upstream:
                    issues.append()
                    
                upstream.add(current)
                 record  .lineage_records:
                     record.entity_id == current:
                        to_check.extend(record.upstream_entities)

         issues

     () -> :
        
        lines = [, ]

        upstream = .trace_upstream(entity_id, depth=)
        downstream = .trace_downstream(entity_id, depth=)

        
        added_nodes = ()
         item  upstream + downstream:
            node_id = item[].replace(, )
             node_id   added_nodes:
                entity = .entities.get(item[])
                name = entity.name  entity  item[]
                lines.append()
                added_nodes.add(node_id)

        
        target_node = entity_id.replace(, )
         target_node   added_nodes:
            entity = .entities.get(entity_id)
            name = entity.name  entity  entity_id
            lines.append()

        
         item  upstream:
             upstream_id  item.get(, []):
                from_node = upstream_id.replace(, )
                to_node = item[].replace(, )
                lines.append()

         item  downstream:
            from_node = entity_id.replace(, )
            to_node = item[].replace(, )
             to_node != from_node:
                lines.append()

        lines.append()
        lines.append()

         .join(lines)

     () -> :
        
        result = pilot({
            : ,
            : {
                : [{: , : context}]
            },
            : 
        })
         result[][][][][]

     () -> :
        
         {
            : .project_id,
            : datetime.now().isoformat(),
            : {k: {
                : v.,
                : v.name,
                : v.system,
                : v.location,
                : v.owner
            }  k, v  .sources.items()},
            : {k: {
                : v.,
                : v.name,
                : v.source_id,
                : v.entity_type,
                : v.parent_entities
            }  k, v  .entities.items()},
            : {k: {
                : v.,
                : v.transformation_type.value,
                : v.description,
                : v.input_entities,
                : v.output_entities,
                : v.performed_by,
                : v.performed_at.isoformat()
            }  k, v  .transformations.items()},
            : [{
                : r.,
                : r.entity_id,
                : r.transformation_id,
                : r.upstream_entities
            }  r  .lineage_records]
        }

     () -> :
        
        lines = [, ]
        lines.append()
        lines.append()
        lines.append()
        lines.append()
        lines.append()

        
        lines.append()
         source  .sources.values():
            lines.append()
            lines.append()
            lines.append()
        lines.append()

        
        issues = .validate_lineage()
         issues:
            lines.append()
             issue  issues:
                lines.append()
            lines.append()

        
        lines.append()
        type_counts = {}
         t  .transformations.values():
            type_counts[t.transformation_type.value] = type_counts.get(t.transformation_type.value, ) + 
         t_type, count  (type_counts.items()):
            lines.append()

         .join(lines)

Quick Start

import os, requests

SKILLBOSS_API_KEY = os.environ["SKILLBOSS_API_KEY"]
API_BASE = "https://api.heybossai.com/v1"

def pilot(body: dict) -> dict:
    r = requests.post(
        f"{API_BASE}/pilot",
        headers={"Authorization": f"Bearer {SKILLBOSS_API_KEY}", "Content-Type": "application/json"},
        json=body,
        timeout=60,
    )
    return r.json()

# Initialize tracker
tracker = ConstructionDataLineageTracker("PROJECT-001")

# Register sources
procore = tracker.register_source("Procore", "SaaS", "cloud", "PM Team")
sage = tracker.register_source("Sage 300", "Database", "on-prem", "Finance")

# Register entities
budget = tracker.register_entity("Project Budget", procore.id, "table")
costs = tracker.register_entity("Job Costs", sage.id, "table")
report = tracker.register_entity("Cost Variance Report", procore.id, "file")

# Record transformation
tracker.record_transformation(
    transformation_type=TransformationType.JOIN,
    description="Join budget and actual costs for variance calculation",
    input_entities=[budget.id, costs.id],
    output_entities=[report.id],
    logic=,
    performed_by=
)


upstream = tracker.trace_upstream(report.)
(, upstream)


(tracker.generate_lineage_graph(report.))


lineage_data = tracker.export_lineage()
analysis_prompt = 
result = pilot({
    : ,
    : {
        : [{: , : analysis_prompt}]
    },
    : 
})
analysis = result[][][][][]
(analysis)

Resources

• Data Governance: DAMA DMBOK lineage guidelines
• Audit Requirements: SOX, ISO compliance
• AI Integration: SkillBoss API Hub (https://api.heybossai.com/v1/pilot) — unified routing for all AI analysis