Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
136 changes: 95 additions & 41 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -108,7 +108,7 @@ print(query_lineage)
# Get source tables
input_nodes = query_lineage.get_input_nodes()
source_tables = {node.table_name for node in input_nodes if node.table_name}
print("-"*60)
print("-" * 60)
print(f"{len(source_tables)} source tables:")
for table in source_tables:
print(f" {table}")
Expand Down Expand Up @@ -151,24 +151,33 @@ ColumnLineageGraph(
from clgraph import Pipeline

queries = [
("raw_events", """
(
"raw_events",
"""
CREATE TABLE raw_events AS
SELECT user_id, event_type, event_timestamp, session_id
FROM source_events
WHERE event_timestamp >= '2024-01-01'
"""),
("daily_active_users", """
""",
),
(
"daily_active_users",
"""
CREATE TABLE daily_active_users AS
SELECT user_id, DATE(event_timestamp) as activity_date, COUNT(*) as event_count
FROM raw_events
GROUP BY user_id, DATE(event_timestamp)
"""),
("user_summary", """
""",
),
(
"user_summary",
"""
CREATE TABLE user_summary AS
SELECT u.name, u.email, dau.activity_date, dau.event_count
FROM users u
JOIN daily_active_users dau ON u.id = dau.user_id
"""),
""",
),
]

pipeline = Pipeline(queries, dialect="bigquery")
Expand Down Expand Up @@ -287,17 +296,23 @@ from clgraph import Pipeline, JSONExporter, CSVExporter, visualize_pipeline_line

# Build pipeline
queries = [
("raw.orders", """
(
"raw.orders",
"""
CREATE TABLE raw.orders AS
SELECT order_id, user_email, amount, order_date
FROM source.orders
"""),
("analytics.revenue", """
""",
),
(
"analytics.revenue",
"""
CREATE TABLE analytics.revenue AS
SELECT user_email, SUM(amount) as total_revenue
FROM raw.orders
GROUP BY user_email
"""),
""",
),
]

pipeline = Pipeline(queries, dialect="bigquery")
Expand Down Expand Up @@ -369,7 +384,10 @@ import json
# Build pipeline
queries = [
("staging", "CREATE TABLE staging.orders AS SELECT id, amount FROM raw.orders"),
("analytics", "CREATE TABLE analytics.totals AS SELECT SUM(amount) as total FROM staging.orders"),
(
"analytics",
"CREATE TABLE analytics.totals AS SELECT SUM(amount) as total FROM staging.orders",
),
]
pipeline = Pipeline.from_tuples(queries, dialect="bigquery")

Expand Down Expand Up @@ -419,17 +437,23 @@ from langchain_ollama import ChatOllama

# Build pipeline
queries = [
("raw.orders", """
(
"raw.orders",
"""
CREATE TABLE raw.orders AS
SELECT order_id, user_email, amount, order_date
FROM source.orders
"""),
("analytics.revenue", """
""",
),
(
"analytics.revenue",
"""
CREATE TABLE analytics.revenue AS
SELECT user_email, SUM(amount) as total_revenue
FROM raw.orders
GROUP BY user_email
"""),
""",
),
]

pipeline = Pipeline(queries, dialect="bigquery")
Expand All @@ -445,9 +469,7 @@ pipeline.generate_all_descriptions(verbose=True)
print("-" * 60)

# View generated descriptions
columns_with_descriptions = [
col for col in pipeline.columns.values() if col.description
]
columns_with_descriptions = [col for col in pipeline.columns.values() if col.description]
print(f"Generated descriptions for {len(columns_with_descriptions)} columns:")
for col in columns_with_descriptions:
print(f" {col.full_name}:")
Expand Down Expand Up @@ -489,17 +511,23 @@ from clgraph.agent import LineageAgent

# Build pipeline
queries = [
("staging.orders", """
(
"staging.orders",
"""
CREATE TABLE staging.orders AS
SELECT order_id, customer_id, amount, order_date
FROM raw.orders WHERE amount > 0
"""),
("analytics.revenue", """
""",
),
(
"analytics.revenue",
"""
CREATE TABLE analytics.revenue AS
SELECT customer_id, SUM(amount) as total_revenue, COUNT(*) as order_count
FROM staging.orders
GROUP BY customer_id
"""),
""",
),
]
pipeline = Pipeline(queries, dialect="bigquery")

Expand Down Expand Up @@ -554,24 +582,30 @@ from clgraph import Pipeline

# Build pipeline with column descriptions (from SQL comments)
queries = [
("customers", """
(
"customers",
"""
CREATE TABLE analytics.customers AS
SELECT
customer_id, -- Unique customer identifier
email, -- Customer email address [pii: true]
signup_date, -- Date customer signed up
lifetime_value -- Total revenue from this customer in USD
FROM raw.customers
"""),
("orders", """
""",
),
(
"orders",
"""
CREATE TABLE analytics.orders AS
SELECT
order_id, -- Unique order identifier
customer_id, -- Reference to customer
amount, -- Order amount in USD
order_date -- Date of order
FROM raw.orders
"""),
""",
),
]
pipeline = Pipeline(queries, dialect="bigquery")

Expand Down Expand Up @@ -633,14 +667,20 @@ from clgraph.tools import (

# Build pipeline
queries = [
("staging.orders", """
(
"staging.orders",
"""
CREATE TABLE staging.orders AS
SELECT order_id, customer_email, amount FROM raw.orders
"""),
("analytics.revenue", """
""",
),
(
"analytics.revenue",
"""
CREATE TABLE analytics.revenue AS
SELECT customer_email, SUM(amount) as total FROM staging.orders GROUP BY 1
"""),
""",
),
]
pipeline = Pipeline(queries, dialect="bigquery")

Expand Down Expand Up @@ -1033,7 +1073,10 @@ from clgraph import Pipeline
# Sample pipeline for examples
queries = [
("raw.orders", "CREATE TABLE raw.orders AS SELECT id, amount FROM source.orders"),
("analytics.metrics", "CREATE TABLE analytics.metrics AS SELECT SUM(amount) as total FROM raw.orders"),
(
"analytics.metrics",
"CREATE TABLE analytics.metrics AS SELECT SUM(amount) as total FROM raw.orders",
),
]
pipeline = Pipeline(queries, dialect="bigquery")

Expand All @@ -1051,8 +1094,14 @@ from clgraph import Pipeline

queries = [
("raw.orders", "CREATE TABLE raw.orders AS SELECT id, amount FROM source.orders"),
("staging.orders", "CREATE TABLE staging.orders AS SELECT id, amount FROM raw.orders WHERE amount > 0"),
("analytics.metrics", "CREATE TABLE analytics.metrics AS SELECT SUM(amount) as total FROM staging.orders"),
(
"staging.orders",
"CREATE TABLE staging.orders AS SELECT id, amount FROM raw.orders WHERE amount > 0",
),
(
"analytics.metrics",
"CREATE TABLE analytics.metrics AS SELECT SUM(amount) as total FROM staging.orders",
),
]
pipeline = Pipeline(queries, dialect="bigquery")

Expand Down Expand Up @@ -1082,7 +1131,10 @@ from clgraph import Pipeline

queries = [
("raw.orders", "CREATE TABLE raw.orders AS SELECT id, amount FROM source.orders"),
("analytics.metrics", "CREATE TABLE analytics.metrics AS SELECT SUM(amount) as total FROM raw.orders"),
(
"analytics.metrics",
"CREATE TABLE analytics.metrics AS SELECT SUM(amount) as total FROM raw.orders",
),
]
pipeline = Pipeline(queries, dialect="bigquery")

Expand All @@ -1091,7 +1143,9 @@ print(f"Columns: {len(pipeline.column_graph.columns)}")
print(f"Edges: {len(pipeline.column_graph.edges)}")

# Backward compatible access (property aliases)
print(f"pipeline.columns == pipeline.column_graph.columns: {pipeline.columns == pipeline.column_graph.columns}")
print(
f"pipeline.columns == pipeline.column_graph.columns: {pipeline.columns == pipeline.column_graph.columns}"
)

# Get source columns (no incoming edges)
source_cols = pipeline.column_graph.get_source_columns()
Expand All @@ -1112,7 +1166,10 @@ from clgraph import Pipeline
queries = [
("raw.orders", "CREATE TABLE raw.orders AS SELECT id, amount FROM source.orders"),
("staging.orders", "CREATE TABLE staging.orders AS SELECT id, amount FROM raw.orders"),
("analytics.metrics", "CREATE TABLE analytics.metrics AS SELECT SUM(amount) as total FROM staging.orders"),
(
"analytics.metrics",
"CREATE TABLE analytics.metrics AS SELECT SUM(amount) as total FROM staging.orders",
),
]
pipeline = Pipeline(queries, dialect="bigquery")

Expand All @@ -1125,10 +1182,7 @@ impacts = pipeline.trace_column_forward("raw.orders", "amount")
print(f"Impacts of raw.orders.amount: {[i.full_name for i in impacts]}")

# Find specific lineage path between two columns (returns edges)
path = pipeline.get_lineage_path(
"raw.orders", "amount",
"analytics.metrics", "total"
)
path = pipeline.get_lineage_path("raw.orders", "amount", "analytics.metrics", "total")
if path:
print(f"Path has {len(path)} edges")
for edge in path:
Expand Down
3 changes: 2 additions & 1 deletion clgraph-simple-diagram.md
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ pipeline = Pipeline.from_sql_files("examples/sql_files/", dialect="duckdb")
# 2. Pipeline object created automatically

# 3. Access graphs
table_deps = pipeline.table_graph # Table dependencies
table_deps = pipeline.table_graph # Table dependencies
col_lineage = pipeline.column_graph # Column lineage

# 4. Use applications
Expand Down Expand Up @@ -90,6 +90,7 @@ dag = pipeline.to_airflow_dag(executor=my_execute_sql, dag_id="my_pipeline")

# LLM-powered descriptions (requires langchain)
from langchain_openai import ChatOpenAI

pipeline.llm = ChatOpenAI()
pipeline.generate_all_descriptions()
```
Expand Down
2 changes: 2 additions & 0 deletions examples/sql_files/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -119,10 +119,12 @@ conn = duckdb.connect(":memory:")
# Build pipeline
pipeline = Pipeline(queries, dialect="duckdb")


# Define executor
def execute_sql(sql: str):
conn.execute(sql)


# Run pipeline in dependency order
result = pipeline.run(executor=execute_sql, max_workers=1, verbose=True)

Expand Down
4 changes: 2 additions & 2 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@ build-backend = "hatchling.build"

[project]
name = "clgraph"
version = "0.0.4"
version = "0.0.5"
description = "Column lineage and pipeline dependency analysis for SQL"
readme = "README.md"
requires-python = ">=3.10"
Expand Down Expand Up @@ -51,7 +51,7 @@ mcp = [
dev = [
"pytest>=7.0.0",
"pytest-cov>=4.0.0",
"ruff>=0.6.0",
"ruff>=0.16.1,<0.17",
"ty>=0.0.1a0",
"jinja2>=3.0.0",
"duckdb>=0.9.0",
Expand Down
Loading
Loading