Publish local runs to analytics_dev on the real database

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

data/README.md

@@ -316,7 +316,8 @@ and they are the first thing to fill in:
 | `LANDING_PREFIX` | `your-name` | `raw` |
 | `LANDING_PATH` | `/Volumes/<catalog>/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 `<catalog>.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.
 

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()

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

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()

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)

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)