Skip to content

Commit dbd2868

Browse files
eric-thomas-dagstereric-thomas-dagster
committed
feat(jde): five components for JD Edwards Orchestrator
Ships the same shape as tm1 / qlik_compose / qlik_replicate: - jde_orchestrator_resource — shared JDE AIS REST auth (HTTP Basic) - jde_orchestration_trigger_job — trigger an orchestration; supports sync (POST returns result) + async (POST returns jobId, poll status) with input parameter overrides - jde_orchestration_status_sensor — poll last-run status via /history?limit=1; trigger Dagster job on SUCCESS / COMPLETED / FAILED transitions; cursor-deduped on (state, jobId) - jde_orchestration_output_ingestion — run an orchestration + emit its row-list output as a DataFrame; handles both flat and nested (e.g. ServiceRequest1.RowSet) response shapes via output_field path - jde_orchestrator_workspace — StateBackedComponent that enumerates every orchestration; per-orchestration selector (Fivetran shape); materializing an asset executes the underlying orchestration Bumps version to 0.10.33.
1 parent e195f1e commit dbd2868

33 files changed

Lines changed: 1114 additions & 2 deletions

File tree

‎assets/ingestion/jde_orchestration_output_ingestion/README.md‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
# JDEOrchestrationOutputIngestionComponent
2+
3+
Execute a **JDE orchestration** and materialize its row-list output as a Dagster DataFrame. Common pattern: orchestrations that wrap JDE Data Services (table reads) or Business Function calls returning tabular results.
4+
5+
Pairs with `jde_orchestrator_resource`.
6+
7+
## Row extraction
8+
9+
JDE orchestration responses vary — some return flat top-level lists, others nest under `ServiceRequest1.RowSet`, `result`, etc. Use `output_field` (dotted path) to point at the right key. If unset, the component walks the response looking for the first list-of-dicts it can find.
10+
11+
## Example
12+
13+
```yaml
14+
type: dagster_community_components.JDEOrchestrationOutputIngestionComponent
15+
attributes:
16+
asset_key: open_ap_invoices
17+
orchestration: JDE_Fetch_Open_APs
18+
inputs:
19+
CompanyCode: "00001"
20+
AsOfDate: "2026-07-10"
21+
output_field: "ServiceRequest1.RowSet"
22+
resource_key: jde_orchestrator_resource
23+
```
24+
25+
## Chaining
26+
27+
- `dataframe_to_snowflake` — land JDE data in a warehouse for BI
28+
- `filter` + `summarize` — analytics on JDE data
29+
- `dataframe_to_csv` — export for downstream teams
30+
31+
## Related
32+
33+
- `jde_orchestrator_resource`
34+
- `jde_orchestration_trigger_job` — no-return sync trigger
35+
- `jde_orchestrator_workspace` — workspace shape

‎assets/ingestion/jde_orchestration_output_ingestion/__init__.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
from .component import JDEOrchestrationOutputIngestionComponent
2+
3+
__all__ = ["JDEOrchestrationOutputIngestionComponent"]

‎assets/ingestion/jde_orchestration_output_ingestion/component.py‎

