Execute Stage¶

The execute stage runs queries and retrieves data for compiled dashboards.

CompiledBoard + Variables → Executor → Query Results (List[Dict])

Entry Points¶

Executor¶

The main class for executing queries:

Executor ¶

Executor(board: CompiledBoard, adapter_registry: Optional[Any] = None, query_registry: Optional[Dict[str, CompiledQuery]] = None)

Executes queries for dashboards.

Stage: EXECUTE (Service Module)

The executor manages query execution for a compiled dashboard. It handles: - Query lookup from the board - Adapter selection based on query type - Result caching for efficiency - Variable substitution

Does NOT: - Compile dashboards (use compile module) - Render charts (use render module)

ATTRIBUTE	DESCRIPTION
`board`	The compiled dashboard
`adapter_registry`	Registry of adapters (optional)
`query_registry`	Complete query registry (for cross-references)

Example

from dataface.compile import compile from dataface.execute import Executor

result = compile(yaml_content) executor = Executor(result.board, query_registry=result.query_registry)

Execute a query¶

data = executor.execute_query("sales", {"year": 2024}) print(data) # [{"date": "2024-01", "amount": 1000}, ...]

Initialize executor.

PARAMETER	DESCRIPTION
`board`	Compiled dashboard TYPE: `CompiledBoard`
`adapter_registry`	Optional adapter registry (uses default if not provided) TYPE: `Optional[Any]` DEFAULT: `None`
`query_registry`	Optional query registry for cross-file references TYPE: `Optional[Dict[str, CompiledQuery]]` DEFAULT: `None`

Source code in dataface/execute/executor.py

def __init__(
    self,
    board: CompiledBoard,
    adapter_registry: Optional[Any] = None,
    query_registry: Optional[Dict[str, CompiledQuery]] = None,
):
    """Initialize executor.

    Args:
        board: Compiled dashboard
        adapter_registry: Optional adapter registry (uses default if not provided)
        query_registry: Optional query registry for cross-file references
    """
    self.board = board
    self.query_registry = query_registry or {}
    self._cache: Dict[str, List[Dict[str, Any]]] = {}

    # Import adapter registry lazily to avoid circular imports
    if adapter_registry is None:
        from dataface.execute.adapters import AdapterRegistry

        self.adapter_registry = AdapterRegistry()
    else:
        self.adapter_registry = adapter_registry

execute_query ¶

execute_query(query_name: str, variables: Optional[VariableValues] = None, use_cache: bool = True) -> List[Dict[str, Any]]

Execute a query and return results.

Stage: EXECUTE (Main Entry Point)

Looks up the query by name, resolves the appropriate adapter, executes the query, and returns the data.

PARAMETER	DESCRIPTION
`query_name`	Name of query to execute (may include "queries." prefix) TYPE: `str`
`variables`	Variable values for query resolution TYPE: `Optional[VariableValues]` DEFAULT: `None`
`use_cache`	Whether to use cached results if available TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`List[Dict[str, Any]]`	List of dictionaries with query results (each dict is a row)

RAISES	DESCRIPTION
`ExecutionError`	If query not found or execution fails

Example

data = executor.execute_query("sales", {"year": 2024}) for row in data: ... print(f"{row['date']}: {row['amount']}")

Source code in dataface/execute/executor.py

