Data Sources¶

Python API 사용에서 Multi-backend을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

개요¶

Python API 사용에서 Truthound, DataSource을(를) 다루는 항목입니다:

Python API 사용에서 Category을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 Sources을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
파일	Python API 사용에서 JSON, CSV, Parquet, NDJSON을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 DataFrames을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 Polars, Pandas을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 SQL, Databases을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 PostgreSQL, SQLite, MySQL, SQL, DuckDB을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 Cloud, Warehouses을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 Databricks, Snowflake, Redshift, BigQuery을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 Enterprise을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 SQL, Oracle, Server을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 Big, Data을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 Apache, Spark을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 SQL, NoSQL을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 MongoDB, Elasticsearch을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 Streaming을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 Apache, Kafka, Kinesis을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

설치 & Dependencies¶

Python API 사용에서 SQLite, JSON, SQL, Python, Core, CSV, Parquet, NDJSON을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Install Multiple Dependencies¶

# Install all enterprise data sources
pip install truthound[enterprise]

# Provider matrix의 모든 SQL driver 설치
pip install truthound[sql-connectors]

# Install specific combinations
pip install truthound[postgresql,bigquery,spark]

# Install all optional dependencies
pip install truthound[all]

SQL row 및 materialization 계약¶

BaseSQLDataSource.execute_query()는 tuple, mapping, _mapping driver row를 cursor column name 기준 dictionary로 정규화합니다. execute_scalar()와 row_count도 같은 계약을 사용합니다.

to_polars_lazyframe()은 무제한 table dump가 아닙니다. fetch_size 단위로 읽고 materialization_row_limit를 넘으면 DataSourceSizeError를 발생시킵니다. SQL source는 다음처럼 공개 facade의 source keyword로 전달합니다.

SQL provider의 table schema 조회는 두 전략 중 하나를 사용합니다.

_get_table_schema_query()를 구현해 공통 DB-API query 경로를 사용합니다.
BigQuery metadata API나 warehouse별 DESCRIBE처럼 provider-native 조회가 필요하면 _fetch_schema()를 구현합니다.

공개 지원 목록의 provider class는 모두 concrete여야 합니다. 사용자 정의 SQL provider는 생성 시 두 schema 전략 중 하나를 제공하는지 검증됩니다. release artifact도 truthound[sql-connectors] 설치 후 SQL provider class를 전수 검사하며, driver import만 성공한 상태는 provider 생성성 통과로 보지 않습니다.

import truthound as th

validation = th.check(source=source)
profile = th.profile(source=source.sample(10_000))

소스 Code Locations¶

BaseDataSource¶

Python API 사용에서 BaseDataSource, BaseDataSource을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Definition¶

from truthound.datasources.base import BaseDataSource

class BaseDataSource(ABC, Generic[ConfigT]):
    """Abstract base class for all data sources."""

    source_type: str = "base"

    @property
    def name(self) -> str:
        """Get the data source name."""

    @property
    @abstractmethod
    def schema(self) -> dict[str, ColumnType]:
        """Column name to type mapping."""

    @property
    def columns(self) -> list[str]:
        """Get list of column names."""

    @property
    def row_count(self) -> int | None:
        """Get row count if efficiently available."""

    @property
    def capabilities(self) -> set[DataSourceCapability]:
        """Get supported capabilities."""

    @abstractmethod
    def to_polars_lazyframe(self) -> pl.LazyFrame:
        """Convert to Polars LazyFrame."""

    @abstractmethod
    def get_execution_engine(self) -> BaseExecutionEngine:
        """Return execution engine."""

    def needs_sampling(self) -> bool:
        """Check if sampling is needed for large datasets."""

    @abstractmethod
    def sample(self, n: int, seed: int | None = None) -> BaseDataSource:
        """Return sampled data source."""

Capabilities¶

from truthound.datasources import DataSourceCapability

