Scheduled API to Postgres ETL with jq, COPY, and Slack Alerts

Ship a complete, production shaped API ingestion pipeline in one YAML file. Every day this Kestra blueprint calls a REST endpoint, flattens the nested JSON response into CSV with jq, bulk loads it into Postgres using the native COPY protocol, verifies the loaded rows with a query, and posts the outcome to Slack. A finally block resets the staging table no matter how the run ends, and a flow level error handler ships the stack trace to Slack when something breaks.

How it works

1. The daily trigger (io.kestra.plugin.core.trigger.Schedule) runs the flow every day at midnight in the Europe/Paris timezone.
2. The call_api task (io.kestra.plugin.core.http.Request) issues a GET to the uri input, which defaults to the Irish parliament members API so the flow works out of the box.
3. The csv task (io.kestra.plugin.scripts.shell.Commands) receives the response body as members.json via inputFiles, installs jq, extracts one CSV line per member, and exposes members.csv as an output file.
4. The load_to_pg task (io.kestra.plugin.jdbc.postgresql.CopyIn) streams the CSV into the house_members table using the Postgres COPY protocol, which is dramatically faster than row inserts. Connection details come from pluginDefaults, declared once for every Postgres task.
5. The query task (io.kestra.plugin.jdbc.postgresql.Query) reads the table back with fetchType: STORE to confirm and expose the loaded data, then slack (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) posts a completion message.
6. The finally block runs clean_data regardless of success or failure, and the errors block sends {{ errorLogs() }} to Slack when the run fails.

What you get

• A scheduled API to warehouse pipeline that runs unattended and reports its own status.
• COPY based bulk loading instead of slow row by row inserts.
• Guaranteed cleanup through finally, keeping the staging table idempotent between runs.
• Centralized Postgres credentials via pluginDefaults, defined once and reused by every task.
• A working public API default so you can run it before wiring your own source.

Who it's for

• Data engineers ingesting third party REST APIs into Postgres on a schedule.
• Teams that want the full ETL lifecycle, including cleanup and alerting, in a single declarative flow.
• Anyone learning the Kestra pattern of schedule, extract, transform, load, verify, notify.

Why orchestrate this with Kestra

The same logic as a cron driven Python script scatters scheduling, secrets, retries, cleanup, and alerting across code and infrastructure. Kestra makes each concern a declarative block: the schedule is a trigger, credentials are secrets injected through plugin defaults, cleanup is a finally block that always runs, and failure alerting is an errors branch with the real stack trace. Every run is logged, replayable, and backfillable from the UI.

Prerequisites

• A Postgres database with a house_members table matching the two CSV columns (or your adapted schema).
• Network access from the Kestra worker to the target API and to Slack.
• Note that clean_data empties the table after each run to keep the demo idempotent, remove or change it to keep the loaded data.

Secrets

• JDBC_CONNECTION_STRING, PG_USERNAME, PG_PASSWORD: Postgres connection shared by all JDBC tasks through pluginDefaults.
• SLACK_WEBHOOK: incoming webhook URL for the success and failure notifications.

Quick start

1. Add the secrets above to your Kestra namespace.
2. Create the house_members staging table in Postgres.
3. Deploy the flow and execute it manually with the default API, or override the uri input.
4. Check the loaded rows in the query task output and the message in your Slack channel.
5. Swap in your own API and adjust the jq expression to your payload.

How to extend

• Replace the finally delete with an upsert or a partition swap to retain history.
• Add pagination by wrapping call_api in io.kestra.plugin.core.flow.ForEach over page numbers.
• Chain io.kestra.plugin.dbt.cli.Run after the load to transform the staged rows into models.
• Add retry at the flow level to absorb flaky API responses automatically.
• Emit the row count in the Slack message using the query task outputs.

Links

• Postgres plugin docs: https://kestra.io/plugins/plugin-jdbc-postgres
• HTTP Request task docs: https://kestra.io/plugins/core/http/io.kestra.plugin.core.http.request
• Slack plugin docs: https://kestra.io/plugins/plugin-notifications