Cloud-agnostic · polyglot · open source

A data-pipeline framework for Java and Python that runs on GCP and AWS.

Sixteen contracts, two languages, one conformance suite every adapter must pass. Swap the cloud, keep the logic.

Culvert is a framework for pipelines defined once against a language-neutral contract set, then run on GCP or AWS by swapping adapters instead of rewriting logic. Sixteen contracts, in Java and Python. GCP today, AWS alongside it, Azure on the way.

Install it

pip install 'culvert[gcp]'

Install the core → The contract spec →

The idea01

How it works

Your pipeline code depends only on Culvert's contracts (BlobStore, Warehouse, Source, Sink, JobControlRepository and eleven more), never on a cloud SDK. Adapters implement them for each cloud, and every adapter must pass the same conformance suite.

How the layers fit

Yours Business
logic

Written once. Depends only on the contracts, never on an SDK.

→↓

The boundary Sixteen
contracts

Java interfaces and Python Protocols. One spec, either runtime.

→↓

Swappable Cloud
adapters

Each one passes the shared conformance suite, or it is not an adapter.

Contracts

Both languages, one spec

16 interfaces plus a metrics record, realised as Java interfaces and Python Protocols. Same seam, either runtime.

Execution

Java owns Beam

The Dataflow/Beam execution layer is Java. Legacy Python Beam is not carried forward; the two languages deliberately don't do the same job.

Transform

dbt, reused

Transformation is SQL and macros, not "Java" or "Python". Packaged once; there is deliberately no Java transform module.

Local-first

Runs on your laptop

The whole stack runs against emulators. Develop and test without a cloud account.

Install02

Install the core, add the clouds you need

One package, extras for the adapters and roles you want. Core installs with no cloud SDKs at all.

shellPython 3.10 or newer

# core contracts only, no cloud SDKs
pip install culvert
# + BigQuery, GCS, Pub/Sub, Secret Manager, observability
pip install 'culvert[gcp]'
# + Airflow-side DAG factory / operators / sensors, or dbt
pip install 'culvert[orchestration]'
pip install 'culvert[transform]'
# + StageClaim on PostgreSQL
pip install 'culvert[postgres]'
# test fixtures, and the conformance suite for your own adapters
pip install 'culvert[tester]'
pip install 'culvert[contract-tests]'
# gcp + orchestration + transform
pip install 'culvert[all]'

pythonauto-discovery

from data_pipeline_core import autoconfig
blob_store = autoconfig.discover().first("blob_store")
# -> GcsBlobStore, discovered via entry points when culvert[gcp] is installed

Contracts03

What each cloud implements

The core contracts and the adapters that satisfy them today. An adapter is only listed once it passes the shared conformance suite.

Contract	GCP	AWS	Azure
`BlobStore`	GCS	S3	skeleton
`Warehouse`	BigQuery	Athena	roadmap
`Source` · `Sink`	Pub/Sub	SQS	roadmap
`JobControlRepository`	BigQuery	DynamoDB · Athena	roadmap
Observability	Cloud Monitoring	CloudWatch	roadmap
Secrets	Secret Manager	Secrets Manager	roadmap
Eleven more, plus a metrics record	Full implementation	Java adapter family	roadmap

The authoritative list is the spec itself: CONTRACT.md →. Adapters are discovered through entry points, so installing an extra is all the wiring there is.

Libraries04

Every library, and where it stands

The Java side ships as separate artifacts under com.enrichmeai.culvert on Maven Central; the Python side ships as one culvert package with the extras above. This table is checked against both registries every day.

Maven artifact	Area	What it is	Released
`data-pipeline-core`	Core	The cloud-neutral kernel: every contract, the shared records, governance policies and auto-discovery.	0.6.1
`data-pipeline-contract-tests`	Core	The conformance suite. Abstract JUnit tests every adapter must pass.	0.6.1
`data-pipeline-tester`	Core	Mockito fixture builders for the contracts, for your own unit tests.	0.6.1
`data-pipeline-it-support`	Core	Testcontainers emulators (BigQuery, GCS) for integration tests.	0.6.1
`data-pipeline-orchestration`	Core	A cloud-neutral DAG model, rendered to Airflow, Composer or other substrates.	0.6.1
`data-pipeline-console`	Core	Read-only operator views over `JobControlRepository`, and the `culvert` command line. No cloud SDK.	0.6.1
`data-pipeline-postgres`	PostgreSQL	Cloud SQL, RDS or self-hosted: `JobControlRepository`, plus the optional `StageClaim` and `InputReadiness`.	0.6.1
`data-pipeline-gcp-gcs`	GCP	Cloud Storage: `BlobStore`.	0.6.1
`data-pipeline-gcp-bigquery`	GCP	BigQuery: `Warehouse`, `JobControlRepository`, FinOps and audit sinks.	0.6.1
`data-pipeline-gcp-pubsub`	GCP	Pub/Sub: `Source` and `Sink`.	0.6.1
`data-pipeline-gcp-secrets`	GCP	Secret Manager: `SecretProvider`.	0.6.1
`data-pipeline-gcp-observability`	GCP	Cloud Trace, Cloud Monitoring and Data Catalog lineage.	0.6.1
`data-pipeline-gcp-dataflow`	GCP	Dataflow: the Beam execution layer (`Pipeline`).	0.6.1
`data-pipeline-aws-s3`	AWS	S3: `BlobStore`.	0.6.1
`data-pipeline-aws-athena`	AWS	Athena: `Warehouse` and `JobControlRepository`.	0.6.1
`data-pipeline-aws-dynamodb`	AWS	DynamoDB: `JobControlRepository`.	0.6.1
`data-pipeline-aws-sqs`	AWS	SQS: `Source` and `Sink`.	0.6.1
`data-pipeline-aws-secrets`	AWS	Secrets Manager: `SecretProvider`.	0.6.1
`data-pipeline-aws-cloudwatch`	AWS	CloudWatch: observability and stage metrics.	0.6.1
`data-pipeline-azure-blob`	Azure	Blob Storage `BlobStore`, a skeleton: it proves the seam, and its methods throw until it is built out.	0.6.1
`data-pipeline-pdf-text`	Documents	PDF to text on PDFBox 3.0.7: `PdfTextExtract.extract(byte[])` returns `null` instead of throwing, for a scanned PDF or one it cannot read. No cloud SDK, so an application and a Dataflow job run the same extraction.	0.6.1

