fastflowtransform.settings

load_profiles

load_profiles(project_dir)

Read project.yml/profiles.yml and return a dict per environment. Expected format: { "dev": {"engine":"duckdb", "duckdb":{"path":":memory:"}}, ... }

Source code in src/fastflowtransform/settings.py

def load_profiles(project_dir: Path) -> dict:
    """
    Read project.yml/profiles.yml and return a dict per environment.
    Expected format:
      { "dev": {"engine":"duckdb", "duckdb":{"path":":memory:"}}, ... }
    """
    pf_path = project_dir / "profiles.yml"
    if not pf_path.exists():
        return {}
    raw_text = pf_path.read_text(encoding="utf-8")
    rendered = _render_profiles_template(raw_text, project_dir)
    return yaml.safe_load(rendered) or {}

resolve_utest_profile

resolve_utest_profile(project_dir, base_env_name, env)

Resolve the utest profile for a given base env, e.g. base_env_name = "dev_duckdb" -> profile "dev_duckdb_utest"

Requirements

• base profile (base_env_name) is resolved with env overrides (FF_*).
• utest profile (base_env_name + "_utest") is resolved from profiles.yml WITHOUT env overrides so it cannot accidentally point at the same DB/schema.
• utest profile MUST exist and MUST be isolated from the base one.

Source code in src/fastflowtransform/settings.py

def resolve_utest_profile(project_dir: Path, base_env_name: str, env: EnvSettings) -> Profile:
    """
    Resolve the *utest* profile for a given base env, e.g.
      base_env_name = "dev_duckdb"  -> profile "dev_duckdb_utest"

    Requirements:
      - base profile (base_env_name) is resolved with env overrides (FF_*).
      - utest profile (base_env_name + "_utest") is resolved from profiles.yml
        WITHOUT env overrides so it cannot accidentally point at the same DB/schema.
      - utest profile MUST exist and MUST be isolated from the base one.
    """
    profiles: dict[str, dict[str, Any]] = load_profiles(project_dir)

    # 1) Resolve the *base* profile normally (with env overrides).
    base_prof: Profile = resolve_profile(project_dir, base_env_name, env)

    # 2) Load the raw utest profile from YAML (no env overrides here).
    utest_env_name = f"{base_env_name}_utest"
    raw_utest = profiles.get(utest_env_name)

    if raw_utest is None:
        raise ProfileConfigError(
            f"Unit-test profile '{utest_env_name}' not found in profiles.yml. "
            f"Define it explicitly to run 'fft utest' for env '{base_env_name}'."
        )

    # Work on a copy and DO NOT call _apply_env_overrides().
    raw_utest_copy = deepcopy(raw_utest)

    # --- Inherit SAFE connection fields from base profile ----------------
    # The idea:
    #   - we inherit things that *do not define isolation*
    #     (e.g. DSN, project, account, warehouse, etc.)
    #   - but we DO NOT inherit things like schema/dataset/path that we want to be different.
    eng = base_prof.engine

    if eng == "postgres":
        base_pg = cast(PostgresProfile, base_prof)
        base_dsn = base_pg.postgres.dsn
        if base_dsn:
            pg_cfg = raw_utest_copy.setdefault("postgres", {})
            pg_cfg.setdefault("dsn", base_dsn)

    elif eng == "bigquery":
        base_bq = cast(BigQueryProfile, base_prof)
        bq_cfg = raw_utest_copy.setdefault("bigquery", {})

        # Safe to inherit: project & location & allow_create_dataset
        if base_bq.bigquery.project is not None:
            bq_cfg.setdefault("project", base_bq.bigquery.project)
        if base_bq.bigquery.location is not None:
            bq_cfg.setdefault("location", base_bq.bigquery.location)
        # dataset is the isolation dimension → MUST be set explicitly in the utest profile
        # and will be checked by _assert_utest_isolated (in CLI/bootstrap).
        if "allow_create_dataset" in base_bq.bigquery.__dict__:
            bq_cfg.setdefault("allow_create_dataset", base_bq.bigquery.allow_create_dataset)

    elif eng == "duckdb":
        base_ddb = cast(DuckDBProfile, base_prof)
        ddb_cfg = raw_utest_copy.setdefault("duckdb", {})
        # Safe-ish to inherit catalog; we do NOT inherit path (isolation) or schema
        if base_ddb.duckdb.catalog is not None:
            ddb_cfg.setdefault("catalog", base_ddb.duckdb.catalog)
        # path & db_schema must be explicitly configured for the utest profile.

    elif eng == "databricks_spark":
        base_dbr = cast(DatabricksSparkProfile, base_prof)
        dbr_cfg = raw_utest_copy.setdefault("databricks_spark", {})
        # Safe to inherit connectivity bits:
        if base_dbr.databricks_spark.master is not None:
            dbr_cfg.setdefault("master", base_dbr.databricks_spark.master)
        if base_dbr.databricks_spark.app_name is not None:
            dbr_cfg.setdefault("app_name", base_dbr.databricks_spark.app_name)
        if base_dbr.databricks_spark.warehouse_dir is not None:
            dbr_cfg.setdefault("warehouse_dir", base_dbr.databricks_spark.warehouse_dir)
        if base_dbr.databricks_spark.catalog is not None:
            dbr_cfg.setdefault("catalog", base_dbr.databricks_spark.catalog)
        # database is the isolation dimension → must differ and will be checked elsewhere.

    elif eng == "snowflake_snowpark":
        base_sf = cast(SnowflakeSnowparkProfile, base_prof)
        sf_cfg = raw_utest_copy.setdefault("snowflake_snowpark", {})
        # Safe to inherit: account/user/password/warehouse/database/role/allow_create_schema
        for attr in (
            "account",
            "user",
            "password",
            "warehouse",
            "database",
            "role",
            "allow_create_schema",
        ):
            val = getattr(base_sf.snowflake_snowpark, attr, None)
            if val is not None:
                sf_cfg.setdefault(attr, val)
        # db_schema (schema) must be explicitly set for the utest profile.

    # 3) Validate the resulting utest profile
    utest_prof: Profile = TypeAdapter(Profile).validate_python(raw_utest_copy)
    _sanity_check_profile(utest_prof)

    return utest_prof