Breaking Data Out of Walled Gardens: 9 Collectors for a Personal Data Warehouse

My data lives in 9 different services. Garmin tracks my steps. Navidrome plays my music. Miniflux holds my RSS feeds. GitHub has my commits. Ghost publishes my blog. Audiobookshelf logs my listening. Uptime Kuma monitors my infrastructure. Open-Meteo records the weather. Actual Budget tracks my finances.

None of them talk to each other. None of them share a common data format, authentication scheme, or API style. My health data is locked behind a reverse-engineered session cookie. My music history requires JWT auth against a Subsonic-compatible API. My weather data is free and unauthenticated but arrives in WMO codes that nobody memorizes.

So I built a pipeline. Nine collectors, one abstract base class, four ingest endpoints, and a Kubernetes CronJob that runs every 15 minutes. All of it feeds into Life Hub, a FastAPI app backed by SQLite that serves as my personal data warehouse.

Here's how it works.

The Architecture

The data flow is deliberately simple:

Source API → Collector → Ingest API → SQLite

Each collector is a Python class that knows how to talk to exactly one external service. It fetches records, normalizes them into a common schema, and POSTs them to one of four ingest endpoints on the Life Hub API. The ingest layer validates the data, deduplicates it against what's already in the database, and returns a count of what was inserted versus skipped.

The entire pipeline runs inside a single Kubernetes CronJob. Same container image as the main app, different entrypoint. Every 15 minutes, it spins up, runs all 9 collectors sequentially, and logs a summary table before exiting.

The BaseCollector Pattern

Every collector extends a single abstract base class. Here's the core of it:

class BaseCollector(ABC):
    name: str = "base"

    @abstractmethod
    async def collect(self) -> list[dict[str, Any]]:
        """Fetch data from the source. Returns a list of record dicts."""
        ...

    @property
    def ingest_path(self) -> str:
        return "/api/v1/ingest/activity"

    async def run(self, api_url: str, api_key: str) -> dict[str, Any]:
        try:
            records = await self.collect()
        except Exception as exc:
            return {"error": str(exc), "inserted": 0, "skipped": 0}

        if not records:
            return {"inserted": 0, "skipped": 0}

        async with httpx.AsyncClient(timeout=30.0) as client:
            resp = await client.post(url, json=records, headers=headers)
            return resp.json()

The design decisions that matter:

• Error isolation. If collect() throws, the error is caught, logged, and returned. The next collector runs regardless. One broken API should never take down the whole pipeline.
• Default ingest path. Most collectors push to /api/v1/ingest/activity. The few that don't (health metrics, infra snapshots, finance) override the ingest_path property.
• Async throughout. Every HTTP call uses httpx.AsyncClient with a 30-second timeout. The collectors could run in parallel with asyncio.gather(), but sequential execution keeps the logs readable and the API load predictable.

The 9 Collectors

1. Garmin Connect: The Interesting One

Garmin doesn't have a public API. The garminconnect Python package reverse-engineers the session-based auth that the web app uses. This means authentication is fragile. Garmin can break it with any update.

To avoid re-authenticating every 15 minutes (96 logins per day), I cache the session token to a JSON file on the persistent volume:

if os.path.isfile(_SESSION_FILE):
    try:
        with open(_SESSION_FILE, "r") as f:
            session_data = json.load(f)
        client.login(session_data)
    except Exception:
        client.login()  # Fresh login if token expired
        _save_session(client)

The collector pulls daily summaries across 4 separate API calls: daily stats (steps, calories, active minutes, floors, distance), heart rate, weight/body composition, and SpO2. Each metric becomes its own record, so a partial failure (say the weight API times out) doesn't lose the step count.

One subtlety: Garmin returns weight in grams and active time in seconds. The collector normalizes these (weight / 1000 for kg, seconds / 60 for minutes) before ingesting.

2. Navidrome: Music Listening

Navidrome exposes a native REST API alongside its Subsonic-compatible interface. The collector hits the native API because it returns richer metadata: genre, year, play count, and actual play timestamps.

Auth is JWT-based: POST to /auth/login, get a token, use it as a bearer token for subsequent requests. Fresh login every run: no session caching needed since auth is fast and reliable.

It fetches the last 50 played songs, filtering out any that have never been played (empty playDate). Each song becomes an activity event with metadata for artist, album, genre, duration, and play count.