def execute_query(
    self,
    query_name: str,
    variables: Optional[VariableValues] = None,
    use_cache: bool = True,
) -> List[Dict[str, Any]]:
    """Execute a query and return results.

    Stage: EXECUTE (Main Entry Point)

    Looks up the query by name, resolves the appropriate adapter,
    executes the query, and returns the data.

    Args:
        query_name: Name of query to execute (may include "queries." prefix)
        variables: Variable values for query resolution
        use_cache: Whether to use cached results if available

    Returns:
        List of dictionaries with query results (each dict is a row)

    Raises:
        ExecutionError: If query not found or execution fails

    Example:
        >>> data = executor.execute_query("sales", {"year": 2024})
        >>> for row in data:
        ...     print(f"{row['date']}: {row['amount']}")
    """
    # ────────────────────────────────────────────────────────────────
    # Step 1: Normalize query name (strip "queries." prefix)
    # ────────────────────────────────────────────────────────────────
    if query_name.startswith("queries."):
        query_name = query_name[8:]

    # ────────────────────────────────────────────────────────────────
    # Step 1b: Merge variables with board defaults
    # ────────────────────────────────────────────────────────────────
    # Build variable registry from board tree
    from dataface.render.variable_registry import (
        build_variable_registry_from_board,
        build_variable_types_from_registry,
        build_variable_defaults_from_registry,
    )

    variable_registry = build_variable_registry_from_board(self.board)
    variable_types = build_variable_types_from_registry(variable_registry)
    variable_defaults = build_variable_defaults_from_registry(variable_registry)

    # Start with all variables from global registry (set to None if no default)
    # This ensures all variables are in context for template resolution
    all_variables: Dict[str, Any] = {}
    for var_name in variable_types.keys():
        all_variables[var_name] = None
    # Override with defaults
    all_variables.update(variable_defaults)
    # Override with provided values
    # Parse JSON strings in variables (from URL parameters)
    parsed_variables = _parse_variable_json_strings(variables or {})

    # Validate variables against their expected types
    validated_variables, validation_errors = _validate_variables(
        parsed_variables, variable_types
    )
    if validation_errors:
        # Log validation errors but don't fail execution
        import warnings

        for error in validation_errors:
            warnings.warn(f"Variable validation: {error}", UserWarning)

    merged_variables = {**all_variables, **validated_variables}
    variables = merged_variables

    # ────────────────────────────────────────────────────────────────
    # Step 2: Look up query
    # ────────────────────────────────────────────────────────────────
    query = self._get_query(query_name)

    # ────────────────────────────────────────────────────────────────
    # Step 3: Check cache
    # ────────────────────────────────────────────────────────────────
    cache_key = self._cache_key(query_name, variables)
    if use_cache and cache_key in self._cache:
        return self._cache[cache_key]

    # ────────────────────────────────────────────────────────────────
    # Step 4: Resolve query references in SQL
    # ────────────────────────────────────────────────────────────────
    query = self._resolve_query_references(query, variables)

    # ────────────────────────────────────────────────────────────────
    # Step 5: Execute via adapter
    # ────────────────────────────────────────────────────────────────
    try:
        result = self.adapter_registry.execute(query, variables)
    except Exception as e:
        raise QueryError(str(e), query_name)

    if not result.is_success:
        raise QueryError(result.error or "Unknown error", query_name)

    # ────────────────────────────────────────────────────────────────
    # Step 6: Cache and return
    # ────────────────────────────────────────────────────────────────
    if use_cache:
        self._cache[cache_key] = result.data

    return result.data

execute_chart ¶

execute_chart(chart: Union[CompiledChart, str], variables: Optional[VariableValues] = None, use_cache: bool = True) -> List[Dict[str, Any]]

Execute the query for a chart.

Convenience method that handles chart → query lookup.

PARAMETER	DESCRIPTION
`chart`	CompiledChart or chart name string TYPE: `Union[CompiledChart, str]`
`variables`	Variable values TYPE: `Optional[VariableValues]` DEFAULT: `None`
`use_cache`	Whether to use cache TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`List[Dict[str, Any]]`	Query results for the chart

Example

chart = board.charts["revenue"] data = executor.execute_chart(chart, {"year": 2024})

Source code in dataface/execute/executor.py

