diff --git a/data/.env.example b/data/.env.example index 7773688..258a855 100644 --- a/data/.env.example +++ b/data/.env.example @@ -71,23 +71,30 @@ DATABRICKS_TOKEN= # cannot. Nothing you run on your machine needs it. # --- Backend database ------------------------------------------------------- -# You own the `analytics` schema and publish marts into it. The backend owns -# `app` and you can read it. Neither side can write to the other's, which is -# enforced by two roles rather than by agreement. +# You publish to the real database, not to a copy on your laptop. Connecting to +# the real one is what exercises TLS, the firewall and your grants, and those +# are exactly what breaks on a first scheduled run if nobody tried them. # -# `scripts/db-setup.py` at the repository root creates the database, both -# schemas and both roles, and prints the `analytics_user` password once. Run it -# against the Postgres that `docker compose up -d db` starts, and you have a -# stand-in for the real database before the real one exists: +# Three schemas, and which you may write is enforced by your role rather than +# by agreement: # -# pip install "psycopg[binary]" -# python ../scripts/db-setup.py --host localhost --port 5432 \ -# --admin-user admin --admin-password password -BACKEND_PG_HOST=localhost +# analytics the scheduled run writes it, you can read it +# analytics_dev you write it, the scheduled run cannot even read it +# app the backend writes it, you can read it +# +# Same table names in both analytics schemas, so promoting a mart changes where +# it lives and nothing the backend selects. You and your teammates share +# `analytics_dev`, so the last publish wins: the table carries a comment saying +# which schema it came from and when, which is how you tell whose run you are +# looking at. +BACKEND_PG_HOST= # your teacher gives you this BACKEND_PG_PORT=5432 BACKEND_PG_DB=project_db -BACKEND_PG_USER=analytics_user -BACKEND_PG_PASSWORD= # printed by db-setup.py, shown once -BACKEND_PG_PUBLISH_SCHEMA=analytics -# The local container has no TLS certificate; Azure requires one. -BACKEND_PG_SSLMODE=prefer +# Not `analytics_user`. That role writes production and its password is +# readable only by your team's Airflow VM. +BACKEND_PG_USER=analytics_dev_user +BACKEND_PG_PASSWORD= # your teacher gives you this +BACKEND_PG_PUBLISH_SCHEMA=analytics_dev +# Azure requires TLS. Leave this alone unless you are pointing at a container +# on your own machine, which has no certificate. +BACKEND_PG_SSLMODE=require diff --git a/data/README.md b/data/README.md index 839e135..bc9f1f5 100644 --- a/data/README.md +++ b/data/README.md @@ -316,7 +316,8 @@ and they are the first thing to fill in: | `LANDING_PREFIX` | `your-name` | `raw` | | `LANDING_PATH` | `/Volumes//landing/dev/your-name/postings` | `.../landing/raw/postings` | | `DBT_SCHEMA` | `dev_yourname` | `analytics` | -| `BACKEND_PG_HOST` | your own Postgres in Docker | the backend's database | +| `BACKEND_PG_PUBLISH_SCHEMA` | `analytics_dev` | `analytics` | +| `BACKEND_PG_USER` | `analytics_dev_user` | `analytics_user` | This is not a naming convention you have to remember. It is what your account is allowed to do. You can write the `dev` container and only read `landing`. @@ -359,8 +360,21 @@ docker exec -it $(docker ps -qf name=scheduler) \ ``` It reads `.dev_yourname.fct_postings_enriched` and writes -`analytics.fct_postings` in your own Postgres, the one `scripts/db-setup.py` -created. `dbt_build` runs the same way. The `ingest` task does not: it starts a +`analytics_dev.fct_postings` in the real backend database. Same table name as +production, one schema across, so promoting it later changes nothing the +backend selects. `dbt_build` runs the same way. + +You share `analytics_dev` with your teammates, so the last publish wins. The +table carries a comment saying where the rows came from, which is how you tell +whose run you are looking at: + +``` +\d+ analytics_dev.fct_postings --> from team_a.dev_alex at 2026-08-13T11:21Z +``` + +Point `BACKEND_PG_PUBLISH_SCHEMA` at `analytics` by mistake and the run stops +with a permission error. `analytics_dev_user` cannot write production, which is +the point of it being a separate role. The `ingest` task does not: it starts a Container Apps job, which needs the VM's identity, so run that one as the script above. diff --git a/data/airflow/dags/pipeline_dag.py b/data/airflow/dags/pipeline_dag.py index e400b2b..e6893c5 100644 --- a/data/airflow/dags/pipeline_dag.py +++ b/data/airflow/dags/pipeline_dag.py @@ -231,10 +231,16 @@ def publish_to_backend() -> int: ) return publish( dsn, + # Your own runs publish to `analytics_dev`, which you may write and + # the scheduled run cannot even read. The default is production, + # because on the VM there is no .env to say otherwise. setting("BACKEND_PG_PUBLISH_SCHEMA", "analytics"), + # The same table name in both schemas, so promotion changes where + # the table lives and never what the backend selects. "fct_postings", columns, rows, + source=f"{Warehouse.from_env().catalog}.{setting('DBT_SCHEMA')}", ) ingest() >> dbt_build() >> publish_to_backend() diff --git a/data/airflow/docker-compose.override.yml b/data/airflow/docker-compose.override.yml index 545f48e..157cd78 100644 --- a/data/airflow/docker-compose.override.yml +++ b/data/airflow/docker-compose.override.yml @@ -6,12 +6,18 @@ # folder, so there is exactly one copy of each. The bind mounts make them # visible inside the Airflow containers rather than duplicating them. # -# 2. The containers join the "finalproject" network created by -# ../../docker-compose.yml, so they reach the database as "db" on port -# 5432. Going through host.docker.internal instead would break on any -# machine that already runs something on port 5432. +# 2. Nothing else. The containers reach the outside world normally, which is +# all they need: the publish task writes `analytics_dev` on the team's real +# Postgres, dbt talks to Databricks, and neither is on this machine. +# +# There used to be a third thing here. The stack joined a "finalproject" +# network and pinned BACKEND_PG_HOST to the local `db` container, back when +# local runs published to a Postgres in Docker. That is gone: overriding the +# host here would silently send a task to a different database than the one +# `uv run` writes, which is the opposite of what this file is for. It also +# meant `astro dev start` failed unless you had first started a database the +# data pipeline no longer touches. # -# Start the database first: (cd ../.. && docker compose up -d db) # When you deploy, bake these paths into the image with COPY. x-local: &local volumes: @@ -33,15 +39,6 @@ x-local: &local # Astro mounts the project here; the team VM uses /opt/airflow. The DAG # reads this rather than hardcoding either one. DBT_PROJECT_DIR: /usr/local/airflow/include/dbt - # Reach the database by its service name: your .env says localhost, which - # inside this container would mean the container itself. - BACKEND_PG_HOST: db - BACKEND_PG_PORT: "5432" - # The local Postgres has no TLS certificate. - BACKEND_PG_SSLMODE: prefer - networks: - - default - - finalproject services: scheduler: @@ -52,7 +49,3 @@ services: <<: *local triggerer: <<: *local - -networks: - finalproject: - external: true diff --git a/data/src/publishing/sync.py b/data/src/publishing/sync.py index e96ef2a..b62b13d 100644 --- a/data/src/publishing/sync.py +++ b/data/src/publishing/sync.py @@ -5,10 +5,11 @@ """ import logging +from datetime import UTC, datetime from typing import LiteralString import psycopg -from psycopg.sql import SQL, Identifier, Placeholder +from psycopg.sql import SQL, Identifier, Literal, Placeholder from ..common.warehouse import Queryable @@ -63,12 +64,23 @@ def read_backend_table(dsn: str, table: str, schema: str = "app") -> list[dict]: def publish( - dsn: str, schema: str, table: str, columns: list[tuple[str, str]], rows: list[list] + dsn: str, + schema: str, + table: str, + columns: list[tuple[str, str]], + rows: list[list], + source: str | None = None, ) -> int: """Replace the backend's copy of the table, return the row count written. Load into staging, then swap inside one transaction, so a reader sees the whole old version or the whole new one and never a half-written table. + + `source` is the warehouse schema the rows came from, and it is recorded as a + comment on the table. One shared `analytics_dev` means the last publish wins, + which is the right behaviour for a place two tracks meet but leaves nobody + able to say why the columns changed this morning. The comment answers it in + any client: `from team_a.dev_alex at 2026-08-13T11:02Z`. """ if not rows: raise ValueError("refusing to publish zero rows over an existing table") @@ -99,6 +111,16 @@ def publish( # when there is nothing to replace yet. cursor.execute(SQL("drop table if exists {}").format(published)) cursor.execute(SQL("alter table {} rename to {}").format(staging, Identifier(table))) + if source: + # A comment, not a column: it describes the table rather than + # every row in it, and it survives the swap without widening + # what the backend has to select. + stamp = datetime.now(tz=UTC).strftime("%Y-%m-%dT%H:%MZ") + cursor.execute( + SQL("comment on table {} is {}").format( + published, Literal(f"from {source} at {stamp}") + ) + ) connection.commit() finally: connection.close() diff --git a/data/tests/publishing/test_sync.py b/data/tests/publishing/test_sync.py index 55acdc7..39fed22 100644 --- a/data/tests/publishing/test_sync.py +++ b/data/tests/publishing/test_sync.py @@ -127,3 +127,29 @@ def test_reading_an_empty_mart_is_refused(): warehouse = FakeWarehouse() with pytest.raises(ValueError, match="no rows"): sync.read_mart(warehouse, "main", "fct_postings_enriched") + + +def test_the_source_schema_is_stamped_on_the_table(connection): + """One shared `analytics_dev` means the last publish wins, which is right for + a place two tracks meet but leaves nobody able to say why the columns changed. + The comment names the warehouse schema the rows came from.""" + sync.publish("dsn", "analytics_dev", "fct_postings", COLUMNS, ROWS, + source="team_a.dev_alex") + + comment = connection.log[index_of(connection.log, "comment on table")] + assert '"analytics_dev"."fct_postings"' in comment + assert "from team_a.dev_alex at " in comment + + +def test_the_stamp_lands_after_the_swap(connection): + """Comment the published table, not the staging one: the rename would carry + the comment across, but only by accident of ordering.""" + sync.publish("dsn", "analytics_dev", "fct_postings", COLUMNS, ROWS, source="s") + assert index_of(connection.log, "rename to") < index_of(connection.log, "comment on table") + + +def test_no_source_means_no_comment(connection): + """Callers that do not know where the rows came from should not write a + misleading stamp, and an unstamped table is better than a wrong one.""" + sync.publish("dsn", "analytics", "fct_postings", COLUMNS, ROWS) + assert not any("comment on table" in statement for statement in connection.log) diff --git a/scripts/db-setup.py b/scripts/db-setup.py index 169e418..c621b6a 100755 --- a/scripts/db-setup.py +++ b/scripts/db-setup.py @@ -1,9 +1,16 @@ #!/usr/bin/env python3 -"""Set up the Postgres database with an 'app' and an 'analytics' schema. +"""Set up the Postgres database with 'app', 'analytics' and 'analytics_dev'. -Creates the database, both schemas, and one login role per schema. Each role -administers its own schema and gets read-only access to the other one, for both -existing and future objects. +Creates the database, all three schemas, and one login role per schema. Each +role administers its own schema and gets read-only access to the others it +needs, for both existing and future objects. + +The third schema is where the data track publishes while it is still building. +It exists so a trainee can run the publish step against the real database +rather than a stand-in on their laptop, which is what exercises TLS, the +firewall and the grants before the first scheduled run does. Its role cannot +write `analytics`, so an `.env` still pointing at production fails with a +permission error instead of replacing what the backend serves. The script is idempotent: re-running it never changes existing state, so a failed run can simply be repeated. Existing roles keep their current password @@ -43,20 +50,37 @@ APP_SCHEMA = "app" ANALYTICS_SCHEMA = "analytics" +ANALYTICS_DEV_SCHEMA = "analytics_dev" APP_ROLE = "app_user" ANALYTICS_ROLE = "analytics_user" +ANALYTICS_DEV_ROLE = "analytics_dev_user" class SchemaAccess(NamedTuple): - """The schema a role fully administers, and the schema it may only read.""" + """The schema a role fully administers, and the ones it may only read.""" administers: str - read_only: str + reads: tuple[str, ...] +# Who may write what, which is the whole dev/production boundary in one table. +# +# `analytics` is written by exactly one role, and that role's password is +# readable only by the team's Airflow VM. Trainees publish to `analytics_dev` +# instead and can read `analytics` to compare, but not write it. +# +# The backend reads both, so its developers can build against a table the data +# track is still shaping without waiting for it to reach production. ROLE_LAYOUT = { - APP_ROLE: SchemaAccess(administers=APP_SCHEMA, read_only=ANALYTICS_SCHEMA), - ANALYTICS_ROLE: SchemaAccess(administers=ANALYTICS_SCHEMA, read_only=APP_SCHEMA), + APP_ROLE: SchemaAccess( + administers=APP_SCHEMA, reads=(ANALYTICS_SCHEMA, ANALYTICS_DEV_SCHEMA) + ), + # Deliberately cannot read `analytics_dev`: production must never end up + # depending on a table somebody is still editing from a laptop. + ANALYTICS_ROLE: SchemaAccess(administers=ANALYTICS_SCHEMA, reads=(APP_SCHEMA,)), + ANALYTICS_DEV_ROLE: SchemaAccess( + administers=ANALYTICS_DEV_SCHEMA, reads=(APP_SCHEMA, ANALYTICS_SCHEMA) + ), } class Privileges(NamedTuple): @@ -289,12 +313,12 @@ def report(args: argparse.Namespace, passwords: dict[str, str | None]) -> None: unchanged = "(unchanged)" print(f"\n✅ Setup complete on {args.host}:{args.port}\n") print(f" database : {NEW_DATABASE}") - print(f" schemas : {APP_SCHEMA}, {ANALYTICS_SCHEMA}\n") + print(f" schemas : {APP_SCHEMA}, {ANALYTICS_SCHEMA}, {ANALYTICS_DEV_SCHEMA}\n") for role, access in ROLE_LAYOUT.items(): + reads = ", ".join(f"'{schema}'" for schema in access.reads) print(f" {role}") print(f" password : {passwords[role] or unchanged}") - print(f" access : full on '{access.administers}', " - f"read-only on '{access.read_only}'") + print(f" access : full on '{access.administers}', read-only on {reads}") # --- Entry point ----------------------------------------------------------- @@ -327,7 +351,8 @@ def main() -> None: creators = [args.admin_user, *roles] for role, access in ROLE_LAYOUT.items(): grant_access(conn, access.administers, role, FULL_ACCESS, creators) - grant_access(conn, access.read_only, role, READ_ONLY, creators) + for schema in access.reads: + grant_access(conn, schema, role, READ_ONLY, creators) report(args, passwords)