# Data Contracts Schema Governance

> Use when implementing data contracts and schema governance.

- Skill: `loopyluci/data-contracts-schema-governance` (Agent Skill)
- Install (CLI): `npx skillmds@latest add loopyluci/data-contracts-schema-governance`
- Raw SKILL.md: https://api.skillmd.com/api/skills/loopyluci/data-contracts-schema-governance/raw
- Safety review: pending (external: skill-scanner PASS, skillspector PASS)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- License: MIT
- Author: LoopyLuci (https://skillmd.com/u/loopyluci)
- Updated: 2026-09-08
- Page: https://skillmd.com/skills/loopyluci/data-contracts-schema-governance

---


# Data Contracts and Schema Governance

Implementing data contracts and schema governance — from schema registries (Avro, Protobuf) through contract testing, data lineage, and breaking change detection.

## When to Use

- Ensuring data quality between producers and consumers
- Managing schema evolution across microservices
- Detecting breaking schema changes in CI/CD
- Building data lineage and impact analysis
- Establishing data ownership and SLAs

## Data Contract Patterns

```python
DATA_CONTRACT_EXAMPLE = {
    'dataset': 'user_events',
    'owner': 'team-analytics',
    'schema_version': '1.2.0',
    'schema': {
        'type': 'record', 'name': 'UserEvent',
        'fields': [
            {'name': 'user_id', 'type': 'string'},
            {'name': 'event_type', 'type': 'string'},
            {'name': 'timestamp', 'type': 'long', 'logicalType': 'timestamp-millis'},
            {'name': 'properties', 'type': {'type': 'map', 'values': 'string'}},
        ],
    },
    'slas': {'freshness': '5min', 'completeness': '99.9%', 'volume_min': 1000},
    'consumers': ['team-product', 'team-ml'],
}

class SchemaCompatibility:
    """Check schema compatibility to prevent breaking changes."""
    
    @staticmethod
    def check_backward(original_schema: dict, new_schema: dict) -> bool:
        """Backward compatible: new reader can read old data."""
        original_fields = {f['name']: f for f in original_schema.get('fields', [])}
        new_fields = {f['name']: f for f in new_schema.get('fields', [])}
        
        for name, orig_field in original_fields.items():
            if name not in new_fields:
                return False  # Can't delete fields
            if orig_field['type'] != new_fields[name]['type']:
                # Allowed: widening type (int → long, string → union)
                if not SchemaCompatibility._is_widening(orig_field['type'], new_fields[name]['type']):
                    return False
        return True
```

## Verification Checklist

- [ ] Schema registry deployed (Confluent, Apicurio, or custom)
- [ ] Compatibility mode defined for each topic (backward, forward, full, none)
- [ ] Data contract document defined per dataset (owner, schema, SLAs, consumers)
- [ ] Breaking change detection in CI/CD pipeline
- [ ] Schema evolution documented in data contract versions
- [ ] Data lineage tracked (producer → topic → consumer)
- [ ] Data quality metrics monitored per contract