Add it with Maven: com.enrichmeai.culvert:<artifact>:0.6.1. Source for each: data-pipeline-libraries-java →

What's new05

What each release added

From the changelog, release by release. The full record, including every fix, is CHANGELOG.md →

0.6.1 Java and Python, on Maven Central and PyPI

Fixed: data-pipeline-pdf-text (Java only): a PDF that uses public-key (certificate) encryption now returns null, like any other PDF it cannot read. In the previous release it escaped as an error. The library still does not decrypt such a PDF.
What it cannot do: Other errors from the PDF library, such as OutOfMemoryError or StackOverflowError, still escape. Nothing in this release is breaking, and no library is new.

0.6.0 Java and Python, on Maven Central and PyPI

New library: data-pipeline-pdf-text (Java only): PdfTextExtract.extract(byte[]) turns a PDF into text with PDFBox 3.0.7. It returns null instead of throwing, for a scanned PDF with no text layer and for one it cannot read: corrupt, truncated, not a PDF, or protected by a user password. Its only dependencies are PDFBox and SLF4J.
Schema files: A STRING field can list the values it may hold, such as allowed: [A, I, C], in Java and Python. Any other value fails validation, and the ODP loader quarantines the row. A blank value in an optional field passes.
What it cannot do: In this release, a PDF that uses public-key encryption escaped the PDF library as an error, because PDFBox's BouncyCastle dependency is optional. The next release fixed it. Nothing in this release is breaking.

0.5.0 Java and Python, on Maven Central and PyPI

Schema files: An entity's layout can live in a YAML or JSON file instead of code: its columns in order, and each field's type, whether it is required, its classification, masking and range. Java and Python load it with the same rules and refuse a bad file naming the entry. The ODP loader takes --schemaUri, so any entity loads.
Python: StageClaim on PostgreSQL, through culvert[postgres], on the same tables as Java; no test yet runs a Java and a Python claimant against each other. A BigQuery AuditEventPublisher, with the audit contract suite. The whole BlobStore contract on GcsBlobStore: streamed reads pinned to one object generation, resumable writes and server-side copy. Stage policy decorators @masked, @quality_check and @governed. And run(source, transforms, sink), with a GCS-to-BigQuery quickstart in 14 lines.
Java: Readiness as a gate: a task can wait until every input it expects for the period is ready, as well as for other stages. The Airflow and Composer renderers emit the check, and the pod and Cloud Run renderer refuses such a task.
Changed: The ODP loader reads whole-number, decimal and true/false cells as values before validating them. A blank cell in a required field is now missing, so rows that loaded before can be quarantined. data-pipeline-core gains SnakeYAML (Java) and PyYAML (Python).
Status: Released on Maven Central (all 20 Java libraries) and on PyPI. The proof harness can now drive Cloud Composer, and the changelog says the real GCP run is still to come.

0.4.0 Java and Python, on Maven Central and PyPI

