App data pipeline

Fetch and pipeline.

Readers: people whose actions not only write the local DB, but also pull from a workspace data connection and then assemble a result. Related: App actions · App package layout · example skill pack backend/internal/skillfromchat/bundled/employee-profile/

---

1. When do you need a fetch pipeline?

Employee profile is the standard example: master → contacts → org → contract → pay → ID photo → assemble HTML.

---

2. Recommended code layout

logic/
  handlers.py           # only dispatch actions; write history on success/failure
  generate_xxx.py       # read pipeline, stepwise query_run, call assemble, save_upload
scripts/
  assemble_xxx.py       # pure function: data + template + mapping → HTML string
assets/
  xxx.html              # result layout
references/
  pipeline.json         # step declarations
  field-mapping.json    # how fields fill the template (optional)

Duty split:

• pipeline: declare “what to query, where params come from, which block to save the result as”
• generate_*: run and tolerate failure (optional steps, fallback queries, warnings)
• assemble_*: presentation; does not hit the data connection directly
• handlers: product action boundary and registration

---

3. Core shape of pipeline.json

{
  "version": 1,
  "id": "my-report-v1",
  "display_name": "My report",
  "input": {
    "emp_name": { "type": "string", "required": true, "label": "Employee name" }
  },
  "steps": [
    {
      "id": "resolve_employee",
      "label": "Resolve employee master",
      "query_id": "employee_by_name",
      "params": { "empName": "{{input.emp_name}}" },
      "save_as": "employee",
      "optional": false
    },
    {
      "id": "fetch_org",
      "label": "Org and job",
      "query_id": "employee_org_by_emp_id",
      "params": { "empId": "{{employee.rows[0].empId}}" },
      "save_as": "org",
      "optional": true
    }
  ],
  "assemble": {
    "script_path": "scripts/assemble_xxx.py",
    "template_path": "assets/xxx.html",
    "field_mapping_path": "references/field-mapping.json"
  }
}

3.1 Step fields

In-app orchestration scripts read these fields; keys such as tool / action matter more when skills run from conversation. Inside the app you usually call platform.query_run directly.

3.2 Relation to data connections

• The workspace must already have a data connection, and query ids must match the pipeline.
• The user must be allowed to run those queries.
• You can platform.query_list() first to see which ids exist, to avoid a hard failure.

---

4. What the orchestration script does (logic skeleton)

Pseudocode (same family as employee-profile generate_profile.py):

run_generate(params, ctx):
  validate required input
  platform = ctx["platform"]
  pipeline = read references/pipeline.json
  bindings = { input: params }
  warnings = []

  for step in pipeline.steps:
    resolve params template
    block = platform.query_run(query_id, params)
    if empty and fallback exists → query again
    if still fail and optional → warnings.append(...); block = empty
    if fail and not optional → return error
    bindings[save_as] = block

  html = assemble(bindings, template, field_mapping)
  up = platform.save_upload(filename, html, "text/html; charset=utf-8")
  return { ok, upload_id, warnings, ... }

Disambiguation: when the master has several rows, return code: "disambiguate" + candidates, so the user can fill an employee ID and submit again (the UI already supports showing this kind of error).

---

5. Assemble and field mapping

• Template: assets/*.html; placeholders or script replacement.
• field-mapping.json: describes “this report column comes from which query field”, empty values, masking, and similar (employee profile has a full sample).
• Result page from a picture: see App views §4 path A.

Assemble scripts should avoid platform-bridge calls so they are easy to unit-test and reuse on the skill side.

---

6. How handlers attach the pipeline

from generate_xxx import run_generate

def handle(action, params, ctx):
    if action == "report.generate":
        result = run_generate(params, ctx)
        # Optional: write status / upload_id into the app.db history table
        return result
    ...

UI: action_form → report.generate; history: list + entity with upload_id/status.

---

7. Must-have checklist (fetch-style apps)

• [ ] Workspace data connection works; every non-optional query_id in the pipeline exists
• [ ] logic/handlers.py action ids match app.json
• [ ] Orchestration script uses ctx["platform"]; no forbidden network libraries
• [ ] references/pipeline.json (or equivalent steps in the script) matches real query param names
• [ ] When preview is needed: save_upload + return upload_id
• [ ] Partial failure returns warnings; do not silently drop fields

---

8. App pack panel (browse and allowlisted edit)

After you open the app, “App pack” can show the in-pack file tree and preview contents.

HTTP (members):

• GET .../apps/{appId}/package/files
• GET .../apps/{appId}/package/file?rel_path=
• PUT .../apps/{appId}/package/file body { rel_path, content }

---

9. Do not put these in the pipeline