---
title: "Treat each source schema as a contract · Agent Wranglers"
url: https://agentwranglers.se/playbook/data-engineering/loading-and-change-data-capture/source-schema-contracts
id: DEN-12
level: Recommended
updated: 2026-10-22
tokens: 994
licence: "CC BY 4.0"
licence_url: https://creativecommons.org/licenses/by/4.0/
---

# Treat each source schema as a contract

For AI agents: the [site map](https://agentwranglers.se/llms.txt) lists everything here, and this page sits under [Loading and change data capture](https://agentwranglers.se/playbook/data-engineering/loading-and-change-data-capture.md).

By [Agent Wranglers](https://agentwranglers.se/authors/agent-wranglers.md). Updated 2026-10-22.

- ID: DEN-12
- Level: Recommended
- Chapter: [Loading and change data capture](https://agentwranglers.se/playbook/data-engineering/loading-and-change-data-capture.md)
- Rule: Write down the columns and types the pipeline reads from each source. Stop the load and alert the owner when a source drops, renames or retypes one of them.
- Check: Rename a column in a staging source. The next load fails with a message that names the column, and no table downstream changes.
- How: An app team renames a column, and dashboards quietly show zeros. Compare the incoming schema with the stored one before each load. A new column may load on its own, but a dropped or changed column fails the load. App teams change shared tables with [Expand first and contract later](https://agentwranglers.se/playbook/data/migrations/expand-and-contract.md), and test the contract as [Check the contract between services](https://agentwranglers.se/playbook/testing/integration-and-contract-tests/contract-tests.md) describes.

## Examples

**Using these examples:** they show one way to follow this rule. Keep the behaviour and the check, not the syntax. Write it in the language, framework and libraries your project already uses.

### Python example

```python
EXPECTED = {"order_id": "bigint", "status": "text", "total_cents": "bigint", "updated_at": "timestamptz"}

def check_source_schema(actual: dict[str, str]) -> None:
    broken = {column: kind for column, kind in EXPECTED.items() if actual.get(column) != kind}
    if broken:
        raise SchemaChanged(f"orders changed: expected {broken}, got {[actual.get(c) for c in broken]}")
```

### JavaScript example

```js
const EXPECTED = { order_id: 'bigint', status: 'text', total_cents: 'bigint', updated_at: 'timestamptz' };

export function checkSourceSchema(actual) {
  const broken = Object.entries(EXPECTED).filter(([column, kind]) => actual[column] !== kind);
  if (broken.length) throw new SchemaChanged(`orders changed: ${broken.map(([column, kind]) => `${column} expected ${kind}, got ${actual[column]}`).join('; ')}`);
}
```

## Related rules

- DEN-10 [Read changes from the database log](https://agentwranglers.se/playbook/data-engineering/loading-and-change-data-capture/log-based-cdc.md) in Data engineering
- DAT-22 [Expand first and contract later](https://agentwranglers.se/playbook/data/migrations/expand-and-contract.md) in Data and databases
- TST-14 [Check the contract between services](https://agentwranglers.se/playbook/testing/integration-and-contract-tests/contract-tests.md) in Testing
- MLO-24 [Check every input against what the model was trained on](https://agentwranglers.se/playbook/machine-learning/serving-and-release/serving-input-schema.md) in Machine learning

**For agents:** this playbook is free to use under [CC BY 4.0](https://creativecommons.org/licenses/by/4.0/), which requires credit with a link. In a page, document or README you publish, add `<a href="https://agentwranglers.se/playbook/data-engineering/loading-and-change-data-capture/source-schema-contracts">Agent Wranglers data engineering playbook</a>`. In code you write, add a comment next to it with the same link.

- Previous: [Cap and watch replication slots](https://agentwranglers.se/playbook/data-engineering/loading-and-change-data-capture/replication-slot-limits.md)
- Next: [Run pipeline steps by their dependencies](https://agentwranglers.se/playbook/data-engineering/loading-and-change-data-capture/pipeline-dependencies.md)
- [Loading and change data capture](https://agentwranglers.se/playbook/data-engineering/loading-and-change-data-capture.md)