diff --git a/CLAUDE.md b/CLAUDE.md index 638ea7d..0b72871 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -69,6 +69,18 @@ The worktree shares git history with the main repo but has its own `.venv/`. Alw - Use `database_name` instead of `database`, `schema_name` instead of `schema`, `table_name` instead of `table` - Address git merge conflicts by pulling main, resolving conflicts, and pushing. Favour simplicity over clean commit history — PRs are squash-merged anyway. +## Backwards compatibility + +The major version is 0. While the major version stays 0, clair does not keep backwards +compatibility. clair has no users at this time, so the best design wins against a stable +interface. + +- Change a public name, a file format, or a function signature when the change makes the + system better. +- Do not add a deprecation shim, an alias for an old name, or a migration path. +- Delete the old code path. Do not keep it beside the new one. +- Name each behaviour change in the pull request description. + ## Keep site_docs/ up to date `site_docs/docs/` is the source of truth for behaviour. Before you complete a change, compare diff --git a/README.md b/README.md index 41fe57b..e03033c 100644 --- a/README.md +++ b/README.md @@ -138,7 +138,10 @@ uv add rivage-clair ### 1. Set up an environment -Run `clair init` — it will prompt for your Snowflake connection details and write `~/.clair/environments.yml`. +Run `clair init`. It asks for your Snowflake connection details and writes two files: + +- `~/.clair/environments.yml` — your connection settings. Do not commit this file. +- `/__routing__.py` — the [routing](https://clair.rivage.sh/guides/routing/) entry of each environment. Commit this file. ### 2. Create a project diff --git a/site_docs/docs/cli/compile.md b/site_docs/docs/cli/compile.md index 99cf30d..d0c065f 100644 --- a/site_docs/docs/cli/compile.md +++ b/site_docs/docs/cli/compile.md @@ -50,7 +50,7 @@ _clairtifacts/ | Flag | Default | Description | |------|---------|-------------| | `--project` | `.` | Path to the clair project root | -| `--env` | optional | Environment name. Necessary if clair must apply routing to the generated SQL. | +| `--env` | optional | Environment name. It selects the entry in `__routing__.py` that clair applies to the generated SQL. | | `--select` | all | Glob pattern that filters the Trouves. Repeat the flag to add more patterns. | | `--run-mode` | `full_refresh` | `full_refresh` or `incremental`. Selects which SQL variant clair generates. | diff --git a/site_docs/docs/cli/overview.md b/site_docs/docs/cli/overview.md index 8f6d62d..908ff9a 100644 --- a/site_docs/docs/cli/overview.md +++ b/site_docs/docs/cli/overview.md @@ -3,7 +3,7 @@ All commands share two common flags: - `--project` — path to the clair project root (default: `.`) -- `--env` — environment name from `~/.clair/environments.yml` (default: `CLAIR_ENV` or `dev`) +- `--env` — environment name. It names a key in `~/.clair/environments.yml` and an entry in `__routing__.py` (default: `CLAIR_ENV` or `dev`) ## Commands @@ -13,6 +13,7 @@ All commands share two common flags: | [`clair compile`](compile.md) | Resolve DAG and write SQL to `_clairtifacts/` | Optional (for routing) | | [`clair run`](run.md) | Run Trouves against Snowflake in dependency order | **Yes** | | [`clair test`](test.md) | Run data quality tests against Snowflake | **Yes** | +| [`clair validate`](validate.md) | Apply the routing entries to every Trouve | No | | [`clair dag`](dag.md) | Print the dependency graph as an indented tree | No | | [`clair docs`](docs.md) | Start a local web UI for the DAG and the documentation | No | | [`clair clean`](clean.md) | Remove compiled artifacts from `_clairtifacts/` | No | @@ -22,6 +23,7 @@ All commands share two common flags: | Variable | Default | Description | |----------|---------|-------------| | `CLAIR_ENV` | `dev` | Active environment name. This is the same as `--env` on every command. `--env` wins if you set both. | +| `CLAIR_USER` | — | The `clair init` routing template reads this name. Give each variable that a routing entry reads the `CLAIR_` prefix. | | `CLAIR_LOG_FORMAT` | _(text)_ | Set to `json` to write structured JSON logs. Use this in CI/CD pipelines and in container environments that read JSON logs. | ## Help diff --git a/site_docs/docs/cli/validate.md b/site_docs/docs/cli/validate.md new file mode 100644 index 0000000..2200d1e --- /dev/null +++ b/site_docs/docs/cli/validate.md @@ -0,0 +1,73 @@ +# `clair validate` + +Apply the routing entries in `__routing__.py` to every Trouve in the project, and report every problem at once. + +This command needs no Snowflake credentials, so CI runs it on every change. + +```bash +clair validate +clair validate --env prod +clair validate --project path/to/project +``` + +## Options + +| Option | Default | Description | +|--------|---------|-------------| +| `--project` | `.` | Path to the clair project root. | +| `--env` | `CLAIR_ENV` or `dev` | The environment to route for. It matches an entry in `__routing__.py`. | + +## What it examines + +- The routing file runs, and it gives a `RoutingTable`. +- The table has one entry for each environment name. +- Every directory name and file name makes a valid address. +- The entry for this environment runs on every TABLE and VIEW Trouve. +- Every address that the entry gives is a valid Snowflake identifier. +- No two Trouves go to one physical target. + +## Output + +A project with no problems: + +``` + environment: dev + routing file: /home/alice/project/__routing__.py + entry: DeveloperRouting(environment_name='dev', user_variable='CLAIR_USER') + Trouves to route: 12 + + ✓ Every routed name is valid. No collisions. +``` + +A project with a problem gives exit code 1: + +``` + environment: dev + routing file: /home/alice/project/__routing__.py + entry: DeveloperRouting(environment_name='dev', user_variable='CLAIR_USER') + Trouves to route: 12 + + ✗ analytics.finance.revenue + The routing entry `DeveloperRouting(environment_name='dev', user_variable='CLAIR_USER')` failed on 'analytics.finance.revenue': KeyError: 'CLAIR_USER' + + 1 problem found. +``` + +`clair compile` and `clair run` stop at the first routing problem, because they must not write to a wrong target. `clair validate` instead reports every problem, so that you correct them together. + +## Exit codes + +| Code | Meaning | +|------|---------| +| 0 | Every routed name is valid, and no two Trouves collide. | +| 1 | clair found one problem or more. | + +## In CI + +```yaml +- name: Validate clair routing + run: | + uv run clair validate --env prod +``` + +See the [routing guide](../guides/routing.md) for the entry types and how to write a `route` method. diff --git a/site_docs/docs/concepts/environments.md b/site_docs/docs/concepts/environments.md index 9f1412e..b3c6f5e 100644 --- a/site_docs/docs/concepts/environments.md +++ b/site_docs/docs/concepts/environments.md @@ -78,7 +78,6 @@ prod: | `role` | — | Default role. If you omit it, Snowflake uses the default role of the user. | | `region` | — | AWS/Azure region. clair needs it for the query URLs in the logs. | | `account_locator` | — | Classic account locator. clair needs it for the query URLs. | -| `routing` | — | Routing policy. See [Routing Policies](../guides/routing.md). | ## Select an environment @@ -106,6 +105,8 @@ In CI, set `CLAIR_ENV` and use key-pair authentication. This method does not nee run: clair run --project . ``` -## Routing policies +## Routing -Each environment can include a routing policy. The policy remaps logical Snowflake names to physical targets. See [Routing Policies](../guides/routing.md). +An environment holds connection settings only. The routing rules are in `__routing__.py`, at the root of your project. clair joins the two files by the environment name. + +An unknown key in an environment block is an error. A `routing:` block from an older version of clair therefore stops the run, and the message tells you to move the rule. See [Routing](../guides/routing.md). diff --git a/site_docs/docs/concepts/index.md b/site_docs/docs/concepts/index.md index 2e2f54d..1af4830 100644 --- a/site_docs/docs/concepts/index.md +++ b/site_docs/docs/concepts/index.md @@ -5,4 +5,4 @@ Read these four concepts before you use clair. - **[Trouve](trouve.md)** — the basic unit. One Python file, one Snowflake object. - **[DAG](dag.md)** — the dependency graph. clair builds it from the Python imports. - **[Project Layout](project-layout.md)** — how the directory structure maps to Snowflake names. -- **[Environments](environments.md)** — Snowflake connection profiles and routing policies. +- **[Environments](environments.md)** — Snowflake connection profiles. diff --git a/site_docs/docs/concepts/project-layout.md b/site_docs/docs/concepts/project-layout.md index b6a4a3a..9c9557a 100644 --- a/site_docs/docs/concepts/project-layout.md +++ b/site_docs/docs/concepts/project-layout.md @@ -19,6 +19,8 @@ A clair project in production usually has 3 or 4 layers: ``` my_project/ +├── __routing__.py # The routing entry of each environment +│ ├── source/ # Tables that already exist — TrouveType.SOURCE │ ├── orders/ │ │ ├── raw.py # source.orders.raw @@ -46,6 +48,7 @@ my_project/ | File | Location | Purpose | |------|----------|---------| +| `__routing__.py` | project root | The [routing](../guides/routing.md) entry of each environment | | `__database_config__.py` | database directory | Warehouse/role defaults for all Trouves in that database | | `__schema_config__.py` | schema directory | Warehouse/role defaults for all Trouves in that schema | @@ -81,4 +84,4 @@ Python imports work in the usual way. A Trouve in `refined/` can import from `so from source.orders.raw import trouve as raw_orders ``` -clair resolves `source.orders.raw` to the Snowflake object at `source.orders.raw`. An active [routing policy](../guides/routing.md) can change this target. +clair resolves `source.orders.raw` to the Snowflake object at `source.orders.raw`. An active [routing entry](../guides/routing.md) can change this target. diff --git a/site_docs/docs/guides/index.md b/site_docs/docs/guides/index.md index a8aaaf2..5a35991 100644 --- a/site_docs/docs/guides/index.md +++ b/site_docs/docs/guides/index.md @@ -6,5 +6,5 @@ These guides show you how to do the usual clair tasks. - **[Incrementality](incrementality.md)** — APPEND and UPSERT strategies for large tables - **[Data Quality Tests](data-quality-tests.md)** — attach tests to Trouves - **[Selectors](selectors.md)** — run only a subset of your project -- **[Routing Policies](routing.md)** — remap Snowflake targets per environment +- **[Routing](routing.md)** — remap the Snowflake target of each environment - **[Per-Database & Schema Config](per-database-schema-config.md)** — warehouse and role overrides per directory diff --git a/site_docs/docs/guides/routing.md b/site_docs/docs/guides/routing.md index fcedfea..d7ac758 100644 --- a/site_docs/docs/guides/routing.md +++ b/site_docs/docs/guides/routing.md @@ -1,91 +1,151 @@ -# Routing Policies +# Routing -Routing remaps logical Snowflake names to different physical targets. clair reads the logical names from your file system. Configure routing for each environment in `~/.clair/environments.yml`. +Routing remaps logical Snowflake names to different physical targets. clair reads the logical names from your file system. You write the routing rules in `__routing__.py`, at the root of your project. The usual use: run a production project against a dev Snowflake database. You do not change a Trouve file. -## SOURCE passthrough +## Where routing lives + +Routing is in the project, not in `~/.clair/environments.yml`: + +| File | Holds | Commit it? | +|---|---|---| +| `~/.clair/environments.yml` | Connection settings and credentials | No | +| `/__routing__.py` | The routing rules | Yes | -SOURCE Trouves always use their logical name. The active routing policy has no effect on them. Routing applies to TABLE and VIEW Trouves only. +The environment name joins the two files. An entry with `environment_name = "dev"` applies when clair loads the `dev` environment from `environments.yml`. -## `database_override` +The routing file holds no credentials, so you commit it and your team reviews it like other code. Because it is Python, one committed rule gives each developer a separate target. -This policy replaces the database part of the name of each TABLE and VIEW Trouve. +!!! warning + A `routing:` block in `environments.yml` is now an error. Move the rule to `__routing__.py`. + +## The three types -```yaml -# ~/.clair/environments.yml -dev: - account: myorg-myaccount - user: alice@example.com - authenticator: externalbrowser - warehouse: dev_warehouse - routing: - policy: database_override - database_name: dev +```python +from clair import RoutingEntry, RoutingTable, TrouveAddress ``` -If this policy is active, clair writes the Trouve at `refined/orders/daily.py` to `dev.orders.daily` in Snowflake. Its logical name stays `refined.orders.daily`. clair still reads the source at `source/orders/raw.py` from `source.orders.raw`. +`TrouveAddress` holds a `database_name`, a `schema_name`, and a `table_name`. It validates each name when you make it, so an address that exists is a valid Snowflake identifier. -**Example mapping:** +`RoutingEntry` is the base class for one environment's rule. You write a subclass with an `environment_name` and a `route` method. -| Logical name | Physical target | -|---|---| -| `source.orders.raw` | `source.orders.raw` (SOURCE — passthrough) | -| `refined.orders.daily` | `dev.orders.daily` | -| `derived.orders.summary` | `dev.orders.summary` | - -## `schema_isolation` - -This policy joins `database.schema.table` into one table name (`DATABASE_SCHEMA_TABLE`), in a database and a schema that you set. Use it to run the projects of many developers in one shared Snowflake schema. Each developer gets different table names, thus the projects do not collide. - -```yaml -# alice's dev environment -dev: - account: myorg-myaccount - user: alice@example.com - authenticator: externalbrowser - warehouse: dev_warehouse - routing: - policy: schema_isolation - database_name: dev - schema_name: alice +`RoutingTable` holds the entries. Your `__routing__.py` makes one and gives it the name `routing`. + +## A routing file + +```python +# /__routing__.py +"""Clair routing -- gives each environment its physical write target.""" + +import os + +from clair import RoutingEntry, RoutingTable, TrouveAddress + + +class DeveloperRouting(RoutingEntry): + """Each person writes to a separate database.""" + + environment_name: str = "dev" + user_variable: str = "CLAIR_USER" + + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + user_name = os.environ[self.user_variable].upper() + return trouve_address.model_copy( + update={"database_name": f"{trouve_address.database_name}_{user_name}"} + ) + + +class ProductionRouting(RoutingEntry): + """Production writes to the logical names, so the address stays the same.""" + + environment_name: str = "prod" + + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + return trouve_address + + +routing = RoutingTable(entries=[DeveloperRouting(), ProductionRouting()]) ``` -**Example mapping:** +With `CLAIR_USER=alice` and the `dev` environment: | Logical name | Physical target | |---|---| | `source.orders.raw` | `source.orders.raw` (SOURCE — passthrough) | -| `refined.orders.daily` | `dev.alice.REFINED_ORDERS_DAILY` | -| `derived.orders.summary` | `dev.alice.DERIVED_ORDERS_SUMMARY` | +| `refined.orders.daily` | `refined_ALICE.orders.daily` | +| `derived.orders.summary` | `derived_ALICE.orders.summary` | -!!! warning - `schema_isolation` joins `database_schema_table` with underscores to make an identifier. Snowflake gives identifiers a limit of 255 characters. A very long Trouve name can go above this limit. clair then raises `InvalidRoutingConfigError`. +## Write a route method + +`route` accepts one `TrouveAddress` and gives one `TrouveAddress`. To change one part, call `model_copy`: + +```python +def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + return trouve_address.model_copy(update={"database_name": "DEV"}) +``` + +To build a new address, name all three parts: + +```python +def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + collapsed_table_name = ( + f"{trouve_address.database_name}_{trouve_address.schema_name}_" + f"{trouve_address.table_name}" + ).upper() + return TrouveAddress( + database_name="DEV", + schema_name="alice", + table_name=collapsed_table_name, + ) +``` + +This second shape puts the projects of many developers in one shared schema. Each developer gets different table names, thus the projects do not collide. + +Add a field for each value that the rule needs. Pydantic validates the fields, and the field values show in the CLI messages. + +## SOURCE passthrough + +SOURCE Trouves always use their logical name. clair never calls `route` for a SOURCE Trouve. Routing applies to TABLE and VIEW Trouves only. + +## No entry for an environment + +An environment with no entry in the table gets passthrough routing: clair writes to the logical names. Those are the production names, so clair warns you first: + +``` +Warning: __routing__.py does not name the environment 'staging'. + Trouves write to their logical (production) names. + The file names: dev, prod +``` + +To make the passthrough deliberate, write an entry that gives the address back, as `ProductionRouting` does above. clair then stays quiet. + +## Validation + +`TrouveAddress` validates every name that it holds. A name must start with a letter or an underscore, hold only letters, digits, underscores or dollar signs, and stay under 255 characters. This applies to the logical names that your directories give, and to the physical names that your rules build. + +Run [`clair validate`](../cli/validate.md) to apply your rules to every Trouve without a Snowflake connection. ## Collision detection If two TABLE or VIEW Trouves route to the same physical target, clair shows a warning before the run: ``` -Warning: Clair found 2 routing collisions (env: dev, policy: database_override → dev) +Warning: Clair found 2 routing collisions (env: dev, entry: DeveloperRouting(environment_name='dev', user_variable='CLAIR_USER')) dev.orders.daily ↳ refined.orders.daily ↳ analytics.orders.daily - Fix: give one Trouve a different name, change the routing policy in environments.yml, + Fix: give one Trouve a different name, change the routing entry in __routing__.py, or use --select to remove one Trouve from this run. ``` -## No routing +## One entry for each environment -Omit the `routing` block. clair then uses the logical names as the physical targets. This is correct for production: +Two entries with one environment name are an error. Only one can win, and a silent choice sends the writes to a target that you do not expect: -```yaml -prod: - account: myorg-myaccount - user: ci_user - private_key_path: ~/.clair/snowflake_key.p8 - warehouse: prod_warehouse - # no routing block — clair uses the logical names +``` +Invalid routing file at __routing__.py: ValidationError: ... +the routing table has more than one entry for: dev. Give each environment one entry. ``` diff --git a/site_docs/mkdocs.yml b/site_docs/mkdocs.yml index 2c17bb8..c49d445 100644 --- a/site_docs/mkdocs.yml +++ b/site_docs/mkdocs.yml @@ -79,7 +79,7 @@ nav: - Incrementality: guides/incrementality.md - Data Quality Tests: guides/data-quality-tests.md - Selectors: guides/selectors.md - - Routing Policies: guides/routing.md + - Routing: guides/routing.md - Per-Database & Schema Config: guides/per-database-schema-config.md - CLI: - cli/overview.md @@ -87,6 +87,7 @@ nav: - clair run: cli/run.md - clair compile: cli/compile.md - clair test: cli/test.md + - clair validate: cli/validate.md - clair dag: cli/dag.md - clair docs: cli/docs.md - clair clean: cli/clean.md diff --git a/src/clair/__init__.py b/src/clair/__init__.py index 96fb145..970fe84 100644 --- a/src/clair/__init__.py +++ b/src/clair/__init__.py @@ -43,6 +43,7 @@ if TYPE_CHECKING: from clair.environments.environments import Environment +from clair.environments.routing import RoutingEntry, RoutingTable, TrouveAddress from clair.trouves.column import Column, ColumnType from clair.trouves.config import DatabaseDefaults, SchemaDefaults from clair.trouves.run_config import ( @@ -84,6 +85,8 @@ "ColumnType", "DatabaseDefaults", "IncrementalMode", + "RoutingEntry", + "RoutingTable", "RunConfig", "RunMode", "SchemaDefaults", @@ -94,6 +97,7 @@ "TestUnique", "TestUniqueColumns", "Trouve", + "TrouveAddress", "TrouveType", "UpsertConfig", ] diff --git a/src/clair/cli/main.py b/src/clair/cli/main.py index 480ef9d..1cdc1c3 100644 --- a/src/clair/cli/main.py +++ b/src/clair/cli/main.py @@ -2,6 +2,7 @@ from __future__ import annotations +import os import re import shutil import sys @@ -31,8 +32,24 @@ from clair.docs.catalog import build_catalog from clair.docs.server import serve from clair.environments.environments import load_environment -from clair.environments.routing import DatabaseOverrideRouting, SchemaIsolationRouting -from clair.exceptions import ClairError, CompileError, EnvironmentsFileNotFoundError +from clair.environments.project_routing import ( + ROUTING_FILE_NAME, + ProjectRouting, + load_project_routing, +) +from clair.environments.routing import ( + collect_routing_problems, + describe_routing, + detect_routing_collisions, + route, +) +from clair.exceptions import ( + ClairError, + CompileError, + EnvironmentsFileNotFoundError, + InvalidRoutingConfigError, + InvalidTrouveAddressError, +) from clair.trouves.run_config import RunMode from clair.trouves.trouve import ExecutionType, TrouveType @@ -119,23 +136,42 @@ def init(project: str | None) -> None: click.echo("") +def _resolve_project_routing(project_root: Path, env_name: str) -> ProjectRouting: + """Load the project routing entry and warn about an absent entry. + + A routing file that does not name the active environment is almost always a + typo. Passthrough routing then writes to the production names, so clair + tells the user before any SQL runs. + """ + project_routing = load_project_routing(project_root, env_name) + + if project_routing.is_unnamed_environment: + click.echo( + click.style( + f"\nWarning: {ROUTING_FILE_NAME} does not name the environment " + f"'{env_name}'.", + fg="yellow", + bold=True, + ) + ) + click.echo( + f" Trouves write to their logical (production) names.\n" + f" The file names: {', '.join(project_routing.environment_names) or 'nothing'}\n" + ) + + return project_routing + + def _print_routing_collision_warnings(trouves: list, env_name: str, routing) -> None: """Show a clear warning about each routing collision, before the SQL starts.""" collisions = find_routing_collisions(trouves) if not collisions: return - policy_desc = "" - if routing is not None: - if isinstance(routing, DatabaseOverrideRouting): - policy_desc = f"database_override → {routing.database_name}" - elif isinstance(routing, SchemaIsolationRouting): - policy_desc = f"schema_isolation → {routing.database_name}.{routing.schema_name}" - n = len(collisions) header = "1 routing collision" if n == 1 else f"{n} routing collisions" - if policy_desc: - header += f" (env: {env_name}, policy: {policy_desc})" + if routing is not None: + header += f" (env: {env_name}, entry: {describe_routing(routing)})" else: header += f" (env: {env_name})" @@ -147,8 +183,8 @@ def _print_routing_collision_warnings(trouves: list, env_name: str, routing) -> click.echo(f" ↳ {source}") click.echo( - "\n Fix: give one Trouve a different name, change the routing policy in " - "environments.yml,\n or use --select to remove one Trouve from this run.\n" + f"\n Fix: give one Trouve a different name, change the routing entry in " + f"{ROUTING_FILE_NAME},\n or use --select to remove one Trouve from this run.\n" ) @@ -261,14 +297,18 @@ def compile_cmd(select: tuple[str, ...], exclude: tuple[str, ...], project: str, run_mode_enum = RunMode(run_mode) run_id = uuid6.uuid7().hex - routing = None environment = None env_name = env or "dev" try: env_name, environment = load_environment(env) - routing = environment.routing except EnvironmentsFileNotFoundError: - logger.warning("compile.no_environments_file", detail="Clair compiles without routing. Run `clair init` to make environments.yml.") + logger.warning("compile.no_environments_file", detail="Clair compiles without an environment. Run `clair init` to make environments.yml.") + except ClairError as e: + logger.error("compile.error", error=str(e)) + sys.exit(1) + + try: + routing = _resolve_project_routing(project_root, env_name).entry except ClairError as e: logger.error("compile.error", error=str(e)) sys.exit(1) @@ -307,11 +347,85 @@ def _on_node_compiled(node_info: CompiledNodeInfo) -> None: write_compile_output(dag, selected, project_root, on_node_compiled=_on_node_compiled, run_mode=run_mode_enum, run_id=run_id) logger.info("compile.complete", run_id=run_id, artifacts_dir=str(artifacts_dir)) + except (InvalidRoutingConfigError, InvalidTrouveAddressError) as e: + logger.error("compile.routing_error", error=str(e)) + click.echo("\n Run `clair validate` to see every routing problem.\n", err=True) + sys.exit(1) except ClairError as e: logger.error("compile.error", error=str(e)) sys.exit(1) +@cli.command() +@click.option( + "--project", + default=".", + help="Path to the Clair project root (defaults to current directory)", +) +@click.option( + "--env", + default=None, + help="Environment name to route for; matches an entry in __routing__.py", +) +def validate(project: str, env: str | None) -> None: + """Apply the project routing entries to every Trouve. + + This command needs no Snowflake credentials, so CI runs it on every change. + """ + project_root = Path(project).resolve() + env_name = env or os.environ.get("CLAIR_ENV") or "dev" + + try: + project_routing = _resolve_project_routing(project_root, env_name) + # Find the Trouves with routing off. A bad entry then reports as a + # routing problem, and does not stop discovery at the first Trouve. + discovered = discover_project(project_root, routing=None) + except ClairError as e: + logger.error("validate.error", error=str(e)) + sys.exit(1) + + routing = project_routing.entry + # Keep the (logical name, type) pair, not the Trouve. The name is the only + # part that the collision report needs, and here it is never None. + routable: list[tuple[str, TrouveType]] = [ + (trouve.compiled.logical_name, trouve.type) + for trouve in discovered + if trouve.compiled is not None and trouve.type != TrouveType.SOURCE + ] + + click.echo(f"\n environment: {env_name}") + click.echo(f" routing file: {project_routing.file_path or 'none'}") + click.echo(f" entry: {describe_routing(routing)}") + click.echo(f" Trouves to route: {len(routable)}\n") + + problems = collect_routing_problems(discovered, routing) + for logical_name, problem in problems: + click.echo(click.style(f" ✗ {logical_name}", fg="red", bold=True)) + click.echo(f" {problem}\n") + + collisions: list[tuple[str, list[str]]] = [] + if not problems: + logical_to_routed = { + logical_name: route(logical_name, trouve_type, routing) + for logical_name, trouve_type in routable + } + collisions = detect_routing_collisions(logical_to_routed) + for routed_target, logical_sources in collisions: + click.echo(click.style(f" ✗ {routed_target}", fg="red", bold=True)) + click.echo(" Two or more Trouves route to this one target:") + for source in logical_sources: + click.echo(f" ↳ {source}") + click.echo("") + + problem_count = len(problems) + len(collisions) + if problem_count: + label = "problem" if problem_count == 1 else "problems" + click.echo(click.style(f" {problem_count} {label} found.\n", fg="red", bold=True)) + sys.exit(1) + + click.echo(click.style(" ✓ Every routed name is valid. No collisions.\n", fg="green")) + + @cli.command() @click.option( "--select", @@ -447,8 +561,9 @@ def run(select: tuple[str, ...], exclude: tuple[str, ...], project: str, env: st "warehouse": environment.warehouse, "role": environment.role, } - discovered = discover_project(project_root, profile_defaults, routing=environment.routing, environment=environment, run_mode=run_mode_enum) - _print_routing_collision_warnings(discovered, env_name, environment.routing) + routing = _resolve_project_routing(project_root, env_name).entry + discovered = discover_project(project_root, profile_defaults, routing=routing, environment=environment, run_mode=run_mode_enum) + _print_routing_collision_warnings(discovered, env_name, routing) dag = build_dag(discovered) # Keep only the Trouves that the selector gives. @@ -502,6 +617,10 @@ def on_node_success(node_name: str) -> bool: finally: adapter.close() + except (InvalidRoutingConfigError, InvalidTrouveAddressError) as e: + logger.error("run.routing_error", error=str(e)) + click.echo("\n Run `clair validate` to see every routing problem.\n", err=True) + sys.exit(1) except ClairError as e: logger.error("run.error", error=str(e)) sys.exit(1) @@ -543,14 +662,15 @@ def test( try: # Load the environment. - _, environment = load_environment(env) + env_name, environment = load_environment(env) # Find the Trouves and make the DAG. profile_defaults = { "warehouse": environment.warehouse, "role": environment.role, } - discovered = discover_project(project_root, profile_defaults, routing=environment.routing, environment=environment) + routing = _resolve_project_routing(project_root, env_name).entry + discovered = discover_project(project_root, profile_defaults, routing=routing, environment=environment) dag = build_dag(discovered) # Keep the Trouves that the selector gives. Keep each SOURCE too, so diff --git a/src/clair/core/discovery.py b/src/clair/core/discovery.py index e39f517..2978840 100644 --- a/src/clair/core/discovery.py +++ b/src/clair/core/discovery.py @@ -17,7 +17,7 @@ if TYPE_CHECKING: from clair.environments.environments import Environment -from clair.environments.routing import RoutingConfig, detect_routing_collisions, route +from clair.environments.routing import RoutingEntry, detect_routing_collisions, route from clair.trouves._refs import THIS_PLACEHOLDER, TROUVE_PLACEHOLDER_PREFIX from clair.trouves._refs import clear as clear_refs from clair.trouves.config import DatabaseDefaults, ResolvedConfig, SchemaDefaults @@ -148,7 +148,7 @@ def _detect_imports( def discover_project( project_root: Path, profile_defaults: dict[str, str | None] | None = None, - routing: RoutingConfig | None = None, + routing: RoutingEntry | None = None, environment: Environment | None = None, run_mode: RunMode | None = None, ) -> list[Trouve]: @@ -161,7 +161,7 @@ def discover_project( Args: project_root: The absolute path of the project root directory. profile_defaults: The default warehouse and role from the active profile. - routing: The routing configuration for the physical names. + routing: The routing entry for the physical names, from __routing__.py. environment: The active environment. Clair puts it in ``clair.env``. Thus a Trouve module can read it at load time, for a feature flag. run_mode: The run mode that the user asks for: FULL_REFRESH or diff --git a/src/clair/core/scaffold.py b/src/clair/core/scaffold.py index 4f70dcd..8cc3869 100644 --- a/src/clair/core/scaffold.py +++ b/src/clair/core/scaffold.py @@ -4,6 +4,8 @@ from pathlib import Path +from clair.environments.project_routing import ROUTING_FILE_NAME + # --------------------------------------------------------------------------- # The file templates. # --------------------------------------------------------------------------- @@ -16,9 +18,54 @@ ) ''' +_ROUTING_TEMPLATE = '''\ +"""Clair routing -- gives each environment its physical write target. + +Each entry names one environment. The name matches a top-level key in +~/.clair/environments.yml. The route method accepts the logical TrouveAddress +and gives the physical TrouveAddress. SOURCE Trouves never route. + +Commit this file. It holds no credentials. +Run `clair validate` to apply the entries to every Trouve in the project. +""" + +import os + +from clair import RoutingEntry, RoutingTable, TrouveAddress + + +class DeveloperRouting(RoutingEntry): + """Each person writes to a separate database. + + Set CLAIR_USER to your name before you run clair. + """ + + environment_name: str = "dev" + user_variable: str = "CLAIR_USER" + + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + user_name = os.environ[self.user_variable].upper() + return trouve_address.model_copy( + update={"database_name": f"{trouve_address.database_name}_{user_name}"} + ) + + +class ProductionRouting(RoutingEntry): + """Production writes to the logical names, so the address stays the same.""" + + environment_name: str = "prod" + + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + return trouve_address + + +routing = RoutingTable(entries=[DeveloperRouting(), ProductionRouting()]) +''' + _ENVIRONMENTS_TEMPLATE = '''\ # The Clair environments. Each environment has its own connection settings. -# Reference: https://github.com/your-org/clair +# Routing is not here. It lives in the project __routing__.py file. +# Reference: https://github.com/rivage-sh/clair dev: account: your-org-your-account # for example, myorg-myaccount @@ -62,7 +109,8 @@ def scaffold_project( ) -> list[tuple[str, str]]: """Make a new Clair project in *project_dir*. - The function writes an example source Trouve file. It also writes the global + The function writes an example source Trouve file and a project + ``__routing__.py`` file. It also writes the global ``~/.clair/environments.yml`` file, if that file does not exist. Args: @@ -81,6 +129,7 @@ def scaffold_project( # Each project file, as a (relative_path, template_content) pair. project_files: list[tuple[str, str]] = [ (f"{source_database_name}/{source_schema_name}/{source_table_name}.py", _SOURCE_TROUVE_TEMPLATE), + (ROUTING_FILE_NAME, _ROUTING_TEMPLATE), ] results: list[tuple[str, str]] = [] diff --git a/src/clair/environments/environments.py b/src/clair/environments/environments.py index c46f416..862e0d6 100644 --- a/src/clair/environments/environments.py +++ b/src/clair/environments/environments.py @@ -9,21 +9,39 @@ import yaml from pydantic import BaseModel, ConfigDict, ValidationError -from clair.environments.routing import Routing from clair.exceptions import ( EnvironmentNotFoundError, EnvironmentsFileNotFoundError, - InvalidRoutingConfigError, - InvalidRoutingPolicyError, + InvalidEnvironmentError, ) DEFAULT_ENVIRONMENTS_PATH = Path.home() / ".clair" / "environments.yml" +def _first_error_message(exc: ValidationError) -> str: + """Take the first message out of a Pydantic error. + + Pydantic prints a report of many lines. A CLI message needs one sentence. + """ + errors = exc.errors() + if not errors: + return str(exc) + first = errors[0] + key_name = ".".join(str(part) for part in first.get("loc", ())) + message = str(first.get("msg", "")) + return f"'{key_name}': {message}" if key_name else message + + class Environment(BaseModel): - """One environment from environments.yml.""" + """One environment from environments.yml. - model_config = ConfigDict(populate_by_name=True) + An environment holds connection settings only. Routing lives in the project + ``__routing__.py``, under the same environment name. + """ + + # "forbid" makes an unknown key an error. A leftover routing block, or a + # misspelt key, would otherwise disappear without a word. + model_config = ConfigDict(populate_by_name=True, extra="forbid") # The identity of the environment. name: str @@ -44,9 +62,6 @@ class Environment(BaseModel): region: str | None = None account_locator: str | None = None - # The routing policy. - routing: Routing | None = None - def to_connection_dict(self) -> dict[str, Any]: """Give the connection dict that SnowflakeAdapter.connect() needs.""" d: dict[str, Any] = { @@ -68,21 +83,6 @@ def to_connection_dict(self) -> dict[str, Any]: return d -def _validate_routing_block(routing_raw: dict[str, Any]) -> None: - """Examine the routing block before Pydantic reads it. - - This function finds an absent policy value and an unknown policy value. Then - it raises a clair error type that the CLI already knows. - """ - if "policy" not in routing_raw: - raise InvalidRoutingConfigError("the routing block must have a 'policy' value") - - policy = routing_raw["policy"] - valid_policies = {"database_override", "schema_isolation"} - if policy not in valid_policies: - raise InvalidRoutingPolicyError(policy) - - def load_environment( env_name: str | None = None, environments_path: Path | None = None, @@ -105,8 +105,7 @@ def load_environment( Raises: EnvironmentsFileNotFoundError: If environments.yml does not exist. EnvironmentNotFoundError: If environments.yml has no such environment. - InvalidRoutingPolicyError: If the file names an unknown routing policy. - InvalidRoutingConfigError: If the routing block has a bad structure. + InvalidEnvironmentError: If the environment block holds an unknown key. """ resolved_name = env_name or os.environ.get("CLAIR_ENV") or "dev" path = environments_path or DEFAULT_ENVIRONMENTS_PATH @@ -125,14 +124,11 @@ def load_environment( env_data: dict[str, Any] = raw[resolved_name] - routing_raw = env_data.get("routing") - if isinstance(routing_raw, dict): - _validate_routing_block(routing_raw) - try: - environment = Environment(name=resolved_name, **env_data) - return resolved_name, environment + return resolved_name, Environment(name=resolved_name, **env_data) except ValidationError as exc: - # Show each Pydantic error as a clair error that the CLI already knows. - # One example is an absent schema_name for the schema_isolation policy. - raise InvalidRoutingConfigError(str(exc)) from exc + # An unknown key is almost always a typo, or a routing block that the user + # did not move to __routing__.py. Both send writes to the wrong target. + raise InvalidEnvironmentError( + resolved_name, str(path), _first_error_message(exc) + ) from exc diff --git a/src/clair/environments/project_routing.py b/src/clair/environments/project_routing.py new file mode 100644 index 0000000..1838118 --- /dev/null +++ b/src/clair/environments/project_routing.py @@ -0,0 +1,147 @@ +"""Load the project routing file, ``__routing__.py``. + +Routing lives in the project, not in ``~/.clair/environments.yml``: + +* Routing is not a secret. A team commits it and reviews it like other code. +* Routing gains the most from Python. An entry reads an environment variable, so + one committed entry gives each developer a separate target. +* A project-local file matches the clair version that the project pins. + +The file defines a ``routing`` name and gives it a ``RoutingTable``. Each entry +in the table names one environment. That name is the join key: it matches a +top-level key in ``~/.clair/environments.yml``. +""" + +from __future__ import annotations + +import hashlib +import importlib.util +import sys +from pathlib import Path +from typing import NamedTuple + +from clair.environments.routing import RoutingEntry, RoutingTable +from clair.exceptions import InvalidRoutingFileError + +ROUTING_FILE_NAME = "__routing__.py" +ROUTING_TABLE_ATTRIBUTE = "routing" + +# Cache key is (resolved path, modification time), so an edit reloads the file +# but repeated loads in one process do not run the file again. An entry that +# reads a keychain or a secret store must not run two times. +_routing_table_cache: dict[tuple[str, int], RoutingTable] = {} + + +class ProjectRouting(NamedTuple): + """The outcome of a routing file lookup for one environment.""" + + entry: RoutingEntry | None + file_path: Path | None + environment_names: list[str] + has_entry: bool = False + + @property + def file_exists(self) -> bool: + """Tell the caller if the project has a ``__routing__.py``.""" + return self.file_path is not None + + @property + def is_unnamed_environment(self) -> bool: + """Tell the caller if the table omits this environment. + + An absent entry is almost always a typo. Clair then writes to the + logical names, which are the production names. + """ + return self.file_exists and not self.has_entry + + +def _module_name_for(path: Path) -> str: + """Build a module name that is unique per routing file path.""" + digest = hashlib.md5(str(path).encode()).hexdigest()[:8] + return f"_clair_routing_{digest}" + + +def _load_routing_table(path: Path) -> RoutingTable: + """Run a routing file and give back its routing table. + + Args: + path: The path of the ``__routing__.py`` file. + + Returns: + The ``RoutingTable`` that the file defines. + + Raises: + InvalidRoutingFileError: If clair cannot run the file, or the file does + not define a ``RoutingTable``. + """ + cache_key = (str(path), path.stat().st_mtime_ns) + cached = _routing_table_cache.get(cache_key) + if cached is not None: + return cached + + module_name = _module_name_for(path) + spec = importlib.util.spec_from_file_location(module_name, path) + if spec is None or spec.loader is None: + raise InvalidRoutingFileError(str(path), "clair cannot read this file") + + module = importlib.util.module_from_spec(spec) + # Register the module before execution. A class or a dataclass in the file + # then keeps one identity across loads. + sys.modules[module_name] = module + try: + spec.loader.exec_module(module) + except Exception as exc: + del sys.modules[module_name] + raise InvalidRoutingFileError( + str(path), f"{type(exc).__name__}: {exc}" + ) from exc + + if not hasattr(module, ROUTING_TABLE_ATTRIBUTE): + raise InvalidRoutingFileError( + str(path), f"the file must define a '{ROUTING_TABLE_ATTRIBUTE}' dict" + ) + + table = getattr(module, ROUTING_TABLE_ATTRIBUTE) + if not isinstance(table, RoutingTable): + raise InvalidRoutingFileError( + str(path), + f"'{ROUTING_TABLE_ATTRIBUTE}' must be a RoutingTable, " + f"but it is a {type(table).__name__}", + ) + + _routing_table_cache[cache_key] = table + return table + + +def load_project_routing(project_root: Path, env_name: str) -> ProjectRouting: + """Find the routing entry for one environment. + + A project without a ``__routing__.py`` gets passthrough routing. An + environment without an entry in the table also gets passthrough routing, and + the caller warns about it, because passthrough writes to production names. + + Args: + project_root: The root directory of the clair project. + env_name: The resolved environment name, such as "dev". + + Returns: + A ``ProjectRouting`` with the entry, the file path, and all the + environment names that the table holds. + + Raises: + InvalidRoutingFileError: If the file exists but clair cannot use it. + """ + path = project_root / ROUTING_FILE_NAME + if not path.exists(): + return ProjectRouting( + entry=None, file_path=None, environment_names=[], has_entry=False + ) + + table = _load_routing_table(path) + entry = table.entry_for(env_name) + return ProjectRouting( + entry=entry, + file_path=path, + environment_names=table.environment_names, + has_entry=entry is not None, + ) diff --git a/src/clair/environments/routing.py b/src/clair/environments/routing.py index f38aa7b..550bd73 100644 --- a/src/clair/environments/routing.py +++ b/src/clair/environments/routing.py @@ -1,115 +1,325 @@ -"""The routing policies. Each policy maps a logical name to a physical target.""" +"""Routing -- remaps the logical address of a Trouve to a physical address. + +Three types make up the routing system: + +* ``TrouveAddress`` holds a database name, a schema name, and a table name. It + validates each name when you make it. An address that exists is a valid one. +* ``RoutingEntry`` is the base class for one environment's rule. A user writes a + subclass and gives it a ``route`` method. +* ``RoutingTable`` holds the entries. The project ``__routing__.py`` makes one. + +The validation happens in ``TrouveAddress``, not after a rule runs. A rule that +gives an address gives a correct address, or it raises an error. + +At this time ``TrouveAddress`` applies the Snowflake identifier rules. +""" from __future__ import annotations import re -from abc import abstractmethod -from typing import Annotated, Literal +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING + +from pydantic import ( + BaseModel, + ConfigDict, + ValidationError, + field_validator, + model_validator, +) + +from clair.exceptions import InvalidRoutingConfigError, InvalidTrouveAddressError +from clair.trouves.trouve import TrouveType -from pydantic import BaseModel, Field +if TYPE_CHECKING: + from clair.trouves.trouve import Trouve -from clair.exceptions import InvalidRoutingConfigError -from clair.trouves.trouve import TrouveType -_VALID_IDENTIFIER = re.compile(r"^[A-Z0-9_]+$") +# Snowflake accepts these characters in an unquoted identifier. +_VALID_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_$]*$") +_MAX_IDENTIFIER_LENGTH = 255 +# Maximum width of an entry description in a CLI message. +_MAX_DESCRIPTION_LENGTH = 200 -class RoutingConfig(BaseModel): - """The parent class of all the routing policies.""" - policy: str +def _first_error_message(exc: ValidationError) -> str: + """Take the first message out of a Pydantic error. - @abstractmethod - def apply(self, logical_name: str) -> str: - """Map a logical full_name to its physical target. + Pydantic prints a report of many lines. A CLI message needs one sentence. + """ + errors = exc.errors() + if not errors: + return str(exc) + first = errors[0] + message = str(first.get("msg", "")).removeprefix("Value error, ") + location = ".".join(str(part) for part in first.get("loc", ())) + return f"{location}: {message}" if location else message + + +class TrouveAddress(BaseModel): + """The full address of one Trouve in the warehouse. + + The model is frozen, so an address is hashable and safe to share. To make a + changed copy, call ``model_copy(update={...})``. + """ + + model_config = ConfigDict(frozen=True) + + database_name: str + schema_name: str + table_name: str + + @field_validator("database_name", "schema_name", "table_name") + @classmethod + def _validate_identifier(cls, value: str) -> str: + """Reject a name that Snowflake cannot use as an unquoted identifier.""" + if len(value) > _MAX_IDENTIFIER_LENGTH: + raise ValueError( + f"'{value}' has {len(value)} characters. " + f"The maximum is {_MAX_IDENTIFIER_LENGTH}." + ) + if not _VALID_IDENTIFIER.match(value): + raise ValueError( + f"'{value}' is not a valid identifier. An identifier starts with a " + "letter or an underscore. The other characters are letters, digits, " + "underscores or dollar signs." + ) + return value + + @classmethod + def parse(cls, full_name: str) -> TrouveAddress: + """Make an address from a "database_name.schema_name.table_name" string. Args: - logical_name: The "database.schema.table" name from the file path. + full_name: The dotted name. Returns: - The routed full_name. + The validated address. Raises: - InvalidRoutingConfigError: If the routed identifier is not correct. + InvalidTrouveAddressError: If the string is not a valid address. """ + parts = full_name.split(".") + if len(parts) != 3: + raise InvalidTrouveAddressError( + full_name, + f"an address needs 3 dot-separated parts, but this name has " + f"{len(parts)}", + ) + try: + return cls( + database_name=parts[0], schema_name=parts[1], table_name=parts[2] + ) + except ValidationError as exc: + raise InvalidTrouveAddressError( + full_name, _first_error_message(exc) + ) from exc + def __str__(self) -> str: + return f"{self.database_name}.{self.schema_name}.{self.table_name}" -class DatabaseOverrideRouting(RoutingConfig): - """Replace the database part of the full_name of each Trouve that is not a SOURCE.""" - policy: Literal["database_override"] = "database_override" - database_name: str +class RoutingEntry(BaseModel, ABC): + """The routing rule for one environment. - def apply(self, logical_name: str) -> str: - _, schema, table = logical_name.split(".") - return f"{self.database_name}.{schema}.{table}" + Write a subclass and give it a ``route`` method. Add a field for each value + that the rule needs. Pydantic validates the fields, and the field values + show in the CLI messages. + Example:: -class SchemaIsolationRouting(RoutingConfig): - """Join database.schema.table into one table name in a constant database and schema.""" + class DeveloperRouting(RoutingEntry): + environment_name: str = "dev" + user_variable: str = "CLAIR_USER" - policy: Literal["schema_isolation"] = "schema_isolation" - database_name: str - schema_name: str + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + user = os.environ[self.user_variable].upper() + return trouve_address.model_copy( + update={"database_name": f"{trouve_address.database_name}_{user}"} + ) + """ - def apply(self, logical_name: str) -> str: - db, schema, table = logical_name.split(".") - new_table = f"{db}_{schema}_{table}".upper() - if not _VALID_IDENTIFIER.match(new_table): - raise InvalidRoutingConfigError( - f"schema_isolation made the incorrect identifier '{new_table}'. " - "Use only the characters A-Z, 0-9 and _." - ) - if len(new_table) > 255: - raise InvalidRoutingConfigError( - f"schema_isolation made the identifier '{new_table}'. " - f"It has {len(new_table)} characters, but the maximum is 255." + # The join key. It matches a top-level key in ~/.clair/environments.yml. + environment_name: str + + @abstractmethod + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + """Give the physical address for one logical address. + + Clair never calls this method for a SOURCE Trouve. A SOURCE always reads + from the address that the file system gives. + + Args: + trouve_address: The logical address of the Trouve. + + Returns: + The physical address to write to. + """ + + +class RoutingTable(BaseModel): + """All the routing entries for one project. + + The project ``__routing__.py`` makes one table and gives it the name + ``routing``. Two entries for one environment name are an error. + """ + + entries: list[RoutingEntry] = [] + + @model_validator(mode="after") + def _reject_duplicate_environment_names(self) -> RoutingTable: + """Stop a second entry for one environment name. + + Only one entry can win. A silent choice between two entries would send + the writes to a target that the user does not expect. + """ + seen: set[str] = set() + duplicates: set[str] = set() + for entry in self.entries: + if entry.environment_name in seen: + duplicates.add(entry.environment_name) + seen.add(entry.environment_name) + if duplicates: + names = ", ".join(sorted(duplicates)) + raise ValueError( + f"the routing table has more than one entry for: {names}. " + "Give each environment one entry." ) - return f"{self.database_name}.{self.schema_name}.{new_table}" + return self + @property + def environment_names(self) -> list[str]: + """Give the sorted names of every environment in the table.""" + return sorted(entry.environment_name for entry in self.entries) -# The tagged union that clair uses to read a routing block from YAML or a dict. -Routing = Annotated[ - DatabaseOverrideRouting | SchemaIsolationRouting, - Field(discriminator="policy"), -] + def entry_for(self, environment_name: str) -> RoutingEntry | None: + """Find the entry for one environment name, or None.""" + for entry in self.entries: + if entry.environment_name == environment_name: + return entry + return None + + +def describe_routing(routing: RoutingEntry | None) -> str: + """Give a short description of a routing entry for a CLI message. + + Pydantic prints the class name and the field values, which tells the reader + why two Trouves went to one target. + """ + if routing is None: + return "none" + description = repr(routing) + if len(description) > _MAX_DESCRIPTION_LENGTH: + return description[: _MAX_DESCRIPTION_LENGTH - 1] + "…" + return description + + +def _apply_routing( + logical_address: TrouveAddress, routing: RoutingEntry +) -> TrouveAddress: + """Run one routing entry and confirm that it gave an address. + + The entry builds a ``TrouveAddress``, so the address rules apply already. + This function adds the context that the address alone does not hold: which + entry ran, and which Trouve it ran on. + """ + entry_text = f"The routing entry `{describe_routing(routing)}`" + try: + physical_address = routing.route(logical_address) + except ValidationError as exc: + raise InvalidRoutingConfigError( + f"{entry_text} built a bad address for '{logical_address}': " + f"{_first_error_message(exc)}" + ) from exc + except (InvalidRoutingConfigError, InvalidTrouveAddressError): + raise + except Exception as exc: + raise InvalidRoutingConfigError( + f"{entry_text} failed on '{logical_address}': {type(exc).__name__}: {exc}" + ) from exc + + if not isinstance(physical_address, TrouveAddress): + raise InvalidRoutingConfigError( + f"{entry_text} gave {type(physical_address).__name__} for " + f"'{logical_address}'. A route method must give a TrouveAddress." + ) + return physical_address def route( logical_name: str, trouve_type: TrouveType, - routing: RoutingConfig | None, + routing: RoutingEntry | None, ) -> str: - """Apply a routing policy to a logical full_name. + """Apply a routing entry to a logical full_name. - A SOURCE Trouve always keeps its name, whatever the routing policy is. + The function validates the logical name first, then applies the entry. A + SOURCE Trouve keeps its logical name, whatever the entry is. Args: - logical_name: The "database.schema.table" name from the file path. + logical_name: The file system name "database_name.schema_name.table_name". trouve_type: SOURCE, TABLE, or VIEW. - routing: The active routing config. Give None to keep the name. + routing: The active routing entry, or None for passthrough. Returns: - The routed full_name. + The physical full_name string. + + Raises: + InvalidTrouveAddressError: If the logical name is not a valid address. + InvalidRoutingConfigError: If the entry fails, or gives a bad address. """ + logical_address = TrouveAddress.parse(logical_name) + if routing is None or trouve_type == TrouveType.SOURCE: - return logical_name - return routing.apply(logical_name) + return str(logical_address) + return str(_apply_routing(logical_address, routing)) -def detect_routing_collisions(logical_to_routed: dict[str, str]) -> list[tuple[str, list[str]]]: - """Give a (target, sources) pair for each routing collision. - A collision occurs when two TABLE or VIEW Trouves route to one physical - target. The last write sets the final content of that target. +def collect_routing_problems( + trouves: list[Trouve], + routing: RoutingEntry | None, +) -> list[tuple[str, str]]: + """Apply a routing entry to every Trouve and collect all the failures. + + ``route()`` stops at the first bad address, which is correct for a run. This + function instead reports every problem at once, so that ``clair validate`` + shows a complete list. + + Args: + trouves: All the Trouves in the project, found with routing off. + routing: The routing entry to test. + + Returns: + A list of (logical_name, problem_text) pairs, in discovery order. + """ + problems: list[tuple[str, str]] = [] + for trouve in trouves: + if not trouve.compiled: + continue + logical_name = trouve.compiled.logical_name + try: + route(logical_name, trouve.type, routing) + except (InvalidRoutingConfigError, InvalidTrouveAddressError) as exc: + problems.append((logical_name, str(exc))) + return problems + + +def detect_routing_collisions( + logical_to_routed: dict[str, str], +) -> list[tuple[str, list[str]]]: + """Give the (target, sources) pairs for the routing collisions. + + A collision happens when two TABLE or VIEW Trouves go to one physical target. + The last write in execution order sets the final state of that target. Args: - logical_to_routed: A map of logical_name to routed_name. It holds each - Trouve that is not a SOURCE. + logical_to_routed: A map of logical_name to routed_name for the Trouves + that are not SOURCE Trouves. Returns: - A list of (routed_target, [logical_source, ...]), one item for each - collision. + A list of (routed_target, [logical_source, ...]) for each collision. """ target_to_sources: dict[str, list[str]] = {} for logical, routed in logical_to_routed.items(): diff --git a/src/clair/exceptions.py b/src/clair/exceptions.py index 459d8a4..380024d 100644 --- a/src/clair/exceptions.py +++ b/src/clair/exceptions.py @@ -34,24 +34,44 @@ def __init__(self, path: str) -> None: self.path = path super().__init__( f"Clair cannot find environments.yml at {path}. " - "Run `clair init` to make one. As an alternative, give your " - "profiles.yml the new name and add a routing block." + "Run `clair init` to make one." ) -class InvalidRoutingPolicyError(ClairError): - """Clair raises this error when the config names an unknown routing policy.""" +class InvalidTrouveAddressError(ClairError): + """Clair raises this error when a name is not a valid Trouve address.""" - def __init__(self, policy: str) -> None: - self.policy = policy + def __init__(self, full_name: str, detail: str) -> None: + self.full_name = full_name + self.detail = detail + super().__init__(f"Clair cannot use '{full_name}' as an address: {detail}") + + +class InvalidEnvironmentError(ClairError): + """Clair raises this error when an environment block holds a bad value.""" + + def __init__(self, env_name: str, path: str, detail: str) -> None: + self.env_name = env_name + self.path = path + self.detail = detail super().__init__( - f"Clair does not know the routing policy '{policy}'. " - "Use database_override or schema_isolation." + f"Clair cannot read the environment '{env_name}' in {path}. " + "An unknown key is a misspelt name, or a routing block that belongs " + f"in the project __routing__.py. Detail: {detail}" ) +class InvalidRoutingFileError(ClairError): + """Clair raises this error when it cannot use the project __routing__.py.""" + + def __init__(self, path: str, detail: str) -> None: + self.path = path + self.detail = detail + super().__init__(f"Invalid routing file at {path}: {detail}") + + class InvalidRoutingConfigError(ClairError): - """Clair raises this error when a routing config block has a bad structure.""" + """Clair raises this error when a routing entry returns an unusable address.""" def __init__(self, detail: str) -> None: super().__init__(detail) diff --git a/tests/conftest.py b/tests/conftest.py index 72f6b9d..10797cf 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -33,7 +33,7 @@ def clean_sys_modules(): for mod_name in after - before: if any( part in mod_name - for part in ("source.", "analytics.", "db.", "tmp_project") + for part in ("source.", "analytics.", "db.", "tmp_project", "_clair_routing_") ): del sys.modules[mod_name] @@ -69,25 +69,22 @@ def tmp_environments(tmp_path: Path) -> Path: private_key_passphrase: s3cr3t warehouse: key_wh -with_routing: +unknown_key: account: test-account user: test-user authenticator: externalbrowser warehouse: test_wh routing: policy: database_override - database_name: OMER_DEV - -with_schema_isolation: - account: test-account - user: test-user - authenticator: externalbrowser - warehouse: test_wh - routing: - policy: schema_isolation - database_name: DEV - schema_name: obaddour """ environments_file = tmp_path / "environments.yml" environments_file.write_text(environments_content) return environments_file + + +@pytest.fixture +def routing_project(tmp_path: Path) -> Path: + """Create a project directory that holds a __routing__.py file.""" + project_dir = tmp_path / "routing_project" + project_dir.mkdir() + return project_dir diff --git a/tests/helpers.py b/tests/helpers.py new file mode 100644 index 0000000..cb17274 --- /dev/null +++ b/tests/helpers.py @@ -0,0 +1,38 @@ +"""Routing entries that the tests share. + +Clair ships no concrete ``RoutingEntry``. A user writes one. These two entries +give the tests the same two shapes that a user writes most often. +""" + +from __future__ import annotations + +from clair.environments.routing import RoutingEntry, TrouveAddress + + +class DatabaseOverrideRouting(RoutingEntry): + """Send every non-SOURCE Trouve to one database.""" + + environment_name: str = "dev" + database_name: str + + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + return trouve_address.model_copy(update={"database_name": self.database_name}) + + +class SchemaIsolationRouting(RoutingEntry): + """Collapse the three names into one table name under a fixed schema.""" + + environment_name: str = "dev" + database_name: str + schema_name: str + + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + collapsed_table_name = ( + f"{trouve_address.database_name}_{trouve_address.schema_name}_" + f"{trouve_address.table_name}" + ).upper() + return TrouveAddress( + database_name=self.database_name, + schema_name=self.schema_name, + table_name=collapsed_table_name, + ) diff --git a/tests/unit/test_discovery.py b/tests/unit/test_discovery.py index 97abc11..0fc0371 100644 --- a/tests/unit/test_discovery.py +++ b/tests/unit/test_discovery.py @@ -12,11 +12,11 @@ find_routing_collisions, recompile_for_selection, ) -from clair.environments.routing import DatabaseOverrideRouting, SchemaIsolationRouting from clair.trouves._refs import TROUVE_PLACEHOLDER_PREFIX from clair.trouves.run_config import RunMode from clair.trouves.test import TestSql from clair.trouves.trouve import TrouveType +from tests.helpers import DatabaseOverrideRouting, SchemaIsolationRouting class TestComputeFullName: diff --git a/tests/unit/test_environments.py b/tests/unit/test_environments.py index 36b51c6..225c1e3 100644 --- a/tests/unit/test_environments.py +++ b/tests/unit/test_environments.py @@ -7,12 +7,10 @@ import pytest from clair.environments.environments import load_environment -from clair.environments.routing import SchemaIsolationRouting from clair.exceptions import ( EnvironmentNotFoundError, EnvironmentsFileNotFoundError, - InvalidRoutingConfigError, - InvalidRoutingPolicyError, + InvalidEnvironmentError, ) @@ -57,42 +55,34 @@ def test_encrypted_private_key_environment(self, tmp_environments: Path): assert env.private_key_path == "/secrets/snowflake_key_enc.p8" assert env.private_key_passphrase == "s3cr3t" - def test_routing_parsed_for_database_override(self, tmp_environments: Path): - _, env = load_environment(env_name="with_routing", environments_path=tmp_environments) - assert env.routing is not None - assert env.routing.policy == "database_override" - assert env.routing.database_name == "OMER_DEV" + def test_environment_has_no_routing_attribute(self, tmp_environments: Path): + _, env = load_environment(env_name="dev", environments_path=tmp_environments) + assert not hasattr(env, "routing") - def test_routing_parsed_for_schema_isolation(self, tmp_environments: Path): - _, env = load_environment(env_name="with_schema_isolation", environments_path=tmp_environments) - assert isinstance(env.routing, SchemaIsolationRouting) - assert env.routing.policy == "schema_isolation" - assert env.routing.database_name == "DEV" - assert env.routing.schema_name == "obaddour" - def test_no_routing_returns_none(self, tmp_environments: Path): - _, env = load_environment(env_name="dev", environments_path=tmp_environments) - assert env.routing is None +class TestUnknownKey: + """Routing moved to the project __routing__.py. An old block must not pass silently.""" + def test_unknown_key_raises(self, tmp_environments: Path): + with pytest.raises(InvalidEnvironmentError, match="unknown_key"): + load_environment(env_name="unknown_key", environments_path=tmp_environments) -class TestLoadEnvironmentValidation: - def test_missing_policy_raises(self, tmp_path: Path): - bad = tmp_path / "env.yml" - bad.write_text("dev:\n account: x\n user: y\n warehouse: z\n routing:\n database_name: FOO\n") - with pytest.raises(InvalidRoutingConfigError, match="policy"): - load_environment(environments_path=bad) + def test_the_error_names_the_new_file(self, tmp_environments: Path): + with pytest.raises(InvalidEnvironmentError, match=r"__routing__\.py"): + load_environment(env_name="unknown_key", environments_path=tmp_environments) - def test_unknown_policy_raises(self, tmp_path: Path): + def test_a_misspelt_key_raises(self, tmp_path: Path): bad = tmp_path / "env.yml" - bad.write_text("dev:\n account: x\n user: y\n warehouse: z\n routing:\n policy: nonsense\n database_name: FOO\n") - with pytest.raises(InvalidRoutingPolicyError, match="nonsense"): + bad.write_text("dev:\n account: x\n user: y\n warehouse: z\n wharehouse: z\n") + with pytest.raises(InvalidEnvironmentError, match="wharehouse"): load_environment(environments_path=bad) - def test_schema_isolation_missing_schema_name_raises(self, tmp_path: Path): - bad = tmp_path / "env.yml" - bad.write_text("dev:\n account: x\n user: y\n warehouse: z\n routing:\n policy: schema_isolation\n database_name: DEV\n") - with pytest.raises(InvalidRoutingConfigError, match="schema_name"): - load_environment(environments_path=bad) + def test_environment_without_routing_block_loads(self, tmp_path: Path): + good = tmp_path / "env.yml" + good.write_text("dev:\n account: x\n user: y\n warehouse: z\n") + name, env = load_environment(environments_path=good) + assert name == "dev" + assert env.account == "x" class TestToConnectionDict: diff --git a/tests/unit/test_project_routing.py b/tests/unit/test_project_routing.py new file mode 100644 index 0000000..4b0fe9b --- /dev/null +++ b/tests/unit/test_project_routing.py @@ -0,0 +1,189 @@ +"""Tests for the project __routing__.py loader.""" + +from __future__ import annotations + +import os +import textwrap +from pathlib import Path + +import pytest + +from clair.environments.project_routing import ( + ROUTING_FILE_NAME, + load_project_routing, +) +from clair.environments.routing import route +from clair.exceptions import InvalidRoutingFileError +from clair.trouves.trouve import TrouveType + +# A routing file needs an entry class. This prelude gives the tests one. +_PRELUDE = """\ +import os + +from clair import RoutingEntry, RoutingTable, TrouveAddress + + +class DatabaseOverride(RoutingEntry): + environment_name: str = "dev" + database_name: str + + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + return trouve_address.model_copy(update={"database_name": self.database_name}) + + +""" + + +def _write_routing_file(project_dir: Path, body: str) -> Path: + """Write a routing file that holds only the given body.""" + path = project_dir / ROUTING_FILE_NAME + path.write_text(textwrap.dedent(body)) + return path + + +def _write_with_prelude(project_dir: Path, body: str) -> Path: + """Write a routing file that holds the entry prelude and the given body.""" + path = project_dir / ROUTING_FILE_NAME + path.write_text(_PRELUDE + textwrap.dedent(body)) + return path + + +class TestLoadProjectRouting: + def test_missing_file_gives_passthrough(self, routing_project: Path): + result = load_project_routing(routing_project, "dev") + assert result.entry is None + assert result.file_path is None + assert result.file_exists is False + + def test_loads_an_entry(self, routing_project: Path): + _write_with_prelude(routing_project, ''' + routing = RoutingTable(entries=[DatabaseOverride(database_name="OMER_DEV")]) + ''') + result = load_project_routing(routing_project, "dev") + assert result.entry is not None + # The entry is a user subclass, so read its own field through the model. + assert result.entry.model_dump()["database_name"] == "OMER_DEV" + assert route("a.b.c", TrouveType.TABLE, result.entry) == "OMER_DEV.b.c" + + def test_an_empty_table_gives_passthrough(self, routing_project: Path): + _write_with_prelude(routing_project, ''' + routing = RoutingTable(entries=[]) + ''') + result = load_project_routing(routing_project, "dev") + assert result.entry is None + assert result.file_exists is True + + def test_unknown_environment_reports_the_known_names(self, routing_project: Path): + _write_with_prelude(routing_project, ''' + routing = RoutingTable(entries=[ + DatabaseOverride(environment_name="dev", database_name="D"), + DatabaseOverride(environment_name="staging", database_name="S"), + ]) + ''') + result = load_project_routing(routing_project, "typo") + assert result.entry is None + assert result.environment_names == ["dev", "staging"] + assert result.is_unnamed_environment is True + + def test_a_named_environment_is_not_unnamed(self, routing_project: Path): + _write_with_prelude(routing_project, ''' + routing = RoutingTable(entries=[DatabaseOverride(database_name="D")]) + ''') + result = load_project_routing(routing_project, "dev") + assert result.is_unnamed_environment is False + + def test_an_entry_reads_an_environment_variable( + self, routing_project: Path, monkeypatch + ): + monkeypatch.setenv("CLAIR_USER", "obaddour") + _write_with_prelude(routing_project, ''' + class DeveloperRouting(RoutingEntry): + environment_name: str = "dev" + + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + user_name = os.environ["CLAIR_USER"].upper() + return trouve_address.model_copy(update={ + "database_name": f"{trouve_address.database_name}_{user_name}" + }) + + + routing = RoutingTable(entries=[DeveloperRouting()]) + ''') + result = load_project_routing(routing_project, "dev") + assert route("analytics.finance.revenue", TrouveType.TABLE, result.entry) == ( + "analytics_OBADDOUR.finance.revenue" + ) + + +class TestRoutingFileValidation: + def test_syntax_error_raises(self, routing_project: Path): + _write_routing_file(routing_project, "routing = [\n") + with pytest.raises(InvalidRoutingFileError, match="SyntaxError"): + load_project_routing(routing_project, "dev") + + def test_error_at_import_time_raises(self, routing_project: Path): + _write_routing_file(routing_project, ''' + raise ValueError("boom") + ''') + with pytest.raises(InvalidRoutingFileError, match="boom"): + load_project_routing(routing_project, "dev") + + def test_missing_routing_table_raises(self, routing_project: Path): + _write_routing_file(routing_project, "# no routing table here\n") + with pytest.raises(InvalidRoutingFileError, match="must define a 'routing'"): + load_project_routing(routing_project, "dev") + + def test_routing_that_is_not_a_table_raises(self, routing_project: Path): + _write_routing_file(routing_project, "routing = 'not a table'\n") + with pytest.raises(InvalidRoutingFileError, match="must be a RoutingTable"): + load_project_routing(routing_project, "dev") + + def test_a_dict_routing_table_raises(self, routing_project: Path): + """The old dict format must not pass without a word.""" + _write_routing_file(routing_project, 'routing = {"dev": None}\n') + with pytest.raises(InvalidRoutingFileError, match="must be a RoutingTable"): + load_project_routing(routing_project, "dev") + + def test_a_duplicate_environment_name_raises(self, routing_project: Path): + _write_with_prelude(routing_project, ''' + routing = RoutingTable(entries=[ + DatabaseOverride(database_name="A"), + DatabaseOverride(database_name="B"), + ]) + ''') + with pytest.raises(InvalidRoutingFileError, match="more than one entry"): + load_project_routing(routing_project, "dev") + + +class TestRoutingFileCache: + def test_repeated_loads_run_the_file_one_time(self, routing_project: Path): + _write_with_prelude(routing_project, ''' + os.environ["CLAIR_TEST_LOAD_COUNT"] = str( + int(os.environ.get("CLAIR_TEST_LOAD_COUNT", "0")) + 1 + ) + + routing = RoutingTable(entries=[]) + ''') + os.environ.pop("CLAIR_TEST_LOAD_COUNT", None) + try: + load_project_routing(routing_project, "dev") + load_project_routing(routing_project, "dev") + load_project_routing(routing_project, "dev") + assert os.environ["CLAIR_TEST_LOAD_COUNT"] == "1" + finally: + os.environ.pop("CLAIR_TEST_LOAD_COUNT", None) + + def test_an_edit_reloads_the_file(self, routing_project: Path): + _write_with_prelude(routing_project, ''' + routing = RoutingTable(entries=[DatabaseOverride(database_name="FIRST")]) + ''') + first = load_project_routing(routing_project, "dev") + assert first.entry is not None + assert first.entry.model_dump()["database_name"] == "FIRST" + + _write_with_prelude(routing_project, ''' + routing = RoutingTable(entries=[DatabaseOverride(database_name="SECOND")]) + ''') + second = load_project_routing(routing_project, "dev") + assert second.entry is not None + assert second.entry.model_dump()["database_name"] == "SECOND" diff --git a/tests/unit/test_routing.py b/tests/unit/test_routing.py index a9eaed1..bf0bbc1 100644 --- a/tests/unit/test_routing.py +++ b/tests/unit/test_routing.py @@ -1,18 +1,135 @@ -"""The tests of the routing policies.""" +"""Tests for TrouveAddress, RoutingEntry, RoutingTable, and route().""" from __future__ import annotations +import os + import pytest from pydantic import ValidationError from clair.environments.routing import ( - DatabaseOverrideRouting, - SchemaIsolationRouting, + RoutingEntry, + RoutingTable, + TrouveAddress, + collect_routing_problems, + describe_routing, detect_routing_collisions, route, ) -from clair.exceptions import InvalidRoutingConfigError +from clair.exceptions import InvalidRoutingConfigError, InvalidTrouveAddressError from clair.trouves.trouve import TrouveType +from tests.helpers import DatabaseOverrideRouting, SchemaIsolationRouting + + +class TestTrouveAddress: + def test_parse_splits_the_three_names(self): + address = TrouveAddress.parse("analytics.finance.revenue") + assert address.database_name == "analytics" + assert address.schema_name == "finance" + assert address.table_name == "revenue" + + def test_str_joins_the_three_names(self): + address = TrouveAddress.parse("analytics.finance.revenue") + assert str(address) == "analytics.finance.revenue" + + def test_address_is_frozen(self): + address = TrouveAddress.parse("a.b.c") + with pytest.raises(ValidationError): + setattr(address, "database_name", "other") # noqa: B010 + + def test_address_is_hashable(self): + assert len({TrouveAddress.parse("a.b.c"), TrouveAddress.parse("a.b.c")}) == 1 + + def test_two_parts_raise(self): + with pytest.raises(InvalidTrouveAddressError, match="3 dot-separated parts"): + TrouveAddress.parse("finance.revenue") + + def test_four_parts_raise(self): + with pytest.raises(InvalidTrouveAddressError, match="3 dot-separated parts"): + TrouveAddress.parse("a.b.c.d") + + @pytest.mark.parametrize( + "bad_name", + ["my-db.b.c", "a.my-schema.c", "a.b.my-table", "1db.b.c", "a..c"], + ) + def test_invalid_identifier_raises(self, bad_name: str): + with pytest.raises(InvalidTrouveAddressError, match="not a valid identifier"): + TrouveAddress.parse(bad_name) + + def test_the_message_names_the_bad_part(self): + with pytest.raises(InvalidTrouveAddressError, match="schema_name"): + TrouveAddress.parse("a.bad-schema.c") + + def test_identifier_over_255_characters_raises(self): + with pytest.raises(InvalidTrouveAddressError, match="255"): + TrouveAddress.parse(f"db.schema.{'a' * 256}") + + def test_dollar_sign_and_underscore_are_valid(self): + assert str(TrouveAddress.parse("_db.s$1.T_2")) == "_db.s$1.T_2" + + def test_direct_construction_validates(self): + with pytest.raises(ValidationError): + TrouveAddress(database_name="my-db", schema_name="b", table_name="c") + + +class TestRoutingEntry: + def test_the_base_class_needs_a_route_method(self): + with pytest.raises(TypeError): + RoutingEntry(environment_name="dev") + + def test_a_subclass_keeps_its_own_fields(self): + entry = DatabaseOverrideRouting( + environment_name="dev", database_name="OMER_DEV" + ) + assert entry.database_name == "OMER_DEV" + + def test_pydantic_validates_a_subclass_field(self): + """An absent database_name is an error, not a silent default.""" + with pytest.raises(ValidationError): + DatabaseOverrideRouting.model_validate({"environment_name": "dev"}) + + +class TestRoutingTable: + def test_entry_for_finds_the_named_entry(self): + entry = DatabaseOverrideRouting(environment_name="dev", database_name="DEV") + table = RoutingTable(entries=[entry]) + assert table.entry_for("dev") is entry + + def test_entry_for_gives_none_for_an_absent_name(self): + assert RoutingTable(entries=[]).entry_for("dev") is None + + def test_a_subclass_survives_the_table(self): + """Pydantic must not reduce an entry to its RoutingEntry base class.""" + table = RoutingTable( + entries=[SchemaIsolationRouting(database_name="DEV", schema_name="mine")] + ) + entry = table.entry_for("dev") + assert isinstance(entry, SchemaIsolationRouting) + assert entry.schema_name == "mine" + + def test_environment_names_are_sorted(self): + table = RoutingTable(entries=[ + DatabaseOverrideRouting(environment_name="prod", database_name="P"), + DatabaseOverrideRouting(environment_name="dev", database_name="D"), + ]) + assert table.environment_names == ["dev", "prod"] + + def test_duplicate_environment_name_raises(self): + with pytest.raises(ValidationError, match="more than one entry"): + RoutingTable(entries=[ + DatabaseOverrideRouting(environment_name="dev", database_name="A"), + DatabaseOverrideRouting(environment_name="dev", database_name="B"), + ]) + + def test_the_duplicate_message_names_the_environment(self): + with pytest.raises(ValidationError, match="staging"): + RoutingTable(entries=[ + DatabaseOverrideRouting(environment_name="staging", database_name="A"), + DatabaseOverrideRouting(environment_name="staging", database_name="B"), + ]) + + def test_an_empty_table_is_valid(self): + assert RoutingTable().environment_names == [] def _db_override(database_name: str) -> DatabaseOverrideRouting: @@ -25,64 +142,190 @@ def _schema_isolation(database_name: str, schema_name: str) -> SchemaIsolationRo class TestRoute: def test_passthrough_when_no_routing(self): - result = route("analytics.finance.revenue", TrouveType.TABLE, None) - assert result == "analytics.finance.revenue" + assert route("analytics.finance.revenue", TrouveType.TABLE, None) == ( + "analytics.finance.revenue" + ) def test_source_passthrough_with_routing(self): - routing = _db_override("OMER_DEV") - result = route("analytics.finance.revenue", TrouveType.SOURCE, routing) + result = route( + "analytics.finance.revenue", TrouveType.SOURCE, _db_override("OMER_DEV") + ) assert result == "analytics.finance.revenue" - def test_source_passthrough_with_schema_isolation(self): - routing = _schema_isolation("DEV", "obaddour") - result = route("refined.products.catalog", TrouveType.SOURCE, routing) - assert result == "refined.products.catalog" - def test_database_override_table(self): - routing = _db_override("OMER_DEV") - result = route("analytics.finance.revenue", TrouveType.TABLE, routing) + result = route( + "analytics.finance.revenue", TrouveType.TABLE, _db_override("OMER_DEV") + ) assert result == "OMER_DEV.finance.revenue" def test_database_override_view(self): - routing = _db_override("OMER_DEV") - result = route("analytics.finance.revenue", TrouveType.VIEW, routing) + result = route( + "analytics.finance.revenue", TrouveType.VIEW, _db_override("OMER_DEV") + ) assert result == "OMER_DEV.finance.revenue" - def test_database_override_preserves_schema_and_table(self): - routing = _db_override("MY_DEV_DB") - result = route("warehouse.orders.daily", TrouveType.TABLE, routing) + def test_database_override_keeps_the_schema_and_the_table(self): + result = route( + "warehouse.orders.daily", TrouveType.TABLE, _db_override("MY_DEV_DB") + ) assert result == "MY_DEV_DB.orders.daily" def test_schema_isolation_table(self): - routing = _schema_isolation("DEV", "obaddour") - result = route("refined.products.catalog", TrouveType.TABLE, routing) + result = route( + "refined.products.catalog", + TrouveType.TABLE, + _schema_isolation("DEV", "obaddour"), + ) assert result == "DEV.obaddour.REFINED_PRODUCTS_CATALOG" - def test_schema_isolation_concatenates_all_three_parts(self): - routing = _schema_isolation("DEV", "myschema") - result = route("analytics.finance.revenue", TrouveType.TABLE, routing) - assert result == "DEV.myschema.ANALYTICS_FINANCE_REVENUE" + def test_route_validates_the_logical_name(self): + """A file system name that Snowflake cannot use is an error, with no entry.""" + with pytest.raises(InvalidTrouveAddressError, match="not a valid identifier"): + route("my-db.finance.revenue", TrouveType.TABLE, None) - def test_schema_isolation_identifier_exceeds_255_chars_raises(self): + def test_route_validates_the_logical_name_of_a_source(self): + with pytest.raises(InvalidTrouveAddressError): + route("my-db.finance.revenue", TrouveType.SOURCE, None) + + def test_an_entry_that_makes_a_long_name_raises(self): routing = _schema_isolation("DEV", "myschema") - long_table = "a" * 250 with pytest.raises(InvalidRoutingConfigError, match="255"): - route(f"db.schema.{long_table}", TrouveType.TABLE, routing) + route(f"db.schema.{'a' * 250}", TrouveType.TABLE, routing) + + def test_the_error_names_the_entry_and_the_trouve(self): + routing = _schema_isolation("DEV", "myschema") + with pytest.raises(InvalidRoutingConfigError) as exc_info: + route(f"db.schema.{'a' * 250}", TrouveType.TABLE, routing) + message = str(exc_info.value) + assert "SchemaIsolationRouting" in message + assert "db.schema." in message + - def test_no_routing_passthrough_for_view(self): - result = route("analytics.finance.summary", TrouveType.VIEW, None) - assert result == "analytics.finance.summary" +class TestRouteRejectsABadEntry: + def test_an_entry_that_gives_a_string_raises(self): + class StringRouting(RoutingEntry): + environment_name: str = "dev" + + def route(self, trouve_address): + return "a.b.c" + + with pytest.raises( + InvalidRoutingConfigError, match="must give a TrouveAddress" + ): + route("analytics.finance.revenue", TrouveType.TABLE, StringRouting()) + + def test_an_entry_that_gives_none_raises(self): + class NoneRouting(RoutingEntry): + environment_name: str = "dev" + + def route(self, trouve_address): + return None + + with pytest.raises(InvalidRoutingConfigError, match="NoneType"): + route("analytics.finance.revenue", TrouveType.TABLE, NoneRouting()) + + def test_an_entry_that_raises_is_wrapped(self, monkeypatch): + monkeypatch.delenv("CLAIR_USER", raising=False) + + class UserRouting(RoutingEntry): + environment_name: str = "dev" + + def route(self, trouve_address): + user = os.environ["CLAIR_USER"] + return trouve_address.model_copy( + update={"database_name": f"{trouve_address.database_name}_{user}"} + ) + + with pytest.raises(InvalidRoutingConfigError, match="KeyError"): + route("analytics.finance.revenue", TrouveType.TABLE, UserRouting()) + + def test_an_entry_that_builds_a_bad_address_raises(self): + class DashRouting(RoutingEntry): + environment_name: str = "dev" + + def route(self, trouve_address): + return TrouveAddress( + database_name="my-db", + schema_name=trouve_address.schema_name, + table_name=trouve_address.table_name, + ) + + with pytest.raises(InvalidRoutingConfigError, match="not a valid identifier"): + route("analytics.finance.revenue", TrouveType.TABLE, DashRouting()) + + def test_an_entry_reads_an_environment_variable(self, monkeypatch): + monkeypatch.setenv("CLAIR_USER", "obaddour") + + class UserRouting(RoutingEntry): + environment_name: str = "dev" + + def route(self, trouve_address): + user = os.environ["CLAIR_USER"].upper() + return trouve_address.model_copy( + update={"database_name": f"{trouve_address.database_name}_{user}"} + ) + + result = route("analytics.finance.revenue", TrouveType.TABLE, UserRouting()) + assert result == "analytics_OBADDOUR.finance.revenue" + + +class TestDescribeRouting: + def test_describes_none(self): + assert describe_routing(None) == "none" + + def test_names_the_class_and_the_fields(self): + description = describe_routing(_db_override("OMER_DEV")) + assert "DatabaseOverrideRouting" in description + assert "OMER_DEV" in description + + def test_names_a_second_field(self): + description = describe_routing(_schema_isolation("DEV", "obaddour")) + assert "DEV" in description + assert "obaddour" in description + + def test_the_description_stays_on_one_line(self): + assert "\n" not in describe_routing(_db_override("OMER_DEV")) + + def test_a_long_description_is_cut_short(self): + description = describe_routing(_db_override("A" * 300)) + assert len(description) == 200 + assert description.endswith("…") + + +class TestLogicalNameValidation: + """A directory name that Snowflake cannot use stops every command.""" + + def test_a_bad_directory_name_stops_discovery(self, tmp_path): + from clair.core.discovery import discover_project + + (tmp_path / "my-db" / "finance").mkdir(parents=True) + (tmp_path / "my-db" / "finance" / "revenue.py").write_text( + "from clair import Trouve, TrouveType\n\n" + 'trouve = Trouve(type=TrouveType.TABLE, sql="SELECT 1 AS x")\n' + ) + with pytest.raises(InvalidTrouveAddressError, match="my-db"): + discover_project(tmp_path, routing=None) + + def test_a_good_directory_name_passes_discovery(self, tmp_path): + from clair.core.discovery import discover_project + + (tmp_path / "my_db" / "finance").mkdir(parents=True) + (tmp_path / "my_db" / "finance" / "revenue.py").write_text( + "from clair import Trouve, TrouveType\n\n" + 'trouve = Trouve(type=TrouveType.TABLE, sql="SELECT 1 AS x")\n' + ) + trouves = discover_project(tmp_path, routing=None) + assert collect_routing_problems(trouves, None) == [] class TestDetectRoutingCollisions: def test_no_collision_returns_empty(self): - result = detect_routing_collisions({ + assert detect_routing_collisions({ "analytics.finance.revenue": "OMER_DEV.finance.revenue", "warehouse.orders.daily": "OMER_DEV.orders.daily", - }) - assert result == [] + }) == [] - def test_collision_returns_target_and_sources(self): + def test_collision_gives_the_target_and_the_sources(self): result = detect_routing_collisions({ "analytics.finance.orders": "OMER_DEV.finance.orders", "warehouse.finance.orders": "OMER_DEV.finance.orders", @@ -90,24 +333,15 @@ def test_collision_returns_target_and_sources(self): assert len(result) == 1 target, sources = result[0] assert target == "OMER_DEV.finance.orders" - assert sorted(sources) == ["analytics.finance.orders", "warehouse.finance.orders"] + assert sorted(sources) == [ + "analytics.finance.orders", + "warehouse.finance.orders", + ] def test_empty_dict_returns_empty(self): assert detect_routing_collisions({}) == [] def test_single_entry_returns_empty(self): - assert detect_routing_collisions({"analytics.finance.revenue": "OMER_DEV.finance.revenue"}) == [] - - -class TestRoutingConfigValidation: - def test_schema_isolation_requires_schema_name(self): - with pytest.raises(ValidationError): - SchemaIsolationRouting.model_validate({"database_name": "DEV"}) - - def test_database_override_does_not_require_schema_name(self): - config = DatabaseOverrideRouting(database_name="OMER_DEV") - assert not hasattr(config, "schema_name") - - def test_schema_name_field(self): - config = SchemaIsolationRouting(database_name="DEV", schema_name="myschema") - assert config.schema_name == "myschema" + assert detect_routing_collisions( + {"analytics.finance.revenue": "OMER_DEV.finance.revenue"} + ) == [] diff --git a/tests/unit/test_scaffold.py b/tests/unit/test_scaffold.py index a301993..95e4d8f 100644 --- a/tests/unit/test_scaffold.py +++ b/tests/unit/test_scaffold.py @@ -50,11 +50,17 @@ def test_creates_environments_yml(self, tmp_path: Path) -> None: environments_path = fake_home / ".clair" / "environments.yml" assert environments_path.exists() + def test_creates_routing_file(self, tmp_path: Path) -> None: + project_dir = tmp_path / "my_project" + scaffold_project(project_dir, **DEFAULT_SOURCE_ARGS, home_dir=tmp_path / "home") + + assert (project_dir / "__routing__.py").exists() + def test_returns_all_paths_as_created(self, tmp_path: Path) -> None: results = _run_scaffold(tmp_path) - # 1 project file and 1 environments.yml file. - assert len(results) == 2 + # 1 source Trouve, 1 __routing__.py, and 1 environments.yml file. + assert len(results) == 3 assert all(status == "created" for status, _ in results) @@ -74,7 +80,17 @@ def test_environments_yml_has_dev_environment(self, tmp_path: Path) -> None: assert "dev:" in content assert "account:" in content assert "externalbrowser" in content - assert " routing:" not in content # By default the file has no routing block. The template shows one in a comment. + # Routing lives in the project __routing__.py, never in environments.yml. + assert " routing:" not in content + + def test_routing_file_defines_a_routing_table(self, tmp_path: Path) -> None: + project_dir = tmp_path / "proj" + scaffold_project(project_dir, **DEFAULT_SOURCE_ARGS, home_dir=tmp_path / "home") + + content = (project_dir / "__routing__.py").read_text() + assert "RoutingTable(" in content + assert "CLAIR_USER" in content + assert "TrouveAddress" in content class TestDoesNotOverwriteExistingFiles: diff --git a/tests/unit/test_validate_cli.py b/tests/unit/test_validate_cli.py new file mode 100644 index 0000000..6e2f470 --- /dev/null +++ b/tests/unit/test_validate_cli.py @@ -0,0 +1,242 @@ +"""Tests for the `clair validate` command.""" + +from __future__ import annotations + +import textwrap +from pathlib import Path + +import pytest +from click.testing import CliRunner + +from clair.cli.main import cli + +# Every routing file in these tests needs an entry class. This prelude gives one. +_PRELUDE = """\ +import os + +from clair import RoutingEntry, RoutingTable, TrouveAddress + + +""" + + +@pytest.fixture +def project_with_trouves(tmp_path: Path) -> Path: + """Build a project with one SOURCE Trouve and two TABLE Trouves.""" + project_dir = tmp_path / "proj" + (project_dir / "source" / "raw").mkdir(parents=True) + (project_dir / "source" / "raw" / "orders.py").write_text( + "from clair import Trouve, TrouveType\n\n" + "trouve = Trouve(type=TrouveType.SOURCE)\n" + ) + for schema_name in ("finance", "reports"): + (project_dir / "analytics" / schema_name).mkdir(parents=True) + (project_dir / "analytics" / schema_name / "revenue.py").write_text( + "from clair import Trouve, TrouveType\n\n" + 'trouve = Trouve(type=TrouveType.TABLE, sql="SELECT 1 AS x")\n' + ) + return project_dir + + +def _write_routing(project_dir: Path, body: str, prelude: bool = True) -> None: + content = textwrap.dedent(body) + (project_dir / "__routing__.py").write_text( + _PRELUDE + content if prelude else content + ) + + +def _run_validate(project_dir: Path, *args: str): + return CliRunner().invoke(cli, ["validate", "--project", str(project_dir), *args]) + + +# A routing entry that adds "_dev" to the database name. +_DEV_SUFFIX_ENTRY = ''' + class DevRouting(RoutingEntry): + environment_name: str = "dev" + + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + return trouve_address.model_copy(update={ + "database_name": f"{trouve_address.database_name}_dev" + }) + + + routing = RoutingTable(entries=[DevRouting()]) +''' + + +class TestValidateSucceeds: + def test_no_routing_file_passes(self, project_with_trouves: Path): + result = _run_validate(project_with_trouves) + assert result.exit_code == 0 + assert "Every routed name is valid" in result.output + + def test_a_valid_entry_passes(self, project_with_trouves: Path): + _write_routing(project_with_trouves, _DEV_SUFFIX_ENTRY) + result = _run_validate(project_with_trouves) + assert result.exit_code == 0 + assert "Every routed name is valid" in result.output + + def test_output_names_the_environment_and_the_entry( + self, project_with_trouves: Path + ): + _write_routing(project_with_trouves, _DEV_SUFFIX_ENTRY) + result = _run_validate(project_with_trouves) + assert "environment: dev" in result.output + # The description names the class, not the base class. + assert "DevRouting" in result.output + + def test_counts_only_the_routable_trouves(self, project_with_trouves: Path): + # 2 TABLE Trouves route. The SOURCE Trouve never routes. + result = _run_validate(project_with_trouves) + assert "Trouves to route: 2" in result.output + + +class TestValidateFails: + def test_an_invalid_identifier_fails(self, project_with_trouves: Path): + _write_routing(project_with_trouves, ''' + class DashRouting(RoutingEntry): + environment_name: str = "dev" + + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + return TrouveAddress( + database_name=f"{trouve_address.database_name}-dev", + schema_name=trouve_address.schema_name, + table_name=trouve_address.table_name, + ) + + + routing = RoutingTable(entries=[DashRouting()]) + ''') + result = _run_validate(project_with_trouves) + assert result.exit_code == 1 + assert "not a valid identifier" in result.output + + def test_reports_every_bad_trouve_not_only_the_first( + self, project_with_trouves: Path + ): + _write_routing(project_with_trouves, ''' + class DashRouting(RoutingEntry): + environment_name: str = "dev" + + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + return TrouveAddress( + database_name=f"{trouve_address.database_name}-dev", + schema_name=trouve_address.schema_name, + table_name=trouve_address.table_name, + ) + + + routing = RoutingTable(entries=[DashRouting()]) + ''') + result = _run_validate(project_with_trouves) + assert "analytics.finance.revenue" in result.output + assert "analytics.reports.revenue" in result.output + assert "2 problems found" in result.output + + def test_a_collision_fails(self, project_with_trouves: Path): + _write_routing(project_with_trouves, ''' + class SharedRouting(RoutingEntry): + environment_name: str = "dev" + + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + return trouve_address.model_copy(update={ + "database_name": "DEV", "schema_name": "shared" + }) + + + routing = RoutingTable(entries=[SharedRouting()]) + ''') + result = _run_validate(project_with_trouves) + assert result.exit_code == 1 + assert "DEV.shared.revenue" in result.output + assert "analytics.finance.revenue" in result.output + assert "analytics.reports.revenue" in result.output + + def test_an_entry_that_raises_fails(self, project_with_trouves: Path, monkeypatch): + monkeypatch.delenv("CLAIR_USER", raising=False) + _write_routing(project_with_trouves, ''' + class UserRouting(RoutingEntry): + environment_name: str = "dev" + + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + user_name = os.environ["CLAIR_USER"] + return trouve_address.model_copy(update={ + "database_name": f"{trouve_address.database_name}_{user_name}" + }) + + + routing = RoutingTable(entries=[UserRouting()]) + ''') + result = _run_validate(project_with_trouves) + assert result.exit_code == 1 + assert "CLAIR_USER" in result.output + + def test_a_broken_routing_file_fails(self, project_with_trouves: Path): + _write_routing(project_with_trouves, "routing = [\n", prelude=False) + result = _run_validate(project_with_trouves) + assert result.exit_code == 1 + + def test_a_duplicate_environment_name_fails(self, project_with_trouves: Path): + _write_routing(project_with_trouves, ''' + class DevRouting(RoutingEntry): + environment_name: str = "dev" + + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + return trouve_address + + + routing = RoutingTable(entries=[DevRouting(), DevRouting()]) + ''') + result = _run_validate(project_with_trouves) + assert result.exit_code == 1 + + def test_a_bad_directory_name_fails(self, tmp_path: Path): + """A directory that Snowflake cannot use as a name stops validate.""" + project_dir = tmp_path / "proj" + (project_dir / "my-db" / "finance").mkdir(parents=True) + (project_dir / "my-db" / "finance" / "revenue.py").write_text( + "from clair import Trouve, TrouveType\n\n" + 'trouve = Trouve(type=TrouveType.TABLE, sql="SELECT 1 AS x")\n' + ) + result = _run_validate(project_dir) + assert result.exit_code == 1 + + +class TestUnnamedEnvironmentWarning: + def test_an_absent_environment_warns(self, project_with_trouves: Path): + _write_routing(project_with_trouves, _DEV_SUFFIX_ENTRY) + result = _run_validate(project_with_trouves, "--env", "typo") + assert "does not name the environment 'typo'" in result.output + + def test_a_named_environment_does_not_warn(self, project_with_trouves: Path): + """An entry for the environment is a decision, so it must stay quiet.""" + _write_routing(project_with_trouves, ''' + class ProdRouting(RoutingEntry): + environment_name: str = "prod" + + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + return trouve_address + + + routing = RoutingTable(entries=[ProdRouting()]) + ''') + result = _run_validate(project_with_trouves, "--env", "prod") + assert result.exit_code == 0 + assert "does not name the environment" not in result.output + + def test_the_env_var_selects_the_environment( + self, project_with_trouves: Path, monkeypatch + ): + monkeypatch.setenv("CLAIR_ENV", "staging") + _write_routing(project_with_trouves, ''' + class StagingRouting(RoutingEntry): + environment_name: str = "staging" + + def route(self, trouve_address: TrouveAddress) -> TrouveAddress: + return trouve_address + + + routing = RoutingTable(entries=[StagingRouting()]) + ''') + result = _run_validate(project_with_trouves) + assert "environment: staging" in result.output