def execute_chart(
    self,
    chart: Union[CompiledChart, str],
    variables: Optional[VariableValues] = None,
    use_cache: bool = True,
) -> List[Dict[str, Any]]:
    """Execute the query for a chart.

    Convenience method that handles chart → query lookup.

    Args:
        chart: CompiledChart or chart name string
        variables: Variable values
        use_cache: Whether to use cache

    Returns:
        Query results for the chart

    Example:
        >>> chart = board.charts["revenue"]
        >>> data = executor.execute_chart(chart, {"year": 2024})
    """
    if isinstance(chart, str):
        if chart not in self.board.charts:
            raise ExecutionError(f"Chart '{chart}' not found")
        chart = self.board.charts[chart]

    # Get data
    data = self.execute_query(chart.query_name, variables, use_cache)

    # Apply chart-level filters if present
    if chart.filters:
        data = self._apply_filters(data, chart.filters, variables)

    return data

Adapters¶

Adapters connect to different data sources. Each query type has a corresponding adapter.

Base Adapter¶

BaseAdapter ¶

Bases: ABC

Base interface for query adapters.

All adapters must implement this interface to execute queries against different backends (MetricFlow, SQL, HTTP, CSV, etc.).

The unified interface pattern uses: - supported_types: Property returning set of query types this adapter handles - Type guards for type-safe field access in execute methods

Subclasses must implement

supported_types: Property returning Set[str] of supported query types
_execute(): Perform the actual query execution

Optional override

_can_execute(): Override for custom eligibility logic beyond type matching

Example

class MyAdapter(BaseAdapter): ... @property ... def supported_types(self) -> Set[str]: ... return {"sql"} ... ... def _execute(self, query, variables): ... if is_sql_query(query): ... # Type checker knows query.sql exists ... return self._run_sql(query.sql)

supported_types `abstractmethod` `property` ¶

supported_types: Set[str]

Query types this adapter can execute.

RETURNS	DESCRIPTION
`Set[str]`	Set of query type strings (e.g., {"sql"}, {"csv", "http"})

Example

adapter.supported_types

can_execute ¶

can_execute(query: CompiledQuery) -> bool

Check if this adapter can execute the given query.

Uses supported_types for type-based routing. Override _can_execute() for additional custom eligibility logic.

PARAMETER	DESCRIPTION
`query`	CompiledQuery object (guaranteed by compiler) TYPE: `CompiledQuery`

RETURNS	DESCRIPTION
`bool`	True if this adapter can execute the query, False otherwise

Source code in dataface/execute/adapters/base.py

def can_execute(self, query: CompiledQuery) -> bool:
    """Check if this adapter can execute the given query.

    Uses supported_types for type-based routing. Override _can_execute()
    for additional custom eligibility logic.

    Args:
        query: CompiledQuery object (guaranteed by compiler)

    Returns:
        True if this adapter can execute the query, False otherwise
    """
    # First check if type is supported
    if query.query_type not in self.supported_types:
        return False
    # Then check any additional conditions
    return self._can_execute(query)

execute ¶

execute(query: CompiledQuery, variables: Optional[VariableValues] = None) -> QueryResult

Execute a query and return results.

PARAMETER	DESCRIPTION
`query`	CompiledQuery object (guaranteed by compiler) TYPE: `CompiledQuery`
`variables`	Optional dictionary of variable values for query resolution TYPE: `Optional[VariableValues]` DEFAULT: `None`

RETURNS	DESCRIPTION
`QueryResult`	QueryResult containing data or error information

Source code in dataface/execute/adapters/base.py

def execute(
    self,
    query: CompiledQuery,
    variables: Optional[VariableValues] = None,
) -> QueryResult:
    """Execute a query and return results.

    Args:
        query: CompiledQuery object (guaranteed by compiler)
        variables: Optional dictionary of variable values for query resolution

    Returns:
        QueryResult containing data or error information
    """
    return self._execute(query, variables)

SQL Adapter¶

For executing SQL queries against databases (DuckDB, PostgreSQL, etc.):

SqlAdapter ¶

SqlAdapter(dbt_project_path: Optional[str] = None, use_example_db: bool = False, connection_string: Optional[str] = None)

Bases: BaseAdapter

