aniondocker/conduit-cli repository overview

# Fragile, hard-to-modify script
data = fetch_api_data()
filtered = [item for item in data if item['status'] == 'active']
processed = [transform(item) for item in filtered]
for item in processed:
    print(f"Result: {item}")

- id: conduit.RestApi
  url: "https://api.example.com/data"
- id: conduit.Filter
  condition: "input.status == 'active'"
- id: conduit.Transform
  template: "{{input | process}}"
- id: conduit.Console
  format: "Result: {{input}}"

pip install git+https://github.com/aniongithub/conduit.git

- id: conduit.Input
  data: [{message: "Hello, Conduit!"}]
- id: conduit.Console
  format: "{{input.message}}"

conduit-cli hello.yaml
# Output: Hello, Conduit!

conduit-cli serve --host 0.0.0.0 --port 8000

curl -X POST http://localhost:8000/run \
  -H "Content-Type: application/json" \
  -d '{"pipeline": [
    {"id": "conduit.Input", "data": [{"name": "World"}]},
    {"id": "conduit.Console", "format": "Hello, {{input.name}}!"}
  ]}'

# Find and analyze Python files
- id: conduit.Glob
  pattern: "**/*.py"
- id: conduit.FileInfo
- id: conduit.Console
  format: "{{input.name}}: {{input.size}} bytes"

# Get Pokemon data with custom limit
- id: conduit.RestApi
  url: "https://pokeapi.co/api/v2/pokemon?limit=${limit:-5}"
- id: conduit.JsonQuery
  query: ".results[].name"
- id: conduit.Console
  format: "Pokemon: {{input}}"

conduit-cli pokemon.yaml --args limit=10

# Parallel processing with Fork
- id: conduit.RestApi
  url: "https://api.example.com/data"
- id: conduit.Fork
  paths:
    summary:
      - id: conduit.JsonQuery
        query: ".metadata.title"
    details:
      - id: conduit.JsonQuery
        query: ".content"
      - id: conduit.Filter
        condition: "len(input) > 100"
- id: conduit.Console
  format: "{{input.summary}}: {{input.details[:50]}}..."

# Convert YAML to API request with yq
yq eval '{"pipeline": ., "args": {"limit": "10"}}' examples/pokemon_evolution.yaml -o=json | \
curl -X POST http://localhost:8000/run -H "Content-Type: application/json" -d @-

from dataclasses import dataclass
from typing import Generator, Iterator
from conduit import PipelineElement

@dataclass
class MyInput:
    value: str

class MyElement(PipelineElement):
    def process(self, input: Iterator[MyInput]) -> Generator[str, None, None]:
        for item in input:
            yield f"Processed: {item.value}"

{
  "success": true,
  "results": ["item1", "item2", "item3"],
  "stdout": ["Console output line 1", "Console output line 2"],
  "stderr": [],
  "stats": {
    "duration": 1.23,
    "total_items_processed": 3,
    "throughput": 2.44,
    "element_metrics": [...]
  }
}