feat(qlik_compose): five components for Qlik Compose DW-automation orchestration

Ships the same shape as qlik_replicate and tm1:

- qlik_compose_resource — shared Compose REST auth (session or API-token)
- qlik_compose_workflow_trigger_job — run/stop a Compose workflow with
  optional wait_for_completion polling to terminal state
- qlik_compose_workflow_status_sensor — poll workflow state; trigger
  Dagster job on COMPLETED / ERROR / FAILED / STOPPED transitions;
  cursor-deduped on (state, last_completion_time)
- qlik_compose_workflow_metrics_ingestion — DataFrame per workflow with
  state, duration, rows_loaded, error_count
- qlik_compose_workspace — StateBackedComponent that enumerates every
  Project × Workflow × Data Mart; per-kind Fivetran-shape selectors
  (workflow_selector, data_mart_selector); materializing a Workflow
  asset runs the workflow

Also fixes README cross-links across ALL new components (qlik_replicate,
tm1, qlik_compose) — strips [foo](../../path/) markdown links to plain
`foo` names so Vercel UI's auto-linking (via /c/<id> route) works. The
relative-path form doesn't match Vercel's routing; plain names do.

Bumps version to 0.10.32.

Author: eric-thomas-dagster

assets/ingestion/qlik_compose_workflow_metrics_ingestion/README.md

@@ -0,0 +1,41 @@
+# QlikComposeWorkflowMetricsIngestionComponent
+
+Emit a DataFrame with per-workflow metrics from **Qlik Compose**. One row per workflow per materialization. Foundation for DW-build observability dashboards.
+
+Pairs with `qlik_compose_resource`.
+
+## Emitted schema
+
+| Column | Type | Notes |
+|---|---|---|
+| `project` | string | Compose project (DW) name |
+| `workflow` | string | Workflow name |
+| `state` | string | RUNNING / COMPLETED / STOPPED / ERROR / FAILED |
+| `last_run_duration_seconds` | float | Wall-clock of the last run |
+| `rows_loaded` | int | Total rows written by the workflow |
+| `error_count` | int | Non-null even at zero |
+| `last_completion_time` | string | ISO-8601 |
+| `polled_at` | float | Unix epoch seconds |
+
+## Example
+
+```yaml
+type: dagster_community_components.QlikComposeWorkflowMetricsIngestionComponent
+attributes:
+  asset_key: qlik_compose_metrics
+  projects: [FinanceDW, SalesDW]
+  group_name: qlik_observability
+  resource_key: qlik_compose_resource
+```
+
+## Chaining
+
+- `dataframe_to_snowflake` → BI dashboards
+- `dataframe_to_prometheus` → alert on error rate
+- `filter` (`error_count > 0`) → `dataframe_to_pagerduty` → alert on any errors
+
+## Related
+
+- `qlik_compose_resource`
+- `qlik_compose_workflow_trigger_job`
+- `qlik_compose_workspace`

assets/ingestion/qlik_compose_workflow_metrics_ingestion/__init__.py

@@ -0,0 +1,3 @@
+from .component import QlikComposeWorkflowMetricsIngestionComponent
+
+__all__ = ["QlikComposeWorkflowMetricsIngestionComponent"]

assets/ingestion/qlik_compose_workflow_metrics_ingestion/component.py