Adapter for executing raw SQL queries via dbt's adapter system.

This adapter leverages dbt's adapter API to execute SQL queries. It uses the user's existing dbt configuration (profiles.yml) for database connections, supporting all dbt-compatible databases.

Supported query types: sql

Example

adapter = SqlAdapter() query = SqlQuery(sql="SELECT * FROM users WHERE id = {{ user_id }}") result = adapter.execute(query, {"user_id": 1})

Initialize SQL adapter.

PARAMETER	DESCRIPTION
`dbt_project_path`	Path to dbt project (default: current directory) TYPE: `Optional[str]` DEFAULT: `None`
`use_example_db`	If True, use DuckDB example database instead of dbt config TYPE: `bool` DEFAULT: `False`
`connection_string`	DuckDB connection string (e.g., ":memory:" or file path) TYPE: `Optional[str]` DEFAULT: `None`

Source code in dataface/execute/adapters/sql_adapter.py

def __init__(
    self,
    dbt_project_path: Optional[str] = None,
    use_example_db: bool = False,
    connection_string: Optional[str] = None,
):
    """Initialize SQL adapter.

    Args:
        dbt_project_path: Path to dbt project (default: current directory)
        use_example_db: If True, use DuckDB example database instead of dbt config
        connection_string: DuckDB connection string (e.g., ":memory:" or file path)
    """
    self.dbt_project_path = Path(dbt_project_path) if dbt_project_path else None
    self.use_example_db = use_example_db
    self.connection_string = connection_string or ":memory:"
    self._adapter: Any = None  # Lazy-loaded dbt adapter
    self._connection: Any = None  # Lazy-loaded DuckDB connection
    self._manifest: Optional[dict] = None  # Lazy-loaded dbt manifest

supported_types `property` ¶

supported_types: Set[str]

Return supported query types.

CSV Adapter¶

For loading data from CSV files:

CsvAdapter ¶

CsvAdapter(project_root: Optional[Path] = None)

Bases: BaseAdapter

Adapter for executing queries against CSV files.

Supported query types: csv

Loads data from CSV files and supports: - Column selection - Row filtering (exact match, AND logic) - Result limiting

Example

adapter = CsvAdapter() query = CsvQuery(file="data/sales.csv", columns=["date", "amount"]) result = adapter.execute(query)

Initialize CSV adapter.

PARAMETER	DESCRIPTION
`project_root`	Root directory of the project (for resolving relative paths) TYPE: `Optional[Path]` DEFAULT: `None`

Source code in dataface/execute/adapters/csv_adapter.py

def __init__(self, project_root: Optional[Path] = None):
    """Initialize CSV adapter.

    Args:
        project_root: Root directory of the project (for resolving relative paths)
    """
    self.project_root = project_root or Path.cwd()

supported_types `property` ¶

supported_types: Set[str]

Return supported query types.

HTTP Adapter¶

For fetching data from REST APIs:

HttpAdapter ¶

HttpAdapter(timeout: int = 30)

Bases: BaseAdapter

Adapter for executing HTTP/REST API queries.

Supported query types: http

Fetches data from REST API endpoints with support for: - Multiple HTTP methods (GET, POST, PUT, DELETE, PATCH) - Custom headers - Query parameters - Request body (JSON)

Example

adapter = HttpAdapter() query = HttpQuery( ... url="https://api.example.com/users", ... headers={"Authorization": "Bearer {{ token }}"} ... ) result = adapter.execute(query, {"token": "xyz"})

Initialize HTTP adapter.

PARAMETER	DESCRIPTION
`timeout`	Request timeout in seconds TYPE: `int` DEFAULT: `30`

Source code in dataface/execute/adapters/http_adapter.py

def __init__(self, timeout: int = 30):
    """Initialize HTTP adapter.

    Args:
        timeout: Request timeout in seconds
    """
    self.timeout = timeout

supported_types `property` ¶

supported_types: Set[str]

Return supported query types.

dbt Adapter¶

