Develop
Extending ContractForge.
Extension APIs exist for reusable behavior. Use them when a capability applies to more than one dataset or project.
Extension points
| Extension | API | Use when |
|---|---|---|
| Preset | register_preset | You want reusable defaults for a contract pattern. |
| Source resolver | register_source_resolver | You need to read a new source type. |
| Write mode | register_write_mode | You need a new target write semantics. |
| Quality rule | register_quality_rule | You need a reusable rule beyond built-ins. |
| Hooks | IngestionHooks | You need controlled custom logic around ingestion. |
Source resolver example
from contractforge import SourceResolution, register_source_resolver
def resolve_custom_source(spec, spark):
df = spark.read.format("json").load(spec.path)
return SourceResolution(
dataframe=df,
metadata={"source_connector": "custom_json"},
capabilities={"batch": True},
)
register_source_resolver("custom_json", resolve_custom_source)Design rules
- Do not add dataset-specific branches to core modules.
- Return redacted operational metadata.
- Keep source resolvers focused on reading, not business shaping.
- Prefer declarative configuration over hidden conventions.
- Add tests at the lowest layer that proves the extension contract.