@@ -0,0 +1,112 @@
+"""Qlik Compose Workflow Metrics Ingestion Component.
+
+Emit a DataFrame with per-workflow metrics from Qlik Compose (state, last-run
+duration, rows loaded, error count). One row per workflow per materialization.
+Feed into BI dashboards or alert-input assets.
+"""
+from typing import List, Optional
+
+import dagster as dg
+from pydantic import Field
+
+
+class QlikComposeWorkflowMetricsIngestionComponent(dg.Component, dg.Model, dg.Resolvable):
+    """Emit per-workflow metrics from Qlik Compose.
+
+    Example:
+
+        ```yaml
+        type: dagster_community_components.QlikComposeWorkflowMetricsIngestionComponent
+        attributes:
+          asset_key: qlik_compose_metrics
+          projects: [FinanceDW, SalesDW]
+          group_name: qlik_observability
+          resource_key: qlik_compose_resource
+        ```
+    """
+
+    asset_key: str = Field(description="Asset key for the emitted DataFrame.")
+    projects: List[str] = Field(description="Compose projects to poll.")
+    workflows: Optional[List[str]] = Field(default=None, description="Optional workflow whitelist across all listed projects.")
+    group_name: Optional[str] = Field(default=None)
+    resource_key: str = Field(default="qlik_compose_resource")
+
+    def build_defs(self, context: dg.ComponentLoadContext) -> dg.Definitions:
+        _self = self
+        required_resource_keys = {self.resource_key}
+
+        @dg.asset(
+            name=_self.asset_key,
+            group_name=_self.group_name,
+            required_resource_keys=required_resource_keys,
+            compute_kind="qlik_compose",
+        )
+        def _the_asset(context: dg.AssetExecutionContext):
+            import time
+            try:
+                import pandas as pd
+                import requests
+            except ImportError as e:
+                raise Exception("pandas or requests library not installed") from e
+
+            resource = getattr(context.resources, _self.resource_key)
+            session = requests.Session()
+            session.verify = resource.verify_ssl
+            headers = resource.get_auth_headers()
+
+            login_body = resource.login_body()
+            if login_body is not None:
+                r = session.post(f"{resource.api_base}/login", json=login_body, timeout=30)
+                if r.status_code >= 300:
+                    raise Exception(f"Compose login failed: {r.status_code}")
+
+            wanted_wfs = set(_self.workflows) if _self.workflows else None
+            rows: list[dict] = []
+            polled_at = time.time()
+
+            for project in _self.projects:
+                try:
+                    lr = session.get(f"{resource.project_url(project)}/workflows", headers=headers, timeout=30)
+                    lr.raise_for_status()
+                    wf_list = (lr.json() or {}).get("workflows") or lr.json() or []
+                except Exception as e:
+                    context.log.warning(f"Failed to list workflows for {project}: {e}")
+                    continue
+
+                if isinstance(wf_list, dict):
+                    wf_list = wf_list.get("value", [])
+
+                for wf_entry in wf_list:
+                    wf_name = wf_entry.get("name") if isinstance(wf_entry, dict) else str(wf_entry)
+                    if not wf_name:
+                        continue
+                    if wanted_wfs is not None and wf_name not in wanted_wfs:
+                        continue
+                    try:
+                        dr = session.get(resource.workflow_url(project, wf_name), headers=headers, timeout=30)
+                        dr.raise_for_status()
+                        detail = dr.json() or {}
+                    except Exception as e:
+                        context.log.warning(f"Detail fetch failed for {project}/{wf_name}: {e}")
+                        continue
+
+                    wf = detail.get("workflow") or detail
+                    rows.append({
+                        "project": project,
+                        "workflow": wf_name,
+                        "state": wf.get("state"),
+                        "last_run_duration_seconds": wf.get("last_run_duration_seconds"),
+                        "rows_loaded": wf.get("rows_loaded") or wf.get("total_rows"),
+                        "error_count": wf.get("error_count") or 0,
+                        "last_completion_time": wf.get("last_completion_time") or wf.get("last_run_time"),
+                        "polled_at": polled_at,
+                    })
+
+            df = pd.DataFrame(rows)
+            context.add_output_metadata({
+                "row_count": len(df),
+                "projects_polled": len(_self.projects),
+            })
+            return df
+
+        return dg.Definitions(assets=[_the_asset])

assets/ingestion/qlik_compose_workflow_metrics_ingestion/example.yaml

@@ -0,0 +1,7 @@
+type: dagster_community_components.QlikComposeWorkflowMetricsIngestionComponent
+attributes:
+  asset_key: qlik_compose_metrics
+  projects: [FinanceDW, SalesDW]
+  group_name: qlik_observability
+  resource_key: qlik_compose_resource
+  # workflows: [FullBuildAndPopulate, IncrementalUpdate]      # optional whitelist

assets/ingestion/qlik_compose_workflow_metrics_ingestion/requirements.txt

@@ -0,0 +1,2 @@
+pandas>=2.0.0
+requests>=2.28.0