3. Miniflux: RSS Reading

The simplest collector. Miniflux has a clean REST API with X-Auth-Token authentication. Fetch the 50 most recently read entries, extract the feed title, author, reading time estimate, and URL. Done.

I track articles marked as read rather than subscribed feeds because the read list represents what I actually consumed, not what I intended to consume.

4. GitHub: The Dual-API Workaround

This one taught me something about GitHub's API. Fine-grained personal access tokens (the newer, more secure kind) return PushEvent objects with empty commit arrays from the Events API. The commit messages are just... missing.

The fix: use two APIs in parallel. The Commits API returns full commit messages reliably. The Events API returns non-push events (pull requests, issues, releases, branch creation) that the Commits API doesn't cover.

# Commits API: get actual commit messages
url = f"https://api.github.com/repos/{repo}/commits?per_page=50"

# Events API: PRs, issues, releases (2 pages max)
for page in range(1, 3):
    url = f"https://api.github.com/repos/{repo}/events?per_page=100&page={page}"

Six event types are tracked: PullRequestEvent, IssuesEvent, IssueCommentEvent, CreateEvent, DeleteEvent, and ReleaseEvent. Each gets a descriptive title like "PR opened: Add session caching" that reads naturally in the timeline.

5. Audiobookshelf: Listening Sessions

Audiobookshelf tracks listening sessions with start time, duration, and current position. The API returns timestamps as Unix epoch in milliseconds, which the collector converts to ISO 8601. It also handles a quirk where the API sometimes returns a flat list and sometimes wraps sessions in a nested property.

6. Ghost CMS: Blog Activity

The Ghost Content API uses a key-as-query-parameter auth scheme. The collector fetches the 10 most recent posts with their tags, authors, reading time, and excerpt. Each published or updated post becomes a "writing" activity event.

follow_redirects=True on the HTTP client handles the case where Ghost sits behind Traefik and Cloudflare Tunnel: there's occasionally an extra redirect in the chain.

7. Uptime Kuma: Infrastructure Status

This collector uses the public status page API, which means no authentication: the data is already meant to be publicly visible. But the API has a quirk: monitor names and monitor statuses come from two different endpoints.

# Endpoint 1: monitor names (publicGroupList → monitorList → id/name)
resp = await client.get(f"{base}/api/status-page/{slug}")

# Endpoint 2: heartbeats keyed by monitor ID
resp = await client.get(f"{base}/api/status-page/heartbeat/{slug}")

The collector joins these two datasets by monitor ID to produce labeled records. Status codes are sparse: 1 means up, 0 means down, anything else maps to "unknown". Each record goes to the dedicated /api/v1/ingest/infra endpoint.

8. Open-Meteo: Weather

Free, no auth, no API key. Just hit the endpoint with latitude/longitude coordinates and get back current conditions including temperature, wind speed, humidity, and a WMO weather code.

The interesting pattern here is timestamp deduplication. Since the CronJob runs every 15 minutes but weather doesn't change that fast, the collector rounds timestamps to the nearest 30-minute slot:

now = datetime.now(timezone.utc)
rounded_minute = (now.minute // 30) * 30
ts = now.replace(minute=rounded_minute, second=0, microsecond=0)

Two runs at 10:03 and 10:14 both produce a timestamp of 10:00. The ingest API deduplicates on (source, timestamp, title), so the second run gets skipped. At most 48 weather entries per day, regardless of how often the collector runs.

9. Finance: CSV Import

The only collector that reads a local file instead of hitting an HTTP API. It parses a CSV with csv.DictReader() expecting columns for date, account name, balance, and optional category totals as a JSON string.

This exists because financial data doesn't have good APIs. Bank APIs require OAuth flows and business relationships. Actual Budget's API is local-only. A CSV file is the universal adapter: export from your bank, from a spreadsheet, from any tool that can write text. It's low-tech and it works.

Graceful Degradation

Every collector starts the same way:

if not settings.MINIFLUX_URL or not settings.MINIFLUX_API_KEY:
    logger.info("[miniflux] URL or API key not configured — skipping")
    return []

No config? No problem. Return an empty list, log why, move on.

This means you can deploy the entire Life Hub stack with zero collectors configured and add data sources one at a time. Start with weather (no auth required), add GitHub (just a token), add Garmin later when you've sorted out the credentials. Each collector is independently optional.

The Garmin collector takes this further: it even catches ImportError for the garminconnect package:

try:
    from garminconnect import Garmin
except ImportError:
    logger.info("[garmin] garminconnect package not installed — skipping")
    return []

If the package isn't in your environment, the collector silently opts out.

The Ingest API

Four bulk endpoints, each with a composite deduplication key:

Every endpoint accepts a JSON array of records, validates each against a Pydantic schema, checks the dedup key, and either inserts or skips. The response is always the same shape: {"inserted": 5, "skipped": 12}.

Dedup uses a SELECT-then-INSERT pattern rather than ON CONFLICT clauses. SQLite handles this fine at our scale, and it's easier to reason about across all four record types.

The Orchestrator

The collector CLI (collector_cli.py) does more than just run collectors. The full sequence every 15 minutes:

1. SQLite backup: POST to /api/v1/backup/run. Non-fatal if it fails.
2. Run all 9 collectors: Sequential, error-isolated.
3. Update sync state: PUT to /api/v1/infra/sync-state for each collector with status and timestamp.
4. Refresh digests: Rebuild daily, weekly, and monthly rule-based summaries.
5. AI advisor reviews: Daily reviews run every cycle (server-side cooldown prevents duplicates). Weekly reviews run Sundays.
6. AI brief generation: Weekly briefs on Sundays, monthly briefs on the 1st.
7. Push notifications: Reminders and digest notifications via ntfy.
8. Summary table: Logged to stdout for the CronJob logs.

A typical run looks like this in the logs:

============================================================
Collection run complete in 8.3s
COLLECTOR             INSERTED   SKIPPED  ERROR
------------------------------------------------------------
miniflux                     3        47  OK
navidrome                    2        48  OK
audiobookshelf               0         8  OK
uptime_kuma                 28         0  OK
ghost                        0        10  OK
github                       4        96  OK
weather                      0         1  OK
finance                      0         0  OK
garmin                       8         2  OK
============================================================
Total inserted: 45 | Errors: 0

Most records get skipped on a typical run because they've already been ingested. That's the dedup working correctly. It means idempotency is built in. Run the collectors twice, get the same result.

What I Learned

Dedup at ingest time is non-negotiable. CronJobs run on a schedule. Services restart. Network blips cause retries. Your pipeline will process the same data more than once. If you don't dedup on ingest, you'll spend more time cleaning data than collecting it.

Session caching pays for itself fast. The Garmin collector runs 96 times per day. Without session caching, that's 96 full authentication flows against a reverse-engineered API that could rate-limit or ban you. With caching, it's one login per session expiry, maybe once a day.

The 30-minute rounding trick is reusable. Any time you're collecting periodic data (weather, prices, exchange rates), round the timestamp to your desired granularity before ingesting. The dedup layer handles the rest.

CSV is a legitimate integration pattern. Not every data source has an API. Not every API is worth the complexity. A CSV file on disk is the universal adapter. It works with bank exports, spreadsheet macros, manual entry, and anything that can write text. Don't over-engineer the integration when a flat file will do.

One failure should never poison the pipeline. This is the most important design decision in the whole system. Wrap every collector in error handling. Log the failure. Move on. The summary table at the end makes failures visible without making them blocking.

Make configuration optional. If adding a new data source requires configuring all nine, nobody will do it. Each collector checks its own config independently. Deploy first, configure incrementally. Start with the free ones (weather), add APIs as you get tokens.

What's Next

The collectors currently run sequentially. Moving to asyncio.gather() for parallel execution would cut the total run time roughly in half: the bottleneck is waiting on HTTP responses, and most of these APIs are independent.

I want to add collectors for Home Assistant (temperature, humidity, energy usage), Apple Health via Health Auto Export (iOS app that pushes health data to a webhook), and Todoist for task completion tracking.

The bigger project is historical backfill. Right now, collectors only capture data going forward from when they're first configured. Building import scripts for historical Garmin data, old GitHub commits, and past Navidrome listening history would fill in the gaps and make trend analysis meaningful from day one.

The code is open source at github.com/snaviaux/life-hub. The collector pipeline is one of the more reusable pieces: the BaseCollector pattern works for any "poll an API, push to a database" use case.