class DataSourceCapability(Enum):
    LAZY_EVALUATION = "lazy_evaluation"  # Supports lazy/deferred execution
    SQL_PUSHDOWN = "sql_pushdown"        # Can push operations to database
    SAMPLING = "sampling"                # Supports efficient sampling
    STREAMING = "streaming"              # Supports streaming/chunked reads
    SCHEMA_INFERENCE = "schema_inference"  # Can infer schema without full scan
    ROW_COUNT = "row_count"              # Can get row count efficiently

ColumnType¶

from truthound.datasources import ColumnType

class ColumnType(Enum):
    # Numeric types
    INTEGER = "integer"
    FLOAT = "float"
    DECIMAL = "decimal"

    # String types
    STRING = "string"
    TEXT = "text"

    # Date/Time types
    DATE = "date"
    DATETIME = "datetime"
    TIME = "time"
    DURATION = "duration"

    # Boolean
    BOOLEAN = "boolean"

    # Binary
    BINARY = "binary"

    # Complex types
    LIST = "list"
    STRUCT = "struct"
    JSON = "json"

    # Other
    NULL = "null"
    UNKNOWN = "unknown"

Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

파일 Sources¶

FileDataSource¶

Python API 사용에서 Automatically을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.datasources.polars_source import FileDataSource

# CSV
source = FileDataSource("data.csv")

# Parquet
source = FileDataSource("data.parquet")

# JSON
source = FileDataSource("data.json")

# NDJSON (newline-delimited JSON)
source = FileDataSource("data.ndjson")

FileDataSourceConfig¶

from truthound.datasources.polars_source import FileDataSourceConfig

config = FileDataSourceConfig(
    path="data.csv",
    name="my_data",
    max_rows=1_000_000,
    infer_schema_length=1000,
    n_rows=None,  # Limit rows to read
    encoding="utf-8",
)
source = FileDataSource(config=config)

Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

DataFrame Sources¶

PolarsDataSource¶

from truthound.datasources.polars_source import PolarsDataSource
import polars as pl

# From DataFrame
df = pl.read_csv("data.csv")
source = PolarsDataSource(df)

# From LazyFrame
lf = pl.scan_csv("data.csv")
source = PolarsDataSource(lf)

PandasDataSource¶

from truthound.datasources.pandas_source import PandasDataSource
import pandas as pd

pdf = pd.read_csv("data.csv")
source = PandasDataSource(pdf)

DictDataSource¶

from truthound.datasources.polars_source import DictDataSource

data = {"col1": [1, 2, 3], "col2": ["a", "b", "c"]}
source = DictDataSource(data)

Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

SQL Databases¶

SQLiteDataSource¶

from truthound.datasources.sql import SQLiteDataSource

# From table
source = SQLiteDataSource(
    database="mydb.db",
    table="users",
)

# With custom query
source = SQLiteDataSource(
    database="mydb.db",
    query="SELECT * FROM users WHERE active = 1",
)

DuckDBDataSource¶

Python API 사용에서 Polars, DuckDB을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.datasources.sql import DuckDBDataSource

# From table
source = DuckDBDataSource(
    database="analytics.duckdb",
    table="events",
)

# With custom query
source = DuckDBDataSource(
    database="analytics.duckdb",
    query="SELECT * FROM events WHERE date > '2024-01-01'",
)

# In-memory database
source = DuckDBDataSource(
    database=":memory:",
    query="SELECT 1 as id, 'test' as value",
)

# Read directly from Parquet file (DuckDB feature)
source = DuckDBDataSource.from_parquet("data/*.parquet")

# Read from CSV file
source = DuckDBDataSource.from_csv("data.csv")

# Create from Polars DataFrame
import polars as pl
df = pl.DataFrame({"id": [1, 2, 3], "value": ["a", "b", "c"]})
source = DuckDBDataSource.from_dataframe(df, "my_table")

PostgreSQLDataSource¶