assets/ingestion/qlik_compose_workflow_metrics_ingestion/schema.json

@@ -0,0 +1,13 @@
+{
+  "component_type": "dagster_community_components.QlikComposeWorkflowMetricsIngestionComponent",
+  "description": "Emit a DataFrame with per-workflow metrics from Qlik Compose (state, last-run duration, rows loaded, error count). One row per workflow per materialization. Foundation for DW-build observability dashboards.",
+  "category": "ingestion",
+  "icon": "TableChart",
+  "attributes": {
+    "asset_key": {"type": "string", "label": "Asset Key", "required": true},
+    "projects": {"type": "array", "label": "Projects", "required": true, "description": "Compose projects to poll."},
+    "workflows": {"type": "array", "label": "Workflows", "required": false, "description": "Optional workflow whitelist across all listed projects."},
+    "group_name": {"type": "string", "label": "Group Name", "required": false},
+    "resource_key": {"type": "string", "label": "Resource Key", "required": false, "default": "qlik_compose_resource"}
+  }
+}

assets/ingestion/qlik_replicate_task_metrics_ingestion/README.md

@@ -2,7 +2,7 @@
 
 Materialize per-task metrics from **Qlik Replicate** (via Enterprise Manager REST API) as a DataFrame asset. Each materialization polls the API and emits one row per task with its current state, latency, throughput, and error count.
 
-Pairs with [`qlik_replicate_resource`](../../../resources/qlik_replicate_resource/) for auth.
+Pairs with `qlik_replicate_resource` for auth.
 
 ## When to use
 
@@ -68,7 +68,7 @@ Or use a schedule-based partitioned asset if you want to keep historical rows in
 
 ## Related
 
-- [`qlik_replicate_resource`](../../../resources/qlik_replicate_resource/) — shared auth
-- [`qlik_replicate_task_trigger_job`](../../../jobs/qlik_replicate_task_trigger_job/) — imperative task control (start/stop/reload)
-- [`qlik_replicate_task_status_sensor`](../../../sensors/qlik_replicate_task_status_sensor/) — event-drive on state transitions
-- [`dataframe_to_pagerduty`](../../sinks/dataframe_to_pagerduty/) — chain metrics → alerts on `error_count > 0` rows
+- `qlik_replicate_resource` — shared auth
+- `qlik_replicate_task_trigger_job` — imperative task control (start/stop/reload)
+- `qlik_replicate_task_status_sensor` — event-drive on state transitions
+- `dataframe_to_pagerduty` — chain metrics → alerts on `error_count > 0` rows

assets/ingestion/tm1_cube_data_ingestion/README.md

@@ -2,7 +2,7 @@
 
 Materialize a slice of a **TM1 cube** via MDX as a Dagster DataFrame asset. Turn multidimensional planning cube data into flat rows for downstream Dagster transforms, dbt models, or warehouse loads.
 
-Pairs with [`tm1_resource`](../../../resources/tm1_resource/) for auth.
+Pairs with `tm1_resource` for auth.
 
 ## Query modes
 
@@ -58,6 +58,6 @@ Common downstream patterns:
 
 ## Related
 
-- [`tm1_resource`](../../../resources/tm1_resource/)
-- [`tm1_process_trigger_job`](../../../jobs/tm1_process_trigger_job/) — run a TI process before ingesting
-- [`tm1_workspace`](../../../integrations/tm1_workspace/) — auto-emit assets per cube (workspace shape)
+- `tm1_resource`
+- `tm1_process_trigger_job` — run a TI process before ingesting
+- `tm1_workspace` — auto-emit assets per cube (workspace shape)

dagster_community_components/__init__.py

@@ -1013,6 +1013,12 @@
     "TM1ProcessStatusSensorComponent": "sensors/tm1_process_status_sensor/component.py",
     "TM1CubeDataIngestionComponent": "assets/ingestion/tm1_cube_data_ingestion/component.py",
     "TM1WorkspaceComponent": "integrations/tm1_workspace/component.py",
