tessl/pypi-sqllineage

SQL Lineage Analysis Tool powered by Python

Overview

Eval results

Files

Configuration Management

Name: tessl/pypi-sqllineage
Author: tessl

Thread-safe configuration system for customizing SQLLineage behavior including default schemas, dialect-specific parsing options, and integration settings. The configuration system supports both environment variables and context-based temporary configuration changes.

Capabilities

SQLLineageConfig

Global configuration object providing thread-safe access to configuration settings with support for environment variable overrides and temporary context-based changes.

class _SQLLineageConfigLoader:
    def __init__(self) -> None:
        """Initialize the configuration loader"""
    
    def __call__(self, **kwargs) -> "_SQLLineageConfigLoader":
        """
        Create a configuration context manager for temporary changes.
        
        Parameters:
        - **kwargs: configuration key-value pairs to set temporarily
        
        Returns:
        Context manager for temporary configuration
        """
    
    def __enter__(self) -> None:
        """Enter the configuration context"""
    
    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        """Exit the configuration context and restore previous settings"""
    
    def __getattr__(self, item: str) -> Any:
        """Get configuration value by attribute name"""
    
    def __setattr__(self, key: str, value: Any) -> None:
        """Set configuration value (raises exception - use context manager instead)"""

# Global configuration instance
SQLLineageConfig = _SQLLineageConfigLoader()

Configuration Options

Core Configuration Keys

# Configuration keys and their types/defaults
config = {
    "DIRECTORY": (str, os.path.join(os.path.dirname(__file__), "data")),  # Static files directory (resolved dynamically)
    "DEFAULT_SCHEMA": (str, ""),                     # Default schema name
    "TSQL_NO_SEMICOLON": (bool, False),            # TSQL mode without semicolon
    "LATERAL_COLUMN_ALIAS_REFERENCE": (bool, False) # Support lateral column alias references
}

Environment Variable Support

All configuration options can be set via environment variables with the SQLLINEAGE_ prefix:

SQLLINEAGE_DIRECTORY - Override default directory
SQLLINEAGE_DEFAULT_SCHEMA - Set default schema
SQLLINEAGE_TSQL_NO_SEMICOLON - Enable TSQL no-semicolon mode
SQLLINEAGE_LATERAL_COLUMN_ALIAS_REFERENCE - Enable lateral column aliases

Usage Examples

Basic Configuration Access

from sqllineage.config import SQLLineageConfig

# Access configuration values
print("Default schema:", SQLLineageConfig.DEFAULT_SCHEMA)
print("Static directory:", SQLLineageConfig.DIRECTORY)
print("TSQL no semicolon:", SQLLineageConfig.TSQL_NO_SEMICOLON)

Environment Variable Configuration

# Set environment variables
export SQLLINEAGE_DEFAULT_SCHEMA="analytics"
export SQLLINEAGE_TSQL_NO_SEMICOLON="true"

from sqllineage.config import SQLLineageConfig

# Configuration will use environment variable values
print("Default schema:", SQLLineageConfig.DEFAULT_SCHEMA)  # "analytics"
print("TSQL mode:", SQLLineageConfig.TSQL_NO_SEMICOLON)    # True

Context-Based Configuration

from sqllineage.config import SQLLineageConfig
from sqllineage.runner import LineageRunner

# Temporary configuration change
with SQLLineageConfig(DEFAULT_SCHEMA="staging"):
    # Inside context, DEFAULT_SCHEMA is "staging"
    sql = "SELECT * FROM customers"  # Will be interpreted as staging.customers
    runner = LineageRunner(sql)
    print("Tables:", [str(t) for t in runner.source_tables])

# Outside context, DEFAULT_SCHEMA returns to original value
print("Default schema:", SQLLineageConfig.DEFAULT_SCHEMA)

Multiple Configuration Changes

from sqllineage.config import SQLLineageConfig

# Set multiple configuration options temporarily
with SQLLineageConfig(
    DEFAULT_SCHEMA="production",
    TSQL_NO_SEMICOLON=True,
    LATERAL_COLUMN_ALIAS_REFERENCE=True
):
    # Analyze SQL with custom configuration
    sql = """
    SELECT 
        customer_id,
        revenue,
        revenue * 0.1 as commission
    FROM sales
    WHERE commission > 1000
    """
    runner = LineageRunner(sql, dialect="tsql")
    runner.print_column_lineage()

Schema-Specific Analysis

from sqllineage.config import SQLLineageConfig
from sqllineage.runner import LineageRunner

# Analyze SQL with different default schemas
schemas = ["raw", "staging", "production"]