For integrating with dbt models:

DbtAdapter ¶

DbtAdapter(dbt_project_path: Optional[Path] = None, profile_name: Optional[str] = None, target_name: Optional[str] = None)

Bases: BaseAdapter

Adapter for executing SQL queries via dbt's adapter system.

Supported query types: sql

Leverages dbt's adapter API to execute SQL queries with support for dbt-specific features like ref(), source(), etc.

This adapter requires dbt-core to be installed and a valid dbt project with profiles.yml configuration.

Example

adapter = DbtAdapter(dbt_project_path=Path("./my_dbt_project")) query = SqlQuery(sql="SELECT * FROM {{ ref('customers') }}") result = adapter.execute(query)

Initialize dbt adapter.

PARAMETER	DESCRIPTION
`dbt_project_path`	Path to dbt project (default: current directory) TYPE: `Optional[Path]` DEFAULT: `None`
`profile_name`	dbt profile name (default: from dbt_project.yml) TYPE: `Optional[str]` DEFAULT: `None`
`target_name`	dbt target name (default: 'dev') TYPE: `Optional[str]` DEFAULT: `None`

Source code in dataface/execute/adapters/dbt_adapter.py

def __init__(
    self,
    dbt_project_path: Optional[Path] = None,
    profile_name: Optional[str] = None,
    target_name: Optional[str] = None,
):
    """Initialize dbt adapter.

    Args:
        dbt_project_path: Path to dbt project (default: current directory)
        profile_name: dbt profile name (default: from dbt_project.yml)
        target_name: dbt target name (default: 'dev')
    """
    self.dbt_project_path = (
        Path(dbt_project_path).resolve()
        if dbt_project_path
        else Path.cwd().resolve()
    )
    self.profile_name = profile_name
    self.target_name = target_name or "dev"
    self._adapter: Any = None
    self._manifest: Optional[Dict[str, Any]] = None

supported_types `property` ¶

supported_types: Set[str]

Return supported query types.

MetricFlow Adapter¶

For querying MetricFlow metrics:

MetricFlowAdapter ¶

MetricFlowAdapter(config_path: Optional[str] = None)

Bases: BaseAdapter

Adapter for executing MetricFlow (dbt Semantic Layer) queries.

Supported query types: metricflow

Executes queries against dbt's Semantic Layer using MetricFlow. In a full implementation, this uses the MetricFlow Python API or CLI.

Example

adapter = MetricFlowAdapter() query = MetricFlowQuery( ... metrics=["revenue", "orders"], ... dimensions=["date", "region"] ... ) result = adapter.execute(query)

Initialize MetricFlow adapter.

PARAMETER	DESCRIPTION
`config_path`	Optional path to MetricFlow configuration TYPE: `Optional[str]` DEFAULT: `None`

Source code in dataface/execute/adapters/metricflow_adapter.py

def __init__(self, config_path: Optional[str] = None):
    """Initialize MetricFlow adapter.

    Args:
        config_path: Optional path to MetricFlow configuration
    """
    self.config_path = config_path

supported_types `property` ¶

supported_types: Set[str]

Return supported query types.

Adapter Registry¶

AdapterRegistry ¶

AdapterRegistry()

Registry for managing and selecting query adapters.

The registry uses type-based routing from the unified query interface. Adapters declare their supported types via the supported_types property, and the registry routes queries to the appropriate adapter.

Priority order: 1. First registered adapter that supports the query type wins 2. dbt adapter is registered first for SQL (with dbt features) 3. Fallback SQL adapter for plain SQL without dbt

Example

registry = AdapterRegistry() query = SqlQuery(sql="SELECT 1") result = registry.execute(query)

Initialize adapter registry with default adapters.

Source code in dataface/execute/adapters/adapter_registry.py

def __init__(self) -> None:
    """Initialize adapter registry with default adapters."""
    self._adapters: List[BaseAdapter] = []
    self._type_index: Dict[str, List[BaseAdapter]] = {}
    self._register_defaults()