+    "QlikComposeResource": "resources/qlik_compose_resource/component.py",
+    "QlikComposeResourceComponent": "resources/qlik_compose_resource/component.py",
+    "QlikComposeWorkflowTriggerJobComponent": "jobs/qlik_compose_workflow_trigger_job/component.py",
+    "QlikComposeWorkflowStatusSensorComponent": "sensors/qlik_compose_workflow_status_sensor/component.py",
+    "QlikComposeWorkflowMetricsIngestionComponent": "assets/ingestion/qlik_compose_workflow_metrics_ingestion/component.py",
+    "QlikComposeWorkspaceComponent": "integrations/qlik_compose_workspace/component.py",
 }
 
 _loaded: dict[str, object] = {}

integrations/qlik_compose_workspace/README.md

@@ -0,0 +1,72 @@
+# QlikComposeWorkspaceComponent
+
+Auto-emit one Dagster asset per **Qlik Compose Workflow / Data Mart** across every Project (Data Warehouse). `StateBackedComponent` — discovery cached to disk, refreshed on explicit trigger. Materializing a Workflow asset runs the workflow to terminal state; Data Mart assets are observational.
+
+The workspace-shape peer of the low-level `qlik_compose_*` components.
+
+## Related components
+
+- `qlik_compose_resource` — shared auth
+- `qlik_compose_workflow_trigger_job` — imperative workflow control
+- `qlik_compose_workflow_status_sensor` — event-drive on state
+- `qlik_compose_workflow_metrics_ingestion` — metrics DataFrame
+
+## When to use which
+
+| Goal | Use |
+|---|---|
+| Wrap ALL Compose workflows + data marts as Dagster assets with one YAML | `qlik_compose_workspace` (this) |
+| Run a specific workflow with bespoke config from a Dagster job | `qlik_compose_workflow_trigger_job` |
+| Kick a downstream job when a workflow completes | `qlik_compose_workflow_status_sensor` |
+| Feed DW-build metrics into observability dashboards | `qlik_compose_workflow_metrics_ingestion` |
+
+## Selector shape
+
+Same as Fivetran's `connector_selector`:
+
+```yaml
+workflow_selector:
+  by_name: [FullBuildAndPopulate]
+  by_pattern: ["Incremental_*"]
+  exclude_by_name: [test_workflow]
+  exclude_by_pattern: ["*_deprecated"]
+```
+
+Same shape for `data_mart_selector`. Omit → include everything.
+
+## Example
+
+```yaml
+type: dagster_community_components.QlikComposeWorkspaceComponent
+attributes:
+  base_url_env_var: QLIK_COMPOSE_URL
+  api_token_env_var: QLIK_COMPOSE_TOKEN
+  projects: [FinanceDW, SalesDW]
+  workflow_selector:
+    by_pattern: ["FullBuild*", "Incremental*"]
+  group_name: qlik_compose
+  defs_state:
+    management_type: LOCAL_FILESYSTEM
+    refresh_if_dev: true
+```
+
+## Config reference
+
+| Field | Type | Default | Description |
+|---|---|---|---|
+| `base_url_env_var` | str | — | Env var with Compose base URL. |
+| `api_token_env_var` | str | — | Env var with API token. |
+| `username_env_var` / `password_env_var` | str | — | Session-based auth (dev). |
+| `verify_ssl` | bool | `true` | TLS verification. |
+| `projects` | list[str] | all | Whitelist of Compose projects. |
+| `workflow_selector` | object | — | Filter workflows. Fivetran-shape. |
+| `data_mart_selector` | object | — | Filter data marts. Fivetran-shape. |
+| `group_name` | str | — | Dagster asset group. |
+| `asset_key_prefix` | list[str] | `["qlik_compose"]` | Prefix; final key = `[...prefix, project, workflow|data_mart, name]`. |
+| `compute_kind` | str | `qlik_compose` | UI compute-kind label. |
+| `wait_for_completion` | bool | `true` | For Workflow assets: wait to terminal state. |
+| `poll_interval_seconds` | int | `30` | Seconds between status polls. |
+| `timeout_seconds` | int | `3600` | Deadline for wait_for_completion. |
+| `defs_state` | object | — | `LOCAL_FILESYSTEM` (dev) or `CLOUD_OBJECT_STORE` (Dagster+). |
+
+[Schema](schema.json) · [Example](example.yaml)