A production-grade ELT pipeline that ingests Concept2 Logbook rowing workout data into Postgres through three "medallion" layers, orchestrated by Dagster and loaded by dlt.
Dagster is a modern, open-source orchestration platform designed for data engineering, data science, and data analytics.
dlt (Data Load Tool) is an open-source Python library designed to solve one of the most tedious parts of data engineering: loading data from various sources (APIs, files, databases) into destinations (warehouses, data lakes, or databases). In the traditional ELT (Extract, Load, Transform) paradigm, dlt focuses heavily on the E and the L. dlt is essentially an "EL" framework for Python developers.
Dagster interface
Pace Trend Scatter Chart with Moving Average as generated by this concept2_pipeline application
The Concept2 PM5 Performance Monitor. Source: Concept2. The PM5 transmits your rowing performance details via bluetooth to the ErgData application on your mobile phone.
Concept2 API ──dlt──► concept2_bronze.results (raw)
│
SQL transform
│
▼
concept2_silver.workouts (cleaned)
│
Plotly charts
│
▼
charts_output/*.png / *.html
Medallion architecture is a data design pattern that organizes data within a lakehouse or data lake into three progressive layers—Bronze, Silver, and Gold—to incrementally improve data quality, structure, and trust. The medallion multi-hop approach allows raw data to be refined into business-ready assets while maintaining full traceability and governance.
| Layer | Tool | Postgres schema | Description |
|---|---|---|---|
| Bronze | dlt | concept2_bronze |
Raw API rows, incremental merge on id |
| Silver | SQLAlchemy | concept2_silver |
Typed, cleaned, pace derived |
| Gold | Plotly | (files on disk) | 7 interactive charts + PNGs |
dlt tracks the highest date value seen across runs in its internal state
table (concept2_bronze._dlt_pipeline_state). On every subsequent run only
workouts newer than that cursor are requested from the API:
GET /api/users/me/results?updated_after=<last_seen_date>&per_page=250
This means an account with years of history does a full load only once; every subsequent daily run fetches only the new rows from the past 24 hours.
To force a full reload:
# Delete the dlt state for this pipeline (typically in the ~/.dlt directory) then re-run
python -c "
import dlt
p = dlt.pipeline('concept2_bronze', destination='postgres', dataset_name='concept2_bronze')
p.drop()
"- Python 3.11+
- Postgres running locally (or update
PG_CONN)
# If you haven't installed uv on your local machine yet (MacOS/Linux):
curl -LsSf https://astral.sh/uv/install.sh | shcd concept2_pipeline
# pip install -e .
uv synccp .env.example .env
# Edit .env — fill in C2_CLIENT_ID and C2_CLIENT_SECRETpython -m concept2_pipeline.authYour web browser opens to https://log.concept2.com/login. Enter your username and password.
Web browser login screen
Approve access in the browser; a token is written to the .env file.
Approve access
export $(grep -v '^#' .env | xargs)Edit .dlt/secrets.toml:
[destination.postgres.credentials]
database = "concept2"
username = "concept2"
password = "concept2pass"
host = "localhost"
port = 5432dagster dev -m concept2_pipeline.definitionsOpen http://localhost:3000, go to Assets, and click Materialize All.
cp .env.example .env
# Fill in C2_CLIENT_ID, C2_CLIENT_SECRET, C2_ACCESS_TOKEN
docker compose up --buildDagster UI at http://localhost:3000.
Note: obtain C2_ACCESS_TOKEN on your host first:
pip install requests requests-oauthlib
python -m concept2_pipeline.auth
# Copy token into .env as C2_ACCESS_TOKENThe pipeline is scheduled to run every day at 06:00 UTC via the
concept2_daily_refresh schedule (enabled by default). You can also trigger
it manually from the Dagster UI or CLI:
dagster job execute -m concept2_pipeline.definitions -j concept2_full_pipeline| File | Description |
|---|---|
monthly_distance_bar.png |
Total metres per calendar month |
cumulative_distance_line.png |
Running total distance over time |
pace_trend_scatter.png |
Pace per 500m + 30-workout moving average |
heart_rate_trend.png |
Avg + max HR per workout |
workout_type_breakdown.png |
Donut chart by activity type |
distance_histogram.png |
Distribution of individual workout distances |
weekly_volume_stacked.png |
Weekly volume stacked by activity type |
Each chart is also saved as an interactive .html file in charts_output/.
concept2_pipeline/
├── concept2_pipeline/
│ ├── auth.py # OAuth2 token helper (run once)
│ ├── definitions.py # Dagster Definitions entry point
│ ├── assets/
│ │ ├── bronze.py # dlt_assets → concept2_bronze schema
│ │ ├── silver.py # SQL transform → concept2_silver schema
│ │ └── gold.py # Plotly charts → charts_output/
│ ├── resources/
│ │ └── postgres_resource.py # Configurable SQLAlchemy resource
│ └── sources/
│ └── concept2_source.py # dlt @source + @resource (incremental)
├── charts_output/ # Generated charts (git-ignored)
├── .dlt/
│ └── secrets.toml # dlt Postgres credentials (git-ignored)
├── .env.example # Environment variable template
├── docker-compose.yml
├── Dockerfile
├── workspace.yaml
└── pyproject.toml
| Variable | Description |
|---|---|
C2_CLIENT_ID |
Concept2 OAuth app client ID |
C2_CLIENT_SECRET |
Concept2 OAuth app client secret |
C2_ACCESS_TOKEN |
Bearer token from python -m concept2_pipeline.auth |
PG_CONN |
SQLAlchemy Postgres URL (used by silver + gold layers) |
CHARTS_OUTPUT_DIR |
Override chart output directory (default: charts_output) |
DAGSTER_POSTGRES_* |
Dagster run-history DB (see docker-compose.yml) |
psql -h localhost -p 5432 -U postgres -d postgres
Password for user postgres:
psql (17.10 (Homebrew), server 17.7)
Type "help" for help.
postgres=# \c concept2
psql (17.10 (Homebrew), server 17.7)
You are now connected to database "concept2" as user "postgres".
concept2=# \dt concept2*.*
List of relations
Schema | Name | Type | Owner
-------------------------+---------------------+-------+----------
concept2_bronze | _dlt_loads | table | postgres
concept2_bronze | _dlt_pipeline_state | table | postgres
concept2_bronze | _dlt_version | table | postgres
concept2_bronze | results | table | postgres
concept2_bronze_staging | _dlt_version | table | postgres
concept2_bronze_staging | results | table | postgres
concept2_silver | workouts | table | postgres
(7 rows)
concept2=# \d concept2_silver.workouts
Table "concept2_silver.workouts"
Column | Type | Collation | Nullable | Default
-----------------------+--------------------------+-----------+----------+---------
workout_id | bigint | | not null |
workout_at | timestamp with time zone | | |
workout_date | date | | |
year | integer | | |
month | integer | | |
day_of_week | integer | | |
activity_type | text | | |
activity_type_clean | text | | |
workout_type | character varying | | |
distance_m | double precision | | |
distance_km | numeric | | |
time_tenths | bigint | | |
time_formatted | character varying | | |
duration | interval | | |
pace_seconds_per_500m | numeric | | |
stroke_rate | integer | | |
calories | integer | | |
watts | integer | | |
heart_rate_avg | integer | | |
heart_rate_max | integer | | |
heart_rate_min | integer | | |
weight_class | character varying | | |
verified | boolean | | |
ranked | boolean | | |
source | character varying | | |
loaded_at | timestamp with time zone | | |
Indexes:
"workouts_pkey" PRIMARY KEY, btree (workout_id)
"ix_silver_workouts_date" btree (workout_date)
concept2=# select workout_at, activity_type, distance_m, time_formatted, pace_seconds_per_500m, stroke_rate, verified, ranked, source from concept2_silver.workouts;
workout_at | activity_type | distance_m | time_formatted | pace_seconds_per_500m | stroke_rate | verified | ranked | source
------------------------+---------------+------------+----------------+-----------------------+-------------+----------+--------+-------------
2025-01-31 12:13:00-05 | rower | 892 | 5:06.9 | 172.03 | 18 | t | f | ErgData iOS
2025-02-01 08:25:00-05 | rower | 1690 | 10:06.4 | 179.41 | 18 | t | f | ErgData iOS
2025-02-02 09:51:00-05 | rower | 2513 | 15:07.9 | 180.64 | 17 | t | f | ErgData iOS
2025-02-08 05:54:00-05 | rower | 3142 | 20:06.3 | 191.96 | 16 | t | f | ErgData iOS
2025-02-12 16:01:00-05 | rower | 2364 | 15:07.0 | 191.84 | 16 | t | f | ErgData iOS
2025-02-15 06:09:00-05 | rower | 4871 | 30:08.8 | 185.67 | 16 | t | f | ErgData iOS
2025-02-16 11:49:00-05 | rower | 4949 | 30:08.6 | 182.72 | 17 | t | f | ErgData iOS
2025-02-17 12:59:00-05 | rower | 5130 | 30:06.7 | 176.09 | 17 | t
...