Небольшие workflow

Три сценария для SDK 0.3, schema_version: 2, с явными входами и выходами. Сохраните YAML-файлы в definitions/ своего приложения.

Проверить доступность API

HTTP-запрос → проверка статуса → результат для вызывающего. Вход — url, выходы — status и url. Установите блоки http и validate; вместо внешнего сайта при проверке использован локальный HTTP-сервер.

YAMLdefinitions/http_status.yaml

schema_version: 2
name: Проверить доступность API
input_schema:
  type: object
  additionalProperties: false
  properties:
    url:
      type: string
      minLength: 1
  required:
  - url
output_schema:
  type: object
  additionalProperties: false
  properties:
    status:
      type: integer
    url:
      type: string
  required:
  - status
  - url
steps:
- id: fetch
  block: http
  params:
    url: '{inputs.url}'
    method: GET
- id: check
  block: validate
  params:
    value: '{fetch.status_code}'
    message: API не вернул 200
    operation: check
    op: equals
    expected: 200
outputs:
  status: '{fetch.status_code}'
  url: '{inputs.url}'

Из YAML в граф

Загрузка графа…

Выбрать активные записи

data.filter выбирает активные записи, validate проверяет, что результат непустой. Вход — records, выходы — items и count. Установите блоки data и validate.

YAMLdefinitions/active_records.yaml

schema_version: 2
name: Выбрать активные записи
input_schema:
  type: object
  additionalProperties: false
  properties:
    records:
      type: array
      items:
        type: object
  required:
  - records
output_schema:
  type: object
  additionalProperties: false
  properties:
    items:
      type: array
      items:
        type: object
    count:
      type: integer
  required:
  - items
  - count
steps:
- id: filter
  block: data
  params:
    operation: filter
    items: '{inputs.records}'
    path: /active
    op: equals
    expected: true
- id: check
  block: validate
  params:
    value: '{filter.count}'
    message: Нет активных записей
    operation: check
    op: gte
    expected: 1
outputs:
  items: '{filter.items}'
  count: '{filter.count}'

Из YAML в граф

Загрузка графа…

Результат операции доступен по явным полям {filter.items} и {filter.count}.

Переиспользовать проверку текста

Один компонент проверяет строку; вызывающий workflow использует его для заголовка и тела сообщения. Компонент можно вызвать из другого процесса без копирования его шагов.

YAMLdefinitions/require_text.yaml

schema_version: 2
name: Проверить непустой текст
input_schema:
  type: object
  additionalProperties: false
  properties:
    text:
      type: string
  required:
  - text
output_schema:
  type: object
  additionalProperties: false
  properties:
    text:
      type: string
      minLength: 1
  required:
  - text
steps:
- id: check
  block: validate
  params:
    value: '{inputs.text}'
    operation: check
    op: min_length
    expected: 1
outputs:
  text: '{check.value}'

Из YAML в граф

Загрузка графа…

YAMLdefinitions/message.yaml

schema_version: 2
name: Проверить заголовок и текст сообщения
input_schema:
  type: object
  additionalProperties: false
  properties:
    subject:
      type: string
    body:
      type: string
  required:
  - subject
  - body
output_schema:
  type: object
  additionalProperties: false
  properties:
    subject:
      type: string
      minLength: 1
    body:
      type: string
      minLength: 1
  required:
  - subject
  - body
steps:
- id: subject
  inputs:
    text: '{inputs.subject}'
  call: require_text
- id: body
  inputs:
    text: '{inputs.body}'
  call: require_text
outputs:
  subject: '{subject.text}'
  body: '{body.text}'

Из YAML в граф

Загрузка графа…

Запуск

После установки SDK файлы запускаются одинаково:

PYTHONЗапуск небольших workflow

from workflow_platform import DirectorySource, WorkflowEngine

engine = WorkflowEngine(DirectorySource("./definitions"), secrets={})

# Передайте адрес вашего тестового API.
print(engine.run("http_status", {"url": "http://127.0.0.1:8080/health"}))

print(engine.run("active_records", {"records": [
    {"id": 1, "active": True},
    {"id": 2, "active": False},
]}))
# {"items": [{"id": 1, "active": True}], "count": 1}

print(engine.run("message", {"subject": "Заявка", "body": "Новый запрос"}))
# {"subject": "Заявка", "body": "Новый запрос"}

Первый вызов требует доступного тестового сервера по переданному адресу. Это три независимых запуска: их можно выполнять по одному. Пустой результат фильтра, пустая строка или неожиданный HTTP-статус дают ошибку, а не успешный результат.

Примеры выполняются без базы платформы и без публикации данных. Согласования и устойчивые очереди требуют серверного режима.