Skip to content

Commit be94b2c

Browse files
eric-thomas-dagsterclaude
eric-thomas-dagster
and
claude
committed
refactor(precisely): observation-only — drop unverified submit + list-runs paths
Precisely Connect ETL publishes exactly one REST endpoint: GET /projects/{jobRunId}/status (plain-text response). The components shipped speculative integrations with two other endpoints that Precisely does NOT document publicly: - PreciselyResource.run_job() POSTed to /projects/{job_id}/run - precisely_job_sensor mode-2 GET /api/v1/jobs/{job_id}/runs Both were "best-guess RESTful shapes" — the docstrings admitted as much. In real customer installs they 404 because the API surface just isn't there. precisely_run_asset would never materialize. This change: 1. precisely_run_asset is now an observable_source_asset that polls the documented Job Status endpoint and emits ObserveResult with status + is_terminal/is_success/is_in_progress flags. No submit. Required field is job_run_id (the actual run-id from Precisely's side), not job_id + speculative submit. 2. PreciselyResource drops run_job() and submit_path_template. Only get_run_status() remains — the verified endpoint. 3. precisely_job_sensor drops the speculative list-runs mode (job_id + list_runs_path_template). Only the verified job_run_id-watching mode remains. 4. README + schema.json + example.yaml updated to reflect the new observe-only shape. Pair-of-components story: - precisely_run_asset → catalog presence + lineage (observable_source_asset) - precisely_job_sensor → event-driven downstream Dagster work (RunRequest on terminal SUCCESS) Both poll the same verified endpoint; they're complementary, not redundant — one is a graph node, one is an event source. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent 9b99ca5 commit be94b2c

5 files changed

Lines changed: 154 additions & 546 deletions

File tree

‎assets/infrastructure/precisely_run_asset/README.md‎

Lines changed: 31 additions & 121 deletions
Original file line numberDiff line numberDiff line change
@@ -1,23 +1,23 @@
1-
# Precisely Run Asset
1+
# Precisely Run Asset (observation-only)
22

3-
Triggers a [Precisely Connect ETL](https://www.precisely.com/product/precisely-connect/connect) (formerly Syncsort DMX / DMExpress) job on demand and surfaces the result as a Dagster asset. Dagster owns the schedule and orchestration; Precisely executes the integration.
3+
Observes a [Precisely Connect ETL](https://www.precisely.com/product/precisely-connect/connect) (formerly Syncsort DMX / DMExpress) job run as a Dagster `observable_source_asset`. Precisely owns the schedule; Dagster surfaces the latest status in its catalog.
44

55
## What this does
66

7-
When the asset is materialized:
7+
When the asset is observed (manually, on a schedule, or by Dagster's automation):
88

9-
1. POSTs to the Precisely Connect job-submit endpoint
10-
2. Polls the documented Job Status endpoint until the job reaches a terminal status
11-
3. Returns a `MaterializeResult` with run metadata on success, or raises on failure / timeout
9+
1. Calls the documented Precisely Job Status endpoint (`GET /projects/{jobRunId}/status`)
10+
2. Emits an `ObserveResult` with the run's current status, plus boolean flags for terminal / success / in-progress states
1211

13-
## API verification status
12+
That's it. There is **no submit / trigger** — Precisely doesn't publish a job-submit REST endpoint, so this component never pretends to start runs. Use Precisely's own scheduler to launch the job; use this component to surface its state in the Dagster graph.
1413

15-
| Endpoint | Verification | Override field |
16-
|---|---|---|
17-
| **Job Status** — `GET /projects/{jobRunId}/status` (plain-text response) | **Verified** against Precisely's [public REST API docs](https://help.precisely.com/r/Connect-ETL/pub/Latest/en-US/Connect-ETL-Rest-API-Reference/Job-Status) | (path is fixed) |
18-
| **Submit** — `POST /projects/{job_id}/run` with JSON `{parameters: {...}}` | **Best-guess** RESTful shape — Precisely's submit endpoint isn't in the public REST docs | `submit_path_template` |
14+
For event-driven downstream Dagster work that fires when a Precisely run completes, pair with the [`precisely_job_sensor`](../../sensors/precisely_job_sensor/).
15+
16+
## API verification status
1917

20-
If your Connect ETL install uses a different submit path, override `submit_path_template`. The status-poll path is fixed because it's the documented one.
18+
| Endpoint | Verification |
19+
|---|---|
20+
| **Job Status** — `GET /projects/{jobRunId}/status` (plain-text response) | **Verified** against Precisely's [public REST API docs](https://help.precisely.com/r/Connect-ETL/pub/Latest/en-US/Connect-ETL-Rest-API-Reference/Job-Status) |
2121

2222
## Connection: env vars OR resource
2323

@@ -47,7 +47,7 @@ resource_key: precisely
4747
| Field | Type | Description |
4848
|---|---|---|
4949
| `asset_key` | `str` | Dagster asset key (e.g. 'precisely/etl/load_customers') |
50-
| `job_id` | `str` | Precisely Connect job ID. Find in Precisely UI: Jobs → Job Details → ID. |
50+
| `job_run_id` | `str` | Precisely Connect ETL job-run ID to observe. Use the run-id from your Precisely-side scheduler / UI. |
5151

5252
### Connection
5353

@@ -57,52 +57,16 @@ resource_key: precisely
5757
| `api_token_env_var` | `str` | — | Env var with Precisely API token |
5858
| `resource_key` | `str` | — | Key of a PreciselyResource |
5959

60-
### Execution
61-
62-
| Field | Type | Default | Description |
63-
|---|---|---|---|
64-
| `parameters` | `dict` | — | Job parameters passed at runtime |
65-
| `poll_interval_seconds` | `float` | `10.0` | Seconds between status polls |
66-
| `timeout_seconds` | `int` | `3600` | Max seconds to wait for job completion |
67-
| `submit_path_template` | `str` | `"/projects/{job_id}/run"` | Template for the job-submit endpoint when not using a resource. Default is a best-guess RESTful shape; replace if your Connect ETL install uses a different path. Available placeholder: {job_id}. The status-poll path is fixed at /projects/{jobRunId}/status per Precisely's documented Connect ETL Job Status endpoint. |
68-
6960
### Catalog metadata
7061

7162
| Field | Type | Default | Description |
7263
|---|---|---|---|
7364
| `group_name` | `str` | `"precisely"` | Dagster asset group name |
74-
| `deps` | `list[str]` | — | Upstream asset keys this asset depends on (e.g. ['raw_orders', 'schema/asset']) |
65+
| `deps` | `List[str]` | — | Upstream asset keys this asset depends on |
7566
| `description` | `str` | — | Asset description shown in the Dagster catalog. |
7667
| `owners` | `List[str]` | — | Asset owners — team names ('team:analytics') or email addresses. |
7768
| `asset_tags` | `Dict[str, str]` | — | Additional key-value tags applied to the asset in the Dagster catalog. |
78-
| `kinds` | `List[str]` | — | Asset kinds for the catalog (e.g. ['snowflake', 'python']). Auto-inferred from component name when unset. |
79-
| `column_lineage` | `Dict[str, List[str]]` | — | Column-level lineage: output column → list of upstream columns it derives from, e.g. {'revenue': ['price', 'quantity']}. |
80-
81-
### Freshness
82-
83-
| Field | Type | Default | Description |
84-
|---|---|---|---|
85-
| `freshness_max_lag_minutes` | `int` | — | Maximum acceptable lag in minutes before the asset is considered stale. Builds a FreshnessPolicy when set. |
86-
| `freshness_cron` | `str` | — | Cron schedule string for the freshness policy, e.g. '0 9 * * 1-5' (weekdays 9am). |
87-
88-
### Partitions
89-
90-
| Field | Type | Default | Description |
91-
|---|---|---|---|
92-
| `partition_type` | `str` | — | Partition type: 'daily', 'weekly', 'monthly', 'hourly', 'static', 'multi', or None for unpartitioned. |
93-
| `partition_start` | `str` | — | Partition start date in ISO format (e.g. '2024-01-01'). Required for time-based partition types. |
94-
| `partition_date_column` | `str` | — | Column used to filter the upstream DataFrame to the current date partition key. |
95-
| `partition_values` | `str` | — | Comma-separated values for static or multi partitioning, e.g. 'acme,globex,initech'. |
96-
| `partition_static_dim` | `str` | — | Dimension name for the static axis in multi-partitioning, e.g. 'customer'. |
97-
| `partition_static_column` | `str` | — | Column used to filter the upstream DataFrame to the current static partition value. |
98-
99-
### Retry policy
100-
101-
| Field | Type | Default | Description |
102-
|---|---|---|---|
103-
| `retry_policy_max_retries` | `int` | — | Max retries on failure. Defines a RetryPolicy when set. |
104-
| `retry_policy_delay_seconds` | `int` | — | Seconds between retries (default 1). |
105-
| `retry_policy_backoff` | `str` | `"exponential"` | Backoff strategy: 'linear' or 'exponential'. |
69+
| `kinds` | `List[str]` | — | Asset kinds for the catalog (e.g. ['precisely']). Auto-inferred when unset. |
10670

10771
<!-- FIELDS:END -->
10872

@@ -114,87 +78,43 @@ resource_key: precisely
11478
type: dagster_component_templates.PreciselyRunAssetComponent
11579
attributes:
11680
asset_key: precisely/etl/load_customers
117-
job_id: "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
81+
job_run_id: "abc-123-def-456"
11882
host_env_var: PRECISELY_HOST
11983
api_token_env_var: PRECISELY_API_TOKEN
12084
```
12185

122-
### With job parameters + retry policy + deps
86+
### Wired into downstream Dagster work
12387

12488
```yaml
12589
type: dagster_component_templates.PreciselyRunAssetComponent
12690
attributes:
12791
asset_key: precisely/etl/load_orders
128-
job_id: "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
129-
host_env_var: PRECISELY_HOST
130-
api_token_env_var: PRECISELY_API_TOKEN
131-
parameters:
132-
inputPath: /data/inbound/orders
133-
outputSchema: analytics
134-
poll_interval_seconds: 15
135-
timeout_seconds: 7200
136-
retry_policy_max_retries: 3
137-
retry_policy_backoff: exponential
138-
deps:
139-
- raw_orders
140-
- staging/customers
141-
```
142-
143-
### Daily-partitioned ETL with freshness policy
144-
145-
```yaml
146-
type: dagster_component_templates.PreciselyRunAssetComponent
147-
attributes:
148-
asset_key: precisely/etl/daily_orders
149-
job_id: "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
92+
job_run_id: "abc-123-def-456"
15093
host_env_var: PRECISELY_HOST
15194
api_token_env_var: PRECISELY_API_TOKEN
152-
partition_type: daily
153-
partition_start: "2024-01-01"
154-
partition_date_column: ORDER_DATE
155-
parameters:
156-
runDate: "{{ partition_key }}"
157-
freshness_max_lag_minutes: 90
158-
freshness_cron: "0 9 * * 1-5" # weekday 9am, must be fresh-within-90m
15995
owners:
16096
- team:integration-eng
16197
- data-eng@company.com
98+
deps:
99+
- raw/upstream_signal
162100
```
163101

164-
### Custom submit endpoint
102+
Downstream assets then declare `deps: [precisely/etl/load_orders]` to surface lineage through the Precisely step.
165103

166-
```yaml
167-
type: dagster_component_templates.PreciselyRunAssetComponent
168-
attributes:
169-
asset_key: precisely/etl/load_legacy
170-
job_id: "legacy-job-42"
171-
host_env_var: PRECISELY_HOST
172-
api_token_env_var: PRECISELY_API_TOKEN
173-
submit_path_template: "/api/v2/jobs/{job_id}/execute" # validate vs your install
174-
```
175-
176-
## Materialization metadata
104+
## Observation metadata
177105

178-
On success, the asset records the following in the Dagster catalog:
106+
Each observation records:
179107

180108
| Key | Description |
181109
|---|---|
182-
| `job_id` | Precisely Connect job ID |
183-
| `run_id` | Precisely run ID returned by the submit endpoint |
184-
| `status` | Final terminal status (`COMPLETED` / `COMPLETED_WITH_WARNINGS`) |
110+
| `job_run_id` | The Precisely run-id polled |
111+
| `status` | Current status (`WAITING` / `RUNNING` / `COMPLETED` / `COMPLETED_WITH_WARNINGS` / `COMPLETED_WITH_ERRORS` / `CANCELLED` / `ERRORED` / `LOST_CONTACT`) |
112+
| `is_terminal` | `True` if status is one of the terminal states |
113+
| `is_success` | `True` if status is `COMPLETED` or `COMPLETED_WITH_WARNINGS` |
114+
| `is_in_progress` | `True` if status is `WAITING` or `RUNNING` |
115+
| `precisely_host` | The Precisely host the status was polled from |
185116

186-
## Failure semantics
187-
188-
The asset raises (failing the materialization) if Precisely returns any of these terminal statuses:
189-
190-
| Status | Meaning |
191-
|---|---|
192-
| `COMPLETED_WITH_ERRORS` | Job ran but produced errors |
193-
| `CANCELLED` | Run was cancelled |
194-
| `ERRORED` | Run failed |
195-
| `LOST_CONTACT` | Connect ETL lost contact with the runtime |
196-
197-
A timeout (`timeout_seconds` elapsed without reaching a terminal status) also raises. Combine with `retry_policy_max_retries` for fault tolerance.
117+
The component does **not** raise on `COMPLETED_WITH_ERRORS` / `ERRORED` / `CANCELLED` / `LOST_CONTACT` — observation is reporting, not gating. If you want a failed Precisely run to fail a downstream Dagster job, use the [`precisely_job_sensor`](../../sensors/precisely_job_sensor/) (which only fires `RunRequest` on terminal SUCCESS) and watch the Dagster job for skips.
198118

199119
## Lineage
200120

@@ -206,16 +126,6 @@ deps:
206126
- staging/customers # path-prefixed asset key
207127
```
208128

209-
For column-level lineage in the catalog, set `column_lineage`:
210-
211-
```yaml
212-
column_lineage:
213-
customer_revenue:
214-
- orders.total
215-
- orders.discount
216-
- customers.id
217-
```
218-
219129
Dependencies can also be wired externally via `map_resolved_asset_specs()` in `definitions.py`.
220130

221131
## Requirements
@@ -227,4 +137,4 @@ requests>=2.28.0
227137
## See also
228138
229139
- [`precisely_job_sensor`](../../sensors/precisely_job_sensor/) — fire a Dagster job when a Precisely run reaches terminal success
230-
- [`precisely_validation.md`](https://github.com/eric-thomas-dagster/dagster-community-components-cli/blob/main/examples/precisely_validation.md) — what's verified against Precisely's public REST docs and what's best-guess
140+
- [`precisely_validation.md`](https://github.com/eric-thomas-dagster/dagster-community-components-cli/blob/main/examples/precisely_validation.md) — what's verified against Precisely's public REST docs

0 commit comments

Comments
 (0)