fastflowtransform.config.budgets

BudgetLimit

Bases: BaseModel

Thresholds for a single metric.

After preprocessing, values are integers (e.g. bytes, rows, ms). You can use either bare numbers or numeric strings:

warn: 5000000000
error: "10_000_000_000"

For query_duration_ms only, we additionally support: "10m", "30s", "2h", "1d", "250ms" (these are converted to ms before validation).

Source code in src/fastflowtransform/config/budgets.py

class BudgetLimit(BaseModel):
    """
    Thresholds for a single metric.

    After preprocessing, values are integers (e.g. bytes, rows, ms). You
    can use either bare numbers or numeric strings:

        warn: 5000000000
        error: "10_000_000_000"

    For query_duration_ms only, we additionally support:
        "10m", "30s", "2h", "1d", "250ms"
    (these are converted to ms before validation).
    """

    model_config = ConfigDict(extra="forbid")

    warn: int | None = None
    error: int | None = None

    @field_validator("warn", "error", mode="before")
    @classmethod
    def _normalize_int(cls, v: Any) -> int | None:
        if v is None:
            return None
        if isinstance(v, (int, float)):
            iv = int(v)
            return iv if iv > 0 else None
        if isinstance(v, str):
            text = v.strip().replace("_", "").replace(",", "")
            if not text:
                return None
            if not text.isdigit():
                # At this point we've already tried to parse duration strings
                # for query_duration_ms; non-numeric leftovers here are an error.
                raise ValueError(f"budget limits must be integers or numeric strings, got {v!r}")
            iv = int(text)
            return iv if iv > 0 else None
        raise TypeError("budget limits must be integers or strings")

BudgetMetrics

Bases: BaseModel

Metrics we can budget against; all are optional.

bytes_scanned → sum of bytes across all SQL queries rows → sum of rows across all SQL queries query_duration_ms → sum of query durations (ms), not wall-clock

Source code in src/fastflowtransform/config/budgets.py

class BudgetMetrics(BaseModel):
    """
    Metrics we can budget against; all are optional.

      bytes_scanned     → sum of bytes across all SQL queries
      rows              → sum of rows across all SQL queries
      query_duration_ms → sum of query durations (ms), not wall-clock
    """

    model_config = ConfigDict(extra="forbid")

    bytes_scanned: BudgetLimit | None = None
    rows: BudgetLimit | None = None
    query_duration_ms: BudgetLimit | None = None

QueryLimitConfig

Bases: BaseModel

Per-engine query limit configuration.

Currently only max_bytes is supported.

Source code in src/fastflowtransform/config/budgets.py

class QueryLimitConfig(BaseModel):
    """
    Per-engine query limit configuration.

    Currently only `max_bytes` is supported.
    """

    model_config = ConfigDict(extra="forbid")

    max_bytes: int | None = None

    @field_validator("max_bytes", mode="before")
    @classmethod
    def _normalize_int(cls, v: Any) -> int | None:
        return BudgetLimit._normalize_int(v)

BudgetsConfig

Bases: BaseModel

Strict representation of budgets.yml.

Example:

version: 1

# Global (across all models in fft run)
total:
  bytes_scanned:
    warn: 5000000000
    error: 10000000000

# Per model limits
models:
  fct_events:
    bytes_scanned:
      warn: 1000000000
      error: 2000000000

# Per tag limits (aggregated over all models with that tag)
tags:
  heavy:
    bytes_scanned:
      warn: 5000000000
      error: 8000000000

# Optional per-engine query guard limits
query_limits:
  duckdb:
    max_bytes: 2000000000

Source code in src/fastflowtransform/config/budgets.py

class BudgetsConfig(BaseModel):
    """
    Strict representation of budgets.yml.

    Example:

        version: 1

        # Global (across all models in fft run)
        total:
          bytes_scanned:
            warn: 5000000000
            error: 10000000000

        # Per model limits
        models:
          fct_events:
            bytes_scanned:
              warn: 1000000000
              error: 2000000000

        # Per tag limits (aggregated over all models with that tag)
        tags:
          heavy:
            bytes_scanned:
              warn: 5000000000
              error: 8000000000

        # Optional per-engine query guard limits
        query_limits:
          duckdb:
            max_bytes: 2000000000
    """

    model_config = ConfigDict(extra="forbid")

    version: int = 1

    total: BudgetMetrics | None = None
    models: dict[str, BudgetMetrics] = Field(default_factory=dict)
    tags: dict[str, BudgetMetrics] = Field(default_factory=dict)
    query_limits: dict[str, QueryLimitConfig] = Field(default_factory=dict)

load_budgets_config

load_budgets_config(project_dir)

Read budgets.yml under project_dir and validate it strictly.

Missing file → returns None (no budgets enforced). Invalid file → raises, caller should wrap into a user-friendly error.

Source code in src/fastflowtransform/config/budgets.py

def load_budgets_config(project_dir: Path) -> BudgetsConfig | None:
    """
    Read budgets.yml under `project_dir` and validate it strictly.

    Missing file → returns None (no budgets enforced).
    Invalid file → raises, caller should wrap into a user-friendly error.
    """
    project_dir = Path(project_dir)
    cfg_path = project_dir / "budgets.yml"
    if not cfg_path.exists():
        return None

    raw = yaml.safe_load(cfg_path.read_text(encoding="utf-8")) or {}
    raw = _normalize_duration_limits(raw)
    return BudgetsConfig.model_validate(raw)