Skip to content

Models Reference

sqldeps.models

Data models for SQLDeps.

This module defines the core data structures used by SQLDeps for representing SQL dependencies and outputs.

SQLProfile dataclass

Data class to hold both SQL dependencies and outputs.

Source code in sqldeps/models.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
@dataclass
class SQLProfile:
    """Data class to hold both SQL dependencies and outputs."""

    # Dependencies (input tables/columns required by the query)
    dependencies: dict[str, list[str]]

    # Outputs (tables/columns created or modified by the query)
    outputs: dict[str, list[str]]

    def __post_init__(self) -> None:
        """Sort tables and columns for consistent output."""
        self.dependencies = {
            table: sorted(cols) for table, cols in sorted(self.dependencies.items())
        }
        self.outputs = {
            table: sorted(cols) for table, cols in sorted(self.outputs.items())
        }

    @property
    def dependency_tables(self) -> list[str]:
        """Get list of dependency tables.

        Returns:
            list[str]: Sorted list of table names referenced as dependencies
        """
        return sorted(self.dependencies.keys())

    @property
    def outcome_tables(self) -> list[str]:
        """Get list of outcome tables.

        Returns:
            list[str]: Sorted list of table names referenced as outputs
        """
        return sorted(self.outputs.keys())

    def to_dict(self) -> dict:
        """Convert to dictionary format.

        Returns:
            dict: Dictionary with dependencies and outputs
        """
        return {"dependencies": self.dependencies, "outputs": self.outputs}

    def to_dataframe(self) -> pd.DataFrame:
        """Convert to a DataFrame with type column indicating dependency or outcome.

        Returns:
            pd.DataFrame: DataFrame with columns for type, schema, table, and column
        """
        records = []

        # Add dependencies
        for table, columns in self.dependencies.items():
            schema, table_name = table.split(".") if "." in table else (None, table)
            if columns:
                for column in columns:
                    records.append(
                        {
                            "type": "dependency",
                            "schema": schema,
                            "table": table_name,
                            "column": column,
                        }
                    )
            else:
                records.append(
                    {
                        "type": "dependency",
                        "schema": schema,
                        "table": table_name,
                        "column": None,
                    }
                )

        # Add outputs
        for table, columns in self.outputs.items():
            schema, table_name = table.split(".") if "." in table else (None, table)
            if columns:
                for column in columns:
                    records.append(
                        {
                            "type": "outcome",
                            "schema": schema,
                            "table": table_name,
                            "column": column,
                        }
                    )
            else:
                records.append(
                    {
                        "type": "outcome",
                        "schema": schema,
                        "table": table_name,
                        "column": None,
                    }
                )

        return pd.DataFrame(records)

dependency_tables property

Get list of dependency tables.

Returns:

Type Description
list[str]

list[str]: Sorted list of table names referenced as dependencies

outcome_tables property

Get list of outcome tables.

Returns:

Type Description
list[str]

list[str]: Sorted list of table names referenced as outputs

__post_init__()

Sort tables and columns for consistent output.

Source code in sqldeps/models.py 
22
23
24
25
26
27
28
29
def __post_init__(self) -> None:
    """Sort tables and columns for consistent output."""
    self.dependencies = {
        table: sorted(cols) for table, cols in sorted(self.dependencies.items())
    }
    self.outputs = {
        table: sorted(cols) for table, cols in sorted(self.outputs.items())
    }

to_dataframe()

Convert to a DataFrame with type column indicating dependency or outcome.

Returns:

Type Description
DataFrame

pd.DataFrame: DataFrame with columns for type, schema, table, and column
Source code in sqldeps/models.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def to_dataframe(self) -> pd.DataFrame:
    """Convert to a DataFrame with type column indicating dependency or outcome.

    Returns:
        pd.DataFrame: DataFrame with columns for type, schema, table, and column
    """
    records = []

    # Add dependencies
    for table, columns in self.dependencies.items():
        schema, table_name = table.split(".") if "." in table else (None, table)
        if columns:
            for column in columns:
                records.append(
                    {
                        "type": "dependency",
                        "schema": schema,
                        "table": table_name,
                        "column": column,
                    }
                )
        else:
            records.append(
                {
                    "type": "dependency",
                    "schema": schema,
                    "table": table_name,
                    "column": None,
                }
            )

    # Add outputs
    for table, columns in self.outputs.items():
        schema, table_name = table.split(".") if "." in table else (None, table)
        if columns:
            for column in columns:
                records.append(
                    {
                        "type": "outcome",
                        "schema": schema,
                        "table": table_name,
                        "column": column,
                    }
                )
        else:
            records.append(
                {
                    "type": "outcome",
                    "schema": schema,
                    "table": table_name,
                    "column": None,
                }
            )

    return pd.DataFrame(records)

to_dict()

Convert to dictionary format.

Returns:

Name Type Description
dict dict

Dictionary with dependencies and outputs
Source code in sqldeps/models.py 
49
50
51
52
53
54
55
def to_dict(self) -> dict:
    """Convert to dictionary format.

    Returns:
        dict: Dictionary with dependencies and outputs
    """
    return {"dependencies": self.dependencies, "outputs": self.outputs}