diff --git a/.dockerignore b/.dockerignore index c281542c..a943d4d8 100644 --- a/.dockerignore +++ b/.dockerignore @@ -14,6 +14,8 @@ !graphs/** !models/ !models/** +!examples/ +!examples/** !infra/ !infra/** infra/.DS_Store diff --git a/CHANGELOG.md b/CHANGELOG.md index 51d8ce12..1ca15738 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,17 @@ capabilities enter through the next minor release. ## Unreleased +### Added + +- Add sanitized failure codes and summaries to local and hosted run history, and reconcile stale + non-terminal runs after an expired lease is safely reacquired. +- Give every BigQuery run-scoped staging table a 24-hour expiration at creation while retaining + immediate cleanup for handled completion and failure paths. +- Add plan-first stage-zero and platform commands, source-free immutable image publication, a + read-only permission preflight, and documented least-privilege upgrade and rollback workflows. +- Expand the Salesforce project contract to four independently watermarked CRM endpoints, four + governed staging models, and an Opportunity fact with Account and historical owner dimensions. + ## 0.5.1 — 2026-08-05 (alpha) ### Fixed diff --git a/Dockerfile b/Dockerfile index 73647f53..3e6f19dc 100644 --- a/Dockerfile +++ b/Dockerfile @@ -14,6 +14,7 @@ COPY src ./src COPY connectors ./connectors COPY graphs ./graphs COPY models ./models +COPY examples ./examples COPY infra ./infra RUN uv sync --frozen --no-dev --no-editable diff --git a/HANDOFF.md b/HANDOFF.md index 8d3c8c27..fe7beb28 100644 --- a/HANDOFF.md +++ b/HANDOFF.md @@ -2,42 +2,40 @@ ## Finished -- Isolated the work on `codex/thin-init-cli`; the original `main` checkout remains untouched. -- Replaced the 232-line `dander init` orchestration body with a typed options object and delegate. -- Split manifest resolution, safety decisions, stage-zero sequencing, image publication, platform - configuration, and Terraform composition into `init_command.py`. -- Preserved the command signature, output, bootstrap monkeypatch points, and private helper aliases. -- Kept runtime behavior, package metadata, Terraform, and deployed resources unchanged. +- Reconciled Josh's current `main` with the complete functional beta-readiness tip. +- Preserved Josh's three unique commits, capability contracts, tickets, and secret-scan review. +- Added sanitized run failures, interrupted-run reconciliation, and supported staging expiration. +- Added plan-first operations, permission preflight, upgrade guidance, and rollback guidance. +- Added the four-endpoint Salesforce CRM example and five governed models. ## Try It -```bash -uv run dander init --help -uv run pytest tests/cli/test_init_cli.py -``` +Run `uv run dander validate`, then inspect the combined optional capabilities with +`uv run dander connector inspect PIPELINE`. The Salesforce example remains under +`examples/salesforce/` and can be validated source-free from an installed wheel. ## Checks -- Focused init suite passed: `9 passed`; full suite passed: `748 passed`. -- Repository-wide Ruff, formatting, and strict mypy passed. -- `dander init --help` matched `main` byte-for-byte. -- Dependency audit, platform Terraform, stage-zero Terraform, and generated Terraform passed. -- Wheel/sdist inspection and external installs passed; the container built and its runtime contract - passed. +- Combined full suite passed: `770 passed`. +- Ruff formatting/lint and strict mypy passed. +- Platform and stage-zero Terraform validation passed with backends disabled. +- Wheel/sdist inspection, external install, source-free generation, and validation passed. +- The combined Linux container image built successfully. +- Josh's five GitHub CI jobs passed on the cross-fork draft PR. ## Decisions -- Keep Typer option declarations in `main.py`; the new module never imports `main.py`. -- Inject existing bootstrap symbols from `main.py` to retain current test and development seams. -- Add no duplicate tests because the existing init suite already covers the extracted wiring. +- Merge exact functional tip `7a1378f`; exclude release-metadata commit `7703567`. +- Preserve both sides of the decision log and all of Josh's capability implementation. +- Keep package, lockfile, scaffold, and CI version metadata at `0.5.1`. ## Remaining -- Protected GitHub CI must repeat Linux package/container checks and security scans. -- Merge only after review; no deployment, release, Terraform apply, or GCP mutation is in scope. +- Josh must review the cross-fork draft PR; it must not merge automatically. +- Release publication, live acceptance, retained-project changes, and schedules remain untouched. ## Review First -- `src/dander/cli/init_command.py` -- `src/dander/cli/main.py` -- `tests/cli/test_init_cli.py` +- `docs/decisions.md` +- `src/dander/ingestion/capabilities.py` +- `src/dander/writer/bigquery.py` diff --git a/README.md b/README.md index 1eba5e87..13c9fecc 100644 --- a/README.md +++ b/README.md @@ -217,11 +217,12 @@ credential or employee row is stored in this repository. ### Enterprise authentication templates -`connectors/salesforce_jwt.example.yaml` is a complete read-only Accounts QueryAll slice: OAuth2 -JWT bearer authentication, Salesforce's distinct authorization-server audience, a short assertion, -server-filtered SOQL, bounded streaming Bulk API 2.0 result pages, a declared raw schema, and -soft-delete visibility. Hosted replays use an inclusive `SystemModstamp` watermark and idempotent -SCD1 publication. Dander 0.5 supports the independently installed first-party +`connectors/salesforce_jwt.example.yaml` is a complete read-only CRM slice for Accounts, Contacts, +Opportunities, and Users: OAuth2 JWT authentication, bounded streaming Bulk API 2.0 result pages, +declared raw schemas, and independent inclusive `SystemModstamp` watermarks. QueryAll retains +Account, Contact, and Opportunity tombstones; User deactivation is represented by `IsActive`. +Contact Email and Phone are personal data enabled by default. Dander supports the independently +installed first-party [`dander-connector-salesforce`](https://github.com/harrisonoconnorhover/dander-connector-salesforce) plugin; an exact manifest pin takes precedence over the deprecated built-in fallback. See [`docs/salesforce.md`](docs/salesforce.md). @@ -321,11 +322,12 @@ New users may instead use the [$300/90-day Free Trial](https://docs.cloud.google While the account remains a Free Trial account, Google says usage is not charged to the payment method; manually upgrading makes overages beyond remaining credit and free allowances billable. -`dander init` owns the complete two-stage bootstrap. It creates a hardened/versioned state bucket, -adopts it into Terraform, creates the administrative identity and Artifact Registry, builds and -pushes the current runtime, then applies datasets, per-pipeline IAM/jobs/schedules/secrets, and the -safety policy from `dander.yaml`. Newly generated projects use the ordinary hosted path without the -optional managed cost guard: +`dander init` and `dander init --apply` remain compatible shortcuts. The documented installation +path separates each mutation from its saved Terraform plan: create the state bucket once, run +`init-admin-plan`, review and run `init-admin-apply`, publish an immutable source-free image with +`image-publish`, then run `init-platform-plan`, review, and run `init-platform-apply`. See the +[hosted quickstart](docs/getting-started.md) for copyable commands. Newly generated projects use the +ordinary hosted path without the optional managed cost guard: ```yaml platform: @@ -350,45 +352,37 @@ override flags take precedence only when explicitly supplied. The cost guard def when `require_guarded_free_tier` is true and disabled when it is false; explicit cost-guard flags can override that default when the combination is valid. -```bash -uv run dander init \ - --project my-gcp-project \ - --failure-alert-email operator@example.com \ - --github-repository owner/repository \ - --apply -``` - -The state bucket defaults to `-dander-state`; the active `gcloud` user becomes the -approved administrator; operator-only stage-zero artifacts default to `~/.dander/` and -remain outside the repository. Terraform never receives secret values. Add each named value after -bootstrap with `gcloud secrets versions add`, then execute a paused pipeline manually before -enabling its schedule. `--failure-alert-email` is an operator input rather than a manifest field, -so personal addresses stay out of public repositories; repeat it on later reconciliations to retain -the email channel and per-pipeline Cloud Run failure policies. Plan-only mode remains available for -established environments but requires an existing backend, bootstrap identity, and immutable -`--container-image`. - -The granular `init-admin-*` and `init-platform-*` commands remain available for operators who need -to review/apply each identity boundary separately. The normal path is the single command above. -The approved administrative identity is deliberately separate from runtime identities; only it -can provision project resources. Guarded installations additionally use it to delegate each -runtime's read-only billing visibility. +The standard installer does not need Project Owner. The one-time stage-zero bundle is Service Usage +Admin, Storage Admin, Artifact Registry Administrator, Service Account Admin, and Project IAM +Admin. A cloud administrator can perform that step and grant the operator Service Account Token +Creator only on Dander's bootstrap account; image publication and later platform plans then use +that account through impersonation. GitHub WIF additionally requires Workload Identity Pool Admin. +Terraform never receives secret values. Add each named value after bootstrap with +`gcloud secrets versions add`, then execute a paused pipeline manually before enabling its schedule. +`--failure-alert-email` is an operator input rather than a manifest field, so personal addresses +stay out of public repositories; repeat it on later reconciliations to retain the email channel and +per-pipeline Cloud Run failure policies. + +Every plan command prints its saved-plan location and exact apply command. Every apply command +applies only that saved plan after confirmation. The approved administrative identity is separate +from runtime identities; only it can provision project resources. Guarded installations +additionally use it to delegate each runtime's read-only billing visibility. The default unguarded path does not request billing-account IAM or grant runtime billing/Pub/Sub guard permissions. Dander does not manage, limit, or prevent cloud spending in that configuration. To opt into the managed guard, set `require_guarded_free_tier: true` and pass `--billing-account ABCDEF-123456-ABCDEF`; the caller then needs the additional billing-account permissions required for the reviewed IAM and budget plan. -For an established environment, the equivalent explicit plan is: +For an established environment, produce the complete manifest-aware plan with: ```bash -uv run dander init \ +uv run dander init-platform-plan \ --project my-gcp-project \ --state-bucket my-existing-tfstate-bucket \ --bootstrap-service-account dander-bootstrap@my-gcp-project.iam.gserviceaccount.com \ --container-image us-central1-docker.pkg.dev/my-gcp-project/dander/dander@sha256:DIGEST \ --config dander.yaml \ - --github-repository owner/repository + --failure-alert-email operator@example.com ``` The image must use an immutable SHA-256 digest. `dander.yaml` declares every additive pipeline, diff --git a/connectors/salesforce.yaml b/connectors/salesforce.yaml index 710760a8..7d59d6dc 100644 --- a/connectors/salesforce.yaml +++ b/connectors/salesforce.yaml @@ -1,4 +1,4 @@ -# Read-only Accounts connector for Dander's disposable Salesforce Developer Edition proof org. +# Read-only CRM connector for Dander's disposable Salesforce Developer Edition proof org. # The External Client App consumer key and RSA private key are secret references, never literals. name: salesforce engine: salesforce_bulk2 @@ -32,8 +32,9 @@ endpoints: request_body: operation: queryAll query: >- - SELECT Id, Name, Type, Industry, AnnualRevenue, NumberOfEmployees, - BillingCity, BillingState, BillingCountry, CreatedDate, LastModifiedDate, + SELECT Id, Name, OwnerId, ParentId, Type, Industry, AnnualRevenue, + NumberOfEmployees, Website, Phone, BillingCity, BillingState, + BillingPostalCode, BillingCountry, CreatedDate, LastModifiedDate, SystemModstamp, IsDeleted FROM Account contentType: CSV columnDelimiter: COMMA @@ -54,6 +55,8 @@ endpoints: - name: Name type: STRING mode: REQUIRED + - {name: OwnerId, type: STRING} + - {name: ParentId, type: STRING} - name: Type type: STRING - name: Industry @@ -62,10 +65,13 @@ endpoints: type: NUMERIC - name: NumberOfEmployees type: INT64 + - {name: Website, type: STRING} + - {name: Phone, type: STRING} - name: BillingCity type: STRING - name: BillingState type: STRING + - {name: BillingPostalCode, type: STRING} - name: BillingCountry type: STRING - name: CreatedDate @@ -80,3 +86,113 @@ endpoints: - name: IsDeleted type: BOOL mode: REQUIRED + + - name: contacts + path: /jobs/query + pagination: + kind: header_cursor + next_cursor_header: Sforce-Locator + cursor_param: locator + size_param: maxRecords + page_size: 10000 + terminal_value: "null" + request_body: + operation: queryAll + query: >- + SELECT Id, AccountId, OwnerId, FirstName, LastName, Name, Email, Phone, + Title, Department, MailingCity, MailingState, MailingPostalCode, + MailingCountry, CreatedDate, LastModifiedDate, SystemModstamp, IsDeleted + FROM Contact + contentType: CSV + columnDelimiter: COMMA + lineEnding: LF + primary_key: [Id] + incremental_cursor: SystemModstamp + raw_schema: + - {name: Id, type: STRING, mode: REQUIRED} + - {name: AccountId, type: STRING} + - {name: OwnerId, type: STRING} + - {name: FirstName, type: STRING} + - {name: LastName, type: STRING, mode: REQUIRED} + - {name: Name, type: STRING, mode: REQUIRED} + - {name: Email, type: STRING} + - {name: Phone, type: STRING} + - {name: Title, type: STRING} + - {name: Department, type: STRING} + - {name: MailingCity, type: STRING} + - {name: MailingState, type: STRING} + - {name: MailingPostalCode, type: STRING} + - {name: MailingCountry, type: STRING} + - {name: CreatedDate, type: TIMESTAMP, mode: REQUIRED} + - {name: LastModifiedDate, type: TIMESTAMP, mode: REQUIRED} + - {name: SystemModstamp, type: TIMESTAMP, mode: REQUIRED} + - {name: IsDeleted, type: BOOL, mode: REQUIRED} + + - name: opportunities + path: /jobs/query + pagination: + kind: header_cursor + next_cursor_header: Sforce-Locator + cursor_param: locator + size_param: maxRecords + page_size: 10000 + terminal_value: "null" + request_body: + operation: queryAll + query: >- + SELECT Id, AccountId, OwnerId, Name, StageName, Amount, Probability, + CloseDate, Type, LeadSource, ForecastCategoryName, IsClosed, IsWon, + CreatedDate, LastModifiedDate, SystemModstamp, IsDeleted FROM Opportunity + contentType: CSV + columnDelimiter: COMMA + lineEnding: LF + primary_key: [Id] + incremental_cursor: SystemModstamp + raw_schema: + - {name: Id, type: STRING, mode: REQUIRED} + - {name: AccountId, type: STRING} + - {name: OwnerId, type: STRING} + - {name: Name, type: STRING, mode: REQUIRED} + - {name: StageName, type: STRING, mode: REQUIRED} + - {name: Amount, type: NUMERIC} + - {name: Probability, type: NUMERIC} + - {name: CloseDate, type: DATE, mode: REQUIRED} + - {name: Type, type: STRING} + - {name: LeadSource, type: STRING} + - {name: ForecastCategoryName, type: STRING} + - {name: IsClosed, type: BOOL, mode: REQUIRED} + - {name: IsWon, type: BOOL, mode: REQUIRED} + - {name: CreatedDate, type: TIMESTAMP, mode: REQUIRED} + - {name: LastModifiedDate, type: TIMESTAMP, mode: REQUIRED} + - {name: SystemModstamp, type: TIMESTAMP, mode: REQUIRED} + - {name: IsDeleted, type: BOOL, mode: REQUIRED} + + - name: users + path: /jobs/query + pagination: + kind: header_cursor + next_cursor_header: Sforce-Locator + cursor_param: locator + size_param: maxRecords + page_size: 10000 + terminal_value: "null" + request_body: + operation: query + query: >- + SELECT Id, Name, Alias, UserType, ProfileId, IsActive, CreatedDate, + LastModifiedDate, SystemModstamp FROM User + contentType: CSV + columnDelimiter: COMMA + lineEnding: LF + primary_key: [Id] + incremental_cursor: SystemModstamp + raw_schema: + - {name: Id, type: STRING, mode: REQUIRED} + - {name: Name, type: STRING, mode: REQUIRED} + - {name: Alias, type: STRING, mode: REQUIRED} + - {name: UserType, type: STRING, mode: REQUIRED} + - {name: ProfileId, type: STRING, mode: REQUIRED} + - {name: IsActive, type: BOOL, mode: REQUIRED} + - {name: CreatedDate, type: TIMESTAMP, mode: REQUIRED} + - {name: LastModifiedDate, type: TIMESTAMP, mode: REQUIRED} + - {name: SystemModstamp, type: TIMESTAMP, mode: REQUIRED} diff --git a/connectors/salesforce_jwt.example.yaml b/connectors/salesforce_jwt.example.yaml index c84a0888..f9dd9fcc 100644 --- a/connectors/salesforce_jwt.example.yaml +++ b/connectors/salesforce_jwt.example.yaml @@ -1,5 +1,5 @@ -# Read-only Salesforce Accounts template. Replace YOUR_DOMAIN and the JWT subject with values from -# your Salesforce org. The issuer and RSA private key are secret references, never literal values. +# Read-only Salesforce CRM template. Contact Email and Phone are personal data. Replace YOUR_DOMAIN +# and the JWT subject; keep credential values in Dander's secret store. name: salesforce engine: salesforce_bulk2 base_url: https://YOUR_DOMAIN.my.salesforce.com/services/data/v67.0 @@ -34,8 +34,9 @@ endpoints: request_body: operation: queryAll query: >- - SELECT Id, Name, Type, Industry, AnnualRevenue, NumberOfEmployees, - BillingCity, BillingState, BillingCountry, CreatedDate, LastModifiedDate, + SELECT Id, Name, OwnerId, ParentId, Type, Industry, AnnualRevenue, + NumberOfEmployees, Website, Phone, BillingCity, BillingState, + BillingPostalCode, BillingCountry, CreatedDate, LastModifiedDate, SystemModstamp, IsDeleted FROM Account contentType: CSV columnDelimiter: COMMA @@ -56,6 +57,8 @@ endpoints: - name: Name type: STRING mode: REQUIRED + - {name: OwnerId, type: STRING} + - {name: ParentId, type: STRING} - name: Type type: STRING - name: Industry @@ -64,10 +67,13 @@ endpoints: type: NUMERIC - name: NumberOfEmployees type: INT64 + - {name: Website, type: STRING} + - {name: Phone, type: STRING} - name: BillingCity type: STRING - name: BillingState type: STRING + - {name: BillingPostalCode, type: STRING} - name: BillingCountry type: STRING - name: CreatedDate @@ -82,3 +88,113 @@ endpoints: - name: IsDeleted type: BOOL mode: REQUIRED + + - name: contacts + path: /jobs/query + pagination: + kind: header_cursor + next_cursor_header: Sforce-Locator + cursor_param: locator + size_param: maxRecords + page_size: 10000 + terminal_value: "null" + request_body: + operation: queryAll + query: >- + SELECT Id, AccountId, OwnerId, FirstName, LastName, Name, Email, Phone, + Title, Department, MailingCity, MailingState, MailingPostalCode, + MailingCountry, CreatedDate, LastModifiedDate, SystemModstamp, IsDeleted + FROM Contact + contentType: CSV + columnDelimiter: COMMA + lineEnding: LF + primary_key: [Id] + incremental_cursor: SystemModstamp + raw_schema: + - {name: Id, type: STRING, mode: REQUIRED} + - {name: AccountId, type: STRING} + - {name: OwnerId, type: STRING} + - {name: FirstName, type: STRING} + - {name: LastName, type: STRING, mode: REQUIRED} + - {name: Name, type: STRING, mode: REQUIRED} + - {name: Email, type: STRING} + - {name: Phone, type: STRING} + - {name: Title, type: STRING} + - {name: Department, type: STRING} + - {name: MailingCity, type: STRING} + - {name: MailingState, type: STRING} + - {name: MailingPostalCode, type: STRING} + - {name: MailingCountry, type: STRING} + - {name: CreatedDate, type: TIMESTAMP, mode: REQUIRED} + - {name: LastModifiedDate, type: TIMESTAMP, mode: REQUIRED} + - {name: SystemModstamp, type: TIMESTAMP, mode: REQUIRED} + - {name: IsDeleted, type: BOOL, mode: REQUIRED} + + - name: opportunities + path: /jobs/query + pagination: + kind: header_cursor + next_cursor_header: Sforce-Locator + cursor_param: locator + size_param: maxRecords + page_size: 10000 + terminal_value: "null" + request_body: + operation: queryAll + query: >- + SELECT Id, AccountId, OwnerId, Name, StageName, Amount, Probability, + CloseDate, Type, LeadSource, ForecastCategoryName, IsClosed, IsWon, + CreatedDate, LastModifiedDate, SystemModstamp, IsDeleted FROM Opportunity + contentType: CSV + columnDelimiter: COMMA + lineEnding: LF + primary_key: [Id] + incremental_cursor: SystemModstamp + raw_schema: + - {name: Id, type: STRING, mode: REQUIRED} + - {name: AccountId, type: STRING} + - {name: OwnerId, type: STRING} + - {name: Name, type: STRING, mode: REQUIRED} + - {name: StageName, type: STRING, mode: REQUIRED} + - {name: Amount, type: NUMERIC} + - {name: Probability, type: NUMERIC} + - {name: CloseDate, type: DATE, mode: REQUIRED} + - {name: Type, type: STRING} + - {name: LeadSource, type: STRING} + - {name: ForecastCategoryName, type: STRING} + - {name: IsClosed, type: BOOL, mode: REQUIRED} + - {name: IsWon, type: BOOL, mode: REQUIRED} + - {name: CreatedDate, type: TIMESTAMP, mode: REQUIRED} + - {name: LastModifiedDate, type: TIMESTAMP, mode: REQUIRED} + - {name: SystemModstamp, type: TIMESTAMP, mode: REQUIRED} + - {name: IsDeleted, type: BOOL, mode: REQUIRED} + + - name: users + path: /jobs/query + pagination: + kind: header_cursor + next_cursor_header: Sforce-Locator + cursor_param: locator + size_param: maxRecords + page_size: 10000 + terminal_value: "null" + request_body: + operation: query + query: >- + SELECT Id, Name, Alias, UserType, ProfileId, IsActive, CreatedDate, + LastModifiedDate, SystemModstamp FROM User + contentType: CSV + columnDelimiter: COMMA + lineEnding: LF + primary_key: [Id] + incremental_cursor: SystemModstamp + raw_schema: + - {name: Id, type: STRING, mode: REQUIRED} + - {name: Name, type: STRING, mode: REQUIRED} + - {name: Alias, type: STRING, mode: REQUIRED} + - {name: UserType, type: STRING, mode: REQUIRED} + - {name: ProfileId, type: STRING, mode: REQUIRED} + - {name: IsActive, type: BOOL, mode: REQUIRED} + - {name: CreatedDate, type: TIMESTAMP, mode: REQUIRED} + - {name: LastModifiedDate, type: TIMESTAMP, mode: REQUIRED} + - {name: SystemModstamp, type: TIMESTAMP, mode: REQUIRED} diff --git a/dander.yaml b/dander.yaml index 2ccecfaa..f4ae1202 100644 --- a/dander.yaml +++ b/dander.yaml @@ -39,7 +39,13 @@ pipelines: salesforce_accounts: source: salesforce - models: [stg_salesforce__accounts] + models: + - stg_salesforce__users + - stg_salesforce__accounts + - stg_salesforce__contacts + - stg_salesforce__opportunities + - fct_salesforce__opportunities + publish_dataplex: true schedule: "0 11 * * *" time_zone: America/New_York paused: false diff --git a/docs/decisions.md b/docs/decisions.md index 21ce1e0d..93d1db3b 100644 --- a/docs/decisions.md +++ b/docs/decisions.md @@ -528,3 +528,22 @@ Resolves the "separate product decisions" the two entries above deferred, unbloc where the result isn't a lazily-consumed iterator). No concrete source implements any of the five yet — this ships the mechanism and contract only, per the existing `ConnectorOperation` registry's Open/Closed extension pattern. + +## 2026-08-05 — Hosted installation is plan-first by default + +- The compatibility `dander init` path remains, while public installation and upgrades separate + stage-zero planning, saved-plan application, immutable image publication, and platform planning. +- A read-only permission preflight tests the active identity before stage-zero Terraform. Optional + billing and Workload Identity permissions are checked only when those features are selected. +- A cloud administrator may perform stage zero once; later image and platform operations use the + bootstrap account through operator-scoped impersonation rather than Project Owner. + +## 2026-08-06 — Salesforce becomes the first deep connector + +- Accounts, Contacts, Opportunities, and Users retain separate inclusive `SystemModstamp` cursors + behind the existing `salesforce_bulk2` engine and plugin API v1. +- QueryAll retains visible CRM tombstones; Users retain inactive owners through `IsActive`. + Hard-deleted or purged records and provider write-back remain outside the contract. +- The governed fact excludes deleted Opportunities and Accounts while retaining inactive owner + dimensions. Existing installations may keep the `salesforce_accounts` pipeline resource ID; + new examples use `salesforce_crm`. diff --git a/docs/getting-started.md b/docs/getting-started.md index 39d5b7f5..77f6e4c1 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -7,9 +7,9 @@ Cloud provisioning has no promised duration and may incur charges. ## Prerequisites - An existing, dedicated, disposable GCP project with billing already linked. -- Project Owner on that disposable project (or equivalent permissions to enable services, create - the state bucket and service accounts, and grant the roles listed by the generated stage-zero - Terraform). +- Either a cloud administrator for the one-time stage-zero handoff, or an installer with these + predefined project roles: Service Usage Admin, Storage Admin, Artifact Registry Administrator, + Service Account Admin, and Project IAM Admin. Project Owner is not required. - Python 3.12, `uv`, Terraform 1.9 or newer, Docker with Buildx, and Google Cloud CLI. - An operator email address for hosted failure notifications. @@ -18,6 +18,9 @@ Set the two private operator values in your shell; do not commit them: ```bash export DANDER_PROJECT="your-disposable-project-id" export DANDER_ALERT_EMAIL="operator@example.com" +export DANDER_STATE_BUCKET="$DANDER_PROJECT-dander-state" +export DANDER_ADMIN_MEMBER="user:$(gcloud config get-value account)" +export DANDER_OPERATOR_DIR="$HOME/.dander/$DANDER_PROJECT/bootstrap" gcloud auth login gcloud config set project "$DANDER_PROJECT" @@ -39,31 +42,83 @@ dander run greenhouse_jobs --dry-run --project "$DANDER_PROJECT" The generated manifest keeps the schedule paused. The dry-run performs no credential or network access and should identify one Greenhouse endpoint and one selected model. -## Provision the paused platform +## Create the Terraform backend once -Read the confirmation carefully, then allow Dander to create stage zero, build and push the pinned -runtime image, and apply the reviewed platform plan: +Terraform cannot create the bucket that stores its own state. An administrator must create this one +hardened bucket before the first plan; every later infrastructure change is plan-reviewed: ```bash -dander init \ +gcloud storage buckets create "gs://$DANDER_STATE_BUCKET" \ --project "$DANDER_PROJECT" \ - --failure-alert-email "$DANDER_ALERT_EMAIL" \ - --apply + --location US \ + --uniform-bucket-level-access \ + --public-access-prevention +gcloud storage buckets update "gs://$DANDER_STATE_BUCKET" \ + --versioning \ + --update-labels=managed-by=dander,purpose=terraform-state ``` -The command stores Terraform state in the hardened bucket -`$DANDER_PROJECT-dander-state`, keeps local operator artifacts outside the project directory, and -leaves the Greenhouse scheduler paused. The generated project disables Dander's managed cost guard, -so this path does not require a billing-account ID or billing-account IAM permissions. Dander is not -managing, limiting, or preventing cloud spending in this configuration; disabling the guard does not -prevent or cap cloud charges. +## Plan and apply stage zero + +`init-admin-plan` first performs a read-only permission check and reports exact missing permissions. +It then adopts the backend bucket into stage-zero state and saves a plan outside the project: + +```bash +dander init-admin-plan \ + --project "$DANDER_PROJECT" \ + --state-bucket "$DANDER_STATE_BUCKET" \ + --admin-member "$DANDER_ADMIN_MEMBER" \ + --operator-artifact-dir "$DANDER_OPERATOR_DIR" + +terraform -chdir=infra/bootstrap-admin show -no-color \ + "$DANDER_OPERATOR_DIR/dander-admin-bootstrap.tfplan" + +dander init-admin-apply \ + --project "$DANDER_PROJECT" \ + --state-bucket "$DANDER_STATE_BUCKET" \ + --admin-member "$DANDER_ADMIN_MEMBER" \ + --operator-artifact-dir "$DANDER_OPERATOR_DIR" +``` + +The apply command uses the exact saved plan. Stage zero creates the bootstrap service account, +Artifact Registry repository, and its project-level provisioning roles. The approved operator gets +Token Creator only on that bootstrap account; later image publication and Terraform operations use +impersonation rather than Project Owner. + +## Publish, plan, and apply the paused platform + +```bash +dander image-publish \ + --project "$DANDER_PROJECT" \ + --state-bucket "$DANDER_STATE_BUCKET" \ + --failure-alert-email "$DANDER_ALERT_EMAIL" +``` + +The command confirms before pushing, prints the immutable image digest, and prints the complete +`init-platform-plan` command. Run that printed command, review its saved plan, then apply it: + +```bash +terraform -chdir=infra show -no-color dander-bootstrap.tfplan + +dander init-platform-apply \ + --project "$DANDER_PROJECT" \ + --state-bucket "$DANDER_STATE_BUCKET" \ + --bootstrap-service-account \ + "dander-bootstrap@$DANDER_PROJECT.iam.gserviceaccount.com" +``` + +The generated project leaves the Greenhouse scheduler paused and disables Dander's managed cost +guard. This path needs no billing-account ID or billing-account IAM. Dander is not managing, +limiting, or preventing cloud spending; disabling the guard does not prevent or cap charges. The managed cost guard is optional. To use its simulation-first budget, Pub/Sub notification path, billing-shutoff function, and guarded runtime preflight, set `platform.safety.require_guarded_free_tier` to `true` and pass `--billing-account`. That opt-in path requires permission to grant the bootstrap identity Billing Account Administrator on the linked billing account. Review those additional billing-account and project-level IAM changes before any -apply. The guard remains simulation-first unless `--live-cost-guard` is explicitly supplied. +apply. Selecting GitHub WIF similarly requires Workload Identity Pool Admin during stage zero. +Neither permission is requested by the standard unguarded, non-WIF path. The guard remains +simulation-first unless `--live-cost-guard` is explicitly supplied. ## Run and verify Greenhouse @@ -90,10 +145,11 @@ non-empty raw and staging relation proves the live public extraction and transfo ## Enable the schedule only after proof Change only `pipelines.greenhouse_jobs.paused` to `false` in `dander.yaml`, rerun `dander validate`, -and repeat the same `dander init ... --apply` command. Review the confirmation and resulting plan; -the expected platform change is the scheduler state plus the immutable image generated from the -updated project content. +then repeat `image-publish`, `init-platform-plan`, plan review, and `init-platform-apply`. The +expected platform changes are the scheduler state and immutable image. Repeat `init-platform-plan` +afterward and require Terraform to report `No changes.` -Use [upgrading.md](upgrading.md) for later releases and safe reruns. If installation or the first +Use [upgrading.md](upgrading.md) for later releases and safe reruns, and [rollback.md](rollback.md) +for restoring a known-good image. If installation or the first hosted run cannot be completed from this page, report it as a defect in the current supported release rather than consulting Dander's source. diff --git a/docs/known-limitations.md b/docs/known-limitations.md index 79991be8..5c813822 100644 --- a/docs/known-limitations.md +++ b/docs/known-limitations.md @@ -16,8 +16,9 @@ an unattended production system containing business-critical data. - ServiceNow incidents extraction performs a stably ordered full endpoint read. It does not use a timestamp watermark with offset paging because that combination can skip moving records. - Source hard deletes are not propagated by the first ServiceNow incidents slice. -- The supported Salesforce and ServiceNow plugins each expose one read-only object slice. They do - not provide provider write-back or broad object/table coverage. +- Salesforce reads Accounts, Contacts, Opportunities, and Users only. QueryAll preserves visible + soft-deletion tombstones, but hard-deleted or purged records cannot be recovered and provider + write-back is not supported. ServiceNow remains one read-only incident slice. - NetSuite customers are simulator-validated only. The first SuiteQL slice performs a full read, is capped by NetSuite's 100,000-result SuiteQL REST limit, and has not passed tenant-specific role, field-availability, or authentication acceptance. It is not in the current public support diff --git a/docs/rollback.md b/docs/rollback.md new file mode 100644 index 00000000..df4a1337 --- /dev/null +++ b/docs/rollback.md @@ -0,0 +1,28 @@ +# Roll back a hosted Dander release + +Rollback restores a previously successful project configuration and immutable runtime image. It +does not undo changes made in a source system, remove ingested rows, or destructively reverse a +BigQuery schema. + +## Identify the known-good deployment + +Record the last project commit that completed a hosted run and its exact Artifact Registry +`@sha256:` image digest. Create a new rollback branch and commit that restores the compatible +package/plugin pins and manifest configuration; do not rewrite shared Git history. + +## Pause and restore through reviewed plans + +1. Set every affected pipeline's tracked `paused` value to `true`. +2. Run `dander init-platform-plan` with the known-good image digest and the installation's existing + state bucket, bootstrap service account, alert address, and guarded-runtime inputs. +3. Reject any plan that changes datasets, state, secret values, unrelated jobs, or runtime IAM. +4. Review the saved plan with `terraform -chdir=infra show -no-color dander-bootstrap.tfplan`. +5. Run `dander init-platform-apply`; it applies only that saved plan. +6. Execute each paused job manually. Verify run history, transforms/tests, row counts, cursors, + leases, and staging cleanup. +7. Restore the tracked schedule state through a second reviewed plan and apply. +8. Repeat `init-platform-plan` and require Terraform to report `No changes.` + +If the prior image is incompatible with the current manifest or deployed schema, stop and repair +forward. Never manually edit Terraform state, `_dander_leases`, `_dander_watermarks`, or staging +tables to manufacture a rollback. diff --git a/docs/salesforce.md b/docs/salesforce.md index 9ccb011e..5c8cf082 100644 --- a/docs/salesforce.md +++ b/docs/salesforce.md @@ -1,75 +1,126 @@ -# Salesforce Accounts +# Salesforce CRM -Dander's Salesforce slice is intentionally read-only: one Accounts Bulk API 2.0 `queryAll` job, -server-filtered SOQL, streaming locator pagination, a declared raw schema, SCD1 publication, one -staging model, and the existing transform/test/run-history path. It does not modify Salesforce -business records. +Dander's Salesforce connector is read-only and uses bounded Bulk API 2.0 CSV streaming for four +independently watermarked endpoints: -Salesforce restricts creation of legacy Connected Apps as of Spring '26 and recommends External -Client Apps for new integrations. Configure one External Client App for non-interactive JWT bearer -authentication: +| Endpoint | Raw relation | Deletion state | +| --- | --- | --- | +| Accounts | `raw.salesforce_accounts` | `IsDeleted` from `queryAll` | +| Contacts | `raw.salesforce_contacts` | `IsDeleted` from `queryAll` | +| Opportunities | `raw.salesforce_opportunities` | `IsDeleted` from `queryAll` | +| Users | `raw.salesforce_users` | `IsActive` from `query` | + +Contact Email and Phone are enabled by default and are personal data. Confirm that the destination +project, operators, retention, and downstream access are appropriate before enabling this pipeline. + +## Salesforce setup + +Salesforce recommends External Client Apps for new integrations. Configure one for non-interactive +JWT bearer authentication: 1. Generate a 2048-bit RSA private key and self-signed public certificate outside the repository. 2. In **Setup → External Client App Manager**, create a local app and enable OAuth. 3. Add **Manage user data via APIs (`api`)** and **Perform requests at any time (`refresh_token`, `offline_access`)**, enable **JWT Bearer Flow**, and upload only the public - certificate. Salesforce requires the refresh-token scope when an External Client App uses this - preauthorized JWT flow; Dander does not store or use a refresh token. -4. Set permitted users to **Admin approved users are pre-authorized**, select a narrow permission - set that can read Account and the selected fields, and assign it to the JWT subject user. -5. Copy the consumer key. Store it and the private key as secret values; never place either value - in connector YAML. + certificate. Dander does not store or use a refresh token. +4. Set permitted users to **Admin approved users are pre-authorized**. Assign a narrow permission + set with API access and read access to Account, Contact, Opportunity, User, and every selected + field. +5. Store the consumer key and RSA private key in Dander's configured secret store. Never place a + credential value in connector YAML. -Copy and edit the template: +Newly generated projects include a source-free copy of the complete example. Copy it into the +project root, then edit the connector: ```bash -cp connectors/salesforce_jwt.example.yaml connectors/salesforce.yaml +cp examples/salesforce/dander.yaml dander.yaml +cp examples/salesforce/connectors/salesforce.yaml connectors/salesforce.yaml +cp -R examples/salesforce/models/staging/. models/staging/ +mkdir -p models/marts +cp -R examples/salesforce/models/marts/. models/marts/ +``` + +Those commands intentionally turn a newly generated starter into a Salesforce-only project. In an +existing project, copy only the connector/models and merge the `plugins.salesforce` and +`pipelines.salesforce_crm` blocks instead of replacing `dander.yaml`. + +Replace `YOUR_DOMAIN`, the API version when necessary, and the JWT subject. Production orgs usually +use `https://login.salesforce.com` for the token URL and audience; sandboxes use +`https://test.salesforce.com`. A My Domain host is supported when it matches the org's OAuth setup. + +Custom fields are opt-in. Add each field to both the endpoint SOQL `SELECT` and `raw_schema`, then +add its governed projection to the relevant model. Undeclared fields fail before publication. + +## Project configuration + +After `dander-connector-salesforce 0.3.0rc1` is published, a new candidate project can use: + +```yaml +version: 1 +plugins: + salesforce: + distribution: dander-connector-salesforce + version: 0.3.0rc1 +pipelines: + salesforce_crm: + source: salesforce + models: + - stg_salesforce__users + - stg_salesforce__accounts + - stg_salesforce__contacts + - stg_salesforce__opportunities + - fct_salesforce__opportunities + publish_dataplex: true + schedule: "0 7 * * *" + time_zone: America/New_York + paused: true + secrets: + SALESFORCE_EXTERNAL_CLIENT_APP_ID: salesforce-client-id + SALESFORCE_EXTERNAL_CLIENT_APP_PRIVATE_KEY: salesforce-private-key ``` -Replace `YOUR_DOMAIN`, the API version when necessary, and the JWT `subject`. Production orgs use -`https://login.salesforce.com` as both authorization-server audience and token host; sandboxes use -`https://test.salesforce.com`. My Domain hosts are also supported when both values follow the org's -OAuth configuration. +Existing deployments may keep their current pipeline ID and resource overrides to avoid replacing +Cloud Run and Scheduler resources. The repository's retained project therefore keeps the historical +`salesforce_accounts` ID even though it now runs the complete CRM slice. -First validate the connector shape without resolving secrets or contacting Salesforce: +Install the exact plugin pin and validate without resolving secrets or contacting Salesforce: ```bash -uv run dander run salesforce --dry-run --sandbox --project YOUR_NO_BILLING_GCP_PROJECT +dander plugins install +dander validate +dander run salesforce_crm --dry-run --project YOUR_GCP_PROJECT ``` -That command is configuration-only; it does not prove authentication. For a real local extraction, -authenticate Application Default Credentials to a BigQuery Sandbox GCP project with billing -disabled, then resolve the template's two references from environment variables and omit -`--dry-run`: +## Governed models -```bash -gcloud auth application-default login -export SALESFORCE_EXTERNAL_CLIENT_APP_ID='the-consumer-key' -export SALESFORCE_EXTERNAL_CLIENT_APP_PRIVATE_KEY="$(< /secure/path/dander-salesforce.key)" -uv run dander run salesforce --sandbox --project YOUR_NO_BILLING_GCP_PROJECT -``` +The project includes four staging models plus `marts.fct_salesforce__opportunities`. The fact: + +- excludes deleted Opportunities and Opportunities tied to deleted Accounts; +- joins Account name, type, and industry; +- joins owner name, alias, type, and active status without filtering inactive owners; +- exposes amount, probability, stage, forecast category, close date, and closed/won flags; and +- publishes descriptions, types, tests, relationships, and metrics through Dander's metadata spine. + +Relationship tests ignore nullable keys. Build Users before Accounts, then Contacts and +Opportunities, so owner and Account assertions have an available parent relation. -The caller must be able to read the project's billing status and create/write the BigQuery Sandbox -dataset. The real command authenticates to Salesforce, streams Accounts through bounded CSV -result pages, replaces the raw sandbox table, and records local run/cursor state in -`.dander/state.db`. Sandbox runs intentionally perform a complete read; hosted runs inject the -committed `SystemModstamp` into the next SOQL query. +## Runtime and replay -A hosted pipeline should map those same environment names to two Secret Manager containers in -`dander.yaml`; Terraform manages the containers and least-privilege runtime access, not secret -versions. Review the plan before applying. +Each endpoint creates one asynchronous query job, polls with a fixed upper bound, and reads each CSV +page through Salesforce's opaque `Sforce-Locator`. Pages are streamed record by record rather than +materializing an endpoint in memory. Query jobs are deleted after success and handled failure. -## Current boundary +Hosted runs apply `SystemModstamp >= ` independently to each endpoint. The inclusive +boundary deliberately rereads tied timestamps; Dander's SCD1 writer makes replay duplicate-free and +commits a watermark only after the complete endpoint succeeds. Sandbox runs remain full reads. -The connector creates one asynchronous Bulk API 2.0 query job, polls it with a fixed upper bound, -and reads each CSV page through Salesforce's opaque `Sforce-Locator`. The response is streamed a -record at a time rather than materialized as one endpoint-sized list. The result job is deleted -after success or handled failure. QueryAll preserves soft-delete visibility, and the inclusive -`SystemModstamp >= ` boundary makes tied timestamps replay-safe through hosted SCD1. +Soft-deleted Accounts, Contacts, and Opportunities remain in raw/staging with `is_deleted=true`. +The fact filters those tombstones. User deactivation changes `is_active` but retains the owner for +historical reporting. Records already hard-deleted or purged from Salesforce cannot be discovered. The base SOQL must remain one unfiltered, unordered `SELECT`: Dander owns the watermark predicate, -and omitting `ORDER BY` preserves Salesforce PK chunking for large jobs. Bulk jobs have no -completion SLA; a job that exceeds Dander's bounded polling window fails clearly and can be rerun. +and omitting `ORDER BY` preserves Salesforce's large-query behavior. A job that exceeds Dander's +bounded polling window fails clearly and can be safely rerun. Official Salesforce references: diff --git a/docs/upgrading.md b/docs/upgrading.md index a1768c0c..0fda853c 100644 --- a/docs/upgrading.md +++ b/docs/upgrading.md @@ -20,29 +20,23 @@ Set every hosted pipeline's `paused` field to `true`, update the exact `ARG DAND in the project Dockerfile, then run `dander validate`. Do not use an unpinned package or mutable container tag. -Build and push the source-free project image and resolve its immutable digest: +Build and push the source-free project image through the bootstrap identity: ```bash -export DANDER_REGION="us-central1" -export DANDER_IMAGE_REPOSITORY="$DANDER_REGION-docker.pkg.dev/$DANDER_PROJECT/dander/dander" - -gcloud auth configure-docker "$DANDER_REGION-docker.pkg.dev" --quiet -docker buildx build --platform linux/amd64 --push \ - --tag "$DANDER_IMAGE_REPOSITORY:v$DANDER_TARGET_VERSION" . -export DANDER_IMAGE_DIGEST="$( - gcloud artifacts docker images describe \ - "$DANDER_IMAGE_REPOSITORY:v$DANDER_TARGET_VERSION" \ - --project "$DANDER_PROJECT" \ - --format='value(image_summary.digest)' -)" -test -n "$DANDER_IMAGE_DIGEST" -export DANDER_IMAGE="$DANDER_IMAGE_REPOSITORY@$DANDER_IMAGE_DIGEST" +dander image-publish \ + --project "$DANDER_PROJECT" \ + --state-bucket "$DANDER_PROJECT-dander-state" \ + --failure-alert-email "$DANDER_ALERT_EMAIL" ``` +Copy the immutable digest from the output and run the complete `init-platform-plan` command Dander +prints. Image publication uses the bootstrap service account; the operator needs Token Creator on +that account, not Project Owner or direct Artifact Registry administration. + ## Review and apply the exact plan ```bash -dander init \ +dander init-platform-plan \ --project "$DANDER_PROJECT" \ --state-bucket "$DANDER_PROJECT-dander-state" \ --bootstrap-service-account "dander-bootstrap@$DANDER_PROJECT.iam.gserviceaccount.com" \ @@ -75,6 +69,9 @@ be immediately retried. After every pipeline passes, restore the tracked `paused` values, build the resulting project image, repeat the reviewed plan/apply sequence, and run one final plan that reports no changes. +Use [rollback.md](rollback.md) if the candidate cannot pass its manual smoke run. Rollback reuses a +known-good immutable digest and never edits Terraform state, leases, watermarks, or staging by hand. + On an alert, inspect the Cloud Run execution and Dander run ledger first. Rerun once only after the prior execution is terminal. Never edit `_dander_leases`, `_dander_watermarks`, run-scoped staging, or Terraform state by hand. A cleanup, lease, cursor, schema, or unexplained-drift problem is a diff --git a/examples/salesforce/dander.yaml b/examples/salesforce/dander.yaml new file mode 100644 index 00000000..759ebd9c --- /dev/null +++ b/examples/salesforce/dander.yaml @@ -0,0 +1,29 @@ +version: 1 + +platform: + region: us-central1 + bigquery_location: US + safety: + require_guarded_free_tier: false + +plugins: + salesforce: + distribution: dander-connector-salesforce + version: 0.3.0rc1 + +pipelines: + salesforce_crm: + source: salesforce + models: + - stg_salesforce__users + - stg_salesforce__accounts + - stg_salesforce__contacts + - stg_salesforce__opportunities + - fct_salesforce__opportunities + publish_dataplex: true + schedule: "0 7 * * *" + time_zone: America/New_York + paused: true + secrets: + SALESFORCE_EXTERNAL_CLIENT_APP_ID: salesforce-client-id + SALESFORCE_EXTERNAL_CLIENT_APP_PRIVATE_KEY: salesforce-private-key diff --git a/models/marts/fct_salesforce__opportunities.sql b/models/marts/fct_salesforce__opportunities.sql new file mode 100644 index 00000000..4252a718 --- /dev/null +++ b/models/marts/fct_salesforce__opportunities.sql @@ -0,0 +1,40 @@ +-- Purpose: governed Salesforce opportunity facts with Account and historical owner dimensions. +-- Grain: one non-deleted Opportunity whose Account is either present and active or not assigned. +WITH users AS ( + SELECT * FROM {{ ref('stg_salesforce__users') }} +), +accounts AS ( + SELECT * FROM {{ ref('stg_salesforce__accounts') }} WHERE NOT is_deleted +), +opportunities AS ( + SELECT * FROM {{ ref('stg_salesforce__opportunities') }} WHERE NOT is_deleted +) + +SELECT + opportunity.opportunity_id, + opportunity.opportunity_name, + opportunity.account_id, + account.account_name, + account.account_type, + account.industry, + opportunity.owner_id, + owner.user_name AS owner_name, + owner.user_alias AS owner_alias, + owner.user_type AS owner_type, + owner.is_active AS owner_is_active, + opportunity.stage_name, + opportunity.amount, + opportunity.probability, + opportunity.close_date, + opportunity.opportunity_type, + opportunity.lead_source, + opportunity.forecast_category, + opportunity.is_closed, + opportunity.is_won, + opportunity.created_at, + opportunity.last_modified_at, + opportunity.system_modified_at +FROM opportunities AS opportunity +LEFT JOIN accounts AS account ON opportunity.account_id = account.account_id +LEFT JOIN users AS owner ON opportunity.owner_id = owner.user_id +WHERE opportunity.account_id IS NULL OR account.account_id IS NOT NULL diff --git a/models/marts/fct_salesforce__opportunities.yml b/models/marts/fct_salesforce__opportunities.yml new file mode 100644 index 00000000..5c941d03 --- /dev/null +++ b/models/marts/fct_salesforce__opportunities.yml @@ -0,0 +1,49 @@ +model: fct_salesforce__opportunities +description: Non-deleted Salesforce Opportunity facts with active Account and historical owner data. +owner: data-eng +materialization: table +dataset: marts +source_system: salesforce +sensitivity: confidential + +columns: + - {name: opportunity_id, type: STRING, description: Stable Salesforce Opportunity identifier.} + - {name: opportunity_name, type: STRING, description: Opportunity display name.} + - {name: account_id, type: STRING, description: Related non-deleted Account identifier.} + - {name: account_name, type: STRING, description: Related non-deleted Account name.} + - {name: account_type, type: STRING, description: Related Account classification.} + - {name: industry, type: STRING, description: Related Account industry.} + - {name: owner_id, type: STRING, description: Current or historical owner User identifier.} + - {name: owner_name, type: STRING, description: Owner display name including inactive owners.} + - {name: owner_alias, type: STRING, description: Owner Salesforce alias.} + - {name: owner_type, type: STRING, description: Owner Salesforce User type.} + - {name: owner_is_active, type: BOOL, description: Whether the owner is currently active.} + - {name: stage_name, type: STRING, description: Current Opportunity stage.} + - {name: amount, type: NUMERIC, description: Opportunity revenue amount when present.} + - {name: probability, type: NUMERIC, description: Salesforce win probability percentage.} + - {name: close_date, type: DATE, description: Expected or actual close date.} + - {name: opportunity_type, type: STRING, description: Salesforce Opportunity type.} + - {name: lead_source, type: STRING, description: Lead source when present.} + - {name: forecast_category, type: STRING, description: Salesforce forecast category.} + - {name: is_closed, type: BOOL, description: Whether the Opportunity is closed.} + - {name: is_won, type: BOOL, description: Whether the Opportunity is won.} + - {name: created_at, type: TIMESTAMP, description: Salesforce record creation timestamp.} + - {name: last_modified_at, type: TIMESTAMP, description: User-facing modification timestamp.} + - {name: system_modified_at, type: TIMESTAMP, description: Salesforce system watermark.} + +tests: + - {column: opportunity_id, not_null: true, unique: true} + - {column: account_id, relationships: {to: stg_salesforce__accounts, field: account_id}} + - {column: owner_id, relationships: {to: stg_salesforce__users, field: user_id}} + - {column: is_closed, not_null: true, accepted_values: [true, false]} + - {column: is_won, not_null: true, accepted_values: [true, false]} + +metrics: + - name: active_opportunity_count + description: Number of non-deleted Opportunities tied to active or no Accounts. + aggregation: count_distinct + field: opportunity_id + - name: active_opportunity_amount + description: Total amount across non-deleted Opportunities tied to active or no Accounts. + aggregation: sum + field: amount diff --git a/models/staging/stg_salesforce__accounts.sql b/models/staging/stg_salesforce__accounts.sql index 4cb36715..3dad7212 100644 --- a/models/staging/stg_salesforce__accounts.sql +++ b/models/staging/stg_salesforce__accounts.sql @@ -4,12 +4,17 @@ WITH source AS ( SELECT Id AS account_id, Name AS account_name, + OwnerId AS owner_id, + ParentId AS parent_account_id, Type AS account_type, Industry AS industry, AnnualRevenue AS annual_revenue, NumberOfEmployees AS number_of_employees, + Website AS website, + Phone AS phone, BillingCity AS billing_city, BillingState AS billing_state, + BillingPostalCode AS billing_postal_code, BillingCountry AS billing_country, TIMESTAMP(CreatedDate) AS created_at, TIMESTAMP(LastModifiedDate) AS last_modified_at, @@ -21,12 +26,17 @@ WITH source AS ( SELECT account_id, account_name, + owner_id, + parent_account_id, account_type, industry, annual_revenue, number_of_employees, + website, + phone, billing_city, billing_state, + billing_postal_code, billing_country, created_at, last_modified_at, diff --git a/models/staging/stg_salesforce__accounts.yml b/models/staging/stg_salesforce__accounts.yml index dab01eed..627167a7 100644 --- a/models/staging/stg_salesforce__accounts.yml +++ b/models/staging/stg_salesforce__accounts.yml @@ -15,6 +15,12 @@ columns: - name: account_name type: STRING description: Salesforce Account display name. + - name: owner_id + type: STRING + description: Salesforce User identifier for the current Account owner. + - name: parent_account_id + type: STRING + description: Parent Salesforce Account identifier when the hierarchy is populated. - name: account_type type: STRING description: Salesforce Account type when assigned. @@ -27,12 +33,21 @@ columns: - name: number_of_employees type: INT64 description: Reported employee count when present. + - name: website + type: STRING + description: Account website when present. + - name: phone + type: STRING + description: Account telephone number when present. - name: billing_city type: STRING description: Billing city when present. - name: billing_state type: STRING description: Billing state or region when present. + - name: billing_postal_code + type: STRING + description: Billing postal code when present. - name: billing_country type: STRING description: Billing country when present. @@ -55,6 +70,14 @@ tests: unique: true - column: account_name not_null: true + - column: owner_id + relationships: + to: stg_salesforce__users + field: user_id + - column: parent_account_id + relationships: + to: stg_salesforce__accounts + field: account_id - column: is_deleted accepted_values: [true, false] diff --git a/models/staging/stg_salesforce__contacts.sql b/models/staging/stg_salesforce__contacts.sql new file mode 100644 index 00000000..9daad77d --- /dev/null +++ b/models/staging/stg_salesforce__contacts.sql @@ -0,0 +1,22 @@ +-- Purpose: normalize Salesforce Contacts while preserving soft-deletion state. +-- Grain: one row per Salesforce Contact id. +SELECT + Id AS contact_id, + AccountId AS account_id, + OwnerId AS owner_id, + FirstName AS first_name, + LastName AS last_name, + Name AS contact_name, + Email AS email, + Phone AS phone, + Title AS title, + Department AS department, + MailingCity AS mailing_city, + MailingState AS mailing_state, + MailingPostalCode AS mailing_postal_code, + MailingCountry AS mailing_country, + TIMESTAMP(CreatedDate) AS created_at, + TIMESTAMP(LastModifiedDate) AS last_modified_at, + TIMESTAMP(SystemModstamp) AS system_modified_at, + IsDeleted AS is_deleted +FROM {{ ref('raw_salesforce_contacts') }} diff --git a/models/staging/stg_salesforce__contacts.yml b/models/staging/stg_salesforce__contacts.yml new file mode 100644 index 00000000..fefb3ec9 --- /dev/null +++ b/models/staging/stg_salesforce__contacts.yml @@ -0,0 +1,42 @@ +model: stg_salesforce__contacts +description: Salesforce Contacts normalized with Account and owner relationships and tombstones. +owner: data-eng +materialization: incremental +dataset: staging +source_system: salesforce +sensitivity: restricted +unique_key: [contact_id] +incremental_cursor: system_modified_at + +columns: + - {name: contact_id, type: STRING, description: Stable Salesforce Contact identifier.} + - {name: account_id, type: STRING, description: Related Account identifier when present.} + - {name: owner_id, type: STRING, description: Current Salesforce User owner identifier.} + - {name: first_name, type: STRING, description: Contact first name when present.} + - {name: last_name, type: STRING, description: Contact last name.} + - {name: contact_name, type: STRING, description: Salesforce-composed Contact display name.} + - {name: email, type: STRING, description: Personal email address when present.} + - {name: phone, type: STRING, description: Personal telephone number when present.} + - {name: title, type: STRING, description: Contact job title when present.} + - {name: department, type: STRING, description: Contact department when present.} + - {name: mailing_city, type: STRING, description: Mailing city when present.} + - {name: mailing_state, type: STRING, description: Mailing state or region when present.} + - {name: mailing_postal_code, type: STRING, description: Mailing postal code when present.} + - {name: mailing_country, type: STRING, description: Mailing country when present.} + - {name: created_at, type: TIMESTAMP, description: Salesforce record creation timestamp.} + - {name: last_modified_at, type: TIMESTAMP, description: User-facing modification timestamp.} + - {name: system_modified_at, type: TIMESTAMP, description: Salesforce system watermark.} + - {name: is_deleted, type: BOOL, description: Whether Salesforce soft-deleted the Contact.} + +tests: + - {column: contact_id, not_null: true, unique: true} + - {column: last_name, not_null: true} + - {column: account_id, relationships: {to: stg_salesforce__accounts, field: account_id}} + - {column: owner_id, relationships: {to: stg_salesforce__users, field: user_id}} + - {column: is_deleted, accepted_values: [true, false]} + +metrics: + - name: contact_count + description: Number of Salesforce Contacts represented, including soft-deleted records. + aggregation: count_distinct + field: contact_id diff --git a/models/staging/stg_salesforce__opportunities.sql b/models/staging/stg_salesforce__opportunities.sql new file mode 100644 index 00000000..e3a74a6c --- /dev/null +++ b/models/staging/stg_salesforce__opportunities.sql @@ -0,0 +1,21 @@ +-- Purpose: normalize Salesforce Opportunities for governed revenue analysis. +-- Grain: one row per Salesforce Opportunity id, including soft-deleted records. +SELECT + Id AS opportunity_id, + AccountId AS account_id, + OwnerId AS owner_id, + Name AS opportunity_name, + StageName AS stage_name, + Amount AS amount, + Probability AS probability, + CloseDate AS close_date, + Type AS opportunity_type, + LeadSource AS lead_source, + ForecastCategoryName AS forecast_category, + IsClosed AS is_closed, + IsWon AS is_won, + TIMESTAMP(CreatedDate) AS created_at, + TIMESTAMP(LastModifiedDate) AS last_modified_at, + TIMESTAMP(SystemModstamp) AS system_modified_at, + IsDeleted AS is_deleted +FROM {{ ref('raw_salesforce_opportunities') }} diff --git a/models/staging/stg_salesforce__opportunities.yml b/models/staging/stg_salesforce__opportunities.yml new file mode 100644 index 00000000..3c0e2ee3 --- /dev/null +++ b/models/staging/stg_salesforce__opportunities.yml @@ -0,0 +1,49 @@ +model: stg_salesforce__opportunities +description: Salesforce Opportunities normalized with revenue, stage, ownership, and tombstones. +owner: data-eng +materialization: incremental +dataset: staging +source_system: salesforce +sensitivity: confidential +unique_key: [opportunity_id] +incremental_cursor: system_modified_at + +columns: + - {name: opportunity_id, type: STRING, description: Stable Salesforce Opportunity identifier.} + - {name: account_id, type: STRING, description: Related Account identifier when present.} + - {name: owner_id, type: STRING, description: Current Salesforce User owner identifier.} + - {name: opportunity_name, type: STRING, description: Opportunity display name.} + - {name: stage_name, type: STRING, description: Current Salesforce sales stage.} + - {name: amount, type: NUMERIC, description: Opportunity revenue amount when present.} + - {name: probability, type: NUMERIC, description: Salesforce win probability percentage.} + - {name: close_date, type: DATE, description: Expected or actual close date.} + - {name: opportunity_type, type: STRING, description: Salesforce Opportunity type.} + - {name: lead_source, type: STRING, description: Lead source when present.} + - {name: forecast_category, type: STRING, description: Salesforce forecast category.} + - {name: is_closed, type: BOOL, description: Whether the Opportunity is closed.} + - {name: is_won, type: BOOL, description: Whether the Opportunity is won.} + - {name: created_at, type: TIMESTAMP, description: Salesforce record creation timestamp.} + - {name: last_modified_at, type: TIMESTAMP, description: User-facing modification timestamp.} + - {name: system_modified_at, type: TIMESTAMP, description: Salesforce system watermark.} + - {name: is_deleted, type: BOOL, description: Whether Salesforce soft-deleted the Opportunity.} + +tests: + - {column: opportunity_id, not_null: true, unique: true} + - {column: opportunity_name, not_null: true} + - {column: stage_name, not_null: true} + - {column: close_date, not_null: true} + - {column: account_id, relationships: {to: stg_salesforce__accounts, field: account_id}} + - {column: owner_id, relationships: {to: stg_salesforce__users, field: user_id}} + - {column: is_closed, accepted_values: [true, false]} + - {column: is_won, accepted_values: [true, false]} + - {column: is_deleted, accepted_values: [true, false]} + +metrics: + - name: source_opportunity_count + description: Number of Salesforce Opportunities represented, including tombstones. + aggregation: count_distinct + field: opportunity_id + - name: source_opportunity_amount + description: Total stated amount across Opportunities with a populated amount. + aggregation: sum + field: amount diff --git a/models/staging/stg_salesforce__users.sql b/models/staging/stg_salesforce__users.sql new file mode 100644 index 00000000..9cfd1ca6 --- /dev/null +++ b/models/staging/stg_salesforce__users.sql @@ -0,0 +1,13 @@ +-- Purpose: normalize Salesforce Users for durable historical ownership dimensions. +-- Grain: one row per Salesforce User id, including inactive users. +SELECT + Id AS user_id, + Name AS user_name, + Alias AS user_alias, + UserType AS user_type, + ProfileId AS profile_id, + IsActive AS is_active, + TIMESTAMP(CreatedDate) AS created_at, + TIMESTAMP(LastModifiedDate) AS last_modified_at, + TIMESTAMP(SystemModstamp) AS system_modified_at +FROM {{ ref('raw_salesforce_users') }} diff --git a/models/staging/stg_salesforce__users.yml b/models/staging/stg_salesforce__users.yml new file mode 100644 index 00000000..29240293 --- /dev/null +++ b/models/staging/stg_salesforce__users.yml @@ -0,0 +1,31 @@ +model: stg_salesforce__users +description: Salesforce Users normalized for owner dimensions while retaining inactive users. +owner: data-eng +materialization: incremental +dataset: staging +source_system: salesforce +sensitivity: confidential +unique_key: [user_id] +incremental_cursor: system_modified_at + +columns: + - {name: user_id, type: STRING, description: Stable Salesforce User identifier.} + - {name: user_name, type: STRING, description: Salesforce User display name.} + - {name: user_alias, type: STRING, description: Salesforce User alias.} + - {name: user_type, type: STRING, description: Salesforce User license/type category.} + - {name: profile_id, type: STRING, description: Related Salesforce Profile identifier.} + - {name: is_active, type: BOOL, description: Whether the Salesforce User can currently sign in.} + - {name: created_at, type: TIMESTAMP, description: Salesforce record creation timestamp.} + - {name: last_modified_at, type: TIMESTAMP, description: User-facing modification timestamp.} + - {name: system_modified_at, type: TIMESTAMP, description: Salesforce system watermark.} + +tests: + - {column: user_id, not_null: true, unique: true} + - {column: user_name, not_null: true} + - {column: is_active, not_null: true, accepted_values: [true, false]} + +metrics: + - name: user_count + description: Number of Salesforce Users represented, including inactive users. + aggregation: count_distinct + field: user_id diff --git a/pyproject.toml b/pyproject.toml index ac8d4eab..588116f4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -129,6 +129,18 @@ packages = ["src/dander"] "connectors/greenhouse_job_board.yaml" = "dander/templates/project/connectors/greenhouse_job_board.yaml" "models/staging/stg_greenhouse__jobs.sql" = "dander/templates/project/models/staging/stg_greenhouse__jobs.sql" "models/staging/stg_greenhouse__jobs.yml" = "dander/templates/project/models/staging/stg_greenhouse__jobs.yml" +"examples/salesforce/dander.yaml" = "dander/templates/project/examples/salesforce/dander.yaml" +"connectors/salesforce_jwt.example.yaml" = "dander/templates/project/examples/salesforce/connectors/salesforce.yaml" +"models/staging/stg_salesforce__users.sql" = "dander/templates/project/examples/salesforce/models/staging/stg_salesforce__users.sql" +"models/staging/stg_salesforce__users.yml" = "dander/templates/project/examples/salesforce/models/staging/stg_salesforce__users.yml" +"models/staging/stg_salesforce__accounts.sql" = "dander/templates/project/examples/salesforce/models/staging/stg_salesforce__accounts.sql" +"models/staging/stg_salesforce__accounts.yml" = "dander/templates/project/examples/salesforce/models/staging/stg_salesforce__accounts.yml" +"models/staging/stg_salesforce__contacts.sql" = "dander/templates/project/examples/salesforce/models/staging/stg_salesforce__contacts.sql" +"models/staging/stg_salesforce__contacts.yml" = "dander/templates/project/examples/salesforce/models/staging/stg_salesforce__contacts.yml" +"models/staging/stg_salesforce__opportunities.sql" = "dander/templates/project/examples/salesforce/models/staging/stg_salesforce__opportunities.sql" +"models/staging/stg_salesforce__opportunities.yml" = "dander/templates/project/examples/salesforce/models/staging/stg_salesforce__opportunities.yml" +"models/marts/fct_salesforce__opportunities.sql" = "dander/templates/project/examples/salesforce/models/marts/fct_salesforce__opportunities.sql" +"models/marts/fct_salesforce__opportunities.yml" = "dander/templates/project/examples/salesforce/models/marts/fct_salesforce__opportunities.yml" [tool.hatch.build.targets.sdist] include = [ @@ -146,7 +158,9 @@ include = [ "docs/known-limitations.md", "docs/netsuite-simulator.md", "docs/operator-soak.md", + "docs/rollback.md", "docs/upgrading.md", + "examples", "infra", "graphs", "models", diff --git a/scripts/check_distribution.py b/scripts/check_distribution.py index 6ef016ea..e9bccae2 100644 --- a/scripts/check_distribution.py +++ b/scripts/check_distribution.py @@ -21,10 +21,23 @@ "dander/templates/project/infra/main.tf", "dander/templates/project/models/staging/stg_greenhouse__jobs.sql", "dander/templates/project/models/staging/stg_greenhouse__jobs.yml", + "dander/templates/project/examples/salesforce/dander.yaml", + "dander/templates/project/examples/salesforce/connectors/salesforce.yaml", + "dander/templates/project/examples/salesforce/models/staging/stg_salesforce__users.sql", + "dander/templates/project/examples/salesforce/models/staging/stg_salesforce__users.yml", + "dander/templates/project/examples/salesforce/models/staging/stg_salesforce__accounts.sql", + "dander/templates/project/examples/salesforce/models/staging/stg_salesforce__accounts.yml", + "dander/templates/project/examples/salesforce/models/staging/stg_salesforce__contacts.sql", + "dander/templates/project/examples/salesforce/models/staging/stg_salesforce__contacts.yml", + "dander/templates/project/examples/salesforce/models/staging/stg_salesforce__opportunities.sql", + "dander/templates/project/examples/salesforce/models/staging/stg_salesforce__opportunities.yml", + "dander/templates/project/examples/salesforce/models/marts/fct_salesforce__opportunities.sql", + "dander/templates/project/examples/salesforce/models/marts/fct_salesforce__opportunities.yml", } SDIST_REQUIRED = { "LICENSE", "README.md", + "examples/salesforce/dander.yaml", "infra/main.tf", "pyproject.toml", "src/dander/project/scaffold.py", diff --git a/src/dander/bootstrap/__init__.py b/src/dander/bootstrap/__init__.py index 93e0dd98..1761f98e 100644 --- a/src/dander/bootstrap/__init__.py +++ b/src/dander/bootstrap/__init__.py @@ -1,6 +1,7 @@ """Infrastructure bootstrap adapters.""" from dander.bootstrap.admin import AdministrativeBootstrap, AdministrativeBootstrapError +from dander.bootstrap.permissions import require_stage_zero_permissions from dander.bootstrap.project import ( ProjectBootstrapError, RuntimeImagePublisher, @@ -34,4 +35,5 @@ "write_summary", "active_admin_member", "wait_for_service_account_impersonation", + "require_stage_zero_permissions", ] diff --git a/src/dander/bootstrap/admin.py b/src/dander/bootstrap/admin.py index 5701ce81..b0aba01e 100644 --- a/src/dander/bootstrap/admin.py +++ b/src/dander/bootstrap/admin.py @@ -135,11 +135,31 @@ def execute( self._run("terraform", "apply", str(plan_path)) return plan_path + def apply_saved_plan(self, *, state_bucket: str) -> Path: + """Apply only the previously saved stage-zero plan.""" + if not _BUCKET_NAME.fullmatch(state_bucket): + raise AdministrativeBootstrapError(f"Invalid state bucket: {state_bucket!r}") + self._prepare_operator_directories(preserve_plan=True) + if not self._plan_path.is_file() or self._plan_path.is_symlink(): + raise AdministrativeBootstrapError( + f"No reviewed stage-zero plan exists at {self._plan_path}" + ) + self._run( + "terraform", + "init", + "-reconfigure", + "-input=false", + f"-backend-config=bucket={state_bucket}", + f"-backend-config=prefix={_STAGE_ZERO_STATE_PREFIX}", + ) + self._run("terraform", "apply", str(self._plan_path)) + return self._plan_path + @staticmethod def _is_within(candidate: Path, parent: Path) -> bool: return candidate == parent or parent in candidate.parents - def _prepare_operator_directories(self) -> None: + def _prepare_operator_directories(self, *, preserve_plan: bool = False) -> None: try: if self._tf_data_dir.is_symlink(): raise AdministrativeBootstrapError("terraform-data must not be a symlink") @@ -164,7 +184,8 @@ def _prepare_operator_directories(self) -> None: self._tf_data_dir.chmod(0o700) if self._plan_path.exists(): self._plan_path.chmod(0o600) - self._plan_path.unlink() + if not preserve_plan: + self._plan_path.unlink() except OSError as error: raise AdministrativeBootstrapError( "Could not create or secure the operator artifact directory" diff --git a/src/dander/bootstrap/permissions.py b/src/dander/bootstrap/permissions.py new file mode 100644 index 00000000..b53ceffe --- /dev/null +++ b/src/dander/bootstrap/permissions.py @@ -0,0 +1,165 @@ +"""Focused stage-zero permission preflight for the authenticated GCP caller.""" + +from __future__ import annotations + +import json +import re +import subprocess +from typing import TYPE_CHECKING, Protocol + +import httpx + +from dander.bootstrap.admin import AdministrativeBootstrapError + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + +_PROJECT_ID = re.compile(r"^[a-z][a-z0-9-]{4,28}[a-z0-9]$") +_BILLING_ACCOUNT = re.compile(r"^[0-9A-F]{6}-[0-9A-F]{6}-[0-9A-F]{6}$") +_PROJECT_PERMISSIONS = ( + "artifactregistry.repositories.create", + "iam.serviceAccounts.create", + "iam.serviceAccounts.get", + "iam.serviceAccounts.getIamPolicy", + "iam.serviceAccounts.setIamPolicy", + "resourcemanager.projects.getIamPolicy", + "resourcemanager.projects.setIamPolicy", + "serviceusage.services.enable", + "serviceusage.services.get", + "storage.buckets.create", + "storage.buckets.get", + "storage.buckets.update", +) +_WIF_PERMISSIONS = ( + "iam.workloadIdentityPools.create", + "iam.workloadIdentityPools.get", + "iam.workloadIdentityPools.update", +) +_BILLING_PERMISSIONS = ( + "billing.accounts.get", + "billing.accounts.getIamPolicy", + "billing.accounts.setIamPolicy", +) + + +class _CompletedProcess(Protocol): + returncode: int + stdout: str + + +class _Runner(Protocol): + def __call__( + self, + args: tuple[str, ...], + *, + cwd: Path, + check: bool, + capture_output: bool = False, + text: bool = False, + ) -> _CompletedProcess: ... + + +def require_stage_zero_permissions( + *, + project: str, + cwd: Path, + billing_account_id: str = "", + github_repository: str = "", + runner: _Runner | None = None, + post: Callable[..., httpx.Response] = httpx.post, +) -> None: + """Fail before Terraform when the current identity lacks stage-zero permissions.""" + if not _PROJECT_ID.fullmatch(project): + raise AdministrativeBootstrapError(f"Invalid project: {project!r}") + if billing_account_id and not _BILLING_ACCOUNT.fullmatch(billing_account_id): + raise AdministrativeBootstrapError("Billing account must use XXXXXX-XXXXXX-XXXXXX format") + command_runner = runner or _subprocess_runner + try: + token_result = command_runner( + ("gcloud", "auth", "print-access-token", f"--project={project}"), + cwd=cwd.resolve(), + check=True, + capture_output=True, + text=True, + ) + except (FileNotFoundError, subprocess.CalledProcessError) as error: + raise AdministrativeBootstrapError( + "Could not obtain an access token for the stage-zero permission preflight" + ) from error + token = token_result.stdout.strip() + if not token: + raise AdministrativeBootstrapError( + "The active gcloud identity did not return an access token" + ) + + project_permissions = (*_PROJECT_PERMISSIONS, *(_WIF_PERMISSIONS if github_repository else ())) + missing_project = _missing_permissions( + f"https://cloudresourcemanager.googleapis.com/v1/projects/{project}:testIamPermissions", + project_permissions, + token=token, + post=post, + ) + missing_billing: tuple[str, ...] = () + if billing_account_id: + missing_billing = _missing_permissions( + "https://cloudbilling.googleapis.com/v1/" + f"billingAccounts/{billing_account_id}:testIamPermissions", + _BILLING_PERMISSIONS, + token=token, + post=post, + ) + if not missing_project and not missing_billing: + return + + details: list[str] = [] + if missing_project: + details.append(f"project: {', '.join(missing_project)}") + if missing_billing: + details.append(f"billing account: {', '.join(missing_billing)}") + raise AdministrativeBootstrapError( + "Stage-zero permission preflight failed; missing " + "; ".join(details) + ) + + +def _missing_permissions( + url: str, + required: tuple[str, ...], + *, + token: str, + post: Callable[..., httpx.Response], +) -> tuple[str, ...]: + try: + response = post( + url, + headers={"Authorization": f"Bearer {token}"}, + content=json.dumps({"permissions": required}), + timeout=15.0, + ) + response.raise_for_status() + payload = response.json() + except (httpx.HTTPError, ValueError) as error: + raise AdministrativeBootstrapError( + "Could not complete the read-only stage-zero permission preflight" + ) from error + granted = payload.get("permissions") if isinstance(payload, dict) else None + if not isinstance(granted, list) or not all(isinstance(item, str) for item in granted): + raise AdministrativeBootstrapError("GCP returned an invalid stage-zero permission response") + return tuple(permission for permission in required if permission not in granted) + + +def _subprocess_runner( + args: tuple[str, ...], + *, + cwd: Path, + check: bool, + capture_output: bool = False, + text: bool = False, +) -> subprocess.CompletedProcess[str]: + return subprocess.run( + args, + cwd=cwd, + check=check, + capture_output=capture_output, + text=text, + ) diff --git a/src/dander/bootstrap/project.py b/src/dander/bootstrap/project.py index 560f7b67..457b917b 100644 --- a/src/dander/bootstrap/project.py +++ b/src/dander/bootstrap/project.py @@ -32,6 +32,7 @@ def __call__( check: bool, capture_output: bool = False, text: bool = False, + input: str | None = None, ) -> subprocess.CompletedProcess[str]: """Execute one argument-vector command.""" @@ -43,6 +44,7 @@ def _subprocess_runner( check: bool, capture_output: bool = False, text: bool = False, + input: str | None = None, ) -> subprocess.CompletedProcess[str]: return subprocess.run( args, @@ -50,6 +52,7 @@ def _subprocess_runner( check=check, capture_output=capture_output, text=text, + input=input, ) @@ -149,21 +152,69 @@ def __init__(self, repository_dir: Path, *, runner: _Runner | None = None) -> No self._repository_dir = repository_dir.resolve() self._runner = runner or _subprocess_runner - def publish(self, *, project: str, region: str, tag_prefix: str = "init") -> str: + def publish( + self, + *, + project: str, + region: str, + tag_prefix: str = "init", + impersonate_service_account: str = "", + require_source_free: bool = False, + ) -> str: if not _PROJECT_ID.fullmatch(project) or not _REGION.fullmatch(region): raise ProjectBootstrapError("Invalid runtime image project or region") if not re.fullmatch(r"[a-z][a-z0-9-]{0,31}", tag_prefix): raise ProjectBootstrapError("Invalid runtime image tag prefix") + if impersonate_service_account and not _ACCOUNT.fullmatch(impersonate_service_account): + raise ProjectBootstrapError("Invalid runtime image impersonation account") + if require_source_free and (self._repository_dir / "src").exists(): + raise ProjectBootstrapError( + "image-publish requires a generated source-free project without a src directory" + ) host = f"{region}-docker.pkg.dev" repository = f"{host}/{project}/dander/dander" tag = f"{tag_prefix}-{self._content_digest()[:12]}" tagged_image = f"{repository}:{tag}" try: - self._runner( - ("gcloud", "auth", "configure-docker", host, "--quiet"), - cwd=self._repository_dir, - check=True, - ) + if impersonate_service_account: + token_result = self._runner( + ( + "gcloud", + "auth", + "print-access-token", + f"--impersonate-service-account={impersonate_service_account}", + f"--project={project}", + ), + cwd=self._repository_dir, + check=True, + capture_output=True, + text=True, + ) + token = token_result.stdout.strip() + if not token: + raise ProjectBootstrapError( + "Bootstrap impersonation returned an empty Artifact Registry token" + ) + self._runner( + ( + "docker", + "login", + host, + "--username", + "oauth2accesstoken", + "--password-stdin", + ), + cwd=self._repository_dir, + check=True, + text=True, + input=token, + ) + else: + self._runner( + ("gcloud", "auth", "configure-docker", host, "--quiet"), + cwd=self._repository_dir, + check=True, + ) self._runner( ( "docker", @@ -179,16 +230,22 @@ def publish(self, *, project: str, region: str, tag_prefix: str = "init") -> str cwd=self._repository_dir, check=True, ) + describe_command: tuple[str, ...] = ( + "gcloud", + "artifacts", + "docker", + "images", + "describe", + tagged_image, + "--format=value(image_summary.digest)", + ) + if impersonate_service_account: + describe_command = ( + *describe_command, + f"--impersonate-service-account={impersonate_service_account}", + ) described = self._runner( - ( - "gcloud", - "artifacts", - "docker", - "images", - "describe", - tagged_image, - "--format=value(image_summary.digest)", - ), + describe_command, cwd=self._repository_dir, check=True, capture_output=True, diff --git a/src/dander/cli/init_command.py b/src/dander/cli/init_command.py index 5609d42b..0534aee5 100644 --- a/src/dander/cli/init_command.py +++ b/src/dander/cli/init_command.py @@ -81,7 +81,7 @@ def execute_init( terraform_bootstrap_cls: type[TerraformBootstrap] = TerraformBootstrap, active_admin_member_fn: Callable[..., str] = active_admin_member, wait_for_impersonation_fn: Callable[..., None] = wait_for_service_account_impersonation, -) -> None: +) -> Path: """Resolve, execute, and render one ``dander init`` request.""" try: manifest = load_project_config(options.config) @@ -157,6 +157,7 @@ def execute_init( ).publish( project=options.project, region=platform.region, + impersonate_service_account=bootstrap_account, ) except ProjectBootstrapError as error: raise ClickException(str(error)) from error @@ -200,6 +201,7 @@ def execute_init( action = "applied" if options.apply else "planned" console.print(f"[green]Bootstrap {action}.[/green] Saved plan: {plan_path}") + return plan_path def _bootstrap_administration( diff --git a/src/dander/cli/main.py b/src/dander/cli/main.py index b3248870..7885285e 100644 --- a/src/dander/cli/main.py +++ b/src/dander/cli/main.py @@ -4,6 +4,7 @@ import os import re +import shlex import shutil as shutil import subprocess as subprocess import sys @@ -21,11 +22,13 @@ AdministrativeBootstrapError, DeploymentSummary, DeploymentVerifier, + ProjectBootstrapError, RuntimeImagePublisher, StateBucketBootstrap, TerraformBootstrap, TerraformBootstrapError, active_admin_member, + require_stage_zero_permissions, wait_for_service_account_impersonation, write_summary, ) @@ -698,6 +701,12 @@ def init_admin_plan( ) -> None: """Plan stage-zero state bucket and bootstrap identity resources.""" try: + require_stage_zero_permissions( + project=project, + cwd=infra_dir.resolve().parent.parent, + billing_account_id=billing_account_id, + github_repository=github_repository, + ) plan_path = AdministrativeBootstrap(infra_dir, operator_artifact_dir).execute( project=project, state_bucket=state_bucket, @@ -709,10 +718,38 @@ def init_admin_plan( billing_account_id=billing_account_id, github_repository=github_repository, github_ref=github_ref, + adopt_state_bucket=True, ) except AdministrativeBootstrapError as error: raise ClickException(str(error)) from error console.print(f"[green]Administrative bootstrap planned.[/green] Saved plan: {plan_path}") + console.print( + "Review: " + + shlex.join(("terraform", f"-chdir={infra_dir}", "show", "-no-color", str(plan_path))) + ) + apply_command = [ + "dander", + "init-admin-apply", + "--project", + project, + "--state-bucket", + state_bucket, + "--admin-member", + admin_member, + "--region", + region, + "--state-location", + state_location, + "--bootstrap-service-account-id", + bootstrap_service_account_id, + "--operator-artifact-dir", + str(operator_artifact_dir), + ] + if billing_account_id: + apply_command.extend(("--billing-account", billing_account_id)) + if github_repository: + apply_command.extend(("--github-repository", github_repository, "--github-ref", github_ref)) + console.print("Next after review: " + shlex.join(apply_command)) @app.command("init-admin-apply") @@ -743,64 +780,167 @@ def init_admin_apply( ): raise typer.Abort() try: - plan_path = AdministrativeBootstrap(infra_dir, operator_artifact_dir).execute( - project=project, + plan_path = AdministrativeBootstrap(infra_dir, operator_artifact_dir).apply_saved_plan( state_bucket=state_bucket, - admin_member=admin_member, - apply=True, - region=region, - state_location=state_location, - bootstrap_service_account_id=bootstrap_service_account_id, - billing_account_id=billing_account_id, - github_repository=github_repository, - github_ref=github_ref, ) except AdministrativeBootstrapError as error: raise ClickException(str(error)) from error console.print(f"[green]Administrative bootstrap applied.[/green] Saved plan: {plan_path}") +@app.command("image-publish") +def image_publish( + project: str = typer.Option(..., "--project", help="GCP project id."), + failure_alert_email: str = typer.Option( + ..., + "--failure-alert-email", + help="Failure-alert address forwarded to the next platform-plan command.", + ), + state_bucket: str = typer.Option( + "", "--state-bucket", help="Defaults to -dander-state." + ), + bootstrap_service_account: str = typer.Option( + "", + "--bootstrap-service-account", + help="Bootstrap identity used for Artifact Registry publication.", + ), + region: str | None = typer.Option( + None, "--region", help="Override platform.region from dander.yaml." + ), + billing_account_id: str = typer.Option("", "--billing-account"), + config: Path = typer.Option(_DEFAULT_PROJECT_CONFIG, "--config"), # noqa: B008 + tag_prefix: str = typer.Option("runtime", "--tag-prefix"), +) -> None: + """Build and publish the source-free runtime image through the bootstrap identity.""" + try: + manifest = load_project_config(config) + manifest.validate_references(config.resolve().parent) + except ProjectConfigError as error: + raise ClickException(str(error)) from error + resolved_region = region or manifest.platform.region + resolved_bucket = state_bucket or f"{project}-dander-state" + bootstrap = bootstrap_service_account or (f"dander-bootstrap@{project}.iam.gserviceaccount.com") + if not typer.confirm( + f"Build and push a source-free runtime image to GCP project {project!r}?", + default=False, + ): + raise typer.Abort() + try: + image = RuntimeImagePublisher(config.resolve().parent).publish( + project=project, + region=resolved_region, + tag_prefix=tag_prefix, + impersonate_service_account=bootstrap, + require_source_free=True, + ) + except ProjectBootstrapError as error: + raise ClickException(str(error)) from error + console.print(f"[green]Published immutable runtime image:[/green] {image}") + next_command = [ + "dander", + "init-platform-plan", + "--project", + project, + "--state-bucket", + resolved_bucket, + "--bootstrap-service-account", + bootstrap, + "--container-image", + image, + "--failure-alert-email", + failure_alert_email, + "--config", + str(config), + ] + if billing_account_id: + next_command.extend(("--billing-account", billing_account_id)) + if region is not None: + next_command.extend(("--region", region)) + console.print("Next: " + shlex.join(next_command)) + + @app.command("init-platform-plan") def init_platform_plan( project: str = typer.Option(..., "--project", help="GCP project id."), state_bucket: str = typer.Option(..., "--state-bucket", help="Existing remote-state bucket."), bootstrap_service_account: str = typer.Option(..., "--bootstrap-service-account"), state_prefix: str = typer.Option("dander/state", "--state-prefix"), - region: str = typer.Option("us-central1", "--region"), - bigquery_location: str = typer.Option("US", "--bigquery-location"), + container_image: str = typer.Option( + ..., "--container-image", help="Immutable runtime image ending in @sha256 digest." + ), + failure_alert_email: str = typer.Option(..., "--failure-alert-email"), + config: Path = typer.Option(_DEFAULT_PROJECT_CONFIG, "--config"), # noqa: B008 + region: str | None = typer.Option(None, "--region"), + bigquery_location: str | None = typer.Option(None, "--bigquery-location"), + billing_account_id: str = typer.Option("", "--billing-account"), + druff_container_image: str = typer.Option("", "--druff-container-image"), + secret_ids: list[str] | None = typer.Option(None, "--secret-id"), # noqa: B008 + github_repository: str = typer.Option("", "--github-repository"), + github_ref: str = typer.Option("refs/heads/main", "--github-ref"), + enable_cost_guard: bool | None = typer.Option(None, "--enable-cost-guard/--no-cost-guard"), + cost_guard_budget_name: str = typer.Option("dander-sbx-cap", "--cost-guard-budget-name"), + cost_guard_budget_amount: str = typer.Option("5.00", "--cost-guard-budget-amount"), + live_cost_guard: bool = typer.Option(False, "--live-cost-guard"), infra_dir: Path = typer.Option(_DEFAULT_INFRA_DIR, hidden=True), # noqa: B008 ) -> None: - """Plan platform Terraform while impersonating the stage-zero identity.""" - plan_path = _execute_platform_bootstrap( - project=project, - state_bucket=state_bucket, - state_prefix=state_prefix, - bootstrap_service_account=bootstrap_service_account, - apply=False, - region=region, - bigquery_location=bigquery_location, - runtime_cpu=1, - runtime_memory="512Mi", - runtime_timeout_seconds=300, - runtime_max_retries=1, - runtime_batch_rows=10_000, - require_guarded_free_tier=False, - enable_runtime=False, - billing_account_id="", - container_image="", - druff_container_image="", - pipelines={}, - failure_alert_email="", - secret_ids=(), - github_repository="", - github_ref="refs/heads/main", - enable_cost_guard=False, - cost_guard_budget_name="dander-sbx-cap", - cost_guard_budget_amount="5.00", - live_cost_guard=False, - infra_dir=infra_dir, + """Plan the complete manifest-defined platform through the bootstrap identity.""" + plan_path = execute_init( + InitOptions( + project=project, + state_bucket=state_bucket, + state_prefix=state_prefix, + bootstrap_service_account=bootstrap_service_account, + admin_member="", + operator_artifact_dir=None, + state_location="US", + region=region, + bigquery_location=bigquery_location, + runtime_cpu=None, + runtime_memory=None, + runtime_timeout_seconds=None, + runtime_max_retries=None, + runtime_batch_rows=None, + require_guarded_free_tier=None, + enable_runtime=True, + billing_account_id=billing_account_id, + container_image=container_image, + druff_container_image=druff_container_image, + config=config, + secret_ids=tuple(secret_ids or ()), + github_repository=github_repository, + github_ref=github_ref, + failure_alert_email=failure_alert_email, + enable_cost_guard=enable_cost_guard, + cost_guard_budget_name=cost_guard_budget_name, + cost_guard_budget_amount=cost_guard_budget_amount, + live_cost_guard=live_cost_guard, + apply=False, + infra_dir=infra_dir, + ), + console=console, + terraform_bootstrap_cls=TerraformBootstrap, + ) + console.print( + "Review: " + + shlex.join(("terraform", f"-chdir={infra_dir}", "show", "-no-color", str(plan_path))) + ) + console.print( + "Next after review: " + + shlex.join( + ( + "dander", + "init-platform-apply", + "--project", + project, + "--state-bucket", + state_bucket, + "--bootstrap-service-account", + bootstrap_service_account, + "--state-prefix", + state_prefix, + ) + ) ) - console.print(f"[green]Platform bootstrap planned.[/green] Saved plan: {plan_path}") @app.command("init-platform-apply") @@ -1399,10 +1539,11 @@ def metadata_runs( local=local, state_path=state_path, ) + records = history.recent(limit=limit, pipeline_id=pipeline) table = Table(title="Dander pipeline runs") for column in ("Run", "Pipeline", "Status", "Stage", "Rows", "Models", "Tests", "Assets"): table.add_column(column) - for record in history.recent(limit=limit, pipeline_id=pipeline): + for record in records: table.add_row( record.run_id, record.pipeline_id, @@ -1414,6 +1555,13 @@ def metadata_runs( str(record.assets), ) console.print(table) + failures = [record for record in records if record.failure_code or record.failure_summary] + if failures: + console.print("[bold red]Failures[/bold red]") + for record in failures: + code = record.failure_code or "failed" + summary = record.failure_summary or "Inspect the run logs for details." + console.print(f"[red]{record.run_id} · {code}[/red]: {summary}") def _metadata_snapshots( diff --git a/src/dander/executor.py b/src/dander/executor.py index 1902a2d9..afe073d4 100644 --- a/src/dander/executor.py +++ b/src/dander/executor.py @@ -9,7 +9,7 @@ from dander.catalog import MetadataSpine, SemanticRegistryPublisher from dander.runtime import PipelineRunResult -from dander.state import LeaseHeartbeat, RunStage, RunStatus +from dander.state import LeaseHeartbeat, RunStage, RunStatus, classify_failure from dander.transform import TransformProject, TransformRunResult if TYPE_CHECKING: @@ -135,6 +135,10 @@ def execute(self) -> PipelineExecutionResult: ) heartbeat = LeaseHeartbeat(self._leases, lease) heartbeat.__enter__() + self._history.reconcile_interrupted( + self._pipeline_id, + current_run_id=run_id, + ) ingestion_result = self._ingestion.run( run_id=run_id, ownership=heartbeat, @@ -221,7 +225,8 @@ def execute(self) -> PipelineExecutionResult: assertions=assertions, assets=assets, ) - except Exception: + except Exception as error: + failure = classify_failure(error, stage=stage, run_id=run_id) try: self._history.finish( run_id, @@ -233,6 +238,8 @@ def execute(self) -> PipelineExecutionResult: assertions=assertions, assets=assets, failure_stage=stage, + failure_code=failure.code, + failure_summary=failure.summary, ) except Exception: _LOGGER.exception( diff --git a/src/dander/pipeline/graph_operations.py b/src/dander/pipeline/graph_operations.py index d624943e..5c48c84d 100644 --- a/src/dander/pipeline/graph_operations.py +++ b/src/dander/pipeline/graph_operations.py @@ -448,4 +448,6 @@ def _run_record(record: RunRecord) -> dict[str, object]: "assertions": record.assertions, "assets": record.assets, "failure_stage": record.failure_stage.value if record.failure_stage is not None else None, + "failure_code": record.failure_code, + "failure_summary": record.failure_summary, } diff --git a/src/dander/project/scaffold.py b/src/dander/project/scaffold.py index f1cb1ce0..650e437f 100644 --- a/src/dander/project/scaffold.py +++ b/src/dander/project/scaffold.py @@ -14,10 +14,35 @@ from collections.abc import Iterable _SOURCE_ASSETS = ( - Path("connectors/greenhouse_job_board.yaml"), - Path("graphs/greenhouse_jobs.yaml"), - Path("models/staging/stg_greenhouse__jobs.sql"), - Path("models/staging/stg_greenhouse__jobs.yml"), + (Path("connectors/greenhouse_job_board.yaml"), Path("connectors/greenhouse_job_board.yaml")), + (Path("graphs/greenhouse_jobs.yaml"), Path("graphs/greenhouse_jobs.yaml")), + ( + Path("models/staging/stg_greenhouse__jobs.sql"), + Path("models/staging/stg_greenhouse__jobs.sql"), + ), + ( + Path("models/staging/stg_greenhouse__jobs.yml"), + Path("models/staging/stg_greenhouse__jobs.yml"), + ), + (Path("examples/salesforce/dander.yaml"), Path("examples/salesforce/dander.yaml")), + ( + Path("connectors/salesforce_jwt.example.yaml"), + Path("examples/salesforce/connectors/salesforce.yaml"), + ), + *( + ( + Path(f"models/{folder}/{name}.{suffix}"), + Path(f"examples/salesforce/models/{folder}/{name}.{suffix}"), + ) + for folder, name in ( + ("staging", "stg_salesforce__users"), + ("staging", "stg_salesforce__accounts"), + ("staging", "stg_salesforce__contacts"), + ("staging", "stg_salesforce__opportunities"), + ("marts", "fct_salesforce__opportunities"), + ) + for suffix in ("sql", "yml") + ), ) @@ -41,16 +66,16 @@ def scaffold_project(destination: Path) -> Path: with resources.as_file(template) as template_path: shutil.copytree(template_path, staging) source_root = _source_checkout_root() - for relative_path in _SOURCE_ASSETS: - packaged = staging / relative_path + for source_path, destination_path in _SOURCE_ASSETS: + packaged = staging / destination_path if packaged.is_file(): continue - if source_root is None or not (source_root / relative_path).is_file(): + if source_root is None or not (source_root / source_path).is_file(): raise ProjectScaffoldError( - f"Installed package is missing starter asset: {relative_path}" + f"Installed package is missing starter asset: {destination_path}" ) packaged.parent.mkdir(parents=True, exist_ok=True) - shutil.copy2(source_root / relative_path, packaged) + shutil.copy2(source_root / source_path, packaged) _render_distribution_version(staging / "Dockerfile") if not (staging / "infra" / "main.tf").is_file(): if source_root is None: diff --git a/src/dander/state/__init__.py b/src/dander/state/__init__.py index 87404a6c..b525a355 100644 --- a/src/dander/state/__init__.py +++ b/src/dander/state/__init__.py @@ -1,5 +1,6 @@ """State module: watermark / control tracking for idempotent restarts.""" +from dander.state.failure import FailureDetails, classify_failure from dander.state.lease import ( BigQueryLeaseStore, LeaseHandle, @@ -26,6 +27,7 @@ "LeaseHeartbeat", "LeaseLostError", "LeaseStore", + "FailureDetails", "RunHistoryStore", "RunRecord", "RunStage", @@ -34,4 +36,5 @@ "SqliteLeaseStore", "SqliteWatermarkStore", "WatermarkStore", + "classify_failure", ] diff --git a/src/dander/state/failure.py b/src/dander/state/failure.py new file mode 100644 index 00000000..9be350bb --- /dev/null +++ b/src/dander/state/failure.py @@ -0,0 +1,113 @@ +"""Sanitized operator-facing failure summaries for the durable run ledger.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from dander.state.run_history import RunStage + +_MAX_SUMMARY_LENGTH = 512 + + +@dataclass(frozen=True) +class FailureDetails: + """A stable code and non-sensitive message safe to persist.""" + + code: str + summary: str + + +def classify_failure(error: Exception, *, stage: RunStage, run_id: str) -> FailureDetails: + """Classify an exception without copying unrestricted exception text into history.""" + chain = _exception_chain(error) + names = {item.__class__.__name__ for item in chain} + statuses = {status for item in chain if (status := _status_code(item)) is not None} + + if 401 in statuses or names & {"Unauthenticated", "Unauthorized"}: + return _details( + "authentication_failed", + "Authentication failed. Verify the configured secret and provider credentials.", + ) + if 403 in statuses or names & {"Forbidden", "PermissionDenied"}: + return _details( + "permission_denied", + "Permission was denied. Verify the provider and GCP permissions for this pipeline.", + ) + if 429 in statuses or names & {"TooManyRequests", "ResourceExhausted"}: + return _details( + "rate_limited", + "The source throttled this run after bounded retries. Retry after its limit resets.", + ) + if names & {"LeaseLostError", "WatermarkConflictError"}: + return _details( + "lease_failed", + "Pipeline ownership or cursor fencing was lost. A fresh run can retry safely.", + ) + if names & {"RawSchemaError", "CursorValueError"}: + return _details( + "source_schema_failed", + f"A source record did not match the declared schema. Inspect logs for run {run_id}.", + ) + if "BigQueryWriteError" in names: + return _details( + "destination_write_failed", + f"BigQuery rejected a destination write. Inspect logs for run {run_id}.", + ) + if stage.value == "transform": + code = ( + "test_failed" if _has_known_phrase(chain, "data tests failed") else "transform_failed" + ) + action = "Data tests failed" if code == "test_failed" else "A transform failed" + return _details(code, f"{action}. Inspect logs for run {run_id}.") + if stage.value == "metadata": + return _details( + "catalog_failed", + f"Metadata or catalog publication failed. Inspect logs for run {run_id}.", + ) + if "EnterpriseSourceError" in names: + if _has_known_phrase(chain, "authentication failed"): + return _details( + "authentication_failed", + "Authentication failed. Verify the configured secret and provider credentials.", + ) + if _has_known_phrase(chain, "permission denied"): + return _details( + "permission_denied", + "Permission was denied. Verify the provider permissions for this pipeline.", + ) + return _details( + "extraction_failed", + f"Source extraction failed after bounded retries. Inspect logs for run {run_id}.", + ) + return _details( + "unexpected_error", + f"The {stage.value} stage failed unexpectedly. Inspect Cloud Run logs for run {run_id}.", + ) + + +def _details(code: str, summary: str) -> FailureDetails: + cleaned = " ".join(summary.split())[:_MAX_SUMMARY_LENGTH] + return FailureDetails(code=code, summary=cleaned) + + +def _exception_chain(error: BaseException) -> tuple[BaseException, ...]: + chain: list[BaseException] = [] + current: BaseException | None = error + seen: set[int] = set() + while current is not None and id(current) not in seen and len(chain) < 8: + chain.append(current) + seen.add(id(current)) + current = current.__cause__ or current.__context__ + return tuple(chain) + + +def _status_code(error: BaseException) -> int | None: + response = getattr(error, "response", None) + value = getattr(response, "status_code", None) + return value if isinstance(value, int) else None + + +def _has_known_phrase(chain: tuple[BaseException, ...], phrase: str) -> bool: + return any(phrase in str(error).lower() for error in chain) diff --git a/src/dander/state/run_history.py b/src/dander/state/run_history.py index 371cb55c..038019b2 100644 --- a/src/dander/state/run_history.py +++ b/src/dander/state/run_history.py @@ -52,10 +52,12 @@ class RunRecord: assertions: int assets: int failure_stage: RunStage | None + failure_code: str | None = None + failure_summary: str | None = None class RunHistoryStore(ABC): - """Persist lifecycle summaries without row values, cursors, or error messages.""" + """Persist lifecycle summaries without row values, cursors, or raw exceptions.""" @abstractmethod def start(self, run_id: str, source: str, *, pipeline_id: str | None = None) -> None: @@ -89,9 +91,15 @@ def finish( assertions: int = 0, assets: int = 0, failure_stage: RunStage | None = None, + failure_code: str | None = None, + failure_summary: str | None = None, ) -> None: """Record a terminal aggregate for a run.""" + def reconcile_interrupted(self, pipeline_id: str, *, current_run_id: str) -> None: + """Mark older active rows failed after this runner has acquired the lease.""" + return None + def recent( self, *, @@ -197,6 +205,8 @@ def finish( assertions: int = 0, assets: int = 0, failure_stage: RunStage | None = None, + failure_code: str | None = None, + failure_summary: str | None = None, ) -> None: self._update( run_id, @@ -213,6 +223,8 @@ def finish( assertions=assertions, assets=assets, failure_stage=failure_stage, + failure_code=failure_code, + failure_summary=failure_summary, ) def _update( @@ -228,6 +240,8 @@ def _update( assertions: int, assets: int, failure_stage: RunStage | None, + failure_code: str | None = None, + failure_summary: str | None = None, ) -> None: self._ensure_table() config = bigquery.QueryJobConfig( @@ -248,6 +262,8 @@ def _update( "STRING", failure_stage.value if failure_stage is not None else None, ), + bigquery.ScalarQueryParameter("failure_code", "STRING", failure_code), + bigquery.ScalarQueryParameter("failure_summary", "STRING", failure_summary), ] ) finished_at = "CURRENT_TIMESTAMP()" if status is not None else "finished_at" @@ -255,7 +271,8 @@ def _update( f"UPDATE `{self._table}` SET status = @status, stage = @stage, " f"finished_at = {finished_at}, endpoints = @endpoints, extracted = @extracted, " "affected = @affected, models = @models, assertions = @assertions, " - "assets = @assets, failure_stage = @failure_stage " + "assets = @assets, failure_stage = @failure_stage, failure_code = @failure_code, " + "failure_summary = @failure_summary " "WHERE run_id = @run_id", job_config=config, ).result() @@ -269,7 +286,8 @@ def _ensure_table(self) -> None: "status STRING NOT NULL, stage STRING, started_at TIMESTAMP NOT NULL, " "finished_at TIMESTAMP, endpoints INT64 NOT NULL, extracted INT64 NOT NULL, " "affected INT64 NOT NULL, models INT64, assertions INT64, assets INT64, " - "failure_stage STRING) CLUSTER BY pipeline_id, status" + "failure_stage STRING, failure_code STRING, failure_summary STRING) " + "CLUSTER BY pipeline_id, status" ).result() for column, data_type in ( ("pipeline_id", "STRING"), @@ -278,6 +296,8 @@ def _ensure_table(self) -> None: ("assertions", "INT64"), ("assets", "INT64"), ("failure_stage", "STRING"), + ("failure_code", "STRING"), + ("failure_summary", "STRING"), ): self._client.query( f"ALTER TABLE `{self._table}` ADD COLUMN IF NOT EXISTS {column} {data_type}" @@ -305,12 +325,32 @@ def recent( "'complete', 'ingest')) AS stage, started_at, finished_at, endpoints, " "extracted, affected, COALESCE(models, 0) AS models, " "COALESCE(assertions, 0) AS assertions, COALESCE(assets, 0) AS assets, " - f"failure_stage FROM `{self._table}`{where} " + "failure_stage, " + "JSON_VALUE(TO_JSON_STRING(history_row), '$.failure_code') AS failure_code, " + "JSON_VALUE(TO_JSON_STRING(history_row), '$.failure_summary') AS failure_summary " + f"FROM `{self._table}` AS history_row{where} " "ORDER BY started_at DESC LIMIT @limit", job_config=bigquery.QueryJobConfig(query_parameters=parameters), ).result() return tuple(_bigquery_run_record(cast("_Row", row)) for row in rows) + def reconcile_interrupted(self, pipeline_id: str, *, current_run_id: str) -> None: + self._ensure_table() + config = bigquery.QueryJobConfig( + query_parameters=[ + bigquery.ScalarQueryParameter("pipeline_id", "STRING", pipeline_id), + bigquery.ScalarQueryParameter("current_run_id", "STRING", current_run_id), + ] + ) + self._client.query( + f"UPDATE `{self._table}` SET status = 'failed', finished_at = CURRENT_TIMESTAMP(), " + "failure_stage = COALESCE(stage, 'ingest'), failure_code = 'interrupted_run', " + "failure_summary = 'A later execution acquired the expired pipeline lease.' " + "WHERE status = 'running' AND COALESCE(pipeline_id, source_name) = @pipeline_id " + "AND run_id != @current_run_id", + job_config=config, + ).result() + class SqliteRunHistoryStore(RunHistoryStore): """Persist sandbox run summaries beside local watermark state.""" @@ -326,7 +366,8 @@ def __init__(self, path: Path) -> None: "endpoints INTEGER NOT NULL DEFAULT 0, extracted INTEGER NOT NULL DEFAULT 0, " "affected INTEGER NOT NULL DEFAULT 0, pipeline_id TEXT, stage TEXT, " "models INTEGER NOT NULL DEFAULT 0, assertions INTEGER NOT NULL DEFAULT 0, " - "assets INTEGER NOT NULL DEFAULT 0, failure_stage TEXT)" + "assets INTEGER NOT NULL DEFAULT 0, failure_stage TEXT, failure_code TEXT, " + "failure_summary TEXT)" ) existing = { str(row[1]) for row in connection.execute("PRAGMA table_info(runs)").fetchall() @@ -338,6 +379,8 @@ def __init__(self, path: Path) -> None: ("assertions", "INTEGER NOT NULL DEFAULT 0"), ("assets", "INTEGER NOT NULL DEFAULT 0"), ("failure_stage", "TEXT"), + ("failure_code", "TEXT"), + ("failure_summary", "TEXT"), ): if column not in existing: connection.execute(f"ALTER TABLE runs ADD COLUMN {column} {declaration}") @@ -390,6 +433,8 @@ def finish( assertions: int = 0, assets: int = 0, failure_stage: RunStage | None = None, + failure_code: str | None = None, + failure_summary: str | None = None, ) -> None: stage = ( RunStage.COMPLETE @@ -400,7 +445,8 @@ def finish( connection.execute( "UPDATE runs SET status = ?, stage = ?, finished_at = CURRENT_TIMESTAMP, " "endpoints = ?, extracted = ?, affected = ?, models = ?, assertions = ?, " - "assets = ?, failure_stage = ? WHERE run_id = ?", + "assets = ?, failure_stage = ?, failure_code = ?, failure_summary = ? " + "WHERE run_id = ?", ( status.value, (stage or RunStage.INGEST).value, @@ -411,6 +457,8 @@ def finish( assertions, assets, failure_stage.value if failure_stage is not None else None, + failure_code, + failure_summary, run_id, ), ) @@ -427,7 +475,8 @@ def recent( "SELECT run_id, COALESCE(pipeline_id, source), source, status, " "COALESCE(stage, CASE WHEN status = 'succeeded' THEN 'complete' ELSE 'ingest' END), " "started_at, finished_at, endpoints, extracted, affected, " - "COALESCE(models, 0), COALESCE(assertions, 0), COALESCE(assets, 0), failure_stage " + "COALESCE(models, 0), COALESCE(assertions, 0), COALESCE(assets, 0), failure_stage, " + "failure_code, failure_summary " "FROM runs" ) parameters: tuple[object, ...] = () @@ -440,6 +489,16 @@ def recent( rows = connection.execute(query, parameters).fetchall() return tuple(_sqlite_run_record(row) for row in rows) + def reconcile_interrupted(self, pipeline_id: str, *, current_run_id: str) -> None: + with sqlite3.connect(self._path) as connection: + connection.execute( + "UPDATE runs SET status = 'failed', finished_at = CURRENT_TIMESTAMP, " + "failure_stage = COALESCE(stage, 'ingest'), failure_code = 'interrupted_run', " + "failure_summary = 'A later execution acquired the expired pipeline lease.' " + "WHERE status = 'running' AND COALESCE(pipeline_id, source) = ? AND run_id != ?", + (pipeline_id, current_run_id), + ) + def _sqlite_run_record(row: tuple[object, ...]) -> RunRecord: return RunRecord( @@ -457,6 +516,8 @@ def _sqlite_run_record(row: tuple[object, ...]) -> RunRecord: assertions=int(cast("int", row[11])), assets=int(cast("int", row[12])), failure_stage=RunStage(str(row[13])) if row[13] is not None else None, + failure_code=str(row[14]) if row[14] is not None else None, + failure_summary=str(row[15]) if row[15] is not None else None, ) @@ -478,4 +539,8 @@ def _bigquery_run_record(row: _Row) -> RunRecord: assertions=int(cast("int", row["assertions"])), assets=int(cast("int", row["assets"])), failure_stage=RunStage(str(failure_stage)) if failure_stage is not None else None, + failure_code=(str(row["failure_code"]) if row["failure_code"] is not None else None), + failure_summary=( + str(row["failure_summary"]) if row["failure_summary"] is not None else None + ), ) diff --git a/src/dander/templates/project/README.md b/src/dander/templates/project/README.md index 6e7cec8e..f2e94472 100644 --- a/src/dander/templates/project/README.md +++ b/src/dander/templates/project/README.md @@ -10,10 +10,11 @@ Job Board pipeline so infrastructure changes remain reviewable before any schedu ```bash dander validate dander run greenhouse_jobs --dry-run --project YOUR_GCP_PROJECT -dander init --project YOUR_GCP_PROJECT --container-image REGION-docker.pkg.dev/PROJECT/dander/dander@sha256:DIGEST +dander image-publish --project YOUR_GCP_PROJECT --failure-alert-email OPERATOR_EMAIL ``` -`dander init` plans by default. Review the saved Terraform plan before using `--apply`. The starter +`image-publish` prints the complete `init-platform-plan` command. Review that saved Terraform plan, +then apply it with `init-platform-apply`. The starter manifest keeps its scheduler paused and does not enable Dander's optional managed cost guard. Disabling the guard does not prevent or cap cloud charges. @@ -21,6 +22,13 @@ Use the public [hosted Greenhouse quickstart](https://github.com/harrisonoconnor for the complete installation, provisioning, manual-run, and schedule-enablement sequence. Follow the [upgrade guide](https://github.com/harrisonoconnorhover/dander/blob/main/docs/upgrading.md) before changing the pinned `DANDER_VERSION` in this project's Dockerfile. +Use the public [rollback guide](https://github.com/harrisonoconnorhover/dander/blob/main/docs/rollback.md) +to restore a known-good commit and immutable image without editing Terraform state. + +`examples/salesforce/` contains a source-free four-object CRM project overlay with its exact plugin +pin, connector schema, four staging models, and Opportunity fact. Follow the public +[Salesforce guide](https://github.com/harrisonoconnorhover/dander/blob/main/docs/salesforce.md) +before copying it; Contact Email and Phone are personal data enabled by default. To host Druff's compiled interface with this platform, also pass its immutable image as `--druff-container-image`. Repeat that input on later full-platform plans to retain the service. diff --git a/src/dander/writer/bigquery.py b/src/dander/writer/bigquery.py index 8c416e78..b94a0225 100644 --- a/src/dander/writer/bigquery.py +++ b/src/dander/writer/bigquery.py @@ -4,7 +4,7 @@ import re from collections.abc import Mapping -from datetime import date, datetime, time +from datetime import UTC, date, datetime, time, timedelta from decimal import Decimal from itertools import islice from typing import TYPE_CHECKING, Any, Protocol, cast @@ -24,6 +24,7 @@ from dander.concurrency import FencingToken _IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") +_STAGING_TTL = timedelta(days=1) class BigQueryWriteError(ValueError): @@ -262,10 +263,10 @@ def write(self, records: Iterable[Mapping[str, Any]], target: WriteTarget) -> in if row[self._snapshot_field] is None: raise BigQueryWriteError(f"Record {index} has a null snapshot value") - _validate_declared_schema(target, self._schema_evolution) + declared = _validate_declared_schema(target, self._schema_evolution) staging_id = _staging_id(target) try: - self._load(rows, staging_id) + self._load(rows, staging_id, schema=declared) self._client.query( _create_snapshot_target_sql( target_id, @@ -298,12 +299,20 @@ def write(self, records: Iterable[Mapping[str, Any]], target: WriteTarget) -> in finally: self._client.delete_table(staging_id, not_found_ok=True) - def _load(self, rows: Sequence[Mapping[str, Any]], staging_id: str) -> None: + def _load( + self, + rows: Sequence[Mapping[str, Any]], + staging_id: str, + *, + schema: Sequence[WriteField], + ) -> None: _load_rows_in_chunks( self._client, rows, staging_id, max_batch_rows=self._max_batch_rows, + expire=True, + schema=schema, ) @@ -340,7 +349,7 @@ def write(self, records: Iterable[Mapping[str, Any]], target: WriteTarget) -> in if collisions: raise BigQueryWriteError(f"SCD2 input uses reserved column {collisions[0]!r}") staged_rows = _deduplicate_keyed(rows, columns, target.business_key) - _validate_declared_schema(target, self._schema_evolution) + declared = _validate_declared_schema(target, self._schema_evolution) staging_id = _staging_id(target) try: @@ -349,6 +358,8 @@ def write(self, records: Iterable[Mapping[str, Any]], target: WriteTarget) -> in staged_rows, staging_id, max_batch_rows=self._max_batch_rows, + expire=True, + schema=declared, ) self._client.query(_create_scd2_target_sql(target_id, staging_id, columns)).result() _apply_schema_evolution( @@ -419,8 +430,12 @@ def write(self, records: Iterable[Mapping[str, Any]], target: WriteTarget) -> in return 0 staging_id = _staging_id(target) try: - _create_declared_target(self._client, staging_id, declared) - self._client.query(_expire_staging_sql(staging_id)).result() + _create_declared_target( + self._client, + staging_id, + declared, + expires=_staging_expiration(), + ) self._client.copy_table( staging_id, target_id, @@ -440,6 +455,13 @@ def write(self, records: Iterable[Mapping[str, Any]], target: WriteTarget) -> in staging_id = _staging_id(target) written = 0 try: + if declared: + _create_declared_target( + self._client, + staging_id, + declared, + expires=_staging_expiration(), + ) batch = first_batch first = True while batch: @@ -459,7 +481,8 @@ def write(self, records: Iterable[Mapping[str, Any]], target: WriteTarget) -> in job_config=_load_config(disposition, declared), ).result() if first: - self._client.query(_expire_staging_sql(staging_id)).result() + if not declared: + self._client.query(_expire_staging_sql(staging_id)).result() first = False written += len(batch) batch = [dict(record) for record in islice(record_iterator, self._max_batch_rows)] @@ -633,10 +656,13 @@ def _create_declared_target( client: _BigQueryClient, target_id: str, schema: Sequence[WriteField], + *, + expires: datetime | None = None, ) -> None: if not schema: return table = bigquery.Table(target_id, schema=_bigquery_schema(schema)) + table.expires = expires created = client.create_table(table, exists_ok=True) _assert_deployed_schema(schema, created.schema) @@ -767,6 +793,13 @@ def _load_rows_in_chunks( schema: Sequence[WriteField] = (), ) -> None: """Bound each load request while preserving one logical truncate-then-append batch.""" + if expire and schema: + _create_declared_target( + client, + destination, + schema, + expires=_staging_expiration(), + ) for offset in range(0, len(rows), max_batch_rows): disposition = ( bigquery.WriteDisposition.WRITE_TRUNCATE @@ -776,9 +809,15 @@ def _load_rows_in_chunks( client.load_table_from_json( [_json_load_row(row) for row in rows[offset : offset + max_batch_rows]], destination, - job_config=_load_config(disposition, schema), + job_config=_load_config( + disposition, + schema, + ), ).result() - if offset == 0 and expire: + if offset == 0 and expire and not schema: + # Legacy schema-inferred writes cannot be precreated safely. Expire them + # immediately after the first successful load without using unsupported + # load-job properties. client.query(_expire_staging_sql(destination)).result() @@ -799,6 +838,10 @@ def _json_load_value(value: object) -> object: return value +def _staging_expiration() -> datetime: + return datetime.now(UTC) + _STAGING_TTL + + def _expire_staging_sql(staging_id: str) -> str: return ( f"ALTER TABLE `{staging_id}` SET OPTIONS " diff --git a/src/dander/writer/storage_write.py b/src/dander/writer/storage_write.py index f044b2f0..d761bf94 100644 --- a/src/dander/writer/storage_write.py +++ b/src/dander/writer/storage_write.py @@ -172,7 +172,10 @@ def write(self, records: Iterable[Mapping[str, Any]], target: WriteTarget) -> in staging_id = _target_id(staging_target) schema_sql = ", ".join(f"`{field.name}` {_field_type_sql(field)}" for field in declared) try: - self._client.query(f"CREATE TABLE `{staging_id}` ({schema_sql})").result() + self._client.query( + f"CREATE TABLE `{staging_id}` ({schema_sql}) OPTIONS " + "(expiration_timestamp = TIMESTAMP_ADD(CURRENT_TIMESTAMP(), INTERVAL 1 DAY))" + ).result() self._backend.append( staged_rows, staging_target, diff --git a/tests/bootstrap/test_admin.py b/tests/bootstrap/test_admin.py index 8c609b82..0aff598b 100644 --- a/tests/bootstrap/test_admin.py +++ b/tests/bootstrap/test_admin.py @@ -110,6 +110,39 @@ def fake_run(args: tuple[str, ...], **kwargs: object) -> subprocess.CompletedPro assert imported[-2:] == ("google_storage_bucket.terraform_state", "unit-state-bucket") +def test_admin_apply_uses_only_the_previously_saved_plan( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + infra_dir = tmp_path / "checkout" / "infra" / "bootstrap-admin" + operator_dir = tmp_path / "operator" + infra_dir.mkdir(parents=True) + commands: list[tuple[str, ...]] = [] + + def fake_run(args: tuple[str, ...], **kwargs: object) -> subprocess.CompletedProcess[str]: + commands.append(args) + for argument in args: + if argument.startswith("-out="): + Path(argument.removeprefix("-out=")).touch() + return subprocess.CompletedProcess(args, 0) + + monkeypatch.setattr(subprocess, "run", fake_run) + bootstrap = AdministrativeBootstrap(infra_dir, operator_dir) + plan = bootstrap.execute( + project="unit-project", + state_bucket="unit-state-bucket", + admin_member="user:operator@example.invalid", + apply=False, + ) + commands.clear() + + applied = bootstrap.apply_saved_plan(state_bucket="unit-state-bucket") + + assert applied == plan + assert commands[-1] == ("terraform", "apply", str(plan)) + assert not any(command[:2] == ("terraform", "plan") for command in commands) + + @pytest.mark.parametrize("artifact_name", ["terraform-data", "dander-admin-bootstrap.tfplan"]) def test_admin_bootstrap_rejects_preexisting_operator_symlinks_before_terraform( monkeypatch: pytest.MonkeyPatch, diff --git a/tests/bootstrap/test_permissions.py b/tests/bootstrap/test_permissions.py new file mode 100644 index 00000000..81cf1281 --- /dev/null +++ b/tests/bootstrap/test_permissions.py @@ -0,0 +1,109 @@ +"""Read-only stage-zero permission preflight tests.""" + +from __future__ import annotations + +import json +import subprocess +from typing import TYPE_CHECKING, Any + +import httpx +import pytest + +from dander.bootstrap import AdministrativeBootstrapError, require_stage_zero_permissions + +if TYPE_CHECKING: + from pathlib import Path + + +class _Runner: + def __call__( + self, + args: tuple[str, ...], + *, + cwd: Path, + check: bool, + capture_output: bool = False, + text: bool = False, + ) -> subprocess.CompletedProcess[str]: + assert args == ( + "gcloud", + "auth", + "print-access-token", + "--project=unit-project", + ) + assert cwd.is_absolute() and check and capture_output and text + return subprocess.CompletedProcess(args, 0, stdout="temporary-access-token\n") + + +def _response(url: str, payload: dict[str, Any]) -> httpx.Response: + return httpx.Response(200, json=payload, request=httpx.Request("POST", url)) + + +def test_preflight_checks_only_core_project_permissions_by_default(tmp_path: Path) -> None: + requests: list[tuple[str, tuple[str, ...], str]] = [] + + def post(url: str, **kwargs: object) -> httpx.Response: + payload = json.loads(str(kwargs["content"])) + permissions = tuple(payload["permissions"]) + requests.append((url, permissions, str(kwargs["headers"]))) + return _response(url, {"permissions": list(permissions)}) + + require_stage_zero_permissions( + project="unit-project", + cwd=tmp_path, + runner=_Runner(), + post=post, + ) + + assert len(requests) == 1 + assert "cloudresourcemanager.googleapis.com" in requests[0][0] + assert not any("workloadIdentityPools" in item for item in requests[0][1]) + assert "temporary-access-token" in requests[0][2] + + +def test_preflight_adds_wif_and_billing_permissions_only_when_requested(tmp_path: Path) -> None: + requested: list[tuple[str, ...]] = [] + + def post(url: str, **kwargs: object) -> httpx.Response: + payload = json.loads(str(kwargs["content"])) + permissions = tuple(payload["permissions"]) + requested.append(permissions) + return _response(url, {"permissions": list(permissions)}) + + require_stage_zero_permissions( + project="unit-project", + cwd=tmp_path, + billing_account_id="ABCDEF-123456-ABCDEF", + github_repository="owner/repository", + runner=_Runner(), + post=post, + ) + + assert len(requested) == 2 + assert any("workloadIdentityPools" in item for item in requested[0]) + assert requested[1] == ( + "billing.accounts.get", + "billing.accounts.getIamPolicy", + "billing.accounts.setIamPolicy", + ) + + +def test_preflight_reports_exact_missing_permissions_without_running_terraform( + tmp_path: Path, +) -> None: + def post(url: str, **kwargs: object) -> httpx.Response: + payload = json.loads(str(kwargs["content"])) + permissions = list(payload["permissions"]) + permissions.remove("resourcemanager.projects.setIamPolicy") + return _response(url, {"permissions": permissions}) + + with pytest.raises( + AdministrativeBootstrapError, + match=r"missing project: resourcemanager\.projects\.setIamPolicy", + ): + require_stage_zero_permissions( + project="unit-project", + cwd=tmp_path, + runner=_Runner(), + post=post, + ) diff --git a/tests/bootstrap/test_project_bootstrap.py b/tests/bootstrap/test_project_bootstrap.py index 1876f8f8..47972863 100644 --- a/tests/bootstrap/test_project_bootstrap.py +++ b/tests/bootstrap/test_project_bootstrap.py @@ -23,6 +23,7 @@ class _Runner: def __init__(self, *, bucket_exists: bool = False) -> None: self.bucket_exists = bucket_exists self.commands: list[tuple[str, ...]] = [] + self.inputs: list[str | None] = [] def __call__( self, @@ -32,15 +33,19 @@ def __call__( check: bool, capture_output: bool = False, text: bool = False, + input: str | None = None, ) -> subprocess.CompletedProcess[str]: assert cwd.is_absolute() self.commands.append(args) + self.inputs.append(input) if args[:4] == ("gcloud", "storage", "buckets", "describe"): return subprocess.CompletedProcess(args, 0 if self.bucket_exists else 1, stdout="") if args[:4] == ("gcloud", "artifacts", "docker", "images"): return subprocess.CompletedProcess(args, 0, stdout="sha256:" + "a" * 64 + "\n") if args[:3] == ("gcloud", "auth", "list"): return subprocess.CompletedProcess(args, 0, stdout="operator@example.invalid\n") + if args[:3] == ("gcloud", "auth", "print-access-token"): + return subprocess.CompletedProcess(args, 0, stdout="temporary-token\n") return subprocess.CompletedProcess(args, 0, stdout="") @@ -100,6 +105,49 @@ def test_runtime_image_publisher_accepts_installed_project_context(tmp_path: Pat assert image.endswith("@sha256:" + "a" * 64) +def test_runtime_image_publisher_uses_bootstrap_impersonation_without_token_in_args( + tmp_path: Path, +) -> None: + for name in ("Dockerfile", "README.md", "dander.yaml"): + (tmp_path / name).write_text(name, encoding="utf-8") + for directory in ("connectors", "models"): + path = tmp_path / directory + path.mkdir() + (path / "content.txt").write_text(directory, encoding="utf-8") + runner = _Runner() + + image = RuntimeImagePublisher(tmp_path, runner=runner).publish( + project="unit-project", + region="us-central1", + impersonate_service_account="dander-bootstrap@unit-project.iam.gserviceaccount.com", + ) + + assert image.endswith("@sha256:" + "a" * 64) + login_index = next( + index for index, command in enumerate(runner.commands) if command[:2] == ("docker", "login") + ) + assert runner.inputs[login_index] == "temporary-token" + assert all("temporary-token" not in part for command in runner.commands for part in command) + describe = next(command for command in runner.commands if "images" in command) + assert ( + "--impersonate-service-account=dander-bootstrap@unit-project.iam.gserviceaccount.com" + in describe + ) + + +def test_runtime_image_publisher_rejects_source_checkout_when_source_free_is_required( + tmp_path: Path, +) -> None: + (tmp_path / "src").mkdir() + + with pytest.raises(ProjectBootstrapError, match="generated source-free project"): + RuntimeImagePublisher(tmp_path, runner=_Runner()).publish( + project="unit-project", + region="us-central1", + require_source_free=True, + ) + + def test_runtime_image_tag_ignores_local_state_but_tracks_infrastructure(tmp_path: Path) -> None: for name in ("Dockerfile", "dander.yaml"): (tmp_path / name).write_text(name, encoding="utf-8") diff --git a/tests/cli/test_init_cli.py b/tests/cli/test_init_cli.py index ed0b3385..5eb1581a 100644 --- a/tests/cli/test_init_cli.py +++ b/tests/cli/test_init_cli.py @@ -92,6 +92,126 @@ def fake_execute(self: object, **kwargs: object) -> Path: assert captured["live_cost_guard"] is False +def test_image_publish_uses_bootstrap_identity_and_prints_platform_plan( + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: dict[str, object] = {} + + def fake_publish(self: object, **kwargs: object) -> str: + captured.update(kwargs) + return "us-central1-docker.pkg.dev/unit-project/dander/dander@sha256:" + "a" * 64 + + monkeypatch.setattr("dander.cli.main.RuntimeImagePublisher.publish", fake_publish) + + result = CliRunner().invoke( + app, + [ + "image-publish", + "--project", + "unit-project", + "--failure-alert-email", + "operator@example.invalid", + "--billing-account", + "ABCDEF-123456-ABCDEF", + ], + input="y\n", + ) + + assert result.exit_code == 0, result.output + assert captured["region"] == "us-central1" + assert captured["impersonate_service_account"] == ( + "dander-bootstrap@unit-project.iam.gserviceaccount.com" + ) + assert captured["require_source_free"] is True + assert "init-platform-plan" in result.output + assert "Published immutable runtime image" in result.output + assert "operator@example.invalid" in result.output + + +def test_platform_plan_resolves_complete_manifest_without_applying( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + captured: dict[str, object] = {} + + def fake_execute(self: object, **kwargs: object) -> Path: + captured.update(kwargs) + return tmp_path / "dander-bootstrap.tfplan" + + monkeypatch.setattr("dander.cli.main.TerraformBootstrap.execute", fake_execute) + + result = CliRunner().invoke( + app, + [ + "init-platform-plan", + "--project", + "unit-project", + "--state-bucket", + "unit-state", + "--bootstrap-service-account", + "dander-bootstrap@unit-project.iam.gserviceaccount.com", + "--container-image", + f"example.invalid/dander@sha256:{'a' * 64}", + "--failure-alert-email", + "operator@example.invalid", + "--billing-account", + "ABCDEF-123456-ABCDEF", + ], + ) + + assert result.exit_code == 0, result.output + assert captured["apply"] is False + assert captured["enable_runtime"] is True + pipelines = captured["pipelines"] + assert isinstance(pipelines, dict) + assert set(pipelines) == { + "greenhouse_jobs", + "greenhouse_jobs_graph", + "hubspot_companies", + "salesforce_accounts", + "servicenow_incidents", + } + assert "terraform -chdir=infra show -no-color" in result.output + assert "init-platform-apply" in result.output + + +def test_admin_plan_runs_read_only_permission_preflight_before_terraform( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + events: list[str] = [] + + def fake_preflight(**kwargs: object) -> None: + events.append("preflight") + assert kwargs["project"] == "unit-project" + + def fake_execute(self: object, **kwargs: object) -> Path: + events.append("terraform") + return tmp_path / "dander-admin-bootstrap.tfplan" + + monkeypatch.setattr("dander.cli.main.require_stage_zero_permissions", fake_preflight) + monkeypatch.setattr("dander.cli.main.AdministrativeBootstrap.execute", fake_execute) + + result = CliRunner().invoke( + app, + [ + "init-admin-plan", + "--project", + "unit-project", + "--state-bucket", + "unit-state", + "--admin-member", + "user:operator@example.invalid", + "--operator-artifact-dir", + str(tmp_path / "operator"), + ], + ) + + assert result.exit_code == 0, result.output + assert events == ["preflight", "terraform"] + assert "init-admin-apply" in result.output + + def test_init_apply_requires_confirmation( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, diff --git a/tests/cli/test_metadata_cli.py b/tests/cli/test_metadata_cli.py index 55e1f883..c7d91fe1 100644 --- a/tests/cli/test_metadata_cli.py +++ b/tests/cli/test_metadata_cli.py @@ -8,7 +8,7 @@ from dander.catalog import SqliteMetadataStore from dander.cli.main import app -from dander.state import RunStatus, SqliteRunHistoryStore +from dander.state import RunStage, RunStatus, SqliteRunHistoryStore if TYPE_CHECKING: from pathlib import Path @@ -113,3 +113,28 @@ def test_metadata_runs_renders_active_run(tmp_path: Path) -> None: assert result.exit_code == 0 assert "running" in result.stdout + + +def test_metadata_runs_renders_safe_failure_summary(tmp_path: Path) -> None: + state = tmp_path / "state.db" + history = SqliteRunHistoryStore(state) + history.start("run-failed", "salesforce", pipeline_id="salesforce_crm") + history.finish( + "run-failed", + RunStatus.FAILED, + endpoints=0, + extracted=0, + affected=0, + failure_stage=RunStage.INGEST, + failure_code="authentication_failed", + failure_summary="Authentication failed. Verify the configured secret.", + ) + + result = CliRunner().invoke( + app, + ["metadata", "runs", "--local", "--state-path", str(state)], + ) + + assert result.exit_code == 0 + assert "authentication" in result.stdout + assert "configured" in result.stdout diff --git a/tests/infra/test_multi_pipeline_runtime.py b/tests/infra/test_multi_pipeline_runtime.py index f11cbebd..dff69f60 100644 --- a/tests/infra/test_multi_pipeline_runtime.py +++ b/tests/infra/test_multi_pipeline_runtime.py @@ -91,8 +91,10 @@ def test_container_carries_the_project_manifest() -> None: dockerignore = (ROOT / ".dockerignore").read_text(encoding="utf-8") assert "COPY dander.yaml ./dander.yaml" in dockerfile + assert "COPY examples ./examples" in dockerfile assert "COPY infra ./infra" in dockerfile assert "!dander.yaml" in dockerignore + assert "!examples/**" in dockerignore assert "!infra/**" in dockerignore assert "infra/.terraform/" in dockerignore diff --git a/tests/ingestion/test_dlt_runtime.py b/tests/ingestion/test_dlt_runtime.py index c561c13f..2dc4556c 100644 --- a/tests/ingestion/test_dlt_runtime.py +++ b/tests/ingestion/test_dlt_runtime.py @@ -187,6 +187,12 @@ def test_salesforce_template_maps_bulk2_query_schema_and_jwt_contract() -> None: "default_expires_in": 300, } assert config.engine is IngestionEngine.SALESFORCE_BULK2 + assert [endpoint.name for endpoint in config.endpoints] == [ + "accounts", + "contacts", + "opportunities", + "users", + ] endpoint = config.endpoints[0] assert endpoint.path == "/jobs/query" assert endpoint.incremental_cursor == "SystemModstamp" @@ -200,9 +206,16 @@ def test_salesforce_template_maps_bulk2_query_schema_and_jwt_contract() -> None: "attributes", "Id", "Name", + "OwnerId", "SystemModstamp", "IsDeleted", } + assert [endpoint.request_body["operation"] for endpoint in config.endpoints] == [ + "queryAll", + "queryAll", + "queryAll", + "query", + ] class _FakeDltSource: diff --git a/tests/pipeline/test_graph_operations.py b/tests/pipeline/test_graph_operations.py index 47b4ea97..59d21b84 100644 --- a/tests/pipeline/test_graph_operations.py +++ b/tests/pipeline/test_graph_operations.py @@ -341,6 +341,8 @@ def run(args: tuple[str, ...]) -> str: "assertions": 0, "assets": 0, "failure_stage": None, + "failure_code": None, + "failure_summary": None, } assert history.calls == [(1, "graph_records")] diff --git a/tests/project/test_config.py b/tests/project/test_config.py index 19350403..ad4b1433 100644 --- a/tests/project/test_config.py +++ b/tests/project/test_config.py @@ -93,6 +93,14 @@ def test_repository_manifest_defines_five_additive_hosted_pipelines() -> None: } assert expanded["salesforce_accounts"]["job_name"] == "dander-salesforce-accounts" assert expanded["salesforce_accounts"]["paused"] is False + assert expanded["salesforce_accounts"]["models"] == [ + "stg_salesforce__users", + "stg_salesforce__accounts", + "stg_salesforce__contacts", + "stg_salesforce__opportunities", + "fct_salesforce__opportunities", + ] + assert expanded["salesforce_accounts"]["publish_dataplex"] is True assert expanded["salesforce_accounts"]["secret_env"] == { "SALESFORCE_EXTERNAL_CLIENT_APP_ID": "salesforce-external-client-app-id", "SALESFORCE_EXTERNAL_CLIENT_APP_PRIVATE_KEY": ( diff --git a/tests/project/test_scaffold.py b/tests/project/test_scaffold.py index dbc35fe5..b70c6cbc 100644 --- a/tests/project/test_scaffold.py +++ b/tests/project/test_scaffold.py @@ -41,8 +41,18 @@ def test_scaffold_creates_complete_paused_project(tmp_path: Path) -> None: "infra/bootstrap-admin/main.tf", "models/staging/stg_greenhouse__jobs.sql", "models/staging/stg_greenhouse__jobs.yml", + "examples/salesforce/dander.yaml", + "examples/salesforce/connectors/salesforce.yaml", + "examples/salesforce/models/staging/stg_salesforce__users.sql", + "examples/salesforce/models/staging/stg_salesforce__accounts.sql", + "examples/salesforce/models/staging/stg_salesforce__contacts.sql", + "examples/salesforce/models/staging/stg_salesforce__opportunities.sql", + "examples/salesforce/models/marts/fct_salesforce__opportunities.sql", ): assert (project / relative).is_file() + salesforce = load_project_config(project / "examples" / "salesforce" / "dander.yaml") + assert salesforce.plugins["salesforce"].version == "0.3.0rc1" + assert set(salesforce.pipelines) == {"salesforce_crm"} assert not list(project.rglob("*.tfplan")) assert not list(project.rglob("*.tfstate")) assert not list(project.rglob(".terraform")) diff --git a/tests/state/test_failure.py b/tests/state/test_failure.py new file mode 100644 index 00000000..fca62304 --- /dev/null +++ b/tests/state/test_failure.py @@ -0,0 +1,44 @@ +"""Operator-safe run failure classification.""" + +from __future__ import annotations + +from dataclasses import dataclass + +from dander.state import RunStage, classify_failure + + +@dataclass +class _Response: + status_code: int + + +class _HttpError(RuntimeError): + def __init__(self, status_code: int, message: str) -> None: + super().__init__(message) + self.response = _Response(status_code) + + +def test_failure_classifier_uses_http_status_without_persisting_exception_text() -> None: + error = RuntimeError("wrapper with token=top-secret") + error.__cause__ = _HttpError(401, "private-key=also-secret") + + failure = classify_failure(error, stage=RunStage.INGEST, run_id="safe-run") + + assert failure.code == "authentication_failed" + assert "top-secret" not in failure.summary + assert "also-secret" not in failure.summary + assert len(failure.summary) <= 512 + + +def test_unknown_failure_is_stage_specific_and_points_to_run_logs() -> None: + failure = classify_failure( + RuntimeError("customer payload must never persist"), + stage=RunStage.METADATA, + run_id="run-123", + ) + + assert failure.code == "catalog_failed" + assert failure.summary == ( + "Metadata or catalog publication failed. Inspect logs for run run-123." + ) + assert "customer payload" not in failure.summary diff --git a/tests/state/test_run_history.py b/tests/state/test_run_history.py index 19b8f11a..c7c3fdd0 100644 --- a/tests/state/test_run_history.py +++ b/tests/state/test_run_history.py @@ -81,6 +81,44 @@ def test_sqlite_run_history_reads_active_run(tmp_path: Path) -> None: assert record.finished_at is None +def test_sqlite_run_history_persists_sanitized_failure_details(tmp_path: Path) -> None: + store = SqliteRunHistoryStore(tmp_path / "state.db") + store.start("run-failed", "salesforce", pipeline_id="salesforce_crm") + + store.finish( + "run-failed", + RunStatus.FAILED, + endpoints=0, + extracted=0, + affected=0, + failure_stage=RunStage.INGEST, + failure_code="authentication_failed", + failure_summary="Authentication failed. Verify the configured secret.", + ) + + record = store.recent(pipeline_id="salesforce_crm")[0] + assert record.failure_code == "authentication_failed" + assert record.failure_summary == "Authentication failed. Verify the configured secret." + + +def test_sqlite_reconciles_only_older_active_runs_for_owned_pipeline(tmp_path: Path) -> None: + store = SqliteRunHistoryStore(tmp_path / "state.db") + store.start("old", "salesforce", pipeline_id="salesforce_crm") + store.checkpoint("old", RunStage.TRANSFORM, endpoints=4, extracted=12, affected=12) + store.start("current", "salesforce", pipeline_id="salesforce_crm") + store.start("other", "greenhouse", pipeline_id="greenhouse_jobs") + + store.reconcile_interrupted("salesforce_crm", current_run_id="current") + + records = {record.run_id: record for record in store.recent(limit=10)} + assert records["old"].status is RunStatus.FAILED + assert records["old"].failure_stage is RunStage.TRANSFORM + assert records["old"].failure_code == "interrupted_run" + assert records["old"].finished_at is not None + assert records["current"].status is RunStatus.RUNNING + assert records["other"].status is RunStatus.RUNNING + + class _QueryJob: def result(self) -> tuple[object, ...]: return () @@ -113,5 +151,25 @@ def test_bigquery_history_can_read_without_creating_or_altering_tables() -> None assert store.recent(limit=1, pipeline_id="graph_records") == () assert len(client.queries) == 1 assert client.queries[0].startswith("SELECT ") + assert "JSON_VALUE(TO_JSON_STRING(history_row), '$.failure_code')" in client.queries[0] + assert "JSON_VALUE(TO_JSON_STRING(history_row), '$.failure_summary')" in client.queries[0] + assert "AS history_row" in client.queries[0] assert "CREATE" not in client.queries[0] assert "ALTER" not in client.queries[0] + + +def test_bigquery_reconciliation_is_bounded_to_owned_pipeline_and_excludes_current() -> None: + client = _BigQueryClient() + store = BigQueryRunHistoryStore( + project="proof-project", + dataset="dander_meta", + client=client, + ) + + store.reconcile_interrupted("salesforce_crm", current_run_id="current") + + update = client.queries[-1] + assert "failure_code = 'interrupted_run'" in update + assert "status = 'running'" in update + assert "COALESCE(pipeline_id, source_name) = @pipeline_id" in update + assert "run_id != @current_run_id" in update diff --git a/tests/test_executor.py b/tests/test_executor.py index 0b4710f6..58ec1974 100644 --- a/tests/test_executor.py +++ b/tests/test_executor.py @@ -72,6 +72,8 @@ def __init__(self) -> None: self.started: tuple[str, str, str] | None = None self.checkpoints: list[RunStage] = [] self.finished: tuple[RunStatus, RunStage | None, int, int, int] | None = None + self.failure: tuple[str | None, str | None] | None = None + self.reconciled: tuple[str, str] | None = None def start(self, run_id: str, source: str, *, pipeline_id: str | None = None) -> None: assert pipeline_id is not None @@ -98,9 +100,15 @@ def finish( assertions: int = 0, assets: int = 0, failure_stage: RunStage | None = None, + failure_code: str | None = None, + failure_summary: str | None = None, ) -> None: assert self.started is not None and run_id == self.started[0] self.finished = (status, failure_stage, models, assertions, assets) + self.failure = (failure_code, failure_summary) + + def reconcile_interrupted(self, pipeline_id: str, *, current_run_id: str) -> None: + self.reconciled = (pipeline_id, current_run_id) class _Metadata(MetadataStore): @@ -247,6 +255,9 @@ def test_executor_marks_transform_failure_without_claiming_ingestion_only_succes assert history.checkpoints == [RunStage.TRANSFORM] assert history.finished == (RunStatus.FAILED, RunStage.TRANSFORM, 0, 0, 0) + assert history.failure is not None + assert history.failure[0] == "transform_failed" + assert "Inspect logs for run" in (history.failure[1] or "") def test_executor_records_active_overlap_as_skipped_without_running_pipeline( @@ -266,6 +277,7 @@ def test_executor_records_active_overlap_as_skipped_without_running_pipeline( assert result.ingestion.endpoints == () assert history.checkpoints == [] assert history.finished == (RunStatus.SKIPPED, None, 0, 0, 0) + assert history.reconciled is None def test_executor_fails_closed_before_transform_when_heartbeat_is_lost( @@ -285,4 +297,6 @@ def test_executor_fails_closed_before_transform_when_heartbeat_is_lost( assert history.checkpoints == [RunStage.TRANSFORM] assert history.finished == (RunStatus.FAILED, RunStage.TRANSFORM, 0, 0, 0) + assert history.failure is not None and history.failure[0] == "lease_failed" + assert history.reconciled is not None assert leases.released diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 29742915..eff1c9c9 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -222,8 +222,17 @@ def finish( assertions: int = 0, assets: int = 0, failure_stage: RunStage | None = None, + failure_code: str | None = None, + failure_summary: str | None = None, ) -> None: - assert (models, assertions, assets, failure_stage) == (0, 0, 0, None) + assert (models, assertions, assets, failure_stage, failure_code, failure_summary) == ( + 0, + 0, + 0, + None, + None, + None, + ) self.finished = (run_id, status, endpoints, extracted, affected) @@ -343,6 +352,7 @@ def test_salesforce_query_record_envelope_satisfies_declared_runtime_schema() -> config = load_source_config( Path(__file__).parents[1] / "connectors" / "salesforce_jwt.example.yaml" ) + config.endpoints = [config.endpoints[0]] class _SalesforceSource(Source): def discover(self) -> Mapping[str, Any]: diff --git a/tests/transform/test_project.py b/tests/transform/test_project.py index ec0c7888..5e0f66c6 100644 --- a/tests/transform/test_project.py +++ b/tests/transform/test_project.py @@ -122,14 +122,40 @@ def test_unknown_selected_model_fails(tmp_path: Path) -> None: project.ordered(["absent"]) -def test_salesforce_accounts_model_loads_and_compiles_raw_relation() -> None: +def test_salesforce_crm_models_compile_in_governed_dependency_order() -> None: models = Path(__file__).parents[2] / "models" project = TransformProject.load(models, project_id="valid-project-123") - model = project.models["stg_salesforce__accounts"] - - assert project.ordered([model.name]) == (model,) - assert "`valid-project-123.raw.salesforce_accounts`" in project.compile(model) - assert [metric.name for metric in model.metadata.metrics] == ["account_count"] + selected = [ + "stg_salesforce__users", + "stg_salesforce__accounts", + "stg_salesforce__contacts", + "stg_salesforce__opportunities", + "fct_salesforce__opportunities", + ] + + assert [model.name for model in project.ordered(selected)] == selected + for endpoint in ("accounts", "contacts", "opportunities", "users"): + model = project.models[f"stg_salesforce__{endpoint}"] + assert f"`valid-project-123.raw.salesforce_{endpoint}`" in project.compile(model) + + fact = project.models["fct_salesforce__opportunities"] + compiled = project.compile(fact) + assert "WHERE NOT is_deleted" in compiled + assert "owner.is_active AS owner_is_active" in compiled + contacts = project.models["stg_salesforce__contacts"] + relationships = { + test.column: (test.relationships.to, test.relationships.field) + for test in contacts.metadata.tests + if test.relationships is not None + } + assert relationships == { + "account_id": ("stg_salesforce__accounts", "account_id"), + "owner_id": ("stg_salesforce__users", "user_id"), + } + assert [metric.name for metric in fact.metadata.metrics] == [ + "active_opportunity_count", + "active_opportunity_amount", + ] def test_servicenow_incidents_model_loads_and_casts_internal_utc_values() -> None: diff --git a/tests/writer/test_bigquery_writer.py b/tests/writer/test_bigquery_writer.py index a285c73e..8ea3ffcd 100644 --- a/tests/writer/test_bigquery_writer.py +++ b/tests/writer/test_bigquery_writer.py @@ -2,7 +2,7 @@ from __future__ import annotations -from datetime import UTC, date, datetime, time +from datetime import UTC, date, datetime, time, timedelta from decimal import Decimal from typing import TYPE_CHECKING, Any @@ -58,7 +58,9 @@ def __init__( self.write_disposition: str | None = None self.write_dispositions: list[str] = [] self.loaded_schemas: list[list[bigquery.SchemaField]] = [] + self.load_configs: list[bigquery.LoadJobConfig] = [] self.created: list[str] = [] + self.created_expirations: dict[str, datetime | None] = {} self.updated: list[tuple[str, list[str]]] = [] self.copies: list[tuple[str, str, str]] = [] self.tables: dict[str, bigquery.Table] = {} @@ -82,6 +84,7 @@ def create_table( return self.tables[table_id] self.tables[table_id] = table self.created.append(table_id) + self.created_expirations[table_id] = table.expires return table def get_table(self, table: str) -> bigquery.Table: @@ -115,6 +118,7 @@ def load_table_from_json( assert job_config.autodetect self.write_disposition = job_config.write_disposition self.write_dispositions.append(job_config.write_disposition) + self.load_configs.append(job_config) self.loaded_rows = [dict(row) for row in json_rows] self.loaded_batches.append(self.loaded_rows) self.destination = destination @@ -175,6 +179,23 @@ def _fenced_target() -> WriteTarget: ) +def _assert_supported_load_config(config: bigquery.LoadJobConfig) -> None: + assert "destinationExpirationTime" not in config.to_api_repr()["load"] + + +def _assert_staging_precreated_with_expiration(client: _Client) -> None: + staging = next(table for table in client.created if "._dander_stage_" in table) + expires = client.created_expirations[staging] + assert expires is not None + remaining = expires - datetime.now(UTC) + assert timedelta(hours=23, minutes=59) < remaining <= timedelta(days=1) + + +def _assert_inferred_staging_expired_after_load(client: _Client) -> None: + _assert_supported_load_config(client.load_configs[0]) + assert client.queries[0].startswith(f"ALTER TABLE `{client.destination}` SET OPTIONS") + + def test_scd1_deduplicates_last_record_and_builds_explicit_merge() -> None: client = _Client() writer = BigQueryScd1Writer(project="unit-project", client=client) @@ -193,7 +214,7 @@ def test_scd1_deduplicates_last_record_and_builds_explicit_merge() -> None: {"id": "one", "label": "new"}, {"id": "two", "label": "other"}, ] - assert client.queries[0].startswith(f"ALTER TABLE `{client.destination}` SET OPTIONS") + _assert_inferred_staging_expired_after_load(client) assert client.queries[1].startswith( "CREATE TABLE IF NOT EXISTS `unit-project.raw.example_widgets`" ) @@ -256,6 +277,8 @@ def test_scd1_encodes_typed_scalars_for_bigquery_json_load() -> None: }, } ] + _assert_supported_load_config(client.load_configs[0]) + _assert_staging_precreated_with_expiration(client) def test_scd1_finalizer_dml_touches_matching_lease_inside_transaction() -> None: @@ -331,10 +354,19 @@ def test_writer_rejects_inconsistent_shape_before_network() -> None: def test_staging_table_is_cleaned_after_load_failure() -> None: client = _Client(load_error=RuntimeError("synthetic load failure")) writer = BigQueryScd1Writer(project="unit-project", client=client) + target = WriteTarget( + project="unit-project", + dataset="raw", + table="example_widgets", + business_key=("id",), + schema=(WriteField(name="id", data_type="STRING", mode="REQUIRED"),), + ) with pytest.raises(RuntimeError, match="synthetic load failure"): - writer.write([{"id": "one"}], _target()) + writer.write([{"id": "one"}], target) + _assert_supported_load_config(client.load_configs[0]) + _assert_staging_precreated_with_expiration(client) assert client.deleted == [client.destination] @@ -347,7 +379,7 @@ def test_replace_writer_stages_then_atomically_replaces_target() -> None: assert affected == 2 assert client.destination.startswith("unit-project.raw._dander_stage_example_widgets_") assert client.write_disposition == "WRITE_TRUNCATE" - assert client.queries[0].startswith(f"ALTER TABLE `{client.destination}` SET OPTIONS") + _assert_inferred_staging_expired_after_load(client) assert client.copies == [ ( client.destination, @@ -393,11 +425,31 @@ def test_replace_writer_bootstraps_declared_schema_for_empty_snapshot() -> None: assert len(client.created) == 1 assert client.created[0].startswith("unit-project.raw._dander_stage_example_widgets_") + expires = client.created_expirations[client.created[0]] + assert expires is not None + assert timedelta(hours=23, minutes=59) < expires - datetime.now(UTC) assert client.tables["unit-project.raw.example_widgets"].schema == [ bigquery.SchemaField("id", "INT64", mode="REQUIRED") ] +def test_replace_writer_precreates_declared_staging_with_expiration() -> None: + client = _Client() + writer = BigQueryReplaceWriter(project="unit-project", client=client) + target = WriteTarget( + project="unit-project", + dataset="raw", + table="example_widgets", + schema=(WriteField(name="id", data_type="INT64", mode="REQUIRED"),), + ) + + assert writer.write([{"id": 1}], target) == 1 + + _assert_supported_load_config(client.load_configs[0]) + _assert_staging_precreated_with_expiration(client) + assert client.queries == [] + + def test_replace_writer_bounds_load_requests_and_appends_after_first_chunk() -> None: client = _Client() writer = BigQueryReplaceWriter( @@ -414,6 +466,9 @@ def test_replace_writer_bounds_load_requests_and_appends_after_first_chunk() -> assert affected == 5 assert [len(batch) for batch in client.loaded_batches] == [2, 2, 1] assert client.write_dispositions == ["WRITE_TRUNCATE", "WRITE_APPEND", "WRITE_APPEND"] + _assert_inferred_staging_expired_after_load(client) + for config in client.load_configs: + _assert_supported_load_config(config) assert client.copies == [ ( client.destination, @@ -782,8 +837,9 @@ def test_snapshot_writer_partitions_and_suppresses_exact_reruns() -> None: ) assert affected == 1 - assert "PARTITION BY DATE(`snapshot_at`)" in client.queries[0] - insert = client.queries[1] + _assert_inferred_staging_expired_after_load(client) + assert "PARTITION BY DATE(`snapshot_at`)" in client.queries[1] + insert = client.queries[2] assert insert.startswith("INSERT INTO") assert "WHERE NOT EXISTS" in insert assert "IS NOT DISTINCT FROM" in insert @@ -837,15 +893,16 @@ def test_scd2_writer_builds_transactional_change_history() -> None: ) assert affected == 2 + _assert_inferred_staging_expired_after_load(client) assert client.loaded_rows == [ {"id": "one", "label": "new"}, {"id": "two", "label": "other"}, ] - create = client.queries[0] + create = client.queries[1] assert "valid_from" in create assert "valid_to" in create assert "is_current" in create - history = client.queries[1] + history = client.queries[2] assert "CREATE TEMP TABLE changed" in history assert "IS DISTINCT FROM" in history assert "BEGIN TRANSACTION" in history diff --git a/tests/writer/test_storage_write_writer.py b/tests/writer/test_storage_write_writer.py index 5d4c00f6..6996c7e5 100644 --- a/tests/writer/test_storage_write_writer.py +++ b/tests/writer/test_storage_write_writer.py @@ -103,6 +103,7 @@ def test_storage_writer_uses_pending_backend_then_idempotent_merge() -> None: assert backend.rows == [{"id": "one", "count": 2}, {"id": "two", "count": 3}] assert backend.max_batch_rows == 2 assert client.queries[0].startswith("CREATE TABLE `unit-project.raw._dander_stage_widgets_") + assert "OPTIONS (expiration_timestamp" in client.queries[0] assert client.queries[-1].startswith("MERGE `unit-project.raw.widgets`") assert backend.target is not None assert client.deleted == [