supported_types `property` ¶

supported_types: Set[str]

Get all query types supported by registered adapters.

RETURNS	DESCRIPTION
`Set[str]`	Set of query type strings

register ¶

register(adapter: BaseAdapter) -> None

Register an adapter.

Adapters are indexed by their supported types for fast lookup.

PARAMETER	DESCRIPTION
`adapter`	Adapter instance to register TYPE: `BaseAdapter`

Source code in dataface/execute/adapters/adapter_registry.py

def register(self, adapter: BaseAdapter) -> None:
    """Register an adapter.

    Adapters are indexed by their supported types for fast lookup.

    Args:
        adapter: Adapter instance to register
    """
    self._adapters.append(adapter)

    # Update type index
    for query_type in adapter.supported_types:
        if query_type not in self._type_index:
            self._type_index[query_type] = []
        self._type_index[query_type].append(adapter)

get_adapters_for_type ¶

get_adapters_for_type(query_type: str) -> List[BaseAdapter]

Get all adapters that support a given query type.

PARAMETER	DESCRIPTION
`query_type`	Query type string (e.g., "sql", "csv") TYPE: `str`

RETURNS	DESCRIPTION
`List[BaseAdapter]`	List of adapters supporting this type (in registration order)

Source code in dataface/execute/adapters/adapter_registry.py

def get_adapters_for_type(self, query_type: str) -> List[BaseAdapter]:
    """Get all adapters that support a given query type.

    Args:
        query_type: Query type string (e.g., "sql", "csv")

    Returns:
        List of adapters supporting this type (in registration order)
    """
    return self._type_index.get(query_type, [])

get_adapter ¶

get_adapter(query: CompiledQuery) -> Optional[BaseAdapter]

Get an adapter that can execute the given query.

Uses type-based routing for fast lookup, then checks additional eligibility via can_execute().

PARAMETER	DESCRIPTION
`query`	CompiledQuery object TYPE: `CompiledQuery`

RETURNS	DESCRIPTION
`Optional[BaseAdapter]`	Adapter instance or None if no adapter can execute the query

Source code in dataface/execute/adapters/adapter_registry.py

def get_adapter(self, query: CompiledQuery) -> Optional[BaseAdapter]:
    """Get an adapter that can execute the given query.

    Uses type-based routing for fast lookup, then checks additional
    eligibility via can_execute().

    Args:
        query: CompiledQuery object

    Returns:
        Adapter instance or None if no adapter can execute the query
    """
    # Fast path: lookup by type
    candidates = self.get_adapters_for_type(query.query_type)

    for adapter in candidates:
        if adapter.can_execute(query):
            return adapter

    # Fallback: check all adapters (for adapters with custom can_execute)
    for adapter in self._adapters:
        if adapter not in candidates and adapter.can_execute(query):
            return adapter

    return None

execute ¶

execute(query: CompiledQuery, variables: Optional[VariableValues] = None) -> QueryResult

Execute a query using the appropriate adapter.

PARAMETER	DESCRIPTION
`query`	CompiledQuery object TYPE: `CompiledQuery`
`variables`	Variable values for query resolution TYPE: `Optional[VariableValues]` DEFAULT: `None`

RETURNS	DESCRIPTION
`QueryResult`	QueryResult with data or error

Source code in dataface/execute/adapters/adapter_registry.py

