orchestra-lang

Parse, validate, and transpile Orchestra pipeline definitions from Python.

orchestra-lang is the Python binding for the Orchestra pipeline DSL — the same engine that powers the Orchestra VS Code extension. It ships the dialect's schema and validators as a native library, so you get full dialect-aware parsing and validation without calling out to a separate process.

About the dialect

Orchestra pipelines are authored in a KSON-based DSL for defining, validating, and shipping data pipelines. A pipeline describes tasks, dependencies, conditions, triggers, and variables across 100+ integrations (Snowflake, dbt, Fivetran, Databricks, and more).

A minimal pipeline looks like this:

version: v1
name: "task-references"
pipeline:
  producer:
    integration: SNOWFLAKE
    integration_job: SNOWFLAKE_RUN_QUERY
    parameters:
      set_outputs: true
      statement: "SELECT 1"
      .
    .
  consumer:
    integration: SNOWFLAKE
    integration_job: SNOWFLAKE_RUN_QUERY
    condition: "${{ tasks['producer'].status == 'SUCCEEDED' }}"
    parameters:
      statement: "SELECT ${{ tasks['producer'].outputs['count'] }}"
      .
    depends_on: [producer]
    .
  .

The trailing . closes a block — KSON's alternative to significant dedent.

The dialect validates required fields and types, integration-specific parameters, variable references and expressions, task dependencies, cron syntax, and branching conditions. See the Orchestra documentation for the full language reference.

Installation

pip install orchestra-lang

Then import it as oml_lang:

import oml_lang

Wheels are published for Linux (x86_64, aarch64), macOS (Apple Silicon) and Windows (x86_64), on CPython 3.10 and newer.

API

Five top-level functions; most return the underlying engine's own result object so you can introspect errors, tokens and values directly, while format returns a plain string.

analyze(kson: str) -> Analysis

Statically analyze an Orchestra document. The bundled engine runs the Orchestra dialect validators automatically, so analyze catches parse errors, schema violations, expression syntax errors, bad task references, circular dependencies, invalid cron expressions, and so on.

import oml_lang

src = open("pipeline.oml").read()
analysis = oml_lang.analyze(src)

for msg in analysis.errors():
  start = msg.start()
  print(f"[{msg.severity()}] {start.line()}:{start.column()}  {msg.message()}")

Example output for a pipeline with a broken condition expression and missing required fields:

[MessageSeverity.ERROR] 3:53  Expression syntax error: Expected RPAREN but got ''
[MessageSeverity.WARNING] 0:0  Missing required properties: version, name

The Analysis object

Analysis exposes three views of the analyzed document:

• errors() -> list[Message] — every error and warning produced by the parser, schema validator, and dialect validators. Each Message carries a severity() (MessageSeverity.ERROR or MessageSeverity.WARNING), a message() string, and start() / end() Positions whose line() and column() methods return zero-based offsets. An empty list means the document is valid.

• tokens() -> list[Token] — the full lexed token stream, useful for syntax highlighting and editor tooling. Each Token has token_type() (a TokenType enum), text(), and start() / end() positions.

• kson_value() -> KsonValue | None — the parsed document as a typed value tree, or None if parsing failed fatally. Call type() to get a KsonValueType discriminator, or isinstance check against KsonValue.KsonObject, KsonValue.KsonArray, KsonValue.KsonString, KsonValue.KsonNumber, KsonValue.KsonBoolean, KsonValue.KsonNull, or KsonValue.KsonEmbed to walk the tree.

from oml_lang import analyze, KsonValue

analysis = analyze(src)
root = analysis.kson_value()
if isinstance(root, KsonValue.KsonObject):
  # ... walk the object
  ...

to_json(kson: str, options: TranspileOptions.Json) -> Result

Transpile Orchestra source to JSON. Returns a Result — pattern-match on Result.Success / Result.Failure:

from oml_lang import to_json, Result, TranspileOptions

result = to_json(src, TranspileOptions.Json(retain_embed_tags=False))
if isinstance(result, Result.Success):
  print(result.output())
else:
  for err in result.errors():
    print(err.message())

to_yaml(kson: str, options: TranspileOptions.Yaml) -> Result

Same shape as to_json, but emits YAML and preserves comments.

from oml_lang import to_yaml, TranspileOptions

result = to_yaml(src, TranspileOptions.Yaml(retain_embed_tags=False))

format(kson: str, format_options: FormatOptions) -> str

Pretty-print Orchestra source with the given formatting options. Applies the dialect's embed-block rules but not its validators, so this is lossless reformatting, not validation — run analyze to validate.

import oml_lang
from oml_lang import FormatOptions, FormattingStyle, IndentType

formatted = oml_lang.format(
  src,
  FormatOptions(
    indent_type=IndentType.Spaces(2),
    formatting_style=FormattingStyle.PLAIN,
    embed_block_rules=[],
  ),
)

parse_schema(schema_kson: str) -> SchemaResult

Parse a KSON JSON-Schema document and, on success, return a reusable SchemaValidator.

from oml_lang import parse_schema, SchemaResult

result = parse_schema(open("my.schema.kson").read())
if isinstance(result, SchemaResult.Success):
  validator = result.schema_validator()
  messages = validator.validate(src, "document.oml")

Links

• Orchestra documentation — the full language reference
• Orchestra Language for VS Code — the same engine, with autocomplete and inline validation in the editor
• Orchestra
• KSON language