Declarative data pipelines over PostgreSQL — with SCD2 versioning, parallel builds and a full audit trail, driven from plain Python files.
A data engineer declares a dataset and the rule that computes it in one
.py file (a spec), applies it through the UI or API, and DataPulse takes
care of the rest: physical tables, row-version history, consistent
incremental recomputation, parallel execution and journaling.
# comment: TASK-101 customers from the CRM
from datapulse import spec, attr, mode, self_db
from datapulse.types import bigint, text, timestamptz
from connect import CRM # your connections, also applied as code
@spec(
dataset="raw.customer",
num=1,
descr="Customers",
parallel=2,
attrs=[
attr("customer_id", bigint, descr="Customer id",
is_primary_key=True, is_chunk=True),
attr("full_name", text, descr="Full name"),
attr("updated_at", timestamptz, descr="Changed at (source)"),
],
modes=[mode("initial", "init", True), mode("increment", "incr", True)],
)
def build(param):
with CRM() as crm, self_db() as db:
src = crm.cursor()
src.execute("select customer_id, full_name, updated_at from crm.customer")
with db.cursor() as dst:
dst.executemany(
"insert into raw.customer_1_$ "
"(customer_id, full_name, updated_at, is_active) "
"values (%s, %s, %s, true)",
src.fetchall(),
)The mapping fills its interface table with plain SQL; the engine compares the slice against the current data and merges the difference as SCD2 row versions. Consumers read the dataset view with two simple predicates:
-- state as of build P -- what changed in (from, to]
where build_id <= P where build_id > :from
and (end_build_id is null and build_id <= :to
or end_build_id >= P)
and is_active- Your code is the model; the database owns the applied state. What you
see in the catalog is your own source, byte for byte — and every applied
version of it is journalled in the warehouse (schema
core), next to the data, in the same transactions. Recreating the container loses nothing. - Atomic upgrades. One or more files apply as a single all-or-nothing batch: validation first, then DDL + audit row + code versions in one transaction. Re-applying the same structure is a hot code replace (any time, even mid-build); structural changes need the system at rest and go through a parallel run; drops and renames are marker files through the same transport.
- Honest increments. Every build freezes its sources at a watermark, so concurrent builds are invisible and increments never lose or re-read an interval — even while a source is committing its chunks. An unreviewed failure freezes its consumers until support marks it fixed.
- Parallel by declaration. A spec names its chunking attribute; the engine fans the merge out across workers with deterministic chunking. Each build runs in its own worker process — stopping one is an instant kill, never a stuck flag.
- Flows as code. A flow template (
@flowspec) is a function over the installed-spec catalog — "everything with an increment mode" stays true as specs come and go; a repeating template is a micro-batch carousel. A failed build holds only its dependent subtree; fix it and the flow picks it up.
The image carries no database of its own. The fastest route is Docker Compose, which brings DataPulse up next to a PostgreSQL 18:
git clone https://github.com/votinvv/datapulse.git
cd datapulse
docker compose up -d
(The compose file also works standalone — it pulls the published image from
Docker Hub, so grabbing the one file is enough. docker compose up -d --build
builds the image from source instead.)
- UI — http://localhost:8000/ (login
admin/admin) - API docs — http://localhost:8000/docs
- project DB —
localhost:5432, databasedatapulse(postgres/postgres)
Or point the published image at your own PostgreSQL — the whole secrets store travels as one JSON env var:
docker run -d --name datapulse -p 8000:8000 \
-e DATAPULSE_SECRETS_JSON='{"self_db": {"host": "db.example.com", "port": 5432,
"user": "postgres", "password": "secret", "dbname": "datapulse"}}' \
votinvv/datapulse:latest
The database itself must exist — DataPulse creates only its core schema
inside it. self_db is the product's own connection; credentials for your
source connections (connect.py) go into the same JSON as further sections,
where the SDK's secret() reads them.
Apply your first spec from the UI (Apply upgrade on the Overview page) or from the command line:
curl -F "file=@raw.customer_1.py" http://localhost:8000/api/upgrade
Spec authors install the SDK for IDE autocompletion and local runs; the install brings the source drivers along (psycopg, oracledb — imported lazily), so a clean machine can run a mapping right away:
pip install votinvv-datapulse
from datapulse import spec, attr, mode, flowspec, self_db, secret — plus
datapulse.dev.run_local(build) to execute a mapping straight from the IDE.
The suite runs end-to-end inside a clean compose stack:
docker compose up -d --build
docker compose cp tests datapulse:/app/
docker compose exec -T datapulse python3 /app/tests/run_all.py
docker compose down -v
- Requirements — what the product does, and the fixed implementation constraints.
- Architecture — the load-bearing patterns, module APIs, HTTP API and the physical data model.
Early stage, developed in main, breaking changes expected. The container is
fully stateless: the applied model and the journals live in your database and
survive container recreation, and the secrets arrive in the
DATAPULSE_SECRETS_JSON env var on every start (Vault is on the roadmap).