Skip to content

Create a data feed you can check

Move two orders from a source file into an output file, with an explicit contract at the loading boundary. Then deliberately break a record and see the contract stop the load. This small example is the starting point for a repeatable feed.

The commands and files below are verified with the published pycharter==0.0.67. You need Python 3.11+ and the dependencies installed below. No cloud account, server, registry or database is required.

1. Make a small project

Create a new directory and work inside it. This example writes accepted.jsonl in that directory and replaces it on a successful run.

mkdir orders-feed
cd orders-feed
python -m pip install "pycharter==0.0.67" "sqlalchemy[asyncio]"

The explicit sqlalchemy[asyncio] installs the greenlet dependency required by this release with SQLAlchemy 2.1. The source checkout already declares it; record-only validation does not need this workaround.

Save these two records as orders.jsonl:

{"order_id": "A-100", "amount": 19.95}
{"order_id": "A-101", "amount": 0}

2. Define what must pass

Save orders-contract.yaml:

api_version: pycharter.io/v2
kind: DataContract
metadata:
  name: orders
  version: "1.0.0"
spec:
  json_schema:
    version: "1.0.0"
    type: object
    properties:
      order_id:
        type: string
        minLength: 1
      amount:
        type: number
        minimum: 0
    required: [order_id, amount]
    additionalProperties: false

The contract requires a nonempty order ID and a nonnegative amount. Its native YAML envelope supplies an identity you can review and version. Start with the rules you need; add richer metadata and business checks as the feed grows.

3. Attach the contract to the load

Save pipeline.yaml alongside the other files:

api_version: pycharter.io/v2
kind: Pipeline
metadata:
  name: orders_feed
  version: "1.0.0"
spec:
  steps:
    - id: read_orders
      type: extract
      source:
        type: file
        path: orders.jsonl
        format: jsonl
    - id: write_orders
      type: load
      input: read_orders
      target:
        type: file
        path: accepted.jsonl
        format: jsonl
        mode: overwrite
      contract:
        ref:
          type: file
          path: orders-contract.yaml
        validate: true
        coerce: true
        on_error: fail

The contract block on write_orders is the enforcement point: records must pass before this loader writes them. A name in a description or a saved association alone would not establish enforcement.

4. Check, run and inspect

Run from the directory containing these files:

pycharter pipeline validate pipeline.yaml
pycharter pipeline run pipeline.yaml

Configuration validation checks the definition; the second command reads the source, checks the records and writes output. Inspect the actual destination:

import json
from pathlib import Path

records = [json.loads(line) for line in Path("accepted.jsonl").read_text().splitlines()]
assert len(records) == 2
assert {record["order_id"] for record in records} == {"A-100", "A-101"}
print(records)

Run the pipeline again. The output still contains two records because this example deliberately uses mode: overwrite.

5. See a failure and recover

Change the second record in orders.jsonl to have "amount": -5, then repeat:

pycharter pipeline run pipeline.yaml

The command fails at the contract check. For this example, the load does not run and the previous accepted.jsonl remains unchanged. That existing file is from the previous successful run; its presence alone does not mean the latest run succeeded. Correct the amount to 0 and run again.

You have now checked the definition, inspected real output, repeated the load and diagnosed an intentional failure. This bounded file example does not establish crash-safe publication or duplicate prevention for other connectors. Choose and verify those guarantees for the destination and deployment profile you adopt.

Continue with your own feed