Реальный compose: Python ETL + Postgres + init.sql

В этом уроке собираем то, что реально нужно junior DE на первом проекте: compose-стенд с Postgres + init.sql + Python ETL-приложение, которое читает из API (моковый JSON) и пишет в БД. На этом примере соберём вместе всё, что разбирали в модулях 8-10: volume для данных, bind mount для кода, healthcheck, depends_on с condition, init-скрипты Postgres.

Первый скрипт: shebang, chmod +x, аргументы

Структура проекта

etl-stack/
  compose.yml
  init/
    01-schema.sql
  app/
    Dockerfile
    requirements.txt
    src/
      main.py

compose.yml

services:
  postgres:
    image: postgres:17
    environment:
      POSTGRES_PASSWORD: secret
      POSTGRES_DB: etl
    volumes:
      - pgdata:/var/lib/postgresql/data
      - ./init:/docker-entrypoint-initdb.d:ro
    ports:
      - "127.0.0.1:5432:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres -d etl"]
      interval: 5s
      timeout: 3s
      retries: 5
      start_period: 10s

  etl:
    build:
      context: ./app
    depends_on:
      postgres:
        condition: service_healthy
    environment:
      DATABASE_URL: postgresql://postgres:secret@postgres:5432/etl
      PYTHONUNBUFFERED: "1"
    volumes:
      - ./app/src:/app/src
    restart: on-failure

volumes:
  pgdata:

Заметные детали:

pgdata — named volume для БД. Переживает рестарт.
./init:/docker-entrypoint-initdb.d:ro — init-скрипты запускаются при первом старте на пустом volume.
ports: 127.0.0.1:5432:5432 — Postgres доступен на localhost для psql, не наружу.
healthcheck + condition: service_healthy — гарантирует, что ETL стартует после готовности БД.
./app/src:/app/src — bind mount кода: live edit без rebuild.
restart: on-failure — если ETL упал — перезапустить (полезно для long-running job’ов).

init/01-schema.sql