from truthound.datasources.sql import PostgreSQLDataSource

# Basic connection
source = PostgreSQLDataSource(
    table="users",
    host="localhost",
    port=5432,
    database="mydb",
    user="postgres",
    password="secret",
)

# With schema
source = PostgreSQLDataSource(
    table="users",
    schema="public",
    host="localhost",
    database="mydb",
    user="postgres",
)

# With connection URL
source = PostgreSQLDataSource.from_connection_string(
    connection_string="postgresql://user:pass@localhost:5432/mydb",
    table="users",
)

# With custom query
source = PostgreSQLDataSource(
    host="localhost",
    database="mydb",
    user="postgres",
    query="SELECT id, email FROM users WHERE created_at > '2024-01-01'",
)

MySQLDataSource¶

from truthound.datasources.sql import MySQLDataSource

source = MySQLDataSource(
    table="orders",
    host="localhost",
    port=3306,
    database="mydb",
    user="root",
    password="secret",
)

Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Cloud Data Warehouses¶

BigQueryDataSource¶

from truthound.datasources.sql import BigQueryDataSource

# Basic usage
source = BigQueryDataSource(
    project="my-gcp-project",
    dataset="analytics",
    table="events",
)

# With credentials file
source = BigQueryDataSource(
    project="my-gcp-project",
    dataset="analytics",
    table="events",
    credentials_path="/path/to/credentials.json",
)

# With custom query
source = BigQueryDataSource(
    project="my-gcp-project",
    query="SELECT * FROM `analytics.events` WHERE date > '2024-01-01'",
)

SnowflakeDataSource¶

from truthound.datasources.sql import SnowflakeDataSource

# Password authentication
source = SnowflakeDataSource(
    account="myaccount",
    user="myuser",
    password="mypassword",
    database="MYDB",
    schema="PUBLIC",
    warehouse="COMPUTE_WH",
    table="USERS",
)

# Key-pair authentication
source = SnowflakeDataSource(
    account="myaccount",
    user="myuser",
    private_key_path="/path/to/key.p8",
    database="MYDB",
    schema="PUBLIC",
    warehouse="COMPUTE_WH",
    table="USERS",
)

RedshiftDataSource¶

from truthound.datasources.sql import RedshiftDataSource

# Password authentication
source = RedshiftDataSource(
    host="cluster.region.redshift.amazonaws.com",
    port=5439,
    database="mydb",
    user="myuser",
    password="mypassword",
    table="users",
)

# IAM authentication
source = RedshiftDataSource(
    host="cluster.region.redshift.amazonaws.com",
    database="mydb",
    iam_role="arn:aws:iam::123456789:role/RedshiftRole",
    table="users",
)

DatabricksDataSource¶

from truthound.datasources.sql import DatabricksDataSource

source = DatabricksDataSource(
    server_hostname="adb-123456789.azuredatabricks.net",
    http_path="/sql/1.0/warehouses/abc123",
    access_token="dapi...",
    catalog="main",
    schema="default",
    table="users",
)

Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Enterprise Databases¶

OracleDataSource¶

from truthound.datasources.sql import OracleDataSource

# With service name
source = OracleDataSource(
    host="localhost",
    port=1521,
    service_name="ORCL",
    user="myuser",
    password="mypassword",
    table="USERS",
)

SQLServerDataSource¶

from truthound.datasources.sql import SQLServerDataSource

# SQL authentication
source = SQLServerDataSource(
    host="localhost",
    port=1433,
    database="mydb",
    user="sa",
    password="mypassword",
    table="users",
)

# Windows authentication
source = SQLServerDataSource(
    host="localhost",
    database="mydb",
    trusted_connection=True,
    table="users",
)

Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Apache Spark¶

SparkDataSource¶

from truthound.datasources.spark_source import SparkDataSource
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("truthound").getOrCreate()
spark_df = spark.read.parquet("data.parquet")

