diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..5663e60 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,30 @@ +name: CI + +on: + push: + branches: ["**"] + pull_request: + branches: ["**"] + +jobs: + test: + name: Lint & Test + runs-on: ubuntu-latest + permissions: + contents: read + + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: pip + + - name: Install dependencies + run: pip install -r requirements.txt + + - name: Run tests + run: pytest tests/ -v --tb=short diff --git a/.github/workflows/daily_migration.yml b/.github/workflows/daily_migration.yml new file mode 100644 index 0000000..5019b85 --- /dev/null +++ b/.github/workflows/daily_migration.yml @@ -0,0 +1,86 @@ +name: Daily DWH Migration (Oracle → PostgreSQL) + +on: + # Run every day at 02:00 UTC + schedule: + - cron: "0 2 * * *" + + # Allow manual runs from the GitHub UI / API + workflow_dispatch: + inputs: + tables: + description: "Comma-separated list of tables to migrate (leave empty for all)" + required: false + default: "" + mode: + description: "Migration mode: full or incremental" + required: false + default: "full" + dry_run: + description: "Dry run (no data written)" + required: false + default: "false" + +env: + # Oracle connection — stored as repository / environment secrets + ORACLE_HOST: ${{ secrets.ORACLE_HOST }} + ORACLE_PORT: ${{ secrets.ORACLE_PORT }} + ORACLE_SERVICE: ${{ secrets.ORACLE_SERVICE }} + ORACLE_USER: ${{ secrets.ORACLE_USER }} + ORACLE_PASSWORD: ${{ secrets.ORACLE_PASSWORD }} + + # PostgreSQL connection — stored as repository / environment secrets + PG_HOST: ${{ secrets.PG_HOST }} + PG_PORT: ${{ secrets.PG_PORT }} + PG_DATABASE: ${{ secrets.PG_DATABASE }} + PG_USER: ${{ secrets.PG_USER }} + PG_PASSWORD: ${{ secrets.PG_PASSWORD }} + + # Migration settings — can be overridden at runtime + MIGRATION_MODE: ${{ github.event.inputs.mode || 'full' }} + MIGRATION_TABLES: ${{ github.event.inputs.tables || '' }} + LOG_LEVEL: INFO + +jobs: + migrate: + name: Migrate Oracle DWH → PostgreSQL + runs-on: ubuntu-latest + environment: production # use a protected environment for secrets approval + permissions: + contents: read + + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: pip + + - name: Install dependencies + run: pip install -r requirements.txt + + - name: Run migration + run: | + EXTRA_FLAGS="" + if [ "${{ github.event.inputs.dry_run }}" = "true" ]; then + EXTRA_FLAGS="--dry-run" + fi + if [ -n "${{ github.event.inputs.tables }}" ]; then + EXTRA_FLAGS="$EXTRA_FLAGS --tables ${{ github.event.inputs.tables }}" + fi + if [ -n "${{ github.event.inputs.mode }}" ]; then + EXTRA_FLAGS="$EXTRA_FLAGS --mode ${{ github.event.inputs.mode }}" + fi + python migrate.py $EXTRA_FLAGS + + - name: Upload migration log + if: always() + uses: actions/upload-artifact@v4 + with: + name: migration-log-${{ github.run_id }} + path: "*.log" + if-no-files-found: ignore + retention-days: 30 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..58df291 --- /dev/null +++ b/.gitignore @@ -0,0 +1,19 @@ +# Python bytecode +__pycache__/ +*.py[cod] +*.pyo + +# Virtual environments +.venv/ +venv/ +env/ + +# Pytest +.pytest_cache/ +.coverage +htmlcov/ + +# Distribution / packaging +dist/ +build/ +*.egg-info/ diff --git a/README.md b/README.md index 00bcb6e..9f81b0e 100644 --- a/README.md +++ b/README.md @@ -1 +1,105 @@ -# test \ No newline at end of file +# Oracle → PostgreSQL DWH Migration Tool + +A Python script that migrates tables from an Oracle Data Warehouse to PostgreSQL, +with a daily GitHub Actions schedule. + +## Project structure + +``` +├── migrate.py – Main migration script +├── type_mapping.py – Oracle-to-PostgreSQL type conversion +├── config.py – Configuration via environment variables +├── requirements.txt – Python dependencies +├── tests/ +│ ├── test_type_mapping.py +│ ├── test_config.py +│ └── test_migrate.py +└── .github/ + └── workflows/ + ├── daily_migration.yml – Scheduled migration (daily @ 02:00 UTC) + └── ci.yml – CI: run tests on every push / PR +``` + +## Quick start + +### 1. Install dependencies + +```bash +pip install -r requirements.txt +``` + +### 2. Set environment variables + +| Variable | Description | Default | +|---|---|---| +| `ORACLE_HOST` | Oracle server hostname | **required** | +| `ORACLE_PORT` | Oracle listener port | `1521` | +| `ORACLE_SERVICE` | Oracle service name or SID | **required** | +| `ORACLE_USER` | Oracle schema/user | **required** | +| `ORACLE_PASSWORD` | Oracle password | **required** | +| `PG_HOST` | PostgreSQL hostname | **required** | +| `PG_PORT` | PostgreSQL port | `5432` | +| `PG_DATABASE` | PostgreSQL database name | **required** | +| `PG_SCHEMA` | Target PostgreSQL schema | `public` | +| `PG_USER` | PostgreSQL user | **required** | +| `PG_PASSWORD` | PostgreSQL password | **required** | +| `MIGRATION_TABLES` | Comma-separated table list (empty = all) | *(all tables)* | +| `MIGRATION_BATCH_SIZE` | Rows fetched per batch | `10000` | +| `MIGRATION_MODE` | `full` (truncate+reload) or `incremental` | `full` | +| `LOG_LEVEL` | Python logging level | `INFO` | + +### 3. Run the migration + +```bash +# Migrate all tables (full reload) +python migrate.py + +# Migrate specific tables +python migrate.py --tables ORDERS,CUSTOMERS,PRODUCTS + +# Incremental mode (append only) +python migrate.py --mode incremental + +# Dry run – connect to both DBs but write nothing +python migrate.py --dry-run +``` + +## GitHub Actions + +### Daily schedule + +The workflow `.github/workflows/daily_migration.yml` runs automatically every day +at **02:00 UTC** (`cron: "0 2 * * *"`). + +You can also trigger it manually from the **Actions** tab, optionally overriding +the table list, mode, and dry-run flag. + +### Secrets required + +Add the following secrets to your repository (or the `production` environment): + +- `ORACLE_HOST`, `ORACLE_PORT`, `ORACLE_SERVICE`, `ORACLE_USER`, `ORACLE_PASSWORD` +- `PG_HOST`, `PG_PORT`, `PG_DATABASE`, `PG_USER`, `PG_PASSWORD` + +### CI + +The workflow `.github/workflows/ci.yml` runs unit tests on every push and pull +request. + +## Running tests + +```bash +pytest tests/ -v +``` + +## How it works + +1. **Schema discovery** – the script queries `ALL_TAB_COLUMNS` in Oracle to + retrieve the full column list (name, data type, precision, scale, nullability). +2. **Type mapping** – Oracle types are converted to the closest PostgreSQL + equivalents (see `type_mapping.py`). +3. **DDL** – `CREATE TABLE IF NOT EXISTS` is executed in PostgreSQL. +4. **Data transfer** – rows are streamed from Oracle in configurable batches and + bulk-inserted into PostgreSQL using `psycopg2.extras.execute_values`. +5. **Full vs incremental** – in `full` mode the target table is truncated before + loading; in `incremental` mode rows are appended. \ No newline at end of file diff --git a/config.py b/config.py new file mode 100644 index 0000000..866e890 --- /dev/null +++ b/config.py @@ -0,0 +1,117 @@ +"""Configuration helpers for the Oracle → PostgreSQL migration tool. + +All sensitive connection parameters are read from environment variables so +that no credentials are stored in source code or configuration files. + +Required environment variables +-------------------------------- +Oracle (source) + ORACLE_HOST – hostname or IP of the Oracle server + ORACLE_PORT – listener port (default: 1521) + ORACLE_SERVICE – service name *or* SID + ORACLE_USER – schema / user name + ORACLE_PASSWORD – password + +PostgreSQL (destination) + PG_HOST – hostname or IP of the PostgreSQL server + PG_PORT – port (default: 5432) + PG_DATABASE – database name + PG_SCHEMA – target schema (default: public) + PG_USER – user name + PG_PASSWORD – password + +Optional + MIGRATION_TABLES – comma-separated list of tables to migrate (default: all) + MIGRATION_BATCH_SIZE – rows fetched per batch (default: 10000) + MIGRATION_MODE – 'full' (truncate+reload) or 'incremental' (default: full) + LOG_LEVEL – Python logging level name (default: INFO) +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field + + +def _require_env(name: str) -> str: + value = os.environ.get(name) + if not value: + raise EnvironmentError( + f"Required environment variable '{name}' is not set." + ) + return value + + +def _optional_env(name: str, default: str) -> str: + return os.environ.get(name, default) + + +@dataclass(frozen=True) +class OracleConfig: + host: str + port: int + service: str + user: str + password: str + + @classmethod + def from_env(cls) -> "OracleConfig": + return cls( + host=_require_env("ORACLE_HOST"), + port=int(_optional_env("ORACLE_PORT", "1521")), + service=_require_env("ORACLE_SERVICE"), + user=_require_env("ORACLE_USER"), + password=_require_env("ORACLE_PASSWORD"), + ) + + @property + def dsn(self) -> str: + return f"{self.host}:{self.port}/{self.service}" + + +@dataclass(frozen=True) +class PostgresConfig: + host: str + port: int + database: str + user: str + password: str + schema: str = "public" + + @classmethod + def from_env(cls) -> "PostgresConfig": + return cls( + host=_require_env("PG_HOST"), + port=int(_optional_env("PG_PORT", "5432")), + database=_require_env("PG_DATABASE"), + user=_require_env("PG_USER"), + password=_require_env("PG_PASSWORD"), + schema=_optional_env("PG_SCHEMA", "public"), + ) + + +@dataclass(frozen=True) +class MigrationConfig: + oracle: OracleConfig + postgres: PostgresConfig + tables: list[str] = field(default_factory=list) + batch_size: int = 10_000 + mode: str = "full" # 'full' | 'incremental' + log_level: str = "INFO" + + @classmethod + def from_env(cls) -> "MigrationConfig": + tables_raw = _optional_env("MIGRATION_TABLES", "") + tables = [t.strip() for t in tables_raw.split(",") if t.strip()] + batch_size = int(_optional_env("MIGRATION_BATCH_SIZE", "10000")) + mode = _optional_env("MIGRATION_MODE", "full").lower() + if mode not in ("full", "incremental"): + raise ValueError(f"MIGRATION_MODE must be 'full' or 'incremental', got '{mode}'") + return cls( + oracle=OracleConfig.from_env(), + postgres=PostgresConfig.from_env(), + tables=tables, + batch_size=batch_size, + mode=mode, + log_level=_optional_env("LOG_LEVEL", "INFO").upper(), + ) diff --git a/migrate.py b/migrate.py new file mode 100644 index 0000000..b8bb364 --- /dev/null +++ b/migrate.py @@ -0,0 +1,319 @@ +"""Oracle to PostgreSQL DWH migration script. + +Usage +----- + python migrate.py [--tables TABLE1,TABLE2] [--mode full|incremental] + [--batch-size N] [--dry-run] + +All database connection parameters are supplied through environment variables +(see config.py). Pass --help for a full list of command-line options. +""" + +from __future__ import annotations + +import argparse +import logging +import sys +import time +from contextlib import contextmanager +from typing import Generator + +import oracledb +import psycopg2 +import psycopg2.extras + +from config import MigrationConfig, OracleConfig, PostgresConfig +from type_mapping import map_column_type + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Connection helpers +# --------------------------------------------------------------------------- + +@contextmanager +def oracle_connection(cfg: OracleConfig) -> Generator[oracledb.Connection, None, None]: + """Yield an Oracle connection and close it when done.""" + conn = oracledb.connect(user=cfg.user, password=cfg.password, dsn=cfg.dsn) + try: + yield conn + finally: + conn.close() + + +@contextmanager +def postgres_connection(cfg: PostgresConfig) -> Generator[psycopg2.extensions.connection, None, None]: + """Yield a PostgreSQL connection and close it when done.""" + conn = psycopg2.connect( + host=cfg.host, + port=cfg.port, + dbname=cfg.database, + user=cfg.user, + password=cfg.password, + ) + conn.autocommit = False + try: + yield conn + finally: + conn.close() + + +# --------------------------------------------------------------------------- +# Schema helpers +# --------------------------------------------------------------------------- + +def get_tables(oracle_cur: oracledb.Cursor, schema: str) -> list[str]: + """Return all user-visible table names in *schema* (upper-cased).""" + oracle_cur.execute( + "SELECT table_name FROM all_tables WHERE owner = :owner ORDER BY table_name", + owner=schema.upper(), + ) + return [row[0] for row in oracle_cur.fetchall()] + + +ColumnInfo = tuple[str, str, int | None, int | None, bool] +# (name, oracle_type, precision, scale, nullable) + + +def get_columns(oracle_cur: oracledb.Cursor, schema: str, table: str) -> list[ColumnInfo]: + """Return column metadata for *schema*.*table* from Oracle.""" + oracle_cur.execute( + """ + SELECT column_name, + data_type, + data_precision, + data_scale, + nullable + FROM all_tab_columns + WHERE owner = :owner + AND table_name = :table_name + ORDER BY column_id + """, + owner=schema.upper(), + table_name=table.upper(), + ) + return [ + (row[0], row[1], row[2], row[3], row[4] == "Y") + for row in oracle_cur.fetchall() + ] + + +def build_create_table_sql(pg_schema: str, table: str, columns: list[ColumnInfo]) -> str: + """Generate a PostgreSQL CREATE TABLE statement from Oracle column metadata.""" + col_defs: list[str] = [] + for name, oracle_type, precision, scale, nullable in columns: + pg_type = map_column_type(oracle_type, precision, scale) + null_clause = "" if nullable else " NOT NULL" + col_defs.append(f' "{name.lower()}" {pg_type}{null_clause}') + cols_sql = ",\n".join(col_defs) + return f'CREATE TABLE IF NOT EXISTS "{pg_schema}"."{table.lower()}" (\n{cols_sql}\n);' + + +# --------------------------------------------------------------------------- +# Migration logic +# --------------------------------------------------------------------------- + +def migrate_table( + oracle_conn: oracledb.Connection, + pg_conn: psycopg2.extensions.connection, + oracle_schema: str, + pg_schema: str, + table: str, + batch_size: int, + mode: str, + dry_run: bool, +) -> int: + """Migrate a single table from Oracle to PostgreSQL. + + Returns the total number of rows copied. + """ + oracle_cur = oracle_conn.cursor() + pg_cur = pg_conn.cursor() + + # Fetch column metadata from Oracle + columns = get_columns(oracle_cur, oracle_schema, table) + if not columns: + logger.warning("Table %s.%s not found in Oracle; skipping.", oracle_schema, table) + return 0 + + col_names = [col[0] for col in columns] + pg_col_names = [f'"{c.lower()}"' for c in col_names] + + # Ensure target table exists in PostgreSQL + create_sql = build_create_table_sql(pg_schema, table, columns) + logger.debug("DDL:\n%s", create_sql) + if not dry_run: + pg_cur.execute(f'CREATE SCHEMA IF NOT EXISTS "{pg_schema}";') + pg_cur.execute(create_sql) + pg_conn.commit() + + # Truncate on full reload + if mode == "full" and not dry_run: + pg_cur.execute(f'TRUNCATE TABLE "{pg_schema}"."{table.lower()}";') + pg_conn.commit() + + # Stream data from Oracle and bulk-insert into PostgreSQL + # Column names are double-quoted to handle reserved words and special characters. + quoted_col_names = [f'"{c}"' for c in col_names] + select_sql = ( + f'SELECT {", ".join(quoted_col_names)} ' + f'FROM "{oracle_schema}"."{table}"' + ) + oracle_cur.arraysize = batch_size + oracle_cur.execute(select_sql) + + insert_sql = ( + f'INSERT INTO "{pg_schema}"."{table.lower()}" ' + f"({', '.join(pg_col_names)}) " + f"VALUES %s" + ) + + total_rows = 0 + while True: + rows = oracle_cur.fetchmany(batch_size) + if not rows: + break + total_rows += len(rows) + if not dry_run: + psycopg2.extras.execute_values(pg_cur, insert_sql, rows, page_size=batch_size) + pg_conn.commit() + logger.debug(" … %d rows inserted into %s.%s", total_rows, pg_schema, table) + + oracle_cur.close() + pg_cur.close() + return total_rows + + +def run_migration(cfg: MigrationConfig, dry_run: bool = False) -> dict[str, int]: + """Run the full migration according to *cfg*. + + Returns a dict mapping table name → row count. + """ + results: dict[str, int] = {} + + with oracle_connection(cfg.oracle) as oracle_conn, \ + postgres_connection(cfg.postgres) as pg_conn: + + oracle_cur = oracle_conn.cursor() + + # Resolve which tables to migrate + if cfg.tables: + tables = [t.upper() for t in cfg.tables] + else: + tables = get_tables(oracle_cur, cfg.oracle.user) + logger.info("Discovered %d tables in schema %s", len(tables), cfg.oracle.user) + + oracle_cur.close() + + for table in tables: + logger.info("Migrating table %s …", table) + start = time.monotonic() + rows = migrate_table( + oracle_conn=oracle_conn, + pg_conn=pg_conn, + oracle_schema=cfg.oracle.user, + pg_schema=cfg.postgres.schema, + table=table, + batch_size=cfg.batch_size, + mode=cfg.mode, + dry_run=dry_run, + ) + elapsed = time.monotonic() - start + logger.info( + " ✓ %s – %d rows in %.1f s (%.0f rows/s)", + table, + rows, + elapsed, + rows / elapsed if elapsed > 0 else 0, + ) + results[table] = rows + + return results + + +# --------------------------------------------------------------------------- +# CLI entry-point +# --------------------------------------------------------------------------- + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Migrate tables from Oracle DWH to PostgreSQL." + ) + parser.add_argument( + "--tables", + default="", + help="Comma-separated list of tables to migrate (default: all tables).", + ) + parser.add_argument( + "--mode", + choices=["full", "incremental"], + default=None, + help="Migration mode: 'full' (truncate+reload) or 'incremental'. " + "Overrides MIGRATION_MODE env var.", + ) + parser.add_argument( + "--batch-size", + type=int, + default=None, + help="Number of rows per batch. Overrides MIGRATION_BATCH_SIZE env var.", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Connect to both databases and enumerate tables/columns but do not write data.", + ) + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + + cfg = MigrationConfig.from_env() + + # CLI arguments override environment variables + if args.tables: + cfg = cfg.__class__( + oracle=cfg.oracle, + postgres=cfg.postgres, + tables=[t.strip() for t in args.tables.split(",") if t.strip()], + batch_size=args.batch_size or cfg.batch_size, + mode=args.mode or cfg.mode, + log_level=cfg.log_level, + ) + elif args.mode or args.batch_size: + cfg = cfg.__class__( + oracle=cfg.oracle, + postgres=cfg.postgres, + tables=cfg.tables, + batch_size=args.batch_size or cfg.batch_size, + mode=args.mode or cfg.mode, + log_level=cfg.log_level, + ) + + logging.basicConfig( + level=getattr(logging, cfg.log_level, logging.INFO), + format="%(asctime)s %(levelname)-8s %(name)s – %(message)s", + datefmt="%Y-%m-%dT%H:%M:%S", + ) + + if args.dry_run: + logger.info("DRY RUN – no data will be written to PostgreSQL.") + + logger.info( + "Starting migration: Oracle %s → PostgreSQL %s (mode=%s, batch=%d)", + cfg.oracle.dsn, + cfg.postgres.database, + cfg.mode, + cfg.batch_size, + ) + + results = run_migration(cfg, dry_run=args.dry_run) + + total = sum(results.values()) + logger.info("Migration complete. %d tables, %d total rows.", len(results), total) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..982c3ab --- /dev/null +++ b/requirements.txt @@ -0,0 +1,8 @@ +# Oracle to PostgreSQL DWH migration tool +# Runtime dependencies +oracledb>=2.0.0 +psycopg2-binary>=2.9.0 + +# Testing +pytest>=8.0.0 +pytest-cov>=5.0.0 diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..d653b43 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,102 @@ +"""Tests for the configuration module.""" + +import os +import pytest + +from config import MigrationConfig, OracleConfig, PostgresConfig + + +ORACLE_ENV = { + "ORACLE_HOST": "oracle-host", + "ORACLE_PORT": "1521", + "ORACLE_SERVICE": "ORCL", + "ORACLE_USER": "dwh_user", + "ORACLE_PASSWORD": "secret", +} + +PG_ENV = { + "PG_HOST": "pg-host", + "PG_PORT": "5432", + "PG_DATABASE": "dwh_db", + "PG_USER": "pg_user", + "PG_PASSWORD": "pg_secret", +} + + +class TestOracleConfig: + def test_from_env_success(self, monkeypatch): + for k, v in ORACLE_ENV.items(): + monkeypatch.setenv(k, v) + cfg = OracleConfig.from_env() + assert cfg.host == "oracle-host" + assert cfg.port == 1521 + assert cfg.service == "ORCL" + assert cfg.dsn == "oracle-host:1521/ORCL" + + def test_missing_required_raises(self, monkeypatch): + monkeypatch.delenv("ORACLE_HOST", raising=False) + with pytest.raises(EnvironmentError, match="ORACLE_HOST"): + OracleConfig.from_env() + + def test_default_port(self, monkeypatch): + for k, v in ORACLE_ENV.items(): + monkeypatch.setenv(k, v) + monkeypatch.delenv("ORACLE_PORT", raising=False) + cfg = OracleConfig.from_env() + assert cfg.port == 1521 + + +class TestPostgresConfig: + def test_from_env_success(self, monkeypatch): + for k, v in PG_ENV.items(): + monkeypatch.setenv(k, v) + cfg = PostgresConfig.from_env() + assert cfg.host == "pg-host" + assert cfg.port == 5432 + assert cfg.database == "dwh_db" + + def test_missing_required_raises(self, monkeypatch): + monkeypatch.delenv("PG_HOST", raising=False) + with pytest.raises(EnvironmentError, match="PG_HOST"): + PostgresConfig.from_env() + + def test_default_port(self, monkeypatch): + for k, v in PG_ENV.items(): + monkeypatch.setenv(k, v) + monkeypatch.delenv("PG_PORT", raising=False) + cfg = PostgresConfig.from_env() + assert cfg.port == 5432 + + +class TestMigrationConfig: + def _set_all_env(self, monkeypatch): + for k, v in {**ORACLE_ENV, **PG_ENV}.items(): + monkeypatch.setenv(k, v) + + def test_defaults(self, monkeypatch): + self._set_all_env(monkeypatch) + monkeypatch.delenv("MIGRATION_TABLES", raising=False) + monkeypatch.delenv("MIGRATION_BATCH_SIZE", raising=False) + monkeypatch.delenv("MIGRATION_MODE", raising=False) + cfg = MigrationConfig.from_env() + assert cfg.tables == [] + assert cfg.batch_size == 10_000 + assert cfg.mode == "full" + + def test_custom_tables(self, monkeypatch): + self._set_all_env(monkeypatch) + monkeypatch.setenv("MIGRATION_TABLES", "ORDERS, CUSTOMERS, PRODUCTS") + cfg = MigrationConfig.from_env() + assert cfg.tables == ["ORDERS", "CUSTOMERS", "PRODUCTS"] + + def test_incremental_mode(self, monkeypatch): + self._set_all_env(monkeypatch) + monkeypatch.setenv("MIGRATION_MODE", "incremental") + cfg = MigrationConfig.from_env() + assert cfg.mode == "incremental" + + def test_invalid_mode_raises(self, monkeypatch): + self._set_all_env(monkeypatch) + monkeypatch.setenv("MIGRATION_MODE", "delta") + with pytest.raises(ValueError, match="MIGRATION_MODE"): + MigrationConfig.from_env() diff --git a/tests/test_migrate.py b/tests/test_migrate.py new file mode 100644 index 0000000..fbaaa99 --- /dev/null +++ b/tests/test_migrate.py @@ -0,0 +1,163 @@ +"""Tests for migrate.py – uses mocks to avoid real DB connections.""" + +from __future__ import annotations + +import pytest +from unittest.mock import MagicMock, patch, call + +import migrate +from migrate import ( + build_create_table_sql, + get_columns, + get_tables, + migrate_table, +) + + +# --------------------------------------------------------------------------- +# build_create_table_sql +# --------------------------------------------------------------------------- + +class TestBuildCreateTableSql: + def test_basic_table(self): + columns = [ + ("ID", "NUMBER", 10, 0, False), + ("NAME", "VARCHAR2", 100, None, True), + ("CREATED_AT", "DATE", None, None, True), + ] + sql = build_create_table_sql("public", "ORDERS", columns) + assert 'CREATE TABLE IF NOT EXISTS "public"."orders"' in sql + assert '"id" BIGINT NOT NULL' in sql + assert '"name" VARCHAR(100)' in sql + assert '"created_at" TIMESTAMP' in sql + + def test_nullable_column_has_no_not_null(self): + columns = [("NOTES", "CLOB", None, None, True)] + sql = build_create_table_sql("public", "T", columns) + assert "NOT NULL" not in sql + + def test_non_nullable_column_has_not_null(self): + columns = [("CODE", "CHAR", 3, None, False)] + sql = build_create_table_sql("public", "T", columns) + assert '"code" CHAR(3) NOT NULL' in sql + + +# --------------------------------------------------------------------------- +# get_tables +# --------------------------------------------------------------------------- + +class TestGetTables: + def test_returns_table_names(self): + mock_cur = MagicMock() + mock_cur.fetchall.return_value = [("ORDERS",), ("CUSTOMERS",)] + result = get_tables(mock_cur, "DWH") + assert result == ["ORDERS", "CUSTOMERS"] + mock_cur.execute.assert_called_once() + # Schema should be upper-cased in query + args = mock_cur.execute.call_args + assert args[1]["owner"] == "DWH" + + +# --------------------------------------------------------------------------- +# get_columns +# --------------------------------------------------------------------------- + +class TestGetColumns: + def test_returns_column_info(self): + mock_cur = MagicMock() + mock_cur.fetchall.return_value = [ + ("ID", "NUMBER", 10, 0, "N"), + ("NAME", "VARCHAR2", 200, None, "Y"), + ] + result = get_columns(mock_cur, "DWH", "ORDERS") + assert len(result) == 2 + assert result[0] == ("ID", "NUMBER", 10, 0, False) + assert result[1] == ("NAME", "VARCHAR2", 200, None, True) + + +# --------------------------------------------------------------------------- +# migrate_table – dry run +# --------------------------------------------------------------------------- + +class TestMigrateTableDryRun: + def _make_oracle_conn(self, columns, rows): + mock_cur = MagicMock() + mock_cur.fetchall.return_value = columns + mock_cur.fetchmany.side_effect = [rows, []] + mock_conn = MagicMock() + mock_conn.cursor.return_value = mock_cur + return mock_conn, mock_cur + + def test_dry_run_does_not_execute_write(self): + columns = [("ID", "NUMBER", 10, 0, "N"), ("VAL", "VARCHAR2", 50, None, "Y")] + rows = [(1, "alpha"), (2, "beta")] + + mock_ora_cur = MagicMock() + # get_columns call + mock_ora_cur.fetchall.return_value = columns + # data fetch + mock_ora_cur.fetchmany.side_effect = [rows, []] + mock_ora_conn = MagicMock() + mock_ora_conn.cursor.return_value = mock_ora_cur + + mock_pg_cur = MagicMock() + mock_pg_conn = MagicMock() + mock_pg_conn.cursor.return_value = mock_pg_cur + + count = migrate_table( + oracle_conn=mock_ora_conn, + pg_conn=mock_pg_conn, + oracle_schema="DWH", + pg_schema="public", + table="ORDERS", + batch_size=1000, + mode="full", + dry_run=True, + ) + + assert count == 2 + # In dry-run mode, no DDL or DML should be executed on Postgres + mock_pg_cur.execute.assert_not_called() + mock_pg_conn.commit.assert_not_called() + + def test_empty_table_skipped_gracefully(self): + mock_ora_cur = MagicMock() + mock_ora_cur.fetchall.return_value = [] # no columns → table not found + mock_ora_conn = MagicMock() + mock_ora_conn.cursor.return_value = mock_ora_cur + + mock_pg_conn = MagicMock() + + count = migrate_table( + oracle_conn=mock_ora_conn, + pg_conn=mock_pg_conn, + oracle_schema="DWH", + pg_schema="public", + table="MISSING_TABLE", + batch_size=1000, + mode="full", + dry_run=True, + ) + assert count == 0 + + +# --------------------------------------------------------------------------- +# CLI argument parsing +# --------------------------------------------------------------------------- + +class TestParseArgs: + def test_defaults(self): + args = migrate.parse_args([]) + assert args.tables == "" + assert args.mode is None + assert args.batch_size is None + assert args.dry_run is False + + def test_dry_run_flag(self): + args = migrate.parse_args(["--dry-run"]) + assert args.dry_run is True + + def test_tables_and_mode(self): + args = migrate.parse_args(["--tables", "A,B", "--mode", "incremental"]) + assert args.tables == "A,B" + assert args.mode == "incremental" diff --git a/tests/test_type_mapping.py b/tests/test_type_mapping.py new file mode 100644 index 0000000..8d4e522 --- /dev/null +++ b/tests/test_type_mapping.py @@ -0,0 +1,92 @@ +"""Tests for the Oracle → PostgreSQL type mapping module.""" + +import pytest + +from type_mapping import map_column_type + + +# --------------------------------------------------------------------------- +# NUMBER mappings +# --------------------------------------------------------------------------- + +class TestNumberMapping: + def test_number_no_precision_returns_numeric(self): + assert map_column_type("NUMBER", None, None) == "NUMERIC" + + def test_number_with_fractional_scale(self): + assert map_column_type("NUMBER", 10, 2) == "NUMERIC(10, 2)" + + def test_number_precision_4_scale_0_is_smallint(self): + assert map_column_type("NUMBER", 4, 0) == "SMALLINT" + + def test_number_precision_9_scale_0_is_integer(self): + assert map_column_type("NUMBER", 9, 0) == "INTEGER" + + def test_number_precision_18_scale_0_is_bigint(self): + assert map_column_type("NUMBER", 18, 0) == "BIGINT" + + def test_number_precision_19_scale_0_is_numeric(self): + result = map_column_type("NUMBER", 19, 0) + assert result == "NUMERIC(19, 0)" + + +# --------------------------------------------------------------------------- +# Character type mappings +# --------------------------------------------------------------------------- + +class TestCharacterMapping: + def test_varchar2_with_length(self): + assert map_column_type("VARCHAR2", 100, None) == "VARCHAR(100)" + + def test_nvarchar2_with_length(self): + assert map_column_type("NVARCHAR2", 200, None) == "VARCHAR(200)" + + def test_char_with_length(self): + assert map_column_type("CHAR", 10, None) == "CHAR(10)" + + def test_clob_maps_to_text(self): + assert map_column_type("CLOB", None, None) == "TEXT" + + def test_long_maps_to_text(self): + assert map_column_type("LONG", None, None) == "TEXT" + + +# --------------------------------------------------------------------------- +# Date / time mappings +# --------------------------------------------------------------------------- + +class TestDateTimeMapping: + def test_date_maps_to_timestamp(self): + assert map_column_type("DATE", None, None) == "TIMESTAMP" + + def test_timestamp_maps_to_timestamp(self): + assert map_column_type("TIMESTAMP", None, None) == "TIMESTAMP" + + def test_timestamp_with_tz(self): + result = map_column_type("TIMESTAMP WITH TIME ZONE", None, None) + assert result == "TIMESTAMP WITH TIME ZONE" + + def test_timestamp_with_local_tz(self): + result = map_column_type("TIMESTAMP WITH LOCAL TIME ZONE", None, None) + assert result == "TIMESTAMP WITH TIME ZONE" + + +# --------------------------------------------------------------------------- +# Binary / other mappings +# --------------------------------------------------------------------------- + +class TestOtherMappings: + def test_blob_maps_to_bytea(self): + assert map_column_type("BLOB", None, None) == "BYTEA" + + def test_float_maps_to_double_precision(self): + assert map_column_type("FLOAT", None, None) == "DOUBLE PRECISION" + + def test_binary_double(self): + assert map_column_type("BINARY_DOUBLE", None, None) == "DOUBLE PRECISION" + + def test_unknown_type_falls_back_to_text(self): + assert map_column_type("SOME_UNKNOWN_TYPE", None, None) == "TEXT" + + def test_case_insensitive(self): + assert map_column_type("varchar2", 50, None) == "VARCHAR(50)" diff --git a/type_mapping.py b/type_mapping.py new file mode 100644 index 0000000..ed88ab1 --- /dev/null +++ b/type_mapping.py @@ -0,0 +1,86 @@ +"""Oracle to PostgreSQL data type mapping.""" + +# Maps Oracle SQL types to PostgreSQL equivalents +ORACLE_TO_POSTGRES: dict[str, str] = { + # Numeric types + "NUMBER": "NUMERIC", + "FLOAT": "DOUBLE PRECISION", + "BINARY_FLOAT": "REAL", + "BINARY_DOUBLE": "DOUBLE PRECISION", + "INTEGER": "INTEGER", + "INT": "INTEGER", + "SMALLINT": "SMALLINT", + # Character types + "CHAR": "CHAR", + "NCHAR": "CHAR", + "VARCHAR2": "VARCHAR", + "NVARCHAR2": "VARCHAR", + "CLOB": "TEXT", + "NCLOB": "TEXT", + "LONG": "TEXT", + # Binary types + "RAW": "BYTEA", + "LONG RAW": "BYTEA", + "BLOB": "BYTEA", + "BFILE": "BYTEA", + # Date / time types + "DATE": "TIMESTAMP", + "TIMESTAMP": "TIMESTAMP", + "TIMESTAMP WITH TIME ZONE": "TIMESTAMP WITH TIME ZONE", + "TIMESTAMP WITH LOCAL TIME ZONE": "TIMESTAMP WITH TIME ZONE", + "INTERVAL YEAR TO MONTH": "INTERVAL", + "INTERVAL DAY TO SECOND": "INTERVAL", + # Other + "XMLTYPE": "XML", + "ROWID": "VARCHAR(18)", + "UROWID": "VARCHAR(4000)", +} + + +def map_column_type(oracle_type: str, precision: int | None, scale: int | None) -> str: + """Return the PostgreSQL column type for a given Oracle column description. + + Args: + oracle_type: The Oracle column type name (upper-cased). + precision: Numeric precision, if applicable. + scale: Numeric scale, if applicable. + + Returns: + A PostgreSQL type string. + """ + oracle_type_upper = oracle_type.upper().strip() + + # NUMBER with scale 0 (or no fractional part) → prefer integer types + if oracle_type_upper == "NUMBER": + if precision is not None and scale == 0: + if precision <= 4: + return "SMALLINT" + if precision <= 9: + return "INTEGER" + if precision <= 18: + return "BIGINT" + if precision is not None: + # scale can be None when Oracle reports precision without scale (e.g. NUMBER(10)); + # default to 0 in that case to produce a valid NUMERIC declaration. + return f"NUMERIC({precision}, {scale if scale is not None else 0})" + return "NUMERIC" + + # VARCHAR2 / NVARCHAR2 / CHAR / NCHAR with length + if oracle_type_upper in ("VARCHAR2", "NVARCHAR2") and precision is not None: + return f"VARCHAR({precision})" + + if oracle_type_upper in ("CHAR", "NCHAR") and precision is not None: + return f"CHAR({precision})" + + # TIMESTAMP with fractional-second precision + if oracle_type_upper.startswith("TIMESTAMP"): + if "WITH TIME ZONE" in oracle_type_upper or "WITH LOCAL TIME ZONE" in oracle_type_upper: + return "TIMESTAMP WITH TIME ZONE" + return "TIMESTAMP" + + mapped = ORACLE_TO_POSTGRES.get(oracle_type_upper) + if mapped: + return mapped + + # Fall back to TEXT for unknown types + return "TEXT"