CREATE TABLE IF NOT EXISTS events (
  id BIGSERIAL PRIMARY KEY,
  event_type TEXT NOT NULL,
  payload JSONB NOT NULL,
  created_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE INDEX IF NOT EXISTS events_event_type_idx ON events(event_type);
CREATE INDEX IF NOT EXISTS events_created_at_idx ON events(created_at);

INSERT INTO events(event_type, payload) VALUES
  ('user_signup', '{"user_id": 1, "email": "[email protected]"}'),
  ('user_login',  '{"user_id": 1}'),
  ('order_placed', '{"user_id": 1, "amount": 42.5, "currency": "USD"}');

Запустится только при первом старте на пустом volume. Если ты пересоздаёшь стенд через compose down (без -v), volume остаётся, init не сработает. С compose down -v — пересоздание с init.

app/Dockerfile

FROM python:3.13-slim

WORKDIR /app

COPY requirements.txt /app/
RUN pip install --no-cache-dir -r requirements.txt

COPY src /app/src

CMD ["python", "-u", "src/main.py"]

Минимально: установить зависимости, запустить main.py. -u для unbuffered stdout (логи сразу видны через compose logs).

app/requirements.txt

psycopg[binary]==3.2.4

Используем psycopg 3 (актуальная major-версия), binary-варианты не требует системных libpq-dev пакетов.

app/src/main.py

import json
import os
import sys
import time
import psycopg


def wait_for_db(url: str, timeout: int = 30) -> psycopg.Connection:
    start = time.time()
    while True:
        try:
            return psycopg.connect(url, autocommit=True)
        except psycopg.OperationalError as e:
            if time.time() - start > timeout:
                raise
            print(f"DB not ready: {e}, retry in 1s", flush=True)
            time.sleep(1)


def main() -> int:
    url = os.environ["DATABASE_URL"]
    print(f"Connecting to {url}", flush=True)

    with wait_for_db(url) as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT count(*) FROM events;")
            count = cur.fetchone()[0]
            print(f"Found {count} existing events", flush=True)

            # Имитация ETL: записываем синтетические event'ы.
            new_events = [
                ("pipeline_run", {"job": "demo", "rows_processed": count}),
            ]
            cur.executemany(
                "INSERT INTO events(event_type, payload) VALUES (%s, %s::jsonb)",
                [(t, json.dumps(p)) for t, p in new_events],
            )
            print(f"Inserted {len(new_events)} new events", flush=True)

    return 0


if __name__ == "__main__":
    sys.exit(main())

Что делает:

Ждёт, пока Postgres ответит на connect (защита от race, даже с healthcheck).
Считает количество event’ов.
Вставляет один новый event (имитация ETL-job).
Выходит с exit 0.

restart: on-failure в compose не сработает, потому что мы выходим успешно. Если бы был лонг-раннер (consumer Kafka, например) — он работал бы постоянно.

Запуск

mkdir -p etl-stack/init etl-stack/app/src
cd etl-stack

# (создаём все файлы как выше)

docker compose up --build

В терминале увидим:

[+] Building 8.2s (10/10) FINISHED
[+] Running 3/3
 [x] Network etl-stack_default       Created
 [x] Volume etl-stack_pgdata         Created
 [x] Container etl-stack-postgres-1  Healthy
 [x] Container etl-stack-etl-1       Started

etl-1       | Connecting to postgresql://postgres:secret@postgres:5432/etl
etl-1       | Found 3 existing events
etl-1       | Inserted 1 new events
etl-1 exited with code 0

Проверим, что вставилось:

docker compose exec postgres psql -U postgres -d etl -c \
  "SELECT id, event_type, payload, created_at FROM events ORDER BY id;"
# 1 | user_signup   | {"email": "[email protected]", "user_id": 1}          | 2026-05-15 ...
# 2 | user_login    | {"user_id": 1}                              | 2026-05-15 ...
# 3 | order_placed  | {"amount": 42.5, "user_id": 1, "currency"...} | 2026-05-15 ...
# 4 | pipeline_run  | {"job": "demo", "rows_processed": 3}        | 2026-05-15 ...

ETL сработал. Запусти ещё раз — увидишь, что count = 4 после первого прогона, потом 5, и так далее.

docker compose run --rm etl

Это альтернативный способ запуска: создаст новый ETL-контейнер, прогоняет main.py, удалит. Удобно, если хочешь повторно дергать pipeline.

Lifecycle volume vs init

Сценарий: «Я хочу пересоздать schema». Если просто перезапустить compose:

docker compose down
docker compose up

— init не сработает, потому что volume pgdata сохранился. БД не пустая, entrypoint Postgres пропускает init.

Чтобы init сработал заново:

docker compose down -v   # удалить и pgdata
docker compose up        # init сработает, schema пересоздаётся

Альтернатива — сменить content init и применить вручную через psql:

docker compose exec postgres psql -U postgres -d etl -f /docker-entrypoint-initdb.d/01-schema.sql

— но IF NOT EXISTS в schema важен, иначе упадёт.

Live edit кода

Меняем src/main.py локально, не перестраивая образ:

# Меняем код в IDE...
echo '# touch' >> app/src/main.py

# Bind mount работает — изменения видны в контейнере сразу.
docker compose run --rm etl
# Запустится свежий код.

Этот паттерн — основа DE-разработки: один раз собрал образ с тяжёлыми зависимостями (numpy, pandas, библиотеки), потом редактируешь Python-код в IDE и быстро прогоняешь.

Dev-цикл с этим стендом

docker compose up --build

Edit src/main.py

docker compose run --rm etl

docker compose exec postgres psql

docker compose down

docker compose down -v

Что мы получили

Этот compose-файл — реальный шаблон для DE-проекта. От него легко вырасти:

Заменить etl-сервис на Airflow DAG-process (через volume mount DAG-папки).
Добавить redis и worker для celery-executor.
Добавить minio для S3-compatible storage.
Добавить kafka для event streaming.

И всё это — в одном compose-файле, с воспроизводимым setup.

Попробуй сам

# 1. Собери всю структуру.
mkdir -p etl-stack/{init,app/src} && cd etl-stack

# init/01-schema.sql
cat > init/01-schema.sql <<'SQL'
CREATE TABLE IF NOT EXISTS events (
  id BIGSERIAL PRIMARY KEY,
  event_type TEXT NOT NULL,
  payload JSONB NOT NULL,
  created_at TIMESTAMPTZ DEFAULT NOW()
);
INSERT INTO events(event_type, payload) VALUES
  ('user_signup', '{"user_id": 1}'),
  ('user_login',  '{"user_id": 1}');
SQL

# app/Dockerfile
cat > app/Dockerfile <<'DOCKER'
FROM python:3.13-slim
WORKDIR /app
COPY requirements.txt /app/
RUN pip install --no-cache-dir -r requirements.txt
COPY src /app/src
CMD ["python", "-u", "src/main.py"]
DOCKER

# app/requirements.txt
echo 'psycopg[binary]==3.2.4' > app/requirements.txt

# app/src/main.py
cat > app/src/main.py <<'PY'
import os, time, json, sys
import psycopg

url = os.environ["DATABASE_URL"]
for _ in range(30):
    try:
        conn = psycopg.connect(url, autocommit=True); break
    except psycopg.OperationalError:
        time.sleep(1)

with conn.cursor() as cur:
    cur.execute("SELECT count(*) FROM events;")
    count = cur.fetchone()[0]
    print(f"events count: {count}")
    cur.execute("INSERT INTO events(event_type, payload) VALUES ('pipeline_run', %s::jsonb)", (json.dumps({"rows": count}),))
print("done")
PY

# compose.yml — как выше.
cat > compose.yml <<'YAML'
services:
  postgres:
    image: postgres:17
    environment: { POSTGRES_PASSWORD: secret, POSTGRES_DB: etl }
    volumes:
      - pgdata:/var/lib/postgresql/data
      - ./init:/docker-entrypoint-initdb.d:ro
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres -d etl"]
      interval: 3s
      retries: 5
      start_period: 5s
  etl:
    build: ./app
    depends_on:
      postgres:
        condition: service_healthy
    environment:
      DATABASE_URL: postgresql://postgres:secret@postgres:5432/etl
      PYTHONUNBUFFERED: "1"
    volumes:
      - ./app/src:/app/src
volumes:
  pgdata:
YAML

# 2. Запусти.
docker compose up --build
# Ожидай: postgres healthy, etl пишет "events count: 2", "done", выходит с 0.

# 3. Прогоняй pipeline ещё раз.
docker compose run --rm etl
# events count: 3

# 4. Проверяй БД.
docker compose exec postgres psql -U postgres -d etl -c "SELECT count(*) FROM events;"

# 5. Cleanup.
docker compose down -v
cd .. && rm -rf etl-stack

TIP

Этот шаблон — отправная точка. В реальном проекте добавь: .env файл для секретов, pgAdmin или adminer для UI к БД (отдельный сервис), Makefile с командами make up, make test, make clean. Это сразу делает onboarding нового члена команды задачей одной команды.

В следующем модуле — продвинутый compose: env-файлы, profiles, override-файлы, secrets, scaling.

Проверка знанийKnowledge check

У тебя стенд: postgres + etl + init.sql, который при первом старте создаёт таблицу events и вставляет 3 строки. Ты сделал docker compose up , всё работает. Потом изменил init/01-schema.sql, добавил новый INSERT, сделал docker compose down и docker compose up . Новый INSERT не появился в БД. Почему, и какие два правильных способа это исправить?

ОтветAnswer

Init-скрипты в /docker-entrypoint-initdb.d/ запускаются Postgres-entrypoint только при первом старте на ПУСТОМ volume. Логика entrypoint: проверка наличия PG_VERSION в PGDATA -> если есть, БД уже инициализирована, скрипты пропускаются. docker compose down (без -v) останавливает и удаляет контейнеры, но named volume pgdata переживает. На повторном compose up новый контейнер видит непустой volume и пропускает init. Новый INSERT в init-скрипте не выполняется. Способ 1 — полный wipe: docker compose down -v, потом docker compose up. -v удалит pgdata, на следующем старте volume пустой, init-скрипты применятся заново. Этот способ уничтожает все данные, что норм для dev/test, не для prod. Способ 2 — применить вручную: docker compose exec postgres psql -U postgres -d etl -c "INSERT INTO events ..." или docker compose exec postgres psql -U postgres -d etl -f /docker-entrypoint-initdb.d/01-schema.sql, но с IF NOT EXISTS / ON CONFLICT, иначе INSERT упадёт по unique constraint. В production-практике это делается через миграции (Alembic, Flyway) — init-скрипты только для первоначальной структуры.