source = SparkDataSource(spark_df)

Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Factory Functions¶

get_datasource()¶

Python API 사용에서 Auto-detect을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.datasources import get_datasource

# File path
source = get_datasource("data.csv")           # FileDataSource
source = get_datasource("data.parquet")       # FileDataSource
source = get_datasource("data.json")          # FileDataSource
source = get_datasource("data.ndjson")        # FileDataSource

# Polars DataFrame
source = get_datasource(polars_df)            # PolarsDataSource

# Pandas DataFrame
source = get_datasource(pandas_df)            # PandasDataSource

# Dictionary
source = get_datasource({"col": [1, 2, 3]})   # DictDataSource

Python API 사용에서 SQLite, SQL, get_datasource(), .db, get_sql_datasource(), Note, NOT을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

get_sql_datasource()¶

Python API 사용에서 SQL, Create, URL을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

from truthound.datasources import get_sql_datasource

# SQLite database file (requires table parameter)
source = get_sql_datasource("mydb.db", table="users")

# PostgreSQL connection URL
source = get_sql_datasource(
    "postgresql://user:pass@localhost/db",
    table="users"
)

Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Usage with Core Functions¶

import truthound as th
from truthound.datasources.sql import PostgreSQLDataSource, BigQueryDataSource, SQLiteDataSource

# Check
source = PostgreSQLDataSource(table="users", host="localhost", database="mydb")
report = th.check(source=source)

# Learn schema from database
source = SQLiteDataSource(database="mydb.db", table="users")
schema = th.learn(source=source)
schema.save("schema.yaml")

# Scan for PII
source = BigQueryDataSource(project="...", dataset="...", table="customers")
pii_report = th.scan(source=source)

# Profile
source = PostgreSQLDataSource(table="orders", host="localhost", database="mydb")
profile = th.profile(source=source)

# Mask
source = PostgreSQLDataSource(table="users", host="localhost", database="mydb")
masked_df = th.mask(source=source, strategy="hash")

Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Query Pushdown¶

Python API 사용에서 Truthound, SQL을(를) 다루는 항목입니다:

from truthound.datasources.sql import PostgreSQLDataSource

source = PostgreSQLDataSource(
    table="large_table",
    host="localhost",
    database="mydb",
)

# Enable pushdown - validations run on the database server
report = th.check(source=source, pushdown=True)

# Example: null_check becomes:
# SELECT COUNT(*) FROM table WHERE column IS NULL

Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Sampling¶

Python API 사용에서 관련 설정과 실행 흐름을(를) 다루는 항목입니다:

source = PostgreSQLDataSource(table="huge_table", host="localhost", database="mydb")

# Check if sampling is needed
if source.needs_sampling():
    print(f"Dataset has {source.row_count:,} rows, sampling recommended")
    source = source.sample(n=100_000, seed=42)

report = th.check(source=source)

Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Connection Pooling¶

Python API 사용에서 SQL을(를) 다루는 항목입니다:

from truthound.datasources.sql.base import SQLDataSourceConfig

config = SQLDataSourceConfig(
    table="users",
    host="localhost",
    database="mydb",
    pool_size=5,
    max_overflow=10,
    pool_timeout=30,
)
source = PostgreSQLDataSource(config=config)

Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

NoSQL Databases¶

MongoDBDataSource (Async)¶

from truthound.datasources import from_mongodb

# Connect to MongoDB
source = await from_mongodb(
    connection_string="mongodb://localhost:27017",
    database="mydb",
    collection="users",
)

async with source:
    lf = await source.to_polars_lazyframe_async()
    report = th.check(lf)

ElasticsearchDataSource (Async)¶

from truthound.datasources import from_elasticsearch

source = await from_elasticsearch(
    hosts=["http://localhost:9200"],
    index="logs",
    query={"match_all": {}},
)

async with source:
    lf = await source.to_polars_lazyframe_async()

Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Streaming 플랫폼¶

