sqlmesh.dbt.util

 1from __future__ import annotations
 2
 3import typing as t
 4
 5import agate
 6from dbt.version import get_installed_version
 7
 8if t.TYPE_CHECKING:
 9    import pandas as pd
10
11
12def _get_dbt_version() -> t.Tuple[int, int, int]:
13    dbt_version = get_installed_version()
14    return (
15        int(dbt_version.major or "0"),
16        int(dbt_version.minor or "0"),
17        int(dbt_version.patch or "0"),
18    )
19
20
21DBT_VERSION = _get_dbt_version()
22
23if DBT_VERSION >= (1, 8, 0):
24    from dbt_common.clients.agate_helper import table_from_data_flat, empty_table, as_matrix  # type: ignore  # noqa: F401
25else:
26    from dbt.clients.agate_helper import table_from_data_flat, empty_table, as_matrix  # type: ignore  # noqa: F401
27
28
29def pandas_to_agate(df: pd.DataFrame) -> agate.Table:
30    """
31    Converts a Pandas DataFrame to an Agate Table
32    """
33
34    return table_from_data_flat(df.to_dict(orient="records"), df.columns.tolist())