for schema in schemas:
    with SQLLineageConfig(DEFAULT_SCHEMA=schema):
        sql = "INSERT INTO customer_summary SELECT * FROM customer_details"
        runner = LineageRunner(sql)
        print(f"Schema {schema}:")
        print(f"  Source: {runner.source_tables}")
        print(f"  Target: {runner.target_tables}")

Dialect-Specific Configuration

from sqllineage.config import SQLLineageConfig

# Configure for SQL Server T-SQL without semicolon separators
with SQLLineageConfig(TSQL_NO_SEMICOLON=True):
    tsql = """
    CREATE TABLE #temp AS SELECT * FROM customers
    INSERT INTO sales_summary 
    SELECT customer_id, SUM(amount) FROM #temp GROUP BY customer_id
    DROP TABLE #temp
    """
    runner = LineageRunner(tsql, dialect="tsql")
    print("T-SQL analysis with no-semicolon mode")

# Configure for databases supporting lateral column references
with SQLLineageConfig(LATERAL_COLUMN_ALIAS_REFERENCE=True):
    sql = """
    SELECT 
        customer_id,
        total_amount,
        total_amount * 0.1 as tax,
        total_amount + tax as final_amount
    FROM orders
    """
    runner = LineageRunner(sql, dialect="redshift")
    runner.print_column_lineage()

Thread Safety

import threading
from sqllineage.config import SQLLineageConfig

def analyze_in_thread(schema_name, sql):
    with SQLLineageConfig(DEFAULT_SCHEMA=schema_name):
        runner = LineageRunner(sql)
        print(f"Thread {schema_name}: {runner.source_tables}")

# Different threads can have different configurations
threads = []
for i, schema in enumerate(["dev", "staging", "prod"]):
    sql = f"SELECT * FROM table_{i}"
    thread = threading.Thread(target=analyze_in_thread, args=(schema, sql))
    threads.append(thread)
    thread.start()

for thread in threads:
    thread.join()

Configuration Validation

from sqllineage.config import SQLLineageConfig
from sqllineage.exceptions import ConfigException

try:
    # Attempting to set configuration directly raises an exception
    SQLLineageConfig.DEFAULT_SCHEMA = "invalid"
except ConfigException as e:
    print(f"Configuration error: {e}")
    print("Use context manager instead:")
    
    with SQLLineageConfig(DEFAULT_SCHEMA="valid"):
        print("Configuration set successfully")

Custom Configuration Patterns

from sqllineage.config import SQLLineageConfig
from sqllineage.runner import LineageRunner

class AnalysisProfile:
    def __init__(self, name, default_schema, dialect, tsql_mode=False):
        self.name = name
        self.default_schema = default_schema
        self.dialect = dialect
        self.tsql_mode = tsql_mode
    
    def analyze(self, sql):
        with SQLLineageConfig(
            DEFAULT_SCHEMA=self.default_schema,
            TSQL_NO_SEMICOLON=self.tsql_mode
        ):
            runner = LineageRunner(sql, dialect=self.dialect)
            return runner

# Define analysis profiles
profiles = {
    "snowflake_prod": AnalysisProfile("Snowflake Prod", "ANALYTICS", "snowflake"),
    "sqlserver_dev": AnalysisProfile("SQL Server Dev", "dbo", "tsql", tsql_mode=True),
    "postgres_staging": AnalysisProfile("PostgreSQL Staging", "staging", "postgres")
}

# Use profiles for analysis
sql = "SELECT customer_id, sum(amount) FROM orders GROUP BY customer_id"

for profile_name, profile in profiles.items():
    runner = profile.analyze(sql)
    print(f"{profile.name}: {runner.source_tables}")

Configuration Debugging

from sqllineage.config import SQLLineageConfig

def debug_configuration():
    print("Current configuration:")
    print(f"  DIRECTORY: {SQLLineageConfig.DIRECTORY}")
    print(f"  DEFAULT_SCHEMA: {SQLLineageConfig.DEFAULT_SCHEMA}")
    print(f"  TSQL_NO_SEMICOLON: {SQLLineageConfig.TSQL_NO_SEMICOLON}")
    print(f"  LATERAL_COLUMN_ALIAS_REFERENCE: {SQLLineageConfig.LATERAL_COLUMN_ALIAS_REFERENCE}")

# Debug before and after configuration changes
debug_configuration()

with SQLLineageConfig(DEFAULT_SCHEMA="test", TSQL_NO_SEMICOLON=True):
    print("\nInside context:")
    debug_configuration()

print("\nAfter context:")
debug_configuration()

Install with Tessl CLI