KafkaDataSource (Async)¶

from truthound.datasources import from_kafka

source = await from_kafka(
    bootstrap_servers="localhost:9092",
    topic="events",
    group_id="truthound-validators",
)

async with source:
    async for batch in source.consume_batches(batch_size=1000):
        report = th.check(batch)

Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Async Factory Functions¶

Python API 사용에서 MongoDB, Elasticsearch, Kafka을(를) 다루는 항목입니다:

from truthound.datasources import get_async_datasource

# Auto-detect async source type
source = await get_async_datasource(
    "mongodb://localhost:27017",
    database="mydb",
    collection="users",
)

# Check if source is async
from truthound.datasources import is_async_source, is_sync_source

if is_async_source(source):
    async with source:
        lf = await source.to_polars_lazyframe_async()

함께 보기¶

Python API 사용에서 Data, Sources, Guide, Detailed을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 DataSources, Architecture, Technical을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 Performance, Guide, Optimization을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Python API 사용에서 Category을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Data 소스	Python API 사용에서 Required, Package을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 Install, Command을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 DataFrame을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 Polars을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 Pandas을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `pandas`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `pip install truthound[pandas]`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 Spark을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `pyspark`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `pip install truthound[spark]`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 SQL, Core을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 SQLite, SQL을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 DuckDB을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `duckdb`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `pip install truthound[duckdb]`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 PostgreSQL, SQL을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `psycopg2`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `pip install truthound[postgresql]`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	MySQL	`pymysql`	`pip install truthound[mysql]`
Python API 사용에서 Cloud을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 BigQuery을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `google-cloud-bigquery`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `pip install truthound[bigquery]`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 Snowflake을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `snowflake-connector-python`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `pip install truthound[snowflake]`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 Redshift을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `redshift-connector`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `pip install truthound[redshift]`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 Databricks을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `databricks-sql-connector`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `pip install truthound[databricks]`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 Enterprise을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 Oracle을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `oracledb`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `pip install truthound[oracle]`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	SQL Server	`pymssql` (`pyodbc`도 사용 가능)	`pip install truthound[sqlserver]`
Python API 사용에서 SQL, NoSQL을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 MongoDB을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `pymongo`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `pip install truthound[mongodb]`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 Elasticsearch을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `elasticsearch`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `pip install truthound[elasticsearch]`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 Streaming을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 Kafka을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `confluent-kafka`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `pip install truthound[kafka]`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 Kinesis을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `aiobotocore`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `pip install truthound[kinesis]`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
파일	Python API 사용에서 JSON, CSV, Parquet, NDJSON, JSONL을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.

Python API 사용에서 Category을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Data 소스	소스 파일
Python API 사용에서 DataFrame을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 Polars을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/polars_source.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 Pandas을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/pandas_source.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 Spark을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/spark_source.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 SQL, Core을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 SQLite, SQL을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/sql/sqlite.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 DuckDB을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/sql/duckdb.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 PostgreSQL, SQL을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/sql/postgresql.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 MySQL, SQL을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/sql/mysql.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 Cloud을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 BigQuery을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/sql/bigquery.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 Snowflake을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/sql/snowflake.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 Redshift을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/sql/redshift.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 Databricks을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/sql/databricks.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 Enterprise을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 Oracle을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/sql/oracle.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 SQL, Server을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/sql/sqlserver.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 SQL, NoSQL을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 MongoDB을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/nosql/mongodb.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 Elasticsearch을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/nosql/elasticsearch.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
Python API 사용에서 Streaming을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 Kafka을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/streaming/kafka.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
	Python API 사용에서 Kinesis을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/streaming/kinesis.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.
파일	Python API 사용에서 관련 설정과 실행 흐름을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.	Python API 사용에서 `datasources/adapters.py`을(를) 기준으로 데이터 품질 검증, 워크플로우 자동화, 결과 해석 방법을 설명합니다.