diff --git a/.env.example b/.env.example index 49f39e7..9911dfc 100644 --- a/.env.example +++ b/.env.example @@ -68,3 +68,29 @@ DUCK_TABLE=sales # Кількість рядків на один INSERT statement (batch size) DUCK_BATCH_SIZE=1000 + +# ============================================================ +# PostgreSQL (опційно — для завантаження даних у PostgreSQL) +# ============================================================ + +# Увімкнення завантаження у PostgreSQL паралельно з Excel/CSV +PG_ENABLED=false + +# Адреса та порт PostgreSQL +# З роботи (через інтернет): PG_HOST=db.lwhs.xyz, PG_PORT=54321 +# З дому (LAN/VPN): PG_HOST=192.168.1.111, PG_PORT=5432 +PG_HOST=localhost +PG_PORT=5432 + +# База даних та автентифікація +PG_DATABASE=analytics +PG_USER=analytics +PG_PASSWORD= + +# Schema та таблиця +PG_SCHEMA=public +PG_TABLE=sales + +# SSL режим: require | verify-ca | verify-full | disable +# require — шифрування без перевірки сертифікату (підходить для self-signed) +PG_SSL_MODE=require diff --git a/docs/superpowers/plans/2026-03-10-postgresql-sink.md b/docs/superpowers/plans/2026-03-10-postgresql-sink.md new file mode 100644 index 0000000..f015dc5 --- /dev/null +++ b/docs/superpowers/plans/2026-03-10-postgresql-sink.md @@ -0,0 +1,568 @@ +# PostgreSQL Sink Implementation Plan + +> **For agentic workers:** REQUIRED: Use superpowers:subagent-driven-development (if subagents available) or superpowers:executing-plans to implement this plan. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Додати `PostgreSQLSink` як третій аналітичний sink для завантаження даних з OLAP у PostgreSQL через `COPY FROM STDIN`. + +**Architecture:** Новий `PostgreSQLConfig` dataclass у `config.py` — за тим самим патерном що і `ClickHouseConfig`/`DuckDBConfig`. `PostgreSQLSink` у `sinks.py` реалізує `AnalyticsSink` ABC з bulk-завантаженням через `psycopg2.copy_expert`. `runner.py` та `cli.py` отримують мінімальні точкові зміни за існуючим патерном. + +**Tech Stack:** Python 3.8-3.13, psycopg2-binary, pandas, numpy + +--- + +## Chunk 1: Конфігурація та залежності + +### Task 1: `PostgreSQLConfig` dataclass + env loading (`config.py`) + +**Files:** +- Modify: `olap_tool/config.py` + +- [ ] **Step 1: Додати `PostgreSQLConfig` dataclass після `DuckDBConfig` (~рядок 103)** + +```python +@dataclass +class PostgreSQLConfig: + """Налаштування підключення до PostgreSQL.""" + enabled: bool = False + host: str = "localhost" + port: int = 5432 + database: str = "analytics" + user: str = "analytics" + password: str = "" + schema: str = "public" + table: str = "sales" + ssl_mode: str = "require" +``` + +- [ ] **Step 2: Додати `postgresql: PostgreSQLConfig` до `AppConfig` (після поля `duckdb`)** + +```python +postgresql: PostgreSQLConfig = field(default_factory=PostgreSQLConfig) +``` + +- [ ] **Step 3: Додати `load_postgres_from_env()` після `load_duckdb_from_env()`** + +```python +def load_postgres_from_env() -> PostgreSQLConfig: + """Читає налаштування PostgreSQL з os.environ.""" + try: + pg_port = int(os.getenv("PG_PORT", "5432")) + except (ValueError, TypeError): + pg_port = 5432 + return PostgreSQLConfig( + enabled=_parse_bool(os.getenv("PG_ENABLED", "false"), False), + host=os.getenv("PG_HOST", "localhost"), + port=pg_port, + database=os.getenv("PG_DATABASE", "analytics"), + user=os.getenv("PG_USER", "analytics"), + password=os.getenv("PG_PASSWORD", ""), + schema=os.getenv("PG_SCHEMA", "public"), + table=os.getenv("PG_TABLE", "sales"), + ssl_mode=os.getenv("PG_SSL_MODE", "require"), + ) +``` + +- [ ] **Step 4: Оновити `apply_profile()` — додати `"postgresql"` до списку секцій (~рядок 267)** + +Знайти рядок: +```python +for section in ("query", "export", "xlsx", "csv", "excel_header", "paths", "display", "clickhouse", "duckdb"): +``` +Замінити на: +```python +for section in ("query", "export", "xlsx", "csv", "excel_header", "paths", "display", "clickhouse", "duckdb", "postgresql"): +``` + +- [ ] **Step 5: Оновити `build_config()` — додати завантаження PG env (після блоку DuckDB, перед `return AppConfig(...)`)** + +```python +# PostgreSQL: env задає defaults, profile може перевизначити +pg_env = load_postgres_from_env() +pg_env_dict = {f.name: getattr(pg_env, f.name) for f in dataclass_fields(pg_env)} +base.setdefault("postgresql", {}) +for k, v in pg_env_dict.items(): + base["postgresql"].setdefault(k, v) +``` + +- [ ] **Step 6: Оновити `return AppConfig(...)` — додати `postgresql` поле** + +```python +postgresql=_build_section(PostgreSQLConfig, base, "postgresql"), +``` + +- [ ] **Step 7: Commit** + +```bash +git add olap_tool/config.py +git commit -m "feat: PostgreSQLConfig dataclass та env loading" +``` + +--- + +### Task 2: Залежності та `.env.example` + +**Files:** +- Modify: `requirements.txt` +- Modify: `.env.example` + +- [ ] **Step 1: Додати `psycopg2-binary` до `requirements.txt` (після рядка `requests>=2.28.0`)** + +``` +psycopg2-binary>=2.9.0 # Для завантаження даних у PostgreSQL через COPY FROM STDIN +``` + +- [ ] **Step 2: Додати секцію PostgreSQL до `.env.example` (в кінець файлу)** + +``` +# ============================================================ +# PostgreSQL (опційно — для завантаження даних у PostgreSQL) +# ============================================================ + +# Увімкнення завантаження у PostgreSQL паралельно з Excel/CSV +PG_ENABLED=false + +# Адреса та порт PostgreSQL +# З роботи (через інтернет): PG_HOST=db.lwhs.xyz, PG_PORT=54321 +# З дому (LAN/VPN): PG_HOST=192.168.1.111, PG_PORT=5432 +PG_HOST=localhost +PG_PORT=5432 + +# База даних та автентифікація +PG_DATABASE=analytics +PG_USER=analytics +PG_PASSWORD= + +# Schema та таблиця +PG_SCHEMA=public +PG_TABLE=sales + +# SSL режим: require | verify-ca | verify-full | disable +# require — шифрування без перевірки сертифікату (підходить для self-signed) +PG_SSL_MODE=require +``` + +- [ ] **Step 3: Встановити psycopg2-binary** + +```bash +pip install psycopg2-binary +``` + +Очікуваний вивід: `Successfully installed psycopg2-binary-2.9.x` + +- [ ] **Step 4: Commit** + +```bash +git add requirements.txt .env.example +git commit -m "feat: psycopg2-binary залежність та PG_* змінні у .env.example" +``` + +--- + +## Chunk 2: PostgreSQLSink + +### Task 3: `PostgreSQLSink` (`sinks.py`) + +**Files:** +- Modify: `olap_tool/sinks.py` + +- [ ] **Step 1: Додати TYPE_CHECKING імпорт для `PostgreSQLConfig` (у блок `if TYPE_CHECKING`)** + +```python +if TYPE_CHECKING: + from .config import ClickHouseConfig + from .config import DuckDBConfig + from .config import PostgreSQLConfig +``` + +- [ ] **Step 2: Додати helper `_pandas_dtype_to_pg()` після `_pandas_dtype_to_duck()` (~рядок 152)** + +```python +def _pandas_dtype_to_pg(dtype) -> str: + """Конвертує pandas dtype у PostgreSQL SQL тип.""" + dtype_str = str(dtype) + if dtype_str.startswith("int") or dtype_str.startswith("uint"): + return "BIGINT" + if dtype_str.startswith("float"): + return "DOUBLE PRECISION" + if dtype_str in ("bool", "boolean"): + return "BOOLEAN" + if dtype_str.startswith("datetime"): + return "TIMESTAMP" + if dtype_str.startswith("date"): + return "DATE" + return "TEXT" +``` + +- [ ] **Step 3: Додати клас `PostgreSQLSink` в кінець `sinks.py`** + +```python +# --------------------------------------------------------------------------- +# PostgreSQL sink (psycopg2 + COPY FROM STDIN) +# --------------------------------------------------------------------------- + +class PostgreSQLSink(AnalyticsSink): + """ + Завантажує DataFrame у PostgreSQL через COPY FROM STDIN. + + Ідемпотентність: DELETE WHERE year_num=X AND week_num=Y → COPY FROM STDIN CSV. + SSL: sslmode=require (шифрування без перевірки self-signed сертифікату). + """ + + def __init__(self, config: "PostgreSQLConfig"): + self._config = config + self._conn = None + self._schema: dict[str, str] | None = None + + def _get_conn(self): + """Повертає активне з'єднання, створює нове якщо потрібно.""" + import psycopg2 + if self._conn is None or self._conn.closed: + self._conn = psycopg2.connect( + host=self._config.host, + port=self._config.port, + dbname=self._config.database, + user=self._config.user, + password=self._config.password, + sslmode=self._config.ssl_mode, + ) + self._conn.autocommit = False + return self._conn + + def _full_table(self) -> str: + """Повертає повну назву таблиці з схемою: "schema"."table".""" + return f'"{self._config.schema}"."{self._config.table}"' + + def _refresh_schema(self) -> None: + """Читає поточну схему таблиці з information_schema.""" + conn = self._get_conn() + with conn.cursor() as cur: + cur.execute( + """ + SELECT column_name, data_type + FROM information_schema.columns + WHERE table_schema = %s AND table_name = %s + ORDER BY ordinal_position + """, + (self._config.schema, self._config.table), + ) + rows = cur.fetchall() + self._schema = {row[0]: row[1] for row in rows} + + def setup(self, df: pd.DataFrame) -> None: + from .utils import print_progress, print_warning + print_progress( + f"Перевірка таблиці PostgreSQL {self._full_table()} " + f"({self._config.host}:{self._config.port})..." + ) + conn = self._get_conn() + cols_ddl = ", ".join( + f'"{col}" {_pandas_dtype_to_pg(df[col].dtype)}' + for col in df.columns + ) + with conn.cursor() as cur: + cur.execute( + f"CREATE TABLE IF NOT EXISTS {self._full_table()} ({cols_ddl})" + ) + conn.commit() + self._refresh_schema() + + # Додаємо нові колонки яких немає в таблиці + for col in df.columns: + if col not in (self._schema or {}): + dtype = _pandas_dtype_to_pg(df[col].dtype) + try: + with conn.cursor() as cur: + cur.execute( + f'ALTER TABLE {self._full_table()} ' + f'ADD COLUMN IF NOT EXISTS "{col}" {dtype}' + ) + conn.commit() + if self._schema is not None: + self._schema[col] = dtype + except Exception as e: + conn.rollback() + print_warning(f"Не вдалося додати колонку `{col}`: {e} — пропускаємо") + + def delete_period(self, year: int, week: int) -> None: + if self._schema is None: + self._refresh_schema() + schema = self._schema or {} + if "year_num" not in schema or "week_num" not in schema: + return + conn = self._get_conn() + with conn.cursor() as cur: + cur.execute( + f"DELETE FROM {self._full_table()} " + f"WHERE year_num = %s AND week_num = %s", + (year, week), + ) + conn.commit() + + def insert(self, df: pd.DataFrame, year: int, week: int) -> int: + import io + if df is None or len(df) == 0: + return 0 + + df = sanitize_df(df) + + # Фільтруємо до колонок що є в таблиці + if self._schema: + cols = [c for c in df.columns if c in self._schema] + df = df[cols] + + if df.empty: + return 0 + + # DataFrame → CSV у пам'яті (None → порожній рядок = NULL у COPY) + buf = io.StringIO() + df.to_csv(buf, index=False, header=False, na_rep="") + buf.seek(0) + + col_list = ", ".join(f'"{c}"' for c in df.columns) + copy_sql = ( + f"COPY {self._full_table()} ({col_list}) " + f"FROM STDIN WITH (FORMAT CSV, NULL '')" + ) + + conn = self._get_conn() + with conn.cursor() as cur: + cur.copy_expert(copy_sql, buf) + conn.commit() + return len(df) + + def close(self) -> None: + if self._conn is not None: + try: + self._conn.close() + except Exception: + pass + self._conn = None +``` + +- [ ] **Step 4: Перевірити синтаксис** + +```bash +python -c "from olap_tool.sinks import PostgreSQLSink; print('OK')" +``` + +Очікуваний вивід: `OK` + +- [ ] **Step 5: Commit** + +```bash +git add olap_tool/sinks.py +git commit -m "feat: PostgreSQLSink з COPY FROM STDIN" +``` + +--- + +## Chunk 3: Інтеграція + +### Task 4: `runner.py` — підключення sink + +**Files:** +- Modify: `olap_tool/runner.py` + +- [ ] **Step 1: Додати імпорт `PostgreSQLSink` (рядок 27 — поряд з іншими sinks)** + +```python +from .sinks import ClickHouseSink, DuckDBSink, PostgreSQLSink +``` + +- [ ] **Step 2: Додати info-блок для PostgreSQL (після DuckDB info-блоку, ~рядок 253)** + +```python +# PostgreSQL налаштування +if config.postgresql.enabled or export_format in ("PG", "POSTGRESQL"): + print( + f" {Fore.CYAN}PostgreSQL: {Fore.WHITE}{config.postgresql.host}:{config.postgresql.port}" + ) + print( + f" {Fore.CYAN}PG Table: {Fore.WHITE}{config.postgresql.schema}.{config.postgresql.table}" + ) +``` + +- [ ] **Step 3: Додати побудову `PostgreSQLSink` (після DuckDB sink, ~рядок 260)** + +```python +if config.postgresql.enabled or export_format in ("PG", "POSTGRESQL"): + sinks.append(PostgreSQLSink(config.postgresql)) +``` + +- [ ] **Step 4: Commit** + +```bash +git add olap_tool/runner.py +git commit -m "feat: PostgreSQLSink інтеграція у runner.py" +``` + +--- + +### Task 5: `cli.py` — нові значення `--format` + +**Files:** +- Modify: `olap_tool/cli.py` + +- [ ] **Step 1: Оновити `choices` у `--format` (~рядок 113)** + +Знайти: +```python +choices=['xlsx', 'csv', 'both'], +help='Формат експорту: xlsx, csv або both (за замовчуванням з config.yaml)' +``` +Замінити на: +```python +choices=['xlsx', 'csv', 'both', 'ch', 'clickhouse', 'duck', 'duckdb', 'pg', 'postgresql'], +help='Формат експорту: xlsx, csv, both або аналітичний sink: ch/clickhouse, duck/duckdb, pg/postgresql' +``` + +- [ ] **Step 2: Перевірити `--help`** + +```bash +python olap.py --help +``` + +Очікуваний вивід: у секції `--format` повинні бути `pg`, `postgresql`. + +- [ ] **Step 3: Перевірити синтаксис конфігу** + +```bash +python -c "from olap_tool.config import AppConfig, PostgreSQLConfig; c = AppConfig(); print(c.postgresql)" +``` + +Очікуваний вивід: `PostgreSQLConfig(enabled=False, host='localhost', ...)` + +- [ ] **Step 4: Commit** + +```bash +git add olap_tool/cli.py +git commit -m "feat: додано pg/postgresql до --format CLI" +``` + +--- + +## Chunk 4: Фінальна перевірка + +### Task 6: Smoke test та підсумковий commit + +- [ ] **Step 1: Перевірити імпорти всіх нових компонентів** + +```bash +python -c " +from olap_tool.config import PostgreSQLConfig, AppConfig, load_postgres_from_env, build_config +from olap_tool.sinks import PostgreSQLSink, _pandas_dtype_to_pg +import pandas as pd +cfg = PostgreSQLConfig(host='db.lwhs.xyz', port=54321, enabled=True) +print('Config OK:', cfg.host, cfg.port) +df = pd.DataFrame({'a': [1, 2], 'b': ['x', 'y']}) +sink = PostgreSQLSink(cfg) +print('Sink created OK') +print('dtype int →', _pandas_dtype_to_pg(df['a'].dtype)) +print('dtype obj →', _pandas_dtype_to_pg(df['b'].dtype)) +" +``` + +Очікуваний вивід: +``` +Config OK: db.lwhs.xyz 54321 +Sink created OK +dtype int → BIGINT +dtype obj → TEXT +``` + +- [ ] **Step 2: Перевірити build_config з PG env змінними** + +```bash +python -c " +import os +os.environ['PG_ENABLED'] = 'true' +os.environ['PG_HOST'] = 'db.lwhs.xyz' +os.environ['PG_PORT'] = '54321' +os.environ['PG_PASSWORD'] = 'test' +from olap_tool.config import build_config +import argparse +args = argparse.Namespace(format=None, filter=None, timeout=None, compress=None, debug=False) +cfg = build_config(args) +print('PG enabled:', cfg.postgresql.enabled) +print('PG host:', cfg.postgresql.host) +print('PG port:', cfg.postgresql.port) +" +``` + +Очікуваний вивід: +``` +PG enabled: True +PG host: db.lwhs.xyz +PG port: 54321 +``` + +- [ ] **Step 3: Перевірити що `sanitize_df` коректно обробляє NaN перед COPY** + +```bash +python -c " +import pandas as pd +import numpy as np +from olap_tool.sinks import sanitize_df +df = pd.DataFrame({'a': [1.0, np.inf, -np.inf, np.nan], 'b__test': ['x', None, 'y', 'z']}) +result = sanitize_df(df) +print(result) +print('columns:', list(result.columns)) +" +``` + +Очікуваний вивід: колонка `b__test` перейменована у `b__test` (без змін — вже безпечна), `inf` → `NaN`. + +- [ ] **Step 4: Перевірити `--format pg` у CLI** + +```bash +python olap.py --help | grep -A2 "format" +``` + +Очікуваний вивід: у рядку choices є `pg` та `postgresql`. + +- [ ] **Step 5: Підсумковий commit (якщо є незакомічені зміни)** + +```bash +git status +git add -A +git commit -m "feat: PostgreSQL sink повна інтеграція (psycopg2 COPY FROM STDIN)" +``` + +--- + +## Використання + +Після реалізації: + +```bash +# .env — додати: +PG_ENABLED=true +PG_HOST=db.lwhs.xyz +PG_PORT=54321 +PG_DATABASE=analytics +PG_USER=analytics +PG_PASSWORD=1c37552e... +PG_SSL_MODE=require + +# Запуск +python olap.py --last-weeks 4 + +# Або sink-only режим +python olap.py --last-weeks 4 --format pg + +# З дому (LAN) +PG_HOST=192.168.1.111 PG_PORT=5432 python olap.py --last-weeks 4 +``` + +**Профіль** `profiles/pg_export.yaml`: +```yaml +name: pg_export +description: "Щотижневий експорт у PostgreSQL" +postgresql: + enabled: true + table: weekly_sales +period: + type: auto + auto_type: last-weeks + auto_value: 4 +``` diff --git a/docs/superpowers/specs/2026-03-10-postgresql-sink-design.md b/docs/superpowers/specs/2026-03-10-postgresql-sink-design.md new file mode 100644 index 0000000..ebdaf5f --- /dev/null +++ b/docs/superpowers/specs/2026-03-10-postgresql-sink-design.md @@ -0,0 +1,101 @@ +# PostgreSQL Sink — Design Document + +**Date:** 2026-03-10 +**Status:** Approved + +## Overview + +Додати `PostgreSQLSink` як третій аналітичний sink поряд із `ClickHouseSink` і `DuckDBSink`. Дані завантажуються через `COPY FROM STDIN` (psycopg2) — найшвидший метод bulk-завантаження у PostgreSQL. + +## Architecture + +### New: `PostgreSQLConfig` dataclass (`config.py`) + +```python +@dataclass +class PostgreSQLConfig: + enabled: bool = False + host: str = "localhost" + port: int = 5432 + database: str = "analytics" + user: str = "analytics" + password: str = "" + schema: str = "public" + table: str = "sales" + ssl_mode: str = "require" +``` + +### `.env` variables + +``` +PG_ENABLED=true +PG_HOST=db.lwhs.xyz +PG_PORT=54321 +PG_DATABASE=analytics +PG_USER=analytics +PG_PASSWORD=... +PG_SCHEMA=public +PG_TABLE=sales +PG_SSL_MODE=require +``` + +`sslmode=require` — шифрування без перевірки self-signed сертифікату. + +### `PostgreSQLSink` (`sinks.py`) + +Реалізує `AnalyticsSink`: + +- **`setup(df)`** — `CREATE TABLE IF NOT EXISTS` + `ALTER TABLE ADD COLUMN IF NOT EXISTS` за схемою DataFrame. Кешує схему через `information_schema.columns`. +- **`delete_period(year, week)`** — `DELETE WHERE year_num=X AND week_num=Y` (ідемпотентність). +- **`insert(df, year, week)`** — `sanitize_df` → DataFrame → `io.StringIO` CSV → `cursor.copy_expert("COPY ... FROM STDIN WITH (FORMAT CSV, NULL '')")`. +- **`close()`** — `conn.close()` у try/except. + +#### pandas dtype → PostgreSQL type mapping + +| pandas dtype | PostgreSQL type | +|--------------|-------------------| +| int/uint | BIGINT | +| float | DOUBLE PRECISION | +| bool | BOOLEAN | +| datetime | TIMESTAMP | +| date | DATE | +| інші | TEXT | + +### Integration (`runner.py`) + +За патерном ClickHouse/DuckDB: +- Виводить info-рядки про підключення якщо `config.postgresql.enabled` або `--format pg/postgresql` +- Додає `PostgreSQLSink(config.postgresql)` до списку `sinks` + +### CLI (`cli.py`) + +`--format` приймає нові значення: `pg`, `postgresql` + +### Profile support + +```yaml +postgresql: + enabled: true + table: weekly_sales +``` + +`apply_profile()` обробляє секцію `postgresql:`. + +## Files Changed + +| File | Change | +|------|--------| +| `olap_tool/config.py` | `PostgreSQLConfig`, `load_postgres_from_env()`, оновлення `AppConfig`, `apply_profile()`, `build_config()` | +| `olap_tool/sinks.py` | `PostgreSQLSink` клас + `_pandas_dtype_to_pg()` helper | +| `olap_tool/runner.py` | Info display + sink instantiation | +| `olap_tool/cli.py` | Нові значення для `--format` | +| `requirements.txt` | `psycopg2-binary` | +| `.env.example` | `PG_*` змінні | + +## Decisions + +- **Прямий psycopg2** (без SQLAlchemy) — менше залежностей, максимальна швидкість +- **COPY FROM STDIN** — найшвидший метод bulk-завантаження (50-100k рядків/сек) +- **Один хост у `.env`** — користувач перемикає `PG_HOST` вручну між роботою і домом +- **`sslmode=require`** — SSL обов'язковий, self-signed cert не перевіряється +- **Sink у `sinks.py`** — не окремий файл, як DuckDB diff --git a/olap_tool/cli.py b/olap_tool/cli.py index ec6810c..0a513da 100644 --- a/olap_tool/cli.py +++ b/olap_tool/cli.py @@ -110,8 +110,8 @@ def parse_arguments() -> argparse.Namespace: export_group.add_argument( '--format', type=str, - choices=['xlsx', 'csv', 'both'], - help='Формат експорту: xlsx, csv або both (за замовчуванням з config.yaml)' + choices=['xlsx', 'csv', 'both', 'ch', 'clickhouse', 'duck', 'duckdb', 'pg', 'postgresql'], + help='Формат експорту: xlsx, csv, both або аналітичний sink: ch/clickhouse, duck/duckdb, pg/postgresql' ) export_group.add_argument( '--filter', diff --git a/olap_tool/config.py b/olap_tool/config.py index d464984..8cf5e80 100644 --- a/olap_tool/config.py +++ b/olap_tool/config.py @@ -102,6 +102,20 @@ class DuckDBConfig: batch_size: int = 1000 +@dataclass +class PostgreSQLConfig: + """Налаштування підключення до PostgreSQL.""" + enabled: bool = False + host: str = "localhost" + port: int = 5432 + database: str = "analytics" + user: str = "analytics" + password: str = "" + schema: str = "public" + table: str = "sales" + ssl_mode: str = "require" # disable|allow|prefer|require|verify-ca|verify-full + + @dataclass class DisplayConfig: ascii_logs: bool = False @@ -121,6 +135,7 @@ class AppConfig: display: DisplayConfig = field(default_factory=DisplayConfig) clickhouse: ClickHouseConfig = field(default_factory=ClickHouseConfig) duckdb: DuckDBConfig = field(default_factory=DuckDBConfig) + postgresql: PostgreSQLConfig = field(default_factory=PostgreSQLConfig) # --------------------------------------------------------------------------- @@ -190,6 +205,25 @@ def load_clickhouse_from_env() -> ClickHouseConfig: ) +def load_postgres_from_env() -> PostgreSQLConfig: + """Читає налаштування PostgreSQL з os.environ.""" + try: + pg_port = int(os.getenv("PG_PORT", "5432")) + except (ValueError, TypeError): + pg_port = 5432 + return PostgreSQLConfig( + enabled=_parse_bool(os.getenv("PG_ENABLED", "false"), False), + host=os.getenv("PG_HOST", "localhost"), + port=pg_port, + database=os.getenv("PG_DATABASE", "analytics"), + user=os.getenv("PG_USER", "analytics"), + password=os.getenv("PG_PASSWORD", ""), + schema=os.getenv("PG_SCHEMA", "public"), + table=os.getenv("PG_TABLE", "sales"), + ssl_mode=os.getenv("PG_SSL_MODE", "require"), + ) + + # --------------------------------------------------------------------------- # Step 2: load config.yaml # --------------------------------------------------------------------------- @@ -264,7 +298,7 @@ def apply_legacy_env_compat(base: dict) -> dict: def apply_profile(base: dict, profile: dict) -> dict: """Deep-merge секцій профілю поверх base.""" - for section in ("query", "export", "xlsx", "csv", "excel_header", "paths", "display", "clickhouse", "duckdb"): + for section in ("query", "export", "xlsx", "csv", "excel_header", "paths", "display", "clickhouse", "duckdb", "postgresql"): if section in profile: base.setdefault(section, {}) base[section].update(profile[section]) @@ -363,6 +397,13 @@ def build_config(args=None, profile_config: Optional[dict] = None) -> AppConfig: for k, v in duck_env_dict.items(): base["duckdb"].setdefault(k, v) + # PostgreSQL: env задає defaults, profile може перевизначити + pg_env = load_postgres_from_env() + pg_env_dict = {f.name: getattr(pg_env, f.name) for f in dataclass_fields(pg_env)} + base.setdefault("postgresql", {}) + for k, v in pg_env_dict.items(): + base["postgresql"].setdefault(k, v) + # 6. Збираємо AppConfig return AppConfig( secrets=secrets, @@ -375,4 +416,5 @@ def build_config(args=None, profile_config: Optional[dict] = None) -> AppConfig: display=_build_section(DisplayConfig, base, "display"), clickhouse=_build_section(ClickHouseConfig, base, "clickhouse"), duckdb=_build_section(DuckDBConfig, base, "duckdb"), + postgresql=_build_section(PostgreSQLConfig, base, "postgresql"), ) diff --git a/olap_tool/queries.py b/olap_tool/queries.py index b9bb48f..3203207 100644 --- a/olap_tool/queries.py +++ b/olap_tool/queries.py @@ -182,10 +182,10 @@ def run_dax_query( export_format = export_config.format.upper() force_csv_only = export_config.force_csv_only streaming_xlsx = xlsx_config.streaming - ch_only = export_format in ("CH", "CLICKHOUSE", "DUCK", "DUCKDB") + sink_only = export_format in ("CH", "CLICKHOUSE", "DUCK", "DUCKDB", "PG", "POSTGRESQL") # Стрімінговий XLSX (НЕ для режиму clickhouse) - if export_format in ("XLSX", "BOTH") and not force_csv_only and streaming_xlsx and not ch_only: + if export_format in ("XLSX", "BOTH") and not force_csv_only and streaming_xlsx and not sink_only: progress.animation_running = False spinner_thread.join(timeout=1.0) xlsx_path = year_dir / f"{year_num}-{week_num:02d}.xlsx" @@ -215,7 +215,7 @@ def run_dax_query( ) return str(xlsx_path) - if (export_format == "CSV" or force_csv_only) and not ch_only: + if (export_format == "CSV" or force_csv_only) and not sink_only: csv_path = year_dir / f"{year_num}-{week_num:02d}.csv" row_count = export_csv_stream( cursor, csv_path, @@ -296,13 +296,13 @@ def run_dax_query( df.rename(columns=renamed_columns, inplace=True) - if export_format not in ["XLSX", "CSV", "BOTH", "CH", "CLICKHOUSE", "DUCK", "DUCKDB"]: + if export_format not in ["XLSX", "CSV", "BOTH", "CH", "CLICKHOUSE", "DUCK", "DUCKDB", "PG", "POSTGRESQL"]: print_warning( f"Невідомий формат експорту: {export_format}. Використовуємо XLSX." ) export_format = "XLSX" - export_xlsx_flag = export_format in ["XLSX", "BOTH"] and not ch_only - export_csv_flag = export_format in ["CSV", "BOTH"] and not ch_only + export_xlsx_flag = export_format in ["XLSX", "BOTH"] and not sink_only + export_csv_flag = export_format in ["CSV", "BOTH"] and not sink_only exported_files = [] if export_xlsx_flag and not force_csv_only: xlsx_path = year_dir / f"{year_num}-{week_num:02d}.xlsx" @@ -354,7 +354,7 @@ def run_dax_query( except Exception as e: print_error(f"Помилка sink {type(sink).__name__}: {e}") - if ch_only: + if sink_only: return None return exported_files[0][0] if exported_files else None except Exception as e: diff --git a/olap_tool/runner.py b/olap_tool/runner.py index 60a8aa7..ba135cf 100644 --- a/olap_tool/runner.py +++ b/olap_tool/runner.py @@ -24,7 +24,7 @@ from .compression import compress_files from .profiles import load_profile, print_profiles_list from .scheduler import start_scheduler, daemon_mode -from .sinks import ClickHouseSink, DuckDBSink +from .sinks import ClickHouseSink, DuckDBSink, PostgreSQLSink CURRENT_YEAR = datetime.datetime.now().year @@ -252,12 +252,23 @@ def main(argv: list[str] | None = None) -> int: f" {Fore.CYAN}DuckDB Table: {Fore.WHITE}{config.duckdb.table}" ) + # PostgreSQL налаштування + if config.postgresql.enabled or export_format in ("PG", "POSTGRESQL"): + print( + f" {Fore.CYAN}PostgreSQL: {Fore.WHITE}{config.postgresql.host}:{config.postgresql.port}" + ) + print( + f" {Fore.CYAN}PG Table: {Fore.WHITE}{config.postgresql.schema}.{config.postgresql.table}" + ) + # Побудова списку активних analytics sinks sinks = [] if config.clickhouse.enabled or export_format in ("CH", "CLICKHOUSE"): sinks.append(ClickHouseSink(config.clickhouse)) if config.duckdb.enabled or export_format in ("DUCK", "DUCKDB"): sinks.append(DuckDBSink(config.duckdb)) + if config.postgresql.enabled or export_format in ("PG", "POSTGRESQL"): + sinks.append(PostgreSQLSink(config.postgresql)) start_time = time.time() files_created: list[str] = [] diff --git a/olap_tool/sinks.py b/olap_tool/sinks.py index a62826d..3a2b873 100644 --- a/olap_tool/sinks.py +++ b/olap_tool/sinks.py @@ -8,7 +8,9 @@ from __future__ import annotations import datetime +import io import re +import threading from abc import ABC, abstractmethod import numpy as np @@ -19,6 +21,7 @@ if TYPE_CHECKING: from .config import ClickHouseConfig from .config import DuckDBConfig + from .config import PostgreSQLConfig # --------------------------------------------------------------------------- @@ -151,6 +154,7 @@ def _pandas_dtype_to_duck(dtype) -> str: return "VARCHAR" + _EXCEL_EPOCH = datetime.date(1899, 12, 30) _DT_RE = re.compile(r"^\d{4}-\d{2}-\d{2}(?: \d{2}:\d{2}:\d{2})?$") @@ -416,3 +420,182 @@ def close(self) -> None: self._session.close() except Exception: pass + + +# --------------------------------------------------------------------------- +# PostgreSQL sink (psycopg2 + COPY FROM STDIN) +# --------------------------------------------------------------------------- + +def _pandas_dtype_to_pg(dtype) -> str: + """Конвертує pandas dtype у PostgreSQL SQL тип.""" + dtype_str = str(dtype) + if dtype_str.startswith("int") or dtype_str.startswith("uint"): + return "BIGINT" + if dtype_str.startswith("float"): + return "DOUBLE PRECISION" + if dtype_str in ("bool", "boolean"): + return "BOOLEAN" + if dtype_str.startswith("datetime"): + return "TIMESTAMP" + if dtype_str.startswith("date"): + return "DATE" + return "TEXT" + +class PostgreSQLSink(AnalyticsSink): + """ + Завантажує DataFrame у PostgreSQL через COPY FROM STDIN. + + Ідемпотентність: DELETE WHERE year_num=X AND week_num=Y → COPY FROM STDIN CSV. + SSL: sslmode=require (шифрування без перевірки self-signed сертифікату). + + NOT thread-safe: psycopg2-з'єднання не підтримують спільне використання між + потоками. Для batch-скриптів з threading створюйте окремий екземпляр на кожен потік. + """ + + def __init__(self, config: "PostgreSQLConfig"): + self._config = config + self._conn = None + self._schema: dict[str, str] | None = None + self._schema_lock = threading.Lock() + + def _get_conn(self): + """Повертає активне з'єднання, створює нове якщо потрібно.""" + import psycopg2 + if self._conn is None or self._conn.closed: + self._conn = psycopg2.connect( + host=self._config.host, + port=self._config.port, + dbname=self._config.database, + user=self._config.user, + password=self._config.password, + sslmode=self._config.ssl_mode, + ) + self._conn.autocommit = False + return self._conn + + def _full_table(self) -> str: + """Повертає повну назву таблиці з схемою: "schema"."table".""" + return f'"{self._config.schema}"."{self._config.table}"' + + def _refresh_schema(self) -> None: + """Читає поточну схему таблиці з information_schema.""" + conn = self._get_conn() + with conn.cursor() as cur: + cur.execute( + """ + SELECT column_name, data_type + FROM information_schema.columns + WHERE table_schema = %s AND table_name = %s + ORDER BY ordinal_position + """, + (self._config.schema, self._config.table), + ) + rows = cur.fetchall() + with self._schema_lock: + self._schema = {row[0]: row[1] for row in rows} + + def setup(self, df: pd.DataFrame) -> None: + from .utils import print_progress, print_warning + print_progress( + f"Перевірка таблиці PostgreSQL {self._full_table()} " + f"({self._config.host}:{self._config.port})..." + ) + conn = self._get_conn() + cols_ddl = ", ".join( + f'"{col}" {_pandas_dtype_to_pg(df[col].dtype)}' + for col in df.columns + ) + try: + with conn.cursor() as cur: + cur.execute( + f"CREATE TABLE IF NOT EXISTS {self._full_table()} ({cols_ddl})" + ) + conn.commit() + except Exception: + conn.rollback() + raise + self._refresh_schema() + with self._schema_lock: + schema = dict(self._schema) if self._schema is not None else {} + + # Додаємо нові колонки яких немає в таблиці + for col in df.columns: + if col not in schema: + dtype = _pandas_dtype_to_pg(df[col].dtype) + try: + with conn.cursor() as cur: + cur.execute( + f'ALTER TABLE {self._full_table()} ' + f'ADD COLUMN IF NOT EXISTS "{col}" {dtype}' + ) + conn.commit() + with self._schema_lock: + if self._schema is not None: + self._schema[col] = dtype + except Exception as e: + conn.rollback() + print_warning(f"Не вдалося додати колонку `{col}`: {e} — пропускаємо") + + def delete_period(self, year: int, week: int) -> None: + if self._schema is None: + self._refresh_schema() + with self._schema_lock: + schema = dict(self._schema) if self._schema is not None else {} + if "year_num" not in schema or "week_num" not in schema: + return + conn = self._get_conn() + try: + with conn.cursor() as cur: + cur.execute( + f"DELETE FROM {self._full_table()} " + f"WHERE year_num = %s AND week_num = %s", + (year, week), + ) + conn.commit() + except Exception: + conn.rollback() + raise + + def insert(self, df: pd.DataFrame, year: int, week: int) -> int: + if df is None or len(df) == 0: + return 0 + + # Фільтруємо до колонок що є в таблиці + with self._schema_lock: + schema = dict(self._schema) if self._schema else {} + if schema: + cols = [c for c in df.columns if c in schema] + df = df[cols] + + if df.empty: + return 0 + + # DataFrame → CSV у пам'яті; \N як sentinel для NULL + # (порожній рядок '' зберігається як '', а не як NULL) + buf = io.StringIO() + df.to_csv(buf, index=False, header=False, na_rep="\\N") + buf.seek(0) + + col_list = ", ".join(f'"{c}"' for c in df.columns) + copy_sql = ( + f"COPY {self._full_table()} ({col_list}) " + r"FROM STDIN WITH (FORMAT CSV, NULL '\N')" + ) + + conn = self._get_conn() + try: + with conn.cursor() as cur: + cur.copy_expert(copy_sql, buf) + conn.commit() + except Exception: + conn.rollback() + raise + return len(df) + + def close(self) -> None: + if self._conn is not None: + try: + self._conn.close() + except Exception: + pass + self._conn = None diff --git a/requirements.txt b/requirements.txt index 0bdf8ae..aaf9608 100644 --- a/requirements.txt +++ b/requirements.txt @@ -18,6 +18,7 @@ PyYAML>=6.0.0 # Для профілів конфігурації (YAML форм schedule>=1.2.0 # Для вбудованого планувальника задач clickhouse-connect>=0.7.0 # Для завантаження даних у ClickHouse requests>=2.28.0 # Для завантаження даних у DuckDB через REST API +psycopg2-binary>=2.9.0 # Для завантаження даних у PostgreSQL через COPY FROM STDIN openpyxl>=3.0.0 # Для читання Excel-файлів (import_xlsx_to_clickhouse.py) python-calamine>=0.1.7 # Rust-based Excel reader, 3-10x швидший за openpyxl rich>=13.0.0 # Красивий термінальний UI: progress bar, панелі, таблиціpyarrow>=14.0.0