def execute(
    self,
    query: CompiledQuery,
    variables: Optional[VariableValues] = None,
) -> QueryResult:
    """Execute a query using the appropriate adapter.

    Args:
        query: CompiledQuery object
        variables: Variable values for query resolution

    Returns:
        QueryResult with data or error
    """
    adapter = self.get_adapter(query)
    if not adapter:
        return QueryResult(
            data=[],
            error=f"No adapter found for query type: {query.query_type}",
        )

    # For SQL queries with profile/target, create a new adapter instance
    # with the correct profile/target configuration
    if query.query_type == "sql":
        from dataface.compile.query_types import is_sql_query
        from dataface.execute.adapters.dbt_adapter import DbtAdapter
        from pathlib import Path

        if is_sql_query(query):
            # Check if adapter needs profile/target configuration
            if hasattr(query, "profile") and query.profile:
                # Create a new DbtAdapter with the specified profile/target
                target = getattr(query, "target", None) or "dev"
                dbt_project_path = (
                    getattr(adapter, "dbt_project_path", None) or Path.cwd()
                )

                # Create adapter with specific profile/target
                profile_adapter = DbtAdapter(
                    dbt_project_path=dbt_project_path,
                    profile_name=query.profile,
                    target_name=target,
                )
                return profile_adapter.execute(query, variables)

    return adapter.execute(query, variables)

Errors¶

errors ¶

Execution error types.

Stage: EXECUTE Purpose: Define error types for query execution failures.

These errors are raised during: - Query execution (QueryError) - Adapter resolution/execution (AdapterError) - Connection failures (ConnectionError)

All errors inherit from ExecutionError for easy catching.

ExecutionError ¶

ExecutionError(message: str, query_name: Optional[str] = None)

Bases: Exception

Base error for all execution failures.

This is the parent class for all execution-related errors. Catch this to handle any execution error.

ATTRIBUTE	DESCRIPTION
`message`	Human-readable error description
`query_name`	Name of query that failed (if applicable)

Source code in dataface/execute/errors.py

def __init__(self, message: str, query_name: Optional[str] = None):
    self.message = message
    self.query_name = query_name
    super().__init__(self._format_message())

QueryError ¶

QueryError(message: str, query_name: Optional[str] = None, sql: Optional[str] = None)

Bases: ExecutionError

Error during query execution.

Raised when: - SQL syntax is invalid - Table/column doesn't exist - Query returns unexpected results

Example

try: ... executor.execute_query("broken_query") ... except QueryError as e: ... print(f"Query failed: {e}")

Source code in dataface/execute/errors.py

def __init__(
    self, message: str, query_name: Optional[str] = None, sql: Optional[str] = None
):
    self.sql = sql
    super().__init__(f"Query execution failed: {message}", query_name)

AdapterError ¶

AdapterError(message: str, adapter_type: Optional[str] = None)

Bases: ExecutionError

Error with adapter resolution or execution.

Raised when: - Adapter not found for query type - Adapter initialization fails - Adapter-specific error occurs

Example

try: ... executor.execute_query("unknown_type_query") ... except AdapterError as e: ... print(f"Adapter error: {e}")

Source code in dataface/execute/errors.py

def __init__(self, message: str, adapter_type: Optional[str] = None):
    self.adapter_type = adapter_type
    super().__init__(f"Adapter error: {message}")

Execute Stage¶

Entry Points¶

Executor¶

Executor ¶

Execute a query¶

execute_query ¶

execute_chart ¶

Adapters¶

Base Adapter¶

BaseAdapter ¶

supported_types abstractmethod property ¶

can_execute ¶

execute ¶

SQL Adapter¶

SqlAdapter ¶

supported_types property ¶

CSV Adapter¶

CsvAdapter ¶

supported_types property ¶

HTTP Adapter¶

HttpAdapter ¶

supported_types property ¶

dbt Adapter¶

DbtAdapter ¶

supported_types property ¶

MetricFlow Adapter¶

MetricFlowAdapter ¶

supported_types property ¶

Adapter Registry¶

AdapterRegistry ¶

supported_types property ¶

register ¶

get_adapters_for_type ¶

get_adapter ¶

execute ¶

Errors¶

errors ¶

ExecutionError ¶

QueryError ¶

AdapterError ¶

supported_types `abstractmethod` `property` ¶

supported_types `property` ¶

supported_types `property` ¶

supported_types `property` ¶

supported_types `property` ¶

supported_types `property` ¶

supported_types `property` ¶