Lines changed: 116 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,116 @@
1+
"""JDE Orchestration Output Ingestion Component.
2+
3+
Execute a JDE orchestration and materialize its result set as a DataFrame.
4+
Common pattern: orchestrations that wrap JDE Data Services (table reads)
5+
or Business Function calls returning tabular results.
6+
"""
7+
from typing import Any, Dict, Optional
8+
9+
import dagster as dg
10+
from pydantic import Field
11+
12+
13+
class JDEOrchestrationOutputIngestionComponent(dg.Component, dg.Model, dg.Resolvable):
14+
"""Execute a JDE orchestration and emit its output rows as a DataFrame.
15+
16+
Example:
17+
18+
```yaml
19+
type: dagster_community_components.JDEOrchestrationOutputIngestionComponent
20+
attributes:
21+
asset_key: open_ap_invoices
22+
orchestration: JDE_Fetch_Open_APs
23+
inputs:
24+
CompanyCode: "00001"
25+
AsOfDate: "2026-07-10"
26+
output_field: RowSet # key in the response JSON that holds rows
27+
group_name: jde_finance
28+
resource_key: jde_orchestrator_resource
29+
```
30+
"""
31+
32+
asset_key: str = Field(description="Asset key for the emitted DataFrame.")
33+
orchestration: str = Field(description="Orchestration name.")
34+
inputs: Optional[Dict[str, Any]] = Field(default=None, description="Orchestration input parameter overrides.")
35+
output_field: Optional[str] = Field(
36+
default=None,
37+
description=(
38+
"Key in the response JSON that holds the row list. Common values: "
39+
"'RowSet', 'ServiceRequest1.RowSet', 'result'. Auto-detects flat "
40+
"top-level lists if unset."
41+
),
42+
)
43+
timeout_seconds: int = Field(default=300)
44+
group_name: Optional[str] = Field(default=None)
45+
resource_key: str = Field(default="jde_orchestrator_resource")
46+
47+
def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
48+
_self = self
49+
required_resource_keys = {self.resource_key}
50+
51+
@dg.asset(
52+
name=_self.asset_key,
53+
group_name=_self.group_name,
54+
required_resource_keys=required_resource_keys,
55+
compute_kind="jde",
56+
)
57+
def _the_asset(context: dg.AssetExecutionContext):
58+
try:
59+
import pandas as pd
60+
import requests
61+
except ImportError as e:
62+
raise Exception("pandas or requests library not installed") from e
63+
64+
resource = getattr(context.resources, _self.resource_key)
65+
session = requests.Session()
66+
session.verify = resource.verify_ssl
67+
headers = resource.get_auth_headers()
68+
69+
url = resource.orchestration_url(_self.orchestration)
70+
body = _self.inputs.copy() if _self.inputs else {}
71+
r = session.post(url, json=body, headers=headers, timeout=_self.timeout_seconds)
72+
if r.status_code >= 400:
73+
raise Exception(
74+
f"JDE orchestration output ingest failed: {r.status_code} {r.text[:200]}"
75+
)
76+
payload = r.json() or {}
77+
78+
# Extract rows.
79+
rows = None
80+
if _self.output_field:
81+
# Walk dot-separated path.
82+
cur: Any = payload
83+
for part in _self.output_field.split("."):
84+
if isinstance(cur, dict):
85+
cur = cur.get(part)
86+
else:
87+
cur = None
88+
break
89+
rows = cur
90+
else:
91+
# Autodetect: look for the first list-of-dicts we find.
92+
for v in payload.values():
93+
if isinstance(v, list) and v and isinstance(v[0], dict):
94+
rows = v
95+
break
96+
if isinstance(v, dict):
97+
for w in v.values():
98+
if isinstance(w, list) and w and isinstance(w[0], dict):
99+
rows = w
100+
break
101+
if rows is not None:
102+
break
103+
104+
if rows is None:
105+
context.log.warning("No row-list found in orchestration response — returning empty DataFrame.")
106+
rows = []
107+
108+
df = pd.DataFrame(rows)
109+
context.add_output_metadata({
110+
"row_count": len(df),
111+
"orchestration": _self.orchestration,
112+
"output_field_used": _self.output_field or "(autodetected)",
113+
})
114+
return df
115+
116+
return dg.Definitions(assets=[_the_asset])

‎assets/ingestion/jde_orchestration_output_ingestion/example.yaml‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
type: dagster_community_components.JDEOrchestrationOutputIngestionComponent
2+
attributes:
3+
asset_key: open_ap_invoices
4+
orchestration: JDE_Fetch_Open_APs
5+
inputs:
6+
CompanyCode: "00001"
7+
AsOfDate: "2026-07-10"
8+
output_field: "ServiceRequest1.RowSet" # dotted path; omit for autodetect
9+
group_name: jde_finance
10+
resource_key: jde_orchestrator_resource

‎assets/ingestion/jde_orchestration_output_ingestion/requirements.txt‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
pandas>=2.0.0
2+
requests>=2.28.0

