Release a dataset only when it meets its data contract

id: datacontract-gated-dataset-release
namespace: company.team

inputs:
  - id: source_url
    type: URI
    displayName: Source CSV
    description: The batch to release. The default is a public sample of 100 orders.
    defaults: https://huggingface.co/datasets/kestra/datasets/raw/main/csv/orders.csv

  - id: inject_bad_rows
    type: BOOL
    displayName: Inject bad rows
    description: Append rows that break the contract (duplicate id, negative price,
      wrong total, invalid email) to watch the release get blocked.
    defaults: false

  - id: notify_slack
    type: BOOL
    displayName: Notify Slack
    description: Post the result to Slack. Needs the SLACK_WEBHOOK_URL secret.
    defaults: false

variables:
  dataset: orders
  published_data: data/published/orders.csv
  published_docs: contracts/orders/index.html
  contract_kv: datacontract_orders_published

concurrency:
  limit: 1

triggers:
  - id: nightly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Release the nightly batch.
    cron: "0 4 * * *"

tasks:
  - id: download_batch
    type: io.kestra.plugin.core.http.Download
    description: Fetch the candidate batch.
    uri: "{{ inputs.source_url }}"

  - id: verify
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: The contract, the previously published contract and the batch share
      one directory.
    inputFiles:
      orders.csv: "{{ outputs.download_batch.uri }}"
      previous.yaml: "{{ kv(vars.contract_kv, errorOnMissing=false) ?? '' }}"
      datacontract.yaml: |
        apiVersion: v3.1.0
        kind: DataContract
        id: orders
        name: Orders
        version: 1.0.0
        status: active
        description:
          purpose: Order lines for revenue reporting. One row per order.
        servers:
          - server: local
            type: local
            path: ./orders.csv
            format: csv
        schema:
          - name: orders
            physicalType: table
            properties:
              - name: order_id
                logicalType: integer
                required: true
                unique: true
              - name: customer_name
                logicalType: string
                required: true
              - name: customer_email
                logicalType: string
                required: true
                logicalTypeOptions:
                  pattern: "^[^@\\s]+@[^@\\s]+\\.[^@\\s]+$"
              - name: product_id
                logicalType: integer
                required: true
              - name: price
                logicalType: number
                required: true
                logicalTypeOptions:
                  minimum: 0.01
              - name: quantity
                logicalType: integer
                required: true
                logicalTypeOptions:
                  minimum: 1
              - name: total
                logicalType: number
                required: true
            quality:
              - type: sql
                description: total equals price times quantity
                query: SELECT count(*) FROM orders WHERE abs(total - price * quantity) > 0.01
                mustBe: 0
              - type: library
                metric: rowCount
                mustBeGreaterOrEqualTo: 50
    outputFiles:
      - orders.csv
      - datacontract.yaml
      - contract.html
      - results.json
    tasks:
      - id: check_contract
        type: io.kestra.plugin.scripts.python.Script
        description: >
          Lint the contract, compare it with the last published contract, test
          the batch against it and export the contract as HTML docs. Outputs one
          verdict.
        containerImage: python:3.12-slim
        beforeCommands:
          - pip install -q --root-user-action=ignore
            "datacontract-cli[duckdb]==1.2.2"
        env:
          INJECT_BAD_ROWS: "{{ inputs.inject_bad_rows }}"
        script: |
          import json, os, yaml
          from datacontract.data_contract import DataContract

          if os.environ["INJECT_BAD_ROWS"] == "true":
              with open("orders.csv", "a") as f:
                  f.write("1,Duplicate Id,dup@example.com,3,10.00,1,10.00\n")
                  f.write("901,Negative Price,neg@example.com,4,-5.00,1,-5.00\n")
                  f.write("902,Wrong Total,total@example.com,5,10.00,2,25.00\n")
                  f.write("903,Bad Email,not-an-email,6,10.00,1,10.00\n")

          new = yaml.safe_load(open("datacontract.yaml"))
          out = {"version": new["version"], "previous_version": "", "breaking": [], "failed_checks": [],
                 "lint_errors": [], "checks": 0}

          lint = DataContract(data_contract_file="datacontract.yaml").lint()
          out["lint_errors"] = [c.reason or c.name for c in lint.checks if str(c.result) != "ResultEnum.passed"]

          if not out["lint_errors"] and os.path.getsize("previous.yaml") > 0:
              old = yaml.safe_load(open("previous.yaml"))
              out["previous_version"] = old["version"]
              result = DataContract(data_contract_file="previous.yaml").breaking(DataContract(data_contract_file="datacontract.yaml"))
              major = lambda v: int(str(v).split(".")[0])
              if result.is_breaking and major(new["version"]) <= major(old["version"]):
                  out["breaking"] = [f"{getattr(e.change_type, 'value', e.change_type)} {e.path}" for e in result.entries if str(getattr(e.level, 'value', e.level)) == "error"]

          if not out["lint_errors"] and not out["breaking"]:
              run = DataContract(data_contract_file="datacontract.yaml").test()
              out["checks"] = len(run.checks)
              out["failed_checks"] = [f"{c.name}: {c.reason}" if c.reason else c.name
                                      for c in run.checks if str(c.result) not in ("ResultEnum.passed", "ResultEnum.info")]
              open("results.json", "w").write(run.model_dump_json(indent=2))
          else:
              open("results.json", "w").write("{}")

          if not out["lint_errors"]:
              open("contract.html", "w").write(DataContract(data_contract_file="datacontract.yaml").export("html"))
          else:
              open("contract.html", "w").write("")

          out["verdict"] = ("INVALID_CONTRACT" if out["lint_errors"] else
                            "BREAKING_CHANGE" if out["breaking"] else
                            "DATA_FAILED" if out["failed_checks"] else "PASSED")
          for k in ("lint_errors", "breaking", "failed_checks"):
              out[k] = "; ".join(out[k])
          print("::" + json.dumps({"outputs": out}) + "::")

  - id: release
    type: io.kestra.plugin.core.flow.Switch
    description: Publish on PASSED. Every other verdict blocks the release and
      leaves the published data and contract as they were.
    value: "{{ outputs.check_contract.vars.verdict }}"
    cases:
      PASSED:
        - id: publish
          type: io.kestra.plugin.core.namespace.UploadFiles
          description: Publish the batch and the contract docs next to each other.
          namespace: "{{ flow.namespace }}"
          filesMap:
            "{{ vars.published_data }}": "{{ outputs.verify.outputFiles['orders.csv'] }}"
            "{{ vars.published_docs }}": "{{ outputs.verify.outputFiles['contract.html'] }}"

        - id: record_contract
          type: io.kestra.plugin.core.kv.Set
          description: The contract that was released becomes the baseline for the next
            breaking change check.
          key: "{{ vars.contract_kv }}"
          value: "{{ read(outputs.verify.outputFiles['datacontract.yaml']) }}"

        - id: log_published
          type: io.kestra.plugin.core.log.Log
          message: >-
            Released {{ vars.dataset }} under contract v{{
            outputs.check_contract.vars.version }}: {{
            outputs.check_contract.vars.checks }} checks passed. Data at {{
            vars.published_data }}, docs at {{ vars.published_docs }}.

        - id: notify_published
          type: io.kestra.plugin.core.flow.If
          condition: "{{ inputs.notify_slack }}"
          then:
            - id: slack_published
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              url: "{{ secret('SLACK_WEBHOOK_URL') }}"
              payload: |
                {
                  "text": "Released {{ vars.dataset }} under data contract v{{ outputs.check_contract.vars.version }}: {{ outputs.check_contract.vars.checks }} checks passed. Execution {{ execution.id }}."
                }
    defaults:
      - id: notify_blocked
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.notify_slack }}"
        then:
          - id: slack_blocked
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": {{ ('Release of ' ~ vars.dataset ~ ' BLOCKED (' ~ outputs.check_contract.vars.verdict ~ '). ' ~ outputs.check_contract.vars.lint_errors ~ outputs.check_contract.vars.breaking ~ outputs.check_contract.vars.failed_checks ~ ' Execution ' ~ execution.id ~ '.') | toJson }}
              }

      - id: block_release
        type: io.kestra.plugin.core.execution.Fail
        description: End the run as FAILED with the reason.
        errorMessage: >-
          {{ outputs.check_contract.vars.verdict }}: {% if
          outputs.check_contract.vars.verdict == 'INVALID_CONTRACT' %}{{
          outputs.check_contract.vars.lint_errors }} {% elseif
          outputs.check_contract.vars.verdict == 'BREAKING_CHANGE' %}contract
          v{{ outputs.check_contract.vars.version }} breaks v{{
          outputs.check_contract.vars.previous_version }} without a major
          version bump: {{ outputs.check_contract.vars.breaking }} {% else %}{{
          outputs.check_contract.vars.failed_checks }}{% endif %}