pycatalyst 0.0.15

A schema-aware data generation and testing platform for Python

PyCatalyst

Schema-aware data generation and testing: recipes (YAML), REST API, optional UI, schema inference, and ML hooks.

Documentation

• Published (GitHub Pages): optophi.github.io/pycatalyst — start with Getting Started → Quick Start.

• Local docs (from repo root):

  pip install "pycatalyst[docs]"
  pycatalyst docs serve
  

  Opens at http://127.0.0.1:5005 by default. Static site: pycatalyst docs build → site/.

• Live API (when the server is running): http://127.0.0.1:8005/docs (Swagger).

Tutorial index in the doc site: Quick Start, API authentication, generate, recipes, infer schema, workbench.

Installation

pip install pycatalyst

API server, DB, and common extras:

pip install "pycatalyst[api,db,inference,faker]"

From source (editable):

git clone https://github.com/optophi/pycatalyst
cd pycatalyst
pip install -e ".[dev]"

Quick Start (CLI)

pycatalyst api --port 8005
pycatalyst generate --recipe path/to/recipe.yaml -n 10

See the Quick Start page in the docs for database setup, UI, and REST examples.

Streaming mock data

Python — same sequence as batch generation for a fixed seed:

from pycatalyst import GenerationEngine, SchemaBuilder

engine = GenerationEngine()
schema = SchemaBuilder("events").add_uuid("id").add_string("kind").build()
for row in engine.iter_records(schema, seed=123, limit=1000):
    process(row)

HTTP (Server-Sent Events) — POST /api/v1/stream/sse with the same field list as /generate, plus max_records and optional interval_ms. Use curl -N and a Bearer token when auth is enabled:

curl -N -H "Authorization: Bearer YOUR_JWT" -H "Content-Type: application/json" \
  -d '{"name":"x","fields":[{"name":"n","type":"int"}],"max_records":5,"seed":1}' \
  http://127.0.0.1:8005/api/v1/stream/sse

CLI — NDJSON to stdout (one JSON object per line):

pycatalyst stream ndjson --schema my.json -n 500 --seed 1

CLI — Kafka (pip install pycatalyst[kafka]):

export PYCATALYST_STREAM_KAFKA_BOOTSTRAP=localhost:9092
export PYCATALYST_STREAM_KAFKA_TOPIC=test-topic
pycatalyst stream kafka --schema my.json -n 2000 --batch 100

Server caps (optional): PYCATALYST_STREAM_MAX_RECORDS, PYCATALYST_STREAM_MAX_INTERVAL_MS.

Development

• Lint & format: ruff check . && ruff format .
• Type check: mypy src/
• Tests: pytest
• Coverage: pytest --cov=pycatalyst --cov-report=term-missing

Publishing

1. Bump version in pyproject.toml and CHANGELOG.md.
2. Create a release tag: git tag v0.1.0 && git push origin v0.1.0.
3. The GitHub Action uses PyPI Trusted Publishing; configure the publisher on PyPI for this repo, then the workflow will publish on tag push.

License

MIT