PostgreSQL: A new library, data-pipeline-postgres: JobControlRepository in plain JDBC, the same on Cloud SQL, RDS or your own server. It passes the shared contract tests unchanged. Each transition locks the run's opening row, so writers racing the same transition are serialised, which the BigQuery, Athena and DynamoDB backends cannot do.
Added: StageClaim: one stage of one unit for one period cannot start twice. A second claimant waits, or is told the stage is held or already done. When a claimant's database session ends, its claim is released. There is no lease, so a claimant that hangs while still connected keeps its claim. Optional, on PostgreSQL only, and Java only for now.
Added: InputReadiness: whether every input a unit expects for a period is produced and validated. It names each input that is missing, failed or still running, rather than counting rows. A failure is replaced only by a later attempt that names it as a retry, and the input is ready once that attempt validates; a later success that does not name it leaves the input failed. On PostgreSQL only, and no gate uses it yet.
Added: Orchestration: a task can list the stages that must be done before it runs. Airflow and Composer renderers given a stage gate make the task fail while any of them is not done; the pod and Cloud Run renderer refuses gated tasks. max_concurrency sets how many of a DAG's tasks may run at once (1 for one at a time), as a ceiling the scheduler can hold lower.
Added: data-pipeline-console: read-only operator views over job control (ConsoleReadService), and a culvert command line with runs, run, entities, failures and adapters. It never writes, and its build fails if a cloud SDK reaches it.
Status: The PostgreSQL code is tested against a real PostgreSQL 16 server. The changelog records no run of it in a deployment. Python has no changes in this release; it moves to 0.4.0 so the two stay on one version.

0.3.0 Java and Python, on Maven Central and PyPI

Python: PyPI goes from 0.1.1 straight to 0.3.0, because Python was not published at 0.2.0. So 0.3.0 also brings 0.2.0's Python changes, including the breaking one: load_from_uri takes a required LoadOptions. Upgrading →
Added: BigQueryWarehouse.merge() works in Java and Python. It upserts with one MERGE, takes the columns from the target table, and refuses a key that is not a target column before anything runs. It used to throw. Athena declares merge unsupported and throws.
Added: Python BlobStore.head(uri), matching Java's field for field: size, version token, last-modified and metadata without reading the object. Implemented for GCS.
Added: Java contract tests for Source and Sink, observability, lineage, FinOps and audit. The Pub/Sub, SQS, Cloud Trace, Cloud Monitoring, CloudWatch, Data Catalog and BigQuery adapters pass them.

0.2.0 Java, on Maven Central

Breaking: Warehouse.loadFromUri takes a required LoadOptions. A defaulted write disposition is what let a re-run extract silently double its data, so there is deliberately no default. LoadOptions.append() keeps the 0.1 behaviour. Backends that cannot honour a disposition now throw instead of appending. Upgrading →
Added: BlobStore.head(uri): size, version token, last-modified and metadata without reading the object, for loaders that must be idempotent by object version. Implemented for GCS and S3.
Added: NoOpJobControlRepository, for deployments whose run state an external orchestrator writes. Registered explicitly, never discovered.
Added: The orchestration substrate (Composer 2 with GKE pods, Composer 3, or Cloud Run jobs) is chosen by configuration rather than hard-coded.
Fixed: A failed load can no longer be reported as succeeded: reconciliation runs before the target is written, and a later row cannot hide a recorded failure.
Licence: The first release under Apache 2.0. Versions published before it remain MIT.

Status06

What's real, stated plainly

What has shipped, stated plainly.

Python: Released: culvert 0.6.1 on PyPI live
Java: Released: com.enrichmeai.culvert:* 0.6.1 on Maven Central live
Clouds: GCP: full implementation live · AWS: Java adapter family S3 · Athena · SQS · DynamoDB · Secrets Manager · CloudWatch · Azure skeleton
Proof: Ran end-to-end on a real GCP project (Cloud Run to BigQuery, event-driven) before release. It caught eight production-only bugs every local test had passed.
Licence: Apache 2.0 for the releases above, in both languages. The 0.1 releases, Java and Python, were published under MIT.

The book07

The book

A book on designing Culvert: twenty-one chapters, in progress. It covers where the contract boundary sits, why Java and Python do different jobs, a review of the author's own code, and building with AI agents held to the same contracts and gates. It publishes once the 0.x line settles.

Every technical claim cites the real source tree.

Essay · starting soon

Why is there no Spring for data pipelines?

Why pipeline logic ends up tied to one cloud's SDK.

Essay · starting soon

Green on emulators, then the first real deploy failed.

What emulators cannot show you, written down as it happened, and why that deploy became the release gate.

Follow along at github.com/enrichmeai/culvert →.

FAQ08

Does this replace Beam, Airflow or dbt?

No. They are engines and Culvert is a framework around them. Beam runs under the Java execution layer, Airflow is wired through the orchestration extra, and dbt is packaged and reused rather than reimplemented.

Why not one language?

Because they do different jobs. Java owns Beam and the execution layer; Python owns the contracts-first pipeline code most teams write. Legacy Python Beam is deliberately not carried forward.

Do I need a cloud account to start?

No. The whole stack runs against emulators on your laptop.

Which clouds does it run on?

Every adapter has to pass the same suite, so a contract behaves the same on either cloud or it is not shipped as an adapter. GCP is fully implemented; AWS has the Java adapter family; Azure has a Blob Storage skeleton and is otherwise roadmap.

Is the 0.x line safe to build on?

The first release shipped only after running end-to-end on real infrastructure, which caught eight bugs the local tests never saw. It is a 0.x line, so expect the contracts to firm up before 1.0.

Does Culvert have anything to do with the AI work?

Not in the code. There is no AI in Culvert. It follows the same release practice: contracts, a conformance suite, and a real deploy before release. The register →