diff --git a/.gitignore b/.gitignore index f1c4b47..f4427fa 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,6 @@ dbt/logs/ # Optional fast-reset dump (generated by scripts/capture_initial_state.sh) data/initial_state.dump + +# Python bytecode from scripts/ (the loader container imports across script modules) +__pycache__/ diff --git a/README.md b/README.md index 08b6f0b..0118749 100644 --- a/README.md +++ b/README.md @@ -1,14 +1,23 @@ -# Data coding exercise +# Data platform onboarding exercise: Meridian Live (MARI portfolio company) -Evaluate the raw source data, identify business metrics a stakeholder would care about, and build dbt models to deliver analysis-ready datasets. +Design and build how a newly-acquired MARI portfolio company's ticketing data lands in TTG's dbt platform alongside TodayTix's own. ## Overview -For this round, you'll complete a short data engineering exercise using a containerized Postgres warehouse and dbt. The exercise runs locally via Docker. This repository contains source data, a starter dbt project structure, and scripts. +TTG's platform has a staging layer over TodayTix's own source data and nothing above it. As TTG scales as part of MARI, portfolio companies with their own ticketing systems, their own schemas, and their own data quality quirks need to land in that same platform so stakeholders and AI tooling can query one unified view — not a pile of one-off tables per source. + +For this round, you'll take on **Meridian Live**, a fictional (but realistic) MARI portfolio ticketing company, and design + build the onboarding of its data. The exercise runs locally via Docker, same as the platform's real dev environment. This repository contains the staging layer, Meridian's raw source data, and the same scripts/tooling engineers use day to day. What sits above staging — the models, the grain, the layering, the names — is yours to decide. + +There are two challenges in this repo. **Challenge 1** is onboarding Meridian — modeling judgment against messy source data. **Challenge 2** is the whitelabel page-tracking sources — a pipeline-design problem. Both hand you a project rather than a task: what the platform should grow on top of what's here is the thing you're working out, including which of the two is worth your time and how far to take it. The exercise will be completed **live with the interviewer(s)**. You'll work locally using this repo. You may use your normal tools, including AI assistants. -**This exercise is about reasoning and approach, not finishing everything.** You are not expected to model every table or build a full production-ready warehouse in the time provided. Focus on a sensible subset of the data, clear model structure, and sound assumptions and explanations. +**This exercise is about architecture and judgment, not finishing everything.** You are not expected to model every Meridian table or handle every edge case in the time provided. We care more about how you reason through the tradeoffs, what you flag as a risk or open question, and how you'd sequence the work than about raw completion. Talk through your thinking as you go — this is as much a design conversation as a coding one. + +## What's already built vs. what you'll build + +- **Already built:** one staging model per TodayTix source table — `stg_accounts`, `stg_events`, `stg_showtimes`, `stg_orders`, `stg_transactions`, `stg_pages`, `stg_identity_merges`. They clean and rename, nothing more. Run `bin/dbt run` right after `init.sh` and confirm they build cleanly before you touch anything. +- **Everything above staging is yours.** There is no intermediate layer and no mart layer. What the platform should expose to stakeholders and to the AI/agent layer, at what grain, under what names, is part of what's being asked — not a template to fill in. ## Getting started @@ -28,7 +37,7 @@ docker compose up -d ./scripts/init.sh ``` -Verify dbt works: +Verify dbt works and the existing platform builds cleanly: ```bash bin/dbt --version @@ -43,48 +52,96 @@ To reset the warehouse to its initial state at any time: ./scripts/reset.sh ``` -## What to build +## Challenge 1: onboard Meridian Live + +Meridian's raw data lands in the same `raw` schema as TTG's (see `dbt/models/sources.yml`), but it isn't a clean drop-in — that's the point. A few problems you'll likely need to reckon with, in no particular order: + +- **Identity across systems.** TTG's unstable `customer_id` is resolvable through the `identity_merges` log. Meridian has no such log, and there's no shared key between TTG and Meridian besides a loosely-formatted email. Some real people plausibly exist in both systems. How would you approach unifying (or deliberately not unifying) identity here, and what are the failure modes of your approach? +- **Currency.** Meridian orders are priced in GBP, EUR, and SEK, formatted inconsistently (symbols, thousands/decimal separators). TTG's amounts are USD. How do you normalize without hardcoding rates in SQL? +- **Grain.** TTG keeps the checkout and the money in separate tables: `orders` is one row per checkout, `transactions` is one row per payment taken against it, and payment is the grain TTG's facts sit at. Meridian has no payments table. The money lives on the order row itself — currency, subtotal, fees, total, status — and `meridian_order_items` splits that same money into line items with their own price and quantity. A Meridian order is therefore not a TTG transaction and not a TTG order either: it's payment detail recorded at checkout grain, with a finer grain underneath it. Read `stg_orders` and `stg_transactions` next to `meridian_orders` and `meridian_order_items` before you commit to anything. Sooner or later someone sums an amount across both sources, and what they get back depends on the grain you chose. +- **Time.** Meridian stores performance start times as naive local timestamps plus a UTC offset in minutes; TTG's `showtimes.start_at` is already a normalized instant. Reconcile these consistently. +- **Order status and cancellations.** Meridian orders can be `paid`, `refunded`, `partial_refund`, or `cancelled` (casing/whitespace inconsistent). What should "revenue" mean once these exist, and does that change what belongs in the fact you expose versus what a stakeholder should query separately? + +You don't need to resolve every one of these perfectly. Pick a defensible position on each, implement what you can in the time available, and be ready to explain what you didn't get to and why. + +## Challenge 2: whitelabel storefronts that onboard themselves + +TTG powers whitelabel storefronts for other MARI brands. Each brand's web tracking lands in its own schema in the warehouse, next to `raw`. Nothing in this project reads them — `stg_pages` covers TodayTix's own pages and stops there. Page loads across TodayTix and the brands are meant to be one stream, and today they aren't. -Explore the source data, consider the relationships and data quality issues, and build dbt models that make the data analysis-ready. The project is set up with three model layers — `staging/` for cleaning raw sources, `intermediate/` for joining and reshaping, and `mart/` for business-ready output. +Most brands follow the tracking standard the team agreed on: the schema is named `wl_`, the page relation is called `pages`, and the columns come from a fixed vocabulary — `page_id`, `visitor_id`, `account_id`, `page_type`, `occurred_at`, `event_id`, `showtime_id`, `utm_source`, `utm_medium`, `brand_code`. Only `page_id` and `occurred_at` are guaranteed. Which of the rest a brand sends depends on the tracker version it launched with, and some brands send extra columns of their own that mean nothing to us. -A large part of this exercise is seeing how you think through the full journey from raw data to business-ready output. We're evaluating your modeling choices, how you handle data quality, what you identify as valuable business insight, and how you structure and test the result. There's no single right answer — show us your approach. +`partner_orpheum` doesn't follow the standard at all — it was onboarded before the standard existed. Unprefixed schema, a `page_views` relation, its own column names, and no page type at all: intent is only readable off the URL path. It predates the standard and it isn't going to be brought onto it, so treat it as a separate problem from the standard-conforming brands. -Use documentation to share your rationale, key definitions, assumptions, and any noteworthy challenges you encountered. +**The problem.** A brand launching is a data-side event: its tracker starts writing into a new schema and nobody on the data team is told. The obvious build — declare a source, write a model per brand — means a pull request per launch, and brands arrive faster than that queue drains. Until it merges, the brand's rows sit in the warehouse invisible to everyone downstream while the brand's team asks why their dashboard is empty. + +**Your task.** Build the page-load path so the brands land alongside TodayTix's pages, and so that a brand launching *after* you finish shows up with no change to this repository at all. Mid-session your interviewer will run `bin/add-partner `, which creates a brand-new `wl_*` schema directly in the warehouse — nothing lands here. You then run `bin/dbt build` against an untouched working tree, and that brand's rows should be there. A list of brand names in a macro is not a solution; it's the same pull request wearing a hat. + +Worth having a position on, and worth saying out loud as you go: + +- What happens the first time a brand appears with a column set nobody anticipated, at 3am, with no one watching. +- What you give up by taking these sources out of dbt's declared source graph, and whether you're willing to pay it. +- How someone debugging this in six months finds out which brands are actually in the pipeline today. +- Which parts should stay hand-written, and how a reader can tell which is which. +- What should happen if discovery returns nothing at all. + +Whatever you build, be able to show rows per brand at the end of it. ## Source data reference -**`dbt/models/sources.yml`** defines the raw source tables and columns. Use it as your starting point. Sources are referenced with `{{ source('raw', 'table_name') }}`. +**`dbt/models/sources.yml`** defines the raw source tables and columns for both TTG and Meridian — use it as your starting point. Sources are referenced with `{{ source('raw', 'table_name') }}`. + +TTG tables (already staged/modeled) — see `dbt/models/staging/` for how they're cleaned: `accounts`, `events`, `showtimes`, `orders`, `transactions`, `pages`, `identity_merges`. -Raw tables live in the `raw` schema and include: +Meridian tables (new, unmodeled): -- **accounts** – Stable account entity (account_id, email, created_at). -- **events** – Shows/productions (e.g. Wicked, Hamilton): event_id, name, slug. -- **showtimes** – A specific performance of an event: showtime_id, event_id, start_at. -- **orders** – Order header: order_id, account_id, showtime_id, created_at, total_amount. -- **transactions** – Payment records: transaction_id, order_id, amount, occurred_at. -- **pages** – Browsing behavior with **stable** `account_id` and **unstable** `customer_id` (may be merged over time); optional event_id, showtime_id. -- **identity_merges** – Merge log for customer_id (from_customer_id → to_customer_id, merged_at). Use to resolve pages to a canonical identity. +- **meridian_customers** – Meridian's account entity: customer_id, full_name (sometimes blank — guest checkout), email, phone, country, created_at, marketing_opt_in. +- **meridian_venues** – Physical venues Meridian sells for. TTG has no venue entity today. +- **meridian_events** – Shows/productions, each tied to one venue: event_id, title, venue_id, category. +- **meridian_performances** – A specific occurrence of an event: performance_id, event_id, **starts_at_local** (naive, no timezone) + **utc_offset_minutes** (must be combined to get a true instant), doors_at_local. +- **meridian_orders** – Order header, **not** 1:1 with a payment: order_id, customer_id (optional — blank for anonymous/gift), performance_id (optional — blank for non-ticket orders), currency, subtotal, fees, total, status, placed_at. +- **meridian_order_items** – Ticket/merch line items within an order, Meridian's natural grain: order_item_id, order_id, seat_section, unit_price, quantity. +- **meridian_web_sessions** – Browsing behavior: session_id, **cookie_id** (unstable, anonymous), customer_id (only populated once known, e.g. at checkout), event_id (optional), page_type, occurred_at. **There is no identity-resolution table for Meridian** — unlike TTG's `identity_merges`, cookie-to-customer linkage only exists where a session happens to convert. + +Whitelabel schemas (Challenge 2) sit outside `raw`, one per brand, and none of them are declared in `sources.yml` — inspect them in the warehouse: + +- **wl_arcadia.pages** – Sends every column in the standard. +- **wl_northgate.pages** – Older tracker: no `showtime_id`, no `utm_*`. Those columns don't exist on the table. +- **wl_lumen.pages** – The standard set plus `consent_state` and `device_type`, which mean nothing to the platform. +- **partner_orpheum.page_views** – Pre-standard: view_id, cookie, member_ref, path, viewed_at, production_ref. +- **wl_sandbox.sessions** – A brand's schema that carries no `pages` relation at all. + +`event_id` and `showtime_id` on whitelabel rows are TTG ids — the brands sell TTG inventory through a TTG-powered storefront. ## Raw data quality -Raw data is intentionally varied. Expect the following issues: +Both sources are intentionally messy, in different ways. + +TTG (existing, already handled in staging — for reference): +- Whitespace and inconsistent casing in text fields. +- `$`/comma-formatted amounts in `orders.total_amount` and `transactions.amount`. +- Sentinel nulls (`N/A`, `NULL` string, empty string) on optional FKs in `pages`. -- **Whitespace** – leading/trailing spaces in text (e.g. event names, emails, page_type). -- **Inconsistent casing** – e.g. `Viewed Product Page` vs `viewed_product_page` vs `VIEWED_PRODUCT_PAGE`. -- **Amount formats** – `150.00`, `$200.50`, or `1,000.00` (with dollar sign or commas) in `orders.total_amount` and `transactions.amount`. -- **Sentinel / null-ish values** – optional FKs may be empty string, `N/A`, or `NULL` instead of SQL NULL. -- **Timestamp consistency** – values are stored as loaded and may need to be cast to a consistent type. +Meridian (new, unhandled): +- **Currency formatting** – `£120.00`, `"95,00 €"` (European decimal comma + symbol), `"1050,00 kr"`, or plain `100.00`, all within the same column. +- **Casing/whitespace** – event categories, order statuses, and page types all vary in casing and padding. +- **Sentinel-ish nulls** – blank, `N/A`, `NULL` string on optional identity/FK fields, same pattern as TTG's `pages` but on different tables. +- **Duplicate identity within Meridian itself** – at least one real person has two `customer_id`s in `meridian_customers` with matching email but slightly different name formatting. +- **No identity_merges equivalent** – see above; this is a real gap, not an oversight to "solve" by inventing data. ## Project layout | Path | Purpose | |------|---------| -| `dbt/models/` | dbt models. `sources.yml` defines raw sources; `staging/`, `intermediate/`, `mart/` are where you add models. | -| `dbt/seeds/` | Mapping seeds (e.g. event_type_mapping.csv); use `ref()` in models. | -| `data/initial/` | CSVs loaded into source tables at init. | -| `data/incremental/` | CSVs appended by `bin/ingest` (e.g. `events/batch_001.csv`). | -| `scripts/` | Init, reset, load_initial_source_data.py, ingest.py. | -| `bin/` | Shims for dbt, ingest, load-initial. | +| `dbt/models/sources.yml` | Declares the `raw` sources — TTG and Meridian. The whitelabel brand schemas are not in here. | +| `dbt/models/staging/` | TTG staging models are already built. Anything else you stage goes here. | +| `dbt/models/intermediate/` | Empty. | +| `dbt/models/mart/` | Empty. | +| `dbt/seeds/` | `event_type_mapping.csv` (existing) and `fx_rates.csv` (new — currency → USD rate, for candidates to reference rather than hardcode). | +| `data/initial/` | CSVs loaded into `raw` at init, both TTG and Meridian. | +| `data/whitelabel/` | CSVs loaded into the per-brand schemas at init. `__.csv` names the relation it becomes. | +| `data/incremental/` | CSVs appended by `bin/ingest` — includes Meridian and whitelabel batches to test re-runs under new data. | +| `scripts/` | Init, reset, load_initial_source_data.py, ingest.py, add_partner.py. | +| `bin/` | Shims for dbt, ingest, load-initial, add-partner. | ## Useful commands @@ -93,9 +150,10 @@ Run these from the repo root. They wrap `docker compose run --rm ...`. No local | Command | Purpose | |--------|---------| | `bin/dbt run` | Run dbt models | -| `bin/dbt seed` | Load dbt seeds (mappings) | +| `bin/dbt seed` | Load dbt seeds (mappings, fx rates) | | `bin/dbt test` | Run dbt tests | | `bin/dbt build` | Run models and tests | | `bin/ingest` | Append `data/incremental/*` into source tables; run `bin/dbt run` after to refresh models | -| `bin/ingest events/batch_001` | Ingest a single batch | -| `bin/load-initial` | Load `data/initial/*` into raw (used by init) | +| `bin/ingest meridian_orders/batch_001` | Ingest a single Meridian batch — useful for testing whether your models handle new/incremental Meridian data cleanly | +| `bin/load-initial` | Load `data/initial/*` into raw and `data/whitelabel/*` into the brand schemas (used by init) | +| `bin/add-partner zephyr` | Launch a new whitelabel brand: creates `wl_zephyr.pages` in the warehouse, nothing in the repo. Add `--full` for the complete column set, `--drop` to remove it | diff --git a/bin/add-partner b/bin/add-partner new file mode 100755 index 0000000..d06ecec --- /dev/null +++ b/bin/add-partner @@ -0,0 +1,6 @@ +#!/usr/bin/env bash +# Shim: create a new whitelabel brand's page-tracking schema in the warehouse. +# Usage: bin/add-partner [--full] [--rows N] | bin/add-partner --drop +set -euo pipefail +cd "$(dirname "$0")/.." +exec docker compose run --rm loader python scripts/add_partner.py "$@" diff --git a/data/incremental/events/.gitkeep b/data/incremental/events/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/data/incremental/meridian_order_items/batch_001.csv b/data/incremental/meridian_order_items/batch_001.csv new file mode 100644 index 0000000..55c41c7 --- /dev/null +++ b/data/incremental/meridian_order_items/batch_001.csv @@ -0,0 +1,3 @@ +order_item_id,order_id,seat_section,unit_price,quantity +oi_22,mo_22,Stalls,70.00,2 +oi_23,mo_23,GA,44.00,2 diff --git a/data/incremental/meridian_orders/batch_001.csv b/data/incremental/meridian_orders/batch_001.csv new file mode 100644 index 0000000..d609402 --- /dev/null +++ b/data/incremental/meridian_orders/batch_001.csv @@ -0,0 +1,3 @@ +order_id,customer_id,performance_id,currency,subtotal,fees,total,status,placed_at +mo_22,MC-1007,mp_12,GBP,£140.00,£9.00,£149.00,paid,2025-02-10T10:00:00 +mo_23,MC-1010,mp_7,GBP,88.00,6.00,94.00,paid,2025-02-11T11:00:00 diff --git a/data/incremental/meridian_web_sessions/batch_001.csv b/data/incremental/meridian_web_sessions/batch_001.csv new file mode 100644 index 0000000..73f1a63 --- /dev/null +++ b/data/incremental/meridian_web_sessions/batch_001.csv @@ -0,0 +1,3 @@ +session_id,cookie_id,customer_id,event_id,page_type,occurred_at +ws_26,ck_9019,,me_windrose,event_view,2025-02-10T09:00:00 +ws_27,ck_9019,MC-1007,,checkout_start,2025-02-10T09:15:00 diff --git a/data/incremental/whitelabel/wl_arcadia__pages/batch_001.csv b/data/incremental/whitelabel/wl_arcadia__pages/batch_001.csv new file mode 100644 index 0000000..dc86518 --- /dev/null +++ b/data/incremental/whitelabel/wl_arcadia__pages/batch_001.csv @@ -0,0 +1,4 @@ +page_id,visitor_id,account_id,page_type,occurred_at,event_id,showtime_id,utm_source,utm_medium,brand_code +arc_p_13,vis_arc_g7,acc_4,home,2025-03-18T08:22:10,,,google,cpc,ARC +arc_p_14,vis_arc_g7,acc_4,VIEWED_PRODUCT_PAGE,2025-03-18T08:25:44,evt_hamilton,,google,cpc,ARC +arc_p_15,vis_arc_g7,acc_4,viewed_showtime,2025-03-18T08:31:02,evt_hamilton,st_evt_hamilton_22,google,cpc,ARC diff --git a/data/incremental/whitelabel/wl_northgate__pages/batch_001.csv b/data/incremental/whitelabel/wl_northgate__pages/batch_001.csv new file mode 100644 index 0000000..a794e93 --- /dev/null +++ b/data/incremental/whitelabel/wl_northgate__pages/batch_001.csv @@ -0,0 +1,3 @@ +page_id,visitor_id,account_id,page_type,occurred_at,event_id,brand_code +ngt_p_10,vis_ngt_f6,,VIEWED PRODUCT PAGE,2025-03-19T12:04:33,evt_lion_king,NGT +ngt_p_11,vis_ngt_f6,acc_16,checkout_start,2025-03-19T12:11:57,evt_lion_king,NGT diff --git a/data/initial/meridian_customers.csv b/data/initial/meridian_customers.csv new file mode 100644 index 0000000..8bbf120 --- /dev/null +++ b/data/initial/meridian_customers.csv @@ -0,0 +1,17 @@ +customer_id,full_name,email,phone,country,created_at,marketing_opt_in +MC-1001, Alice Whitfield , alice.whitfield@example.com ,+44 20 7946 0958,UK,2024-11-02T10:15:00,true +MC-1002,Ben Carter,ben.carter@example.com,+44 161 496 0001,uk,2024-11-05T09:00:00,false +MC-1003,Chiara Rossi,chiara.rossi@example.com,+39 06 4991 0000,Italy,2024-12-01T14:22:00,TRUE +MC-1004,Daniel Kim,daniel.kim@example.com,,South Korea,2025-01-10T08:45:00, +MC-1005,Emma Novak,EMMA.NOVAK@EXAMPLE.COM,+43 1 4000 0000,Austria,2025-01-14T11:00:00,false +MC-1006,Alice Whitfield,alice.whitfield@example.com,+44 20 7946 0958,United Kingdom,2025-01-20T09:30:00,true +MC-1007,Farid Haddad,farid.haddad@example.com,+33 1 4000 0000,France,2025-01-25T16:10:00,false +MC-1008, Grace Lin ,grace.lin@example.com,+44 20 1234 5678,UK,2025-02-01T12:00:00,true +MC-1009,,unknown_guest_09@example.com,,UK,2025-02-03T13:00:00,false +MC-1010,Ivy Chen,ivy.chen@example.com,+44 161 000 0000,uk,2025-02-05T15:45:00,TRUE +MC-1011,Jonas Weber,jonas.weber@example.com,+49 30 1234 5678,Germany,2025-02-08T10:00:00,false +MC-1012,Karolina Nowak,karolina.nowak@example.com,+48 22 000 0000,Poland,2025-02-10T09:15:00,true +MC-1013,Liam O'Sullivan,liam.osullivan@example.com,+353 1 234 5678,Ireland,2025-02-12T14:30:00,false +MC-1014,Maya Patel, maya.patel@example.com ,+44 20 7000 0000,UK,2025-02-15T11:20:00,TRUE +MC-1015,Noah Bergstrom,noah.bergstrom@example.com,+46 8 000 0000,Sweden,2025-02-18T17:00:00,false +MC-1016,Sam Reyes,user4@example.com,+44 20 5000 0000,UK,2025-02-20T09:00:00,true diff --git a/data/initial/meridian_events.csv b/data/initial/meridian_events.csv new file mode 100644 index 0000000..90f0cca --- /dev/null +++ b/data/initial/meridian_events.csv @@ -0,0 +1,6 @@ +event_id,title,venue_id,category +me_starlight, Starlight Revue ,mv_apollo, Theatre +me_windrose,Windrose,mv_globe,theatre +me_comedy_night,Comedy Night Live,mv_forum,COMEDY +me_midnight_opera, Midnight Opera ,mv_konzert,Opera +me_folk_tales,Folk Tales,mv_olympia,Theatre diff --git a/data/initial/meridian_order_items.csv b/data/initial/meridian_order_items.csv new file mode 100644 index 0000000..35117e5 --- /dev/null +++ b/data/initial/meridian_order_items.csv @@ -0,0 +1,25 @@ +order_item_id,order_id,seat_section,unit_price,quantity +oi_1a,mo_1,Stalls,70.00,1 +oi_1b,mo_1,Circle,50.00,1 +oi_2,mo_2, Stalls ,47.50,2 +oi_3a,mo_3,Balcony,47.50,1 +oi_3b,mo_3,Stalls,47.50,1 +oi_4,mo_4,Stalls,70.00,2 +oi_5,mo_5,GA,55.00,2 +oi_6,mo_6,Stalls,65.00,2 +oi_7,mo_7,Balcony,52.50,2 +oi_8,mo_8,GA,49.50,2 +oi_9,mo_9,Stalls,75.00,2 +oi_10,mo_10,Stalls,56.00,2 +oi_11a,mo_11,Balcony,70.00,1 +oi_11b,mo_11,GA,50.00,1 +oi_12,mo_12,GA,52.50,2 +oi_13,mo_13,Stalls,59.00,2 +oi_14,mo_14,Balcony,61.00,2 +oi_15,mo_15,Stalls,67.50,2 +oi_16,mo_16,Stalls,64.00,2 +oi_17,mo_17,GA,49.50,2 +oi_18,mo_18,Stalls,525.00,2 +oi_19,mo_19,GA,50.00,2 +oi_20,mo_20,GA,50.00,1 +oi_21,mo_21,Merch,25.00,1 diff --git a/data/initial/meridian_orders.csv b/data/initial/meridian_orders.csv new file mode 100644 index 0000000..4f9cca4 --- /dev/null +++ b/data/initial/meridian_orders.csv @@ -0,0 +1,22 @@ +order_id,customer_id,performance_id,currency,subtotal,fees,total,status,placed_at +mo_1,MC-1001,mp_1,GBP,£120.00,£8.00,£128.00,paid,2025-01-08T10:15:00 +mo_2,MC-1002,mp_1,GBP,95.00,6.50,101.50, Paid ,2025-01-09T11:00:00 +mo_3,MC-1003,mp_9,EUR,"95,00 €","6,00 €","101,00 €",PAID,2025-01-20T09:30:00 +mo_4,MC-1004,mp_4,GBP,£140.00,£9.50,£149.50,paid,2025-01-05T14:00:00 +mo_5,MC-1005,mp_9,EUR,110.00,7.00,117.00,paid,2025-01-22T16:45:00 +mo_6,MC-1006,mp_2,GBP,£130.00,£8.50,£138.50,paid,2025-01-14T09:00:00 +mo_7,MC-1007,mp_5,GBP,105.00,7.00,112.00,paid,2025-01-16T10:20:00 +mo_8,MC-1008,mp_6,GBP,£99.00,£6.50,£105.50, refunded ,2025-01-24T15:10:00 +mo_9,MC-1009,mp_7,GBP,150.00,10.00,160.00,paid,2025-01-30T18:00:00 +mo_10,MC-1010,mp_8,GBP,£112.00,£7.50,£119.50,PARTIAL_REFUND,2025-02-02T12:00:00 +mo_11,MC-1011,mp_9,EUR,"120,00 €","8,00 €","128,00 €",paid,2025-02-04T09:45:00 +mo_12,MC-1012,mp_10,EUR,105.00,7.00,112.00,paid,2025-02-06T11:15:00 +mo_13,MC-1013,mp_11,GBP,£118.00,£8.00,£126.00,paid,2025-01-13T13:30:00 +mo_14,MC-1014,mp_12,GBP,122.00,8.00,130.00,Paid,2025-01-20T15:50:00 +mo_15,MC-1001,mp_3,GBP,£135.00,£9.00,£144.00,paid,2025-01-22T10:05:00 +mo_16,MC-1016,mp_2,GBP,£128.00,£8.50,£136.50,paid,2025-01-15T09:20:00 +mo_17,MC-1002,mp_5,GBP,99.00,6.50,105.50, cancelled ,2025-01-18T08:40:00 +mo_18,MC-1015,mp_6,SEK,"1050,00 kr","70,00 kr","1120,00 kr",paid,2025-01-27T10:00:00 +mo_19,MC-1003,mp_10,EUR,100.00,6.50,106.50,paid,2025-02-09T14:20:00 +mo_20,,mp_7,GBP,50.00,0.00,50.00,paid,2025-02-01T20:00:00 +mo_21,MC-1013,,GBP,25.00,0.00,25.00,paid,2025-01-29T09:00:00 diff --git a/data/initial/meridian_performances.csv b/data/initial/meridian_performances.csv new file mode 100644 index 0000000..8f6a30d --- /dev/null +++ b/data/initial/meridian_performances.csv @@ -0,0 +1,13 @@ +performance_id,event_id,starts_at_local,utc_offset_minutes,doors_at_local +mp_1,me_starlight,2025-01-10 19:30:00,0,2025-01-10 18:45:00 +mp_2,me_starlight,2025-01-17 19:30:00,0,2025-01-17 18:45:00 +mp_3,me_starlight,2025-01-24 14:30:00,0,2025-01-24 13:45:00 +mp_4,me_windrose,2025-01-12 20:00:00,0,2025-01-12 19:15:00 +mp_5,me_windrose,2025-01-19 20:00:00,0,2025-01-19 19:15:00 +mp_6,me_windrose,2025-01-26 15:00:00,0,2025-01-26 14:15:00 +mp_7,me_comedy_night,2025-02-01 20:30:00,0,2025-02-01 19:45:00 +mp_8,me_comedy_night,2025-02-08 20:30:00,0,2025-02-08 19:45:00 +mp_9,me_midnight_opera,2025-02-05 19:00:00,60,2025-02-05 18:15:00 +mp_10,me_midnight_opera,2025-02-12 19:00:00,60,2025-02-12 18:15:00 +mp_11,me_folk_tales,2025-01-15 19:30:00,0,2025-01-15 18:45:00 +mp_12,me_folk_tales,2025-01-22 19:30:00,0,2025-01-22 18:45:00 diff --git a/data/initial/meridian_venues.csv b/data/initial/meridian_venues.csv new file mode 100644 index 0000000..6e174a6 --- /dev/null +++ b/data/initial/meridian_venues.csv @@ -0,0 +1,6 @@ +venue_id,name,city,country +mv_apollo, Apollo Theatre ,London,UK +mv_globe,Globe Theatre,London,uk +mv_olympia,Olympia Theatre,Dublin,Ireland +mv_forum, The Forum ,Bath,United Kingdom +mv_konzert,Konzerthaus,Vienna,Austria diff --git a/data/initial/meridian_web_sessions.csv b/data/initial/meridian_web_sessions.csv new file mode 100644 index 0000000..39d3323 --- /dev/null +++ b/data/initial/meridian_web_sessions.csv @@ -0,0 +1,26 @@ +session_id,cookie_id,customer_id,event_id,page_type,occurred_at +ws_1,ck_9001,,me_starlight,event_view,2025-01-02T09:00:00 +ws_2,ck_9001,,me_starlight,performance_view,2025-01-02T09:05:00 +ws_3,ck_9001,MC-1001,,checkout_start,2025-01-02T09:12:00 +ws_4,ck_9002,,me_windrose,EVENT_VIEW,2025-01-03T11:00:00 +ws_5,ck_9002,,me_windrose,performance_view,2025-01-03T11:04:00 +ws_6,ck_9003,, me_comedy_night ,event_view,2025-01-05T13:00:00 +ws_7,ck_9004,MC-1004,,checkout_start,2025-01-04T08:50:00 +ws_8,ck_9004,MC-1004,,checkout_complete,2025-01-04T08:55:00 +ws_9,ck_9005,,me_midnight_opera,Event_View,2025-01-06T17:00:00 +ws_10,ck_9006,,me_starlight,event_view,2025-01-07T09:15:00 +ws_11,ck_9006,MC-1006,,checkout_start,2025-01-14T08:45:00 +ws_12,ck_9007,,me_folk_tales,event_view,2025-01-10T12:00:00 +ws_13,ck_9008,,me_windrose,performance_view,2025-01-11T14:30:00 +ws_14,ck_9009,,me_comedy_night,event_view,2025-01-15T09:00:00 +ws_15,ck_9010,N/A,me_midnight_opera,event_view,2025-01-18T10:00:00 +ws_16,ck_9011,,me_starlight,EVENT_VIEW,2025-01-20T11:00:00 +ws_17,ck_9011,MC-1001,,checkout_start,2025-01-22T09:50:00 +ws_18,ck_9012,,me_folk_tales,performance_view,2025-01-21T15:00:00 +ws_19,ck_9013,,,event_view,2025-01-23T09:00:00 +ws_20,ck_9014,MC-1016,,checkout_start,2025-01-15T09:00:00 +ws_21,ck_9014,MC-1016,,checkout_complete,2025-01-15T09:05:00 +ws_22,ck_9015,,me_windrose,event_view,2025-01-25T10:00:00 +ws_23,ck_9016,,me_comedy_night,performance_view,2025-02-01T19:00:00 +ws_24,ck_9017,NULL,me_midnight_opera,event_view,2025-02-03T14:00:00 +ws_25,ck_9018,,me_folk_tales,event_view,2025-01-14T09:00:00 diff --git a/data/whitelabel/partner_orpheum__page_views.csv b/data/whitelabel/partner_orpheum__page_views.csv new file mode 100644 index 0000000..ae7dfb2 --- /dev/null +++ b/data/whitelabel/partner_orpheum__page_views.csv @@ -0,0 +1,9 @@ +view_id,cookie,member_ref,path,viewed_at,production_ref +orp_v_1,ck_orp_a1,acc_6,/,2025-01-16T09:11:40, +orp_v_2,ck_orp_a1,acc_6,/show/wicked,2025-01-16T09:14:02,evt_wicked +orp_v_3,ck_orp_b2,,/show/hamilton,2025-01-30T13:26:18,evt_hamilton +orp_v_4,ck_orp_b2,,/show/hamilton/performances,2025-01-30T13:29:44,evt_hamilton +orp_v_5,ck_orp_c3,acc_12,/checkout,2025-02-12T18:02:57,evt_six +orp_v_6,ck_orp_c3,acc_12,/checkout/confirmation,2025-02-12T18:09:31,evt_six +orp_v_7,ck_orp_d4,,/,2025-02-25T07:44:12, +orp_v_8,ck_orp_e5,acc_20,/show/aladdin,2025-03-07T16:38:25,evt_aladdin diff --git a/data/whitelabel/wl_arcadia__pages.csv b/data/whitelabel/wl_arcadia__pages.csv new file mode 100644 index 0000000..5d4469c --- /dev/null +++ b/data/whitelabel/wl_arcadia__pages.csv @@ -0,0 +1,13 @@ +page_id,visitor_id,account_id,page_type,occurred_at,event_id,showtime_id,utm_source,utm_medium,brand_code +arc_p_1,vis_arc_a1,acc_3,viewed_product_page,2025-01-14T11:02:11,evt_wicked,,google,cpc,ARC +arc_p_2,vis_arc_a1,acc_3,VIEWED_SHOWTIME,2025-01-14T11:04:52,evt_wicked,st_evt_wicked_1,google,cpc,ARC +arc_p_3,vis_arc_b2,,viewed product page,2025-01-19T18:44:03,evt_six,,,,ARC +arc_p_4,vis_arc_b2,N/A,checkout_start,2025-01-19T18:51:20,evt_six,,newsletter,email,ARC +arc_p_5,vis_arc_c3,acc_11,viewed_showtime ,2025-02-02T09:12:44,evt_hamilton,st_evt_hamilton_19,,,ARC +arc_p_6,vis_arc_c3,acc_11,checkout_complete,2025-02-02T09:20:05,evt_hamilton,st_evt_hamilton_19,,,ARC +arc_p_7,vis_arc_d4,NULL,home,2025-02-11T20:01:37,,,facebook,paid_social,ARC +arc_p_8,vis_arc_d4,,VIEWED_PRODUCT_PAGE,2025-02-11T20:03:59,evt_lion_king,,facebook,paid_social,ARC +arc_p_9,vis_arc_e5,acc_7,viewed_product_page,2025-02-24T13:30:12,evt_aladdin,,,,ARC +arc_p_10,vis_arc_e5,acc_7,viewed_showtime,2025-02-24T13:33:48,evt_lion_king,st_evt_lion_king_37,,,ARC +arc_p_11,vis_arc_f6,, home ,2025-03-05T07:55:02,,,google,organic,ARC +arc_p_12,vis_arc_f6,acc_15,checkout_start,2025-03-05T08:10:19,evt_book_of_mormon,,google,organic,ARC diff --git a/data/whitelabel/wl_lumen__pages.csv b/data/whitelabel/wl_lumen__pages.csv new file mode 100644 index 0000000..d16b383 --- /dev/null +++ b/data/whitelabel/wl_lumen__pages.csv @@ -0,0 +1,11 @@ +page_id,visitor_id,account_id,page_type,occurred_at,event_id,showtime_id,utm_source,utm_medium,brand_code,consent_state,device_type +lum_p_1,vis_lum_a1,acc_1,home,2025-01-11T12:00:04,,,direct,none,LUM,granted,mobile +lum_p_2,vis_lum_a1,acc_1,viewed_product_page,2025-01-11T12:02:47,evt_wicked,,direct,none,LUM,granted,mobile +lum_p_3,vis_lum_b2,,VIEWED_PRODUCT_PAGE,2025-01-27T15:31:22,evt_aladdin,,tiktok,paid_social,LUM,denied,mobile +lum_p_4,vis_lum_b2,N/A,viewed_showtime,2025-01-27T15:36:55,evt_wicked,st_evt_wicked_4,tiktok,paid_social,LUM,denied,mobile +lum_p_5,vis_lum_c3,acc_13,checkout_start,2025-02-14T17:48:30,evt_six,,,,LUM,granted,desktop +lum_p_6,vis_lum_c3,acc_13,checkout_complete,2025-02-14T17:55:11,evt_six,,,,LUM,granted,desktop +lum_p_7,vis_lum_d4,NULL,viewed showtime,2025-02-28T09:07:19,evt_hamilton,st_evt_hamilton_20,google,cpc,LUM,unknown,tablet +lum_p_8,vis_lum_e5,acc_18, home ,2025-03-09T22:14:06,,,,,LUM,granted,desktop +lum_p_9,vis_lum_e5,acc_18,viewed_product_page,2025-03-09T22:18:41,evt_deathly_hallows,,,,LUM,granted,desktop +lum_p_10,vis_lum_f6,,viewed_product_page,2025-03-15T11:29:53,evt_lion_king,,newsletter,email,LUM,granted,mobile diff --git a/data/whitelabel/wl_northgate__pages.csv b/data/whitelabel/wl_northgate__pages.csv new file mode 100644 index 0000000..5451eab --- /dev/null +++ b/data/whitelabel/wl_northgate__pages.csv @@ -0,0 +1,10 @@ +page_id,visitor_id,account_id,page_type,occurred_at,event_id,brand_code +ngt_p_1,vis_ngt_a1,acc_2,viewed_product_page,2025-01-08T16:20:44,evt_lion_king,NGT +ngt_p_2,vis_ngt_a1,acc_2,VIEWED SHOWTIME,2025-01-08T16:24:10,evt_lion_king,NGT +ngt_p_3,vis_ngt_b2,,home,2025-01-21T10:05:31,,NGT +ngt_p_4,vis_ngt_b2,N/A,viewed_product_page,2025-01-21T10:09:57,evt_six,NGT +ngt_p_5,vis_ngt_c3,acc_9,checkout_start ,2025-02-06T19:41:02,evt_six,NGT +ngt_p_6,vis_ngt_c3,acc_9,checkout_complete,2025-02-06T19:47:38,evt_six,NGT +ngt_p_7,vis_ngt_d4,NULL,VIEWED_PRODUCT_PAGE,2025-02-19T08:33:15,evt_deathly_hallows,NGT +ngt_p_8,vis_ngt_e5,acc_5,viewed_showtime,2025-03-01T21:12:49,evt_hamilton,NGT +ngt_p_9,vis_ngt_e5,acc_5,home,2025-03-01T21:15:03,,NGT diff --git a/data/whitelabel/wl_sandbox__sessions.csv b/data/whitelabel/wl_sandbox__sessions.csv new file mode 100644 index 0000000..fd27915 --- /dev/null +++ b/data/whitelabel/wl_sandbox__sessions.csv @@ -0,0 +1,4 @@ +session_id,visitor_id,started_at +sbx_s_1,vis_sbx_a1,2025-02-01T10:00:00 +sbx_s_2,vis_sbx_b2,2025-02-01T10:15:22 +sbx_s_3,vis_sbx_c3,2025-02-02T14:41:09 diff --git a/dbt/models/sources.yml b/dbt/models/sources.yml index b3af396..b1d1a2f 100644 --- a/dbt/models/sources.yml +++ b/dbt/models/sources.yml @@ -1,12 +1,17 @@ -# Source tables (raw schema). Initial data loaded by scripts/load_initial_source_data.py; -# incremental data appended by scripts/ingest.py. Candidates model from source('raw', '
'). -# Identity on pages: pages have both stable account_id and unstable customer_id; identity_merges applies to customer_id only. +# Source tables (raw schema). TodayTix tables loaded/appended as in the original platform; +# Meridian tables are the new MARI portfolio company being onboarded (loaded the same way). +# Candidates model from source('raw', '
'). +# Identity on TTG pages: stable account_id + unstable customer_id; identity_merges resolves customer_id only. +# Identity on Meridian: NO merge log is provided. cookie_id is unstable and pre-login; customer_id is only +# populated once a session is tied to an account (e.g. at checkout). There is no table that maps cookie_id +# history the way identity_merges does for TTG — that gap is intentional, see README. version: 2 sources: - name: raw schema: raw description: Source tables loaded from data/initial/ and appended from data/incremental/ tables: + # ---- Existing TodayTix (TTG) platform — already staged; nothing above staging ---- - name: accounts description: Stable account entity (e.g. logged-in user). columns: @@ -62,10 +67,96 @@ sources: - name: showtime_id description: Optional context (e.g. viewed showtime). - name: identity_merges - description: Merge log for customer_id (e.g. Segment Unify). from_customer_id was merged into to_customer_id at merged_at. Use to resolve pages.customer_id to canonical id when building marts. + description: Merge log for TTG customer_id (e.g. Segment Unify). from_customer_id was merged into to_customer_id at merged_at. Use to resolve pages.customer_id to canonical id when building marts. columns: - name: from_customer_id description: Id that was merged away (unstable/deprecated). - name: to_customer_id description: Canonical id that absorbed the merge. - name: merged_at + + # ---- Meridian Live (new MARI portfolio company) — onboarding target for this exercise ---- + - name: meridian_customers + description: Meridian's account entity. No relationship to TTG accounts is provided — a shared real-world person may exist in both systems with no shared key other than (unreliable) email. + columns: + - name: customer_id + description: Primary key, Meridian format (e.g. MC-1001). + - name: full_name + description: Free text; sometimes blank (guest checkout). + - name: email + - name: phone + - name: country + - name: created_at + - name: marketing_opt_in + - name: meridian_venues + description: Physical venues Meridian sells tickets for. TTG's model has no venue entity today. + columns: + - name: venue_id + description: Primary key. + - name: name + - name: city + - name: country + - name: meridian_events + description: Shows/productions on Meridian, each tied to one venue. + columns: + - name: event_id + description: Primary key, Meridian format (e.g. me_starlight). Not the same id space as raw.events. + - name: title + - name: venue_id + description: FK to meridian_venues. + - name: category + description: Free text genre (e.g. Theatre, Comedy, Opera); casing inconsistent. + - name: meridian_performances + description: A specific occurrence of a Meridian event. TTG's equivalent (showtimes) stores start_at already normalized to UTC; Meridian does not. + columns: + - name: performance_id + description: Primary key. + - name: event_id + description: FK to meridian_events. + - name: starts_at_local + description: Naive local timestamp, no timezone. Must be combined with utc_offset_minutes to get a true instant. + - name: utc_offset_minutes + description: Offset from UTC in minutes for this performance's local time (varies by venue and DST). + - name: doors_at_local + - name: meridian_orders + description: Order header. Grain is one row per checkout, not per payment — there is no separate transactions table. Multi-currency; amounts are formatted inconsistently (symbols, thousands/decimal separators). + columns: + - name: order_id + description: Primary key. + - name: customer_id + description: Optional FK to meridian_customers; blank for anonymous/gift purchases. + - name: performance_id + description: Optional FK to meridian_performances; blank for non-ticket orders (e.g. merchandise). + - name: currency + description: ISO-ish currency code (GBP, EUR, SEK, USD). + - name: subtotal + - name: fees + - name: total + description: subtotal + fees, in the order's original currency, as formatted text. + - name: status + description: paid / refunded / partial_refund / cancelled. Casing and whitespace inconsistent. + - name: placed_at + - name: meridian_order_items + description: Ticket/merch line items within an order. An order may have multiple items at different price points — this is Meridian's natural grain, one level finer than meridian_orders. + columns: + - name: order_item_id + description: Primary key. + - name: order_id + description: FK to meridian_orders. + - name: seat_section + - name: unit_price + - name: quantity + - name: meridian_web_sessions + description: Browsing behavior on Meridian. cookie_id is the only pre-login identifier and is never resolved to a customer_id except when a session happens to convert — there is no merge/identity-resolution table like TTG's identity_merges for this source. + columns: + - name: session_id + description: Primary key. + - name: cookie_id + description: Unstable, anonymous browser identifier. + - name: customer_id + description: Populated only once known (e.g. checkout_start/checkout_complete); NULL/blank/N/A/NULL-string otherwise. + - name: event_id + description: Optional context; blank for account-level pages like checkout. + - name: page_type + description: Free text; casing and whitespace inconsistent. + - name: occurred_at diff --git a/dbt/models/staging/stg_accounts.sql b/dbt/models/staging/stg_accounts.sql new file mode 100644 index 0000000..793783a --- /dev/null +++ b/dbt/models/staging/stg_accounts.sql @@ -0,0 +1,6 @@ +-- Existing platform (already built). Do not modify as part of this exercise. +select + account_id, + nullif(trim(email), '') as email, + created_at +from {{ source('raw', 'accounts') }} diff --git a/dbt/models/staging/stg_events.sql b/dbt/models/staging/stg_events.sql new file mode 100644 index 0000000..269d909 --- /dev/null +++ b/dbt/models/staging/stg_events.sql @@ -0,0 +1,6 @@ +-- Existing platform (already built). Do not modify as part of this exercise. +select + event_id, + trim(name) as name, + trim(slug) as slug +from {{ source('raw', 'events') }} diff --git a/dbt/models/staging/stg_identity_merges.sql b/dbt/models/staging/stg_identity_merges.sql new file mode 100644 index 0000000..0e164d4 --- /dev/null +++ b/dbt/models/staging/stg_identity_merges.sql @@ -0,0 +1,6 @@ +-- Existing platform (already built). Do not modify as part of this exercise. +select + from_customer_id, + to_customer_id, + merged_at +from {{ source('raw', 'identity_merges') }} diff --git a/dbt/models/staging/stg_orders.sql b/dbt/models/staging/stg_orders.sql new file mode 100644 index 0000000..3b917a5 --- /dev/null +++ b/dbt/models/staging/stg_orders.sql @@ -0,0 +1,8 @@ +-- Existing platform (already built). Do not modify as part of this exercise. +select + order_id, + account_id, + nullif(trim(showtime_id), '') as showtime_id, + created_at, + replace(replace(trim(total_amount), '$', ''), ',', '')::numeric as total_amount +from {{ source('raw', 'orders') }} diff --git a/dbt/models/staging/stg_pages.sql b/dbt/models/staging/stg_pages.sql new file mode 100644 index 0000000..93790bb --- /dev/null +++ b/dbt/models/staging/stg_pages.sql @@ -0,0 +1,11 @@ +-- Existing platform (already built). Do not modify as part of this exercise. +-- Sentinel nulls ('', 'N/A', 'NULL') collapsed; page_type normalized to snake_case. +select + page_id, + nullif(nullif(nullif(trim(account_id), ''), 'N/A'), 'NULL') as account_id, + nullif(nullif(nullif(trim(customer_id), ''), 'N/A'), 'NULL') as customer_id, + lower(regexp_replace(trim(page_type), '\s+', '_', 'g')) as page_type, + occurred_at, + nullif(nullif(nullif(trim(event_id), ''), 'N/A'), 'NULL') as event_id, + nullif(nullif(nullif(trim(showtime_id), ''), 'N/A'), 'NULL') as showtime_id +from {{ source('raw', 'pages') }} diff --git a/dbt/models/staging/stg_showtimes.sql b/dbt/models/staging/stg_showtimes.sql new file mode 100644 index 0000000..6ff09f8 --- /dev/null +++ b/dbt/models/staging/stg_showtimes.sql @@ -0,0 +1,6 @@ +-- Existing platform (already built). Do not modify as part of this exercise. +select + showtime_id, + event_id, + start_at +from {{ source('raw', 'showtimes') }} diff --git a/dbt/models/staging/stg_transactions.sql b/dbt/models/staging/stg_transactions.sql new file mode 100644 index 0000000..004d3c1 --- /dev/null +++ b/dbt/models/staging/stg_transactions.sql @@ -0,0 +1,7 @@ +-- Existing platform (already built). Do not modify as part of this exercise. +select + transaction_id, + order_id, + replace(replace(trim(amount), '$', ''), ',', '')::numeric as amount, + occurred_at +from {{ source('raw', 'transactions') }} diff --git a/dbt/seeds/.gitkeep b/dbt/seeds/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/dbt/seeds/fx_rates.csv b/dbt/seeds/fx_rates.csv new file mode 100644 index 0000000..b3e97c4 --- /dev/null +++ b/dbt/seeds/fx_rates.csv @@ -0,0 +1,5 @@ +currency,usd_rate +USD,1.00 +GBP,1.27 +EUR,1.08 +SEK,0.096 diff --git a/scripts/add_partner.py b/scripts/add_partner.py new file mode 100644 index 0000000..de094ae --- /dev/null +++ b/scripts/add_partner.py @@ -0,0 +1,101 @@ +""" +Onboard a new whitelabel brand: create wl_.pages in the warehouse and fill it +with page-tracking rows. Nothing is written to the repo — this is the warehouse-side +half of onboarding, the half that happens without a pull request. + +Run from repo root: + docker compose run --rm loader python scripts/add_partner.py [--full] [--rows N] + docker compose run --rm loader python scripts/add_partner.py --drop +""" +import argparse +import os +import sys +from datetime import datetime, timedelta + +import psycopg2 +from psycopg2 import sql + +import whitelabel_sources + +PGHOST = os.environ.get("PGHOST", "warehouse") +PGPORT = int(os.environ.get("PGPORT", "5432")) +PGUSER = os.environ.get("PGUSER", "postgres") +PGPASSWORD = os.environ.get("PGPASSWORD", "postgres") +PGDATABASE = os.environ.get("PGDATABASE", "warehouse") + +# A brand's tracker emits whichever of the standard columns its version knows about. +# The reduced set is what a brand launching on the current tracker sends. +REDUCED_COLUMNS = ["page_id", "visitor_id", "account_id", "page_type", "occurred_at", "event_id", "brand_code"] +FULL_COLUMNS = REDUCED_COLUMNS + ["showtime_id", "utm_source", "utm_medium"] + +PAGE_TYPES = ["home", "VIEWED_PRODUCT_PAGE", "viewed_showtime ", "checkout_start", "checkout_complete"] +EVENT_IDS = ["evt_wicked", "evt_hamilton", "evt_lion_king", "evt_six", ""] +SHOWTIME_IDS = ["st_evt_wicked_2", "st_evt_hamilton_21", "st_evt_lion_king_38", "", ""] +UTM = [("google", "cpc"), ("newsletter", "email"), ("", ""), ("facebook", "paid_social"), ("direct", "none")] + +FIRST_OCCURRED_AT = datetime(2025, 3, 20, 9, 0, 0) + + +def build_rows(brand: str, columns: list[str], count: int) -> list[tuple]: + brand_code = brand[:3].upper() + rows = [] + for i in range(count): + occurred_at = FIRST_OCCURRED_AT + timedelta(hours=7 * i) + utm_source, utm_medium = UTM[i % len(UTM)] + values = { + "page_id": f"{brand}_p_{i + 1}", + "visitor_id": f"vis_{brand}_{i // 2 + 1}", + "account_id": f"acc_{(i % 20) + 1}" if i % 3 else None, + "page_type": PAGE_TYPES[i % len(PAGE_TYPES)], + "occurred_at": occurred_at.isoformat(), + "event_id": EVENT_IDS[i % len(EVENT_IDS)] or None, + "brand_code": brand_code, + "showtime_id": SHOWTIME_IDS[i % len(SHOWTIME_IDS)] or None, + "utm_source": utm_source or None, + "utm_medium": utm_medium or None, + } + rows.append(tuple(values[c] for c in columns)) + return rows + + +def main(): + parser = argparse.ArgumentParser(description="Create a whitelabel brand's page-tracking schema.") + parser.add_argument("brand", help="Brand slug, e.g. zephyr (schema becomes wl_zephyr)") + parser.add_argument("--full", action="store_true", help="Emit every standard column, not just the reduced set") + parser.add_argument("--rows", type=int, default=10, help="How many page rows to generate (default 10)") + parser.add_argument("--drop", action="store_true", help="Drop the brand's schema instead of creating it") + args = parser.parse_args() + + brand = args.brand.strip().lower().removeprefix("wl_") + if not brand.replace("_", "").isalnum(): + print(f"Brand slug must be alphanumeric/underscore, got: {args.brand}", file=sys.stderr) + sys.exit(1) + schema = f"wl_{brand}" + + conn = psycopg2.connect( + host=PGHOST, port=PGPORT, user=PGUSER, password=PGPASSWORD, dbname=PGDATABASE + ) + conn.autocommit = False + + try: + with conn.cursor() as cur: + if args.drop: + cur.execute(sql.SQL("DROP SCHEMA IF EXISTS {} CASCADE").format(sql.Identifier(schema))) + print(f"Dropped schema {schema}") + else: + columns = FULL_COLUMNS if args.full else REDUCED_COLUMNS + rows = build_rows(brand, columns, args.rows) + whitelabel_sources.create_relation(cur, schema, "pages", columns) + n = whitelabel_sources.insert_rows(cur, schema, "pages", columns, rows) + print(f"Created {schema}.pages ({', '.join(columns)}) with {n} rows") + conn.commit() + except Exception as e: + conn.rollback() + print(f"Error: {e}", file=sys.stderr) + sys.exit(1) + finally: + conn.close() + + +if __name__ == "__main__": + main() diff --git a/scripts/ingest.py b/scripts/ingest.py index 36cee7e..8832b0e 100644 --- a/scripts/ingest.py +++ b/scripts/ingest.py @@ -1,7 +1,8 @@ """ -Append CSV batches from data/incremental/ into source tables (raw schema). +Append CSV batches from data/incremental/ into source tables (raw schema), and from +data/incremental/whitelabel/__
/ into the whitelabel brand schemas. Run from repo root: docker compose run --rm loader python scripts/ingest.py [batch] - batch: optional path or name (e.g. pages/batch_001 or pages/batch_001.csv) + batch: optional path or name (e.g. pages/batch_001 or meridian_orders/batch_001.csv) If omitted, processes all CSV files under data/incremental/. """ import csv @@ -11,6 +12,8 @@ import psycopg2 from psycopg2.extras import execute_values +import whitelabel_sources + PGHOST = os.environ.get("PGHOST", "warehouse") PGPORT = int(os.environ.get("PGPORT", "5432")) PGUSER = os.environ.get("PGUSER", "postgres") @@ -152,6 +155,93 @@ def ingest_identity_merges(cur, path: str) -> int: return len(rows) +def ingest_meridian_orders(cur, path: str) -> int: + with open(path, newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + rows = [ + ( + r["order_id"], + _null_if_empty(r.get("customer_id")), + _null_if_empty(r.get("performance_id")), + r["currency"], + r["subtotal"], + r["fees"], + r["total"], + r["status"], + r["placed_at"], + ) + for r in reader + ] + if not rows: + return 0 + execute_values( + cur, + "INSERT INTO raw.meridian_orders (order_id, customer_id, performance_id, currency, subtotal, fees, total, status, placed_at) VALUES %s", + rows, + ) + return len(rows) + + +def ingest_meridian_order_items(cur, path: str) -> int: + with open(path, newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + rows = [ + ( + r["order_item_id"], + r["order_id"], + r["seat_section"], + r["unit_price"], + r["quantity"], + ) + for r in reader + ] + if not rows: + return 0 + execute_values( + cur, + "INSERT INTO raw.meridian_order_items (order_item_id, order_id, seat_section, unit_price, quantity) VALUES %s", + rows, + ) + return len(rows) + + +def ingest_meridian_web_sessions(cur, path: str) -> int: + with open(path, newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + rows = [ + ( + r["session_id"], + r["cookie_id"], + _null_if_empty(r.get("customer_id")), + _null_if_empty(r.get("event_id")), + r["page_type"], + r["occurred_at"], + ) + for r in reader + ] + if not rows: + return 0 + execute_values( + cur, + "INSERT INTO raw.meridian_web_sessions (session_id, cookie_id, customer_id, event_id, page_type, occurred_at) VALUES %s", + rows, + ) + return len(rows) + + +def ingest_whitelabel(cur, path: str, relation: str) -> tuple[str, int]: + """Append a batch to a whitelabel brand relation named by its parent dir, __
.""" + schema, _, table = relation.partition("__") + with open(path, newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + columns = list(reader.fieldnames or []) + rows = [ + tuple(r[c] if r[c] and r[c].strip() else None for c in columns) for r in reader + ] + n = whitelabel_sources.insert_rows(cur, schema, table, columns, rows) + return f"{schema}.{table}", n + + # Map first path segment (entity dir name) to (table_name, ingest_fn) ENTITY_HANDLERS = { "accounts": ("raw.accounts", ingest_accounts), @@ -161,6 +251,9 @@ def ingest_identity_merges(cur, path: str) -> int: "transactions": ("raw.transactions", ingest_transactions), "pages": ("raw.pages", ingest_pages), "identity_merges": ("raw.identity_merges", ingest_identity_merges), + "meridian_orders": ("raw.meridian_orders", ingest_meridian_orders), + "meridian_order_items": ("raw.meridian_order_items", ingest_meridian_order_items), + "meridian_web_sessions": ("raw.meridian_web_sessions", ingest_meridian_web_sessions), } @@ -204,6 +297,10 @@ def main(): rel = os.path.relpath(path, INCREMENTAL_DIR) parts = rel.split(os.sep) entity = parts[0] if parts else None + if entity == "whitelabel" and len(parts) > 2: + relation, n = ingest_whitelabel(cur, path, parts[1]) + print(f"Appended {n} rows to {relation} from {rel}") + continue if entity not in ENTITY_HANDLERS: print(f"Unknown entity dir: {entity}, skipping {rel}", file=sys.stderr) continue diff --git a/scripts/init.sh b/scripts/init.sh index 27d9570..99b34c4 100755 --- a/scripts/init.sh +++ b/scripts/init.sh @@ -18,6 +18,18 @@ docker compose exec -T warehouse psql -U postgres -d warehouse -v ON_ERROR_STOP= DROP SCHEMA IF EXISTS raw CASCADE; DROP SCHEMA IF EXISTS public CASCADE; CREATE SCHEMA public; + +-- Whitelabel brand schemas, including any added mid-session with bin/add-partner. +DO $$ +DECLARE brand_schema text; +BEGIN + FOR brand_schema IN + SELECT nspname FROM pg_namespace + WHERE nspname LIKE 'wl\_%' OR nspname LIKE 'partner\_%' + LOOP + EXECUTE format('DROP SCHEMA IF EXISTS %I CASCADE', brand_schema); + END LOOP; +END $$; SQL echo "Loading initial source data into raw..." diff --git a/scripts/load_initial_source_data.py b/scripts/load_initial_source_data.py index 5849457..c84d48c 100644 --- a/scripts/load_initial_source_data.py +++ b/scripts/load_initial_source_data.py @@ -1,5 +1,6 @@ """ Create raw schema and source tables, then load CSVs from data/initial/. +Also loads the per-brand whitelabel schemas from data/whitelabel/. Run from repo root: docker compose run --rm loader python scripts/load_initial_source_data.py """ import csv @@ -9,6 +10,8 @@ import psycopg2 from psycopg2.extras import execute_values +import whitelabel_sources + # Defaults match docker/dbt/profiles.yml and warehouse service PGHOST = os.environ.get("PGHOST", "warehouse") PGPORT = int(os.environ.get("PGPORT", "5432")) @@ -18,6 +21,7 @@ REPO_ROOT = os.environ.get("REPO_ROOT", "/app") INIT_DIR = os.path.join(REPO_ROOT, "data", "initial") +WHITELABEL_DIR = os.path.join(REPO_ROOT, "data", "whitelabel") def _null_if_empty(s): @@ -219,6 +223,225 @@ def main(): ) print(f"Loaded {len(rows)} rows into raw.identity_merges") + # --- Meridian Live (new MARI portfolio company being onboarded) --- + + # meridian_customers: customer_id, full_name, email, phone, country, created_at, marketing_opt_in + meridian_customers_csv = os.path.join(INIT_DIR, "meridian_customers.csv") + if os.path.isfile(meridian_customers_csv): + cur.execute(""" + CREATE TABLE raw.meridian_customers ( + customer_id TEXT PRIMARY KEY, + full_name TEXT, + email TEXT, + phone TEXT, + country TEXT, + created_at TIMESTAMPTZ, + marketing_opt_in TEXT + ); + """) + with open(meridian_customers_csv, newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + rows = [ + ( + r["customer_id"], + _null_if_empty(r.get("full_name")), + r["email"], + _null_if_empty(r.get("phone")), + r["country"], + r["created_at"], + _null_if_empty(r.get("marketing_opt_in")), + ) + for r in reader + ] + execute_values( + cur, + "INSERT INTO raw.meridian_customers (customer_id, full_name, email, phone, country, created_at, marketing_opt_in) VALUES %s", + rows, + ) + print(f"Loaded {len(rows)} rows into raw.meridian_customers") + + # meridian_venues: venue_id, name, city, country + meridian_venues_csv = os.path.join(INIT_DIR, "meridian_venues.csv") + if os.path.isfile(meridian_venues_csv): + cur.execute(""" + CREATE TABLE raw.meridian_venues ( + venue_id TEXT PRIMARY KEY, + name TEXT, + city TEXT, + country TEXT + ); + """) + with open(meridian_venues_csv, newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + rows = [(r["venue_id"], r["name"], r["city"], r["country"]) for r in reader] + execute_values( + cur, + "INSERT INTO raw.meridian_venues (venue_id, name, city, country) VALUES %s", + rows, + ) + print(f"Loaded {len(rows)} rows into raw.meridian_venues") + + # meridian_events: event_id, title, venue_id, category + meridian_events_csv = os.path.join(INIT_DIR, "meridian_events.csv") + if os.path.isfile(meridian_events_csv): + cur.execute(""" + CREATE TABLE raw.meridian_events ( + event_id TEXT PRIMARY KEY, + title TEXT, + venue_id TEXT, + category TEXT + ); + """) + with open(meridian_events_csv, newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + rows = [ + (r["event_id"], r["title"], r["venue_id"], r["category"]) + for r in reader + ] + execute_values( + cur, + "INSERT INTO raw.meridian_events (event_id, title, venue_id, category) VALUES %s", + rows, + ) + print(f"Loaded {len(rows)} rows into raw.meridian_events") + + # meridian_performances: performance_id, event_id, starts_at_local, utc_offset_minutes, doors_at_local + meridian_performances_csv = os.path.join(INIT_DIR, "meridian_performances.csv") + if os.path.isfile(meridian_performances_csv): + cur.execute(""" + CREATE TABLE raw.meridian_performances ( + performance_id TEXT PRIMARY KEY, + event_id TEXT, + starts_at_local TIMESTAMP, + utc_offset_minutes INTEGER, + doors_at_local TIMESTAMP + ); + """) + with open(meridian_performances_csv, newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + rows = [ + ( + r["performance_id"], + r["event_id"], + r["starts_at_local"], + r["utc_offset_minutes"], + r["doors_at_local"], + ) + for r in reader + ] + execute_values( + cur, + "INSERT INTO raw.meridian_performances (performance_id, event_id, starts_at_local, utc_offset_minutes, doors_at_local) VALUES %s", + rows, + ) + print(f"Loaded {len(rows)} rows into raw.meridian_performances") + + # meridian_orders: order_id, customer_id, performance_id, currency, subtotal, fees, total, status, placed_at + meridian_orders_csv = os.path.join(INIT_DIR, "meridian_orders.csv") + if os.path.isfile(meridian_orders_csv): + cur.execute(""" + CREATE TABLE raw.meridian_orders ( + order_id TEXT PRIMARY KEY, + customer_id TEXT, + performance_id TEXT, + currency TEXT, + subtotal TEXT, + fees TEXT, + total TEXT, + status TEXT, + placed_at TIMESTAMPTZ + ); + """) + with open(meridian_orders_csv, newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + rows = [ + ( + r["order_id"], + _null_if_empty(r.get("customer_id")), + _null_if_empty(r.get("performance_id")), + r["currency"], + r["subtotal"], + r["fees"], + r["total"], + r["status"], + r["placed_at"], + ) + for r in reader + ] + execute_values( + cur, + "INSERT INTO raw.meridian_orders (order_id, customer_id, performance_id, currency, subtotal, fees, total, status, placed_at) VALUES %s", + rows, + ) + print(f"Loaded {len(rows)} rows into raw.meridian_orders") + + # meridian_order_items: order_item_id, order_id, seat_section, unit_price, quantity + meridian_order_items_csv = os.path.join(INIT_DIR, "meridian_order_items.csv") + if os.path.isfile(meridian_order_items_csv): + cur.execute(""" + CREATE TABLE raw.meridian_order_items ( + order_item_id TEXT PRIMARY KEY, + order_id TEXT, + seat_section TEXT, + unit_price TEXT, + quantity INTEGER + ); + """) + with open(meridian_order_items_csv, newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + rows = [ + ( + r["order_item_id"], + r["order_id"], + r["seat_section"], + r["unit_price"], + r["quantity"], + ) + for r in reader + ] + execute_values( + cur, + "INSERT INTO raw.meridian_order_items (order_item_id, order_id, seat_section, unit_price, quantity) VALUES %s", + rows, + ) + print(f"Loaded {len(rows)} rows into raw.meridian_order_items") + + # meridian_web_sessions: session_id, cookie_id, customer_id, event_id, page_type, occurred_at + meridian_web_sessions_csv = os.path.join(INIT_DIR, "meridian_web_sessions.csv") + if os.path.isfile(meridian_web_sessions_csv): + cur.execute(""" + CREATE TABLE raw.meridian_web_sessions ( + session_id TEXT PRIMARY KEY, + cookie_id TEXT, + customer_id TEXT, + event_id TEXT, + page_type TEXT, + occurred_at TIMESTAMPTZ + ); + """) + with open(meridian_web_sessions_csv, newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + rows = [ + ( + r["session_id"], + r["cookie_id"], + _null_if_empty(r.get("customer_id")), + _null_if_empty(r.get("event_id")), + r["page_type"], + r["occurred_at"], + ) + for r in reader + ] + execute_values( + cur, + "INSERT INTO raw.meridian_web_sessions (session_id, cookie_id, customer_id, event_id, page_type, occurred_at) VALUES %s", + rows, + ) + print(f"Loaded {len(rows)} rows into raw.meridian_web_sessions") + + # --- Whitelabel storefronts: one schema per brand, outside raw --- + whitelabel_sources.load_dir(cur, WHITELABEL_DIR) + conn.commit() except Exception as e: conn.rollback() diff --git a/scripts/whitelabel_sources.py b/scripts/whitelabel_sources.py new file mode 100644 index 0000000..d091910 --- /dev/null +++ b/scripts/whitelabel_sources.py @@ -0,0 +1,90 @@ +""" +Load the whitelabel storefront page-tracking schemas. + +Each brand's tracking lands in its own schema, one relation per CSV in data/whitelabel/. +A file named __
.csv becomes exactly that relation: wl_arcadia__pages.csv +loads into wl_arcadia.pages. The column set comes from the CSV header, so brands whose +trackers emit different columns land in the warehouse exactly as they really are. + +Shared by load_initial_source_data.py and add_partner.py. +""" +import csv +import os + +from psycopg2 import sql +from psycopg2.extras import execute_values + + +def column_type(column: str) -> str: + """Columns named *_at hold timestamps; the rest land as text and are cleaned in dbt.""" + return "TIMESTAMPTZ" if column.endswith("_at") else "TEXT" + + +def _null_if_blank(value): + """Blank cells become NULL. Padding and sentinel text ('N/A', 'NULL') are left alone.""" + if value is None or not value.strip(): + return None + return value + + +def create_relation(cur, schema: str, table: str, columns: list[str]) -> None: + """(Re)create schema.table with one column per name, typed by column_type. + + Only the named relation is replaced; a brand's other relations survive. + """ + cur.execute(sql.SQL("CREATE SCHEMA IF NOT EXISTS {}").format(sql.Identifier(schema))) + cur.execute( + sql.SQL("DROP TABLE IF EXISTS {}.{} CASCADE").format( + sql.Identifier(schema), sql.Identifier(table) + ) + ) + cur.execute( + sql.SQL("CREATE TABLE {}.{} ({})").format( + sql.Identifier(schema), + sql.Identifier(table), + sql.SQL(", ").join( + sql.SQL("{} {}").format(sql.Identifier(c), sql.SQL(column_type(c))) + for c in columns + ), + ) + ) + + +def insert_rows(cur, schema: str, table: str, columns: list[str], rows: list[tuple]) -> int: + if not rows: + return 0 + execute_values( + cur, + sql.SQL("INSERT INTO {}.{} ({}) VALUES %s").format( + sql.Identifier(schema), + sql.Identifier(table), + sql.SQL(", ").join(sql.Identifier(c) for c in columns), + ).as_string(cur), + rows, + ) + return len(rows) + + +def load_csv(cur, csv_path: str) -> None: + """Load one __
.csv into the relation its filename names.""" + stem = os.path.basename(csv_path)[: -len(".csv")] + schema, _, table = stem.partition("__") + if not schema or not table: + raise ValueError(f"Expected __
.csv, got {os.path.basename(csv_path)}") + + with open(csv_path, newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + columns = list(reader.fieldnames or []) + rows = [tuple(_null_if_blank(r[c]) for c in columns) for r in reader] + + create_relation(cur, schema, table, columns) + n = insert_rows(cur, schema, table, columns, rows) + print(f"Loaded {n} rows into {schema}.{table}") + + +def load_dir(cur, directory: str) -> None: + if not os.path.isdir(directory): + return + for name in sorted(os.listdir(directory)): + if name.endswith(".csv"): + load_csv(cur, os.path.join(directory, name))