‎assets/ingestion/jde_orchestration_output_ingestion/schema.json‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
{
2+
"component_type": "dagster_community_components.JDEOrchestrationOutputIngestionComponent",
3+
"description": "Execute a JDE orchestration and emit its row-list output as a Dagster DataFrame. Common pattern: orchestrations that wrap JDE Data Services (table reads) or Business Function calls returning tabular results. Foundation for JDE → warehouse pipelines.",
4+
"category": "ingestion",
5+
"icon": "TableChart",
6+
"attributes": {
7+
"asset_key": {"type": "string", "label": "Asset Key", "required": true},
8+
"orchestration": {"type": "string", "label": "Orchestration", "required": true, "description": "Orchestration name."},
9+
"inputs": {"type": "object", "label": "Input Parameters", "required": false, "description": "Orchestration input overrides."},
10+
"output_field": {"type": "string", "label": "Output Field Path", "required": false, "description": "Dotted path to the row-list in the response (e.g. ServiceRequest1.RowSet). Autodetects flat top-level lists if unset."},
11+
"timeout_seconds": {"type": "integer", "label": "Timeout (s)", "required": false, "default": 300},
12+
"group_name": {"type": "string", "label": "Group Name", "required": false},
13+
"resource_key": {"type": "string", "label": "Resource Key", "required": false, "default": "jde_orchestrator_resource"}
14+
}
15+
}

‎dagster_community_components/__init__.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1019,6 +1019,12 @@
10191019
"QlikComposeWorkflowStatusSensorComponent": "sensors/qlik_compose_workflow_status_sensor/component.py",
10201020
"QlikComposeWorkflowMetricsIngestionComponent": "assets/ingestion/qlik_compose_workflow_metrics_ingestion/component.py",
10211021
"QlikComposeWorkspaceComponent": "integrations/qlik_compose_workspace/component.py",
1022+
"JDEOrchestratorResource": "resources/jde_orchestrator_resource/component.py",
1023+
"JDEOrchestratorResourceComponent": "resources/jde_orchestrator_resource/component.py",
1024+
"JDEOrchestrationTriggerJobComponent": "jobs/jde_orchestration_trigger_job/component.py",
1025+
"JDEOrchestrationStatusSensorComponent": "sensors/jde_orchestration_status_sensor/component.py",
1026+
"JDEOrchestrationOutputIngestionComponent": "assets/ingestion/jde_orchestration_output_ingestion/component.py",
1027+
"JDEOrchestratorWorkspaceComponent": "integrations/jde_orchestrator_workspace/component.py",
10221028
}
10231029

10241030
_loaded: dict[str, object] = {}

‎integrations/jde_orchestrator_workspace/README.md‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
# JDEOrchestratorWorkspaceComponent
2+
3+
Auto-emit one Dagster asset per **JDE orchestration** by enumerating via the JDE Orchestrator REST API. `StateBackedComponent` — discovery cached to disk, refreshed on explicit trigger.
4+
5+
The workspace-shape peer of the `jde_orchestrator_*` low-level components.
6+
7+
## Example
8+
9+
```yaml
10+
type: dagster_community_components.JDEOrchestratorWorkspaceComponent
11+
attributes:
12+
base_url_env_var: JDE_AIS_URL
13+
username_env_var: JDE_USER
14+
password_env_var: JDE_PASSWORD
15+
orchestration_selector:
16+
by_pattern: ["JDE_*"]
17+
exclude_by_pattern: ["*_deprecated"]
18+
group_name: jde_orchestrations
19+
async_mode: true
20+
wait_for_completion: true
21+
defs_state:
22+
management_type: LOCAL_FILESYSTEM
23+
refresh_if_dev: true
24+
```
25+
26+
## Selector
27+
28+
Same Fivetran-shape as other vendor workspaces:
29+
30+
```yaml
31+
orchestration_selector:
32+
by_name: [JDE_AR_Recon, JDE_AP_Payment_Run]
33+
by_pattern: ["JDE_*"]
34+
exclude_by_name: [test_orch]
35+
exclude_by_pattern: ["*_deprecated"]
36+
```
37+
38+
## Related
39+
40+
- `jde_orchestrator_resource`
41+
- `jde_orchestration_trigger_job`
42+
- `jde_orchestration_status_sensor`
43+
- `jde_orchestration_output_ingestion`

‎integrations/jde_orchestrator_workspace/__init__.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
from .component import JDEOrchestratorWorkspaceComponent, OrchestrationSelector
2+
3+
__all__ = ["JDEOrchestratorWorkspaceComponent", "OrchestrationSelector"]

0 commit comments

Comments
 (0)