API Reference¶

def cleanup(self: PgUpsert) -> PgUpsert:
    """Drop all temporary tables and views created by the upsert pipeline.

    Issues ``DROP ... IF EXISTS ... CASCADE`` for every known ``ups_*``
    temporary object, including the configurable control table.  Safe to
    call multiple times or when no temporary objects exist.

    This is useful for long-lived connections where you want to reclaim
    temporary object space without closing the connection.

    Returns:
        self, for method chaining.
    """
    for view in self._TEMP_VIEWS:
        self.db.execute(SQL("DROP VIEW IF EXISTS {} CASCADE").format(Identifier(view)))
    for table in self._TEMP_TABLES:
        self.db.execute(SQL("DROP TABLE IF EXISTS {} CASCADE").format(Identifier(table)))
    # The control table name is configurable.
    self.db.execute(SQL("DROP TABLE IF EXISTS {} CASCADE").format(Identifier(self.control_table)))
    # Reset pipeline state so the instance isn't left with stale results.
    self.qa_passed = False
    self._qa_findings = []
    # Invalidate the per-table PK column cache so a subsequent run
    # after a schema change doesn't return stale PKs.
    self._qa._pk_cols_cache.clear()
    return self

def show_control(self: PgUpsert) -> None:
    """Display contents of the control table.

    If the `interactive` flag is set to `True`, the control table will be displayed in a Tkinter window. Otherwise, the results will be logged.

    The control table definition is as follows:

    | column name          | data type | required | description |
    |----------------------|-----------|----------|-------------|
    | `table_name`         | text      | yes      | The name of the table to process. |
    | `exclude_cols`       | text      | no       | A comma-separated list of columns to exclude from the upsert process. |
    | `exclude_null_checks`| text      | no       | A comma-separated list of columns to exclude from the not-null check during the QA process. |
    | `interactive`        | boolean   | yes      | A flag to indicate whether the QA and upsert processes should be interactive. |
    | `null_errors`        | text      | no       | A comma-separated list of columns with null values. |
    | `pk_errors`          | text      | no       | A comma-separated list of primary key errors. |
    | `fk_errors`          | text      | no       | A comma-separated list of foreign key errors. |
    | `ck_errors`          | text      | no       | A comma-separated list of check constraint errors. |
    | `rows_updated`       | integer   | no       | The number of rows updated during the upsert process. |
    | `rows_inserted`      | integer   | no       | The number of rows inserted during the upsert process. |
    """  # noqa: E501
    self._validate_control()
    self._control.show(self.interactive)

def qa_all(self: PgUpsert) -> PgUpsert:
    """Performs QA checks for nulls in non-null columns, for duplicated
    primary key values, for invalid foreign keys, and invalid check constraints
    in a set of staging tables to be loaded into base tables.
    If there are failures in the QA checks, loading is not attempted.
    If the loading step is carried out, it is done within a transaction.

    The `null_errors`, `pk_errors`, `fk_errors`, `ck_errors` columns of the
    control table will be updated to identify any errors that occur,
    so that this information is available to the caller.

    The `rows_updated` and `rows_inserted` columns of the control table
    will be updated with counts of the number of rows affected by the
    upsert operation for each table.

    When the upsert operation updates the base table, all columns of the
    base table that are also in the staging table are updated.  The
    update operation does not test to see if column contents are different,
    and so does not update only those values that are different.

    This method runs [`PgUpsert`](pg_upsert.md) methods in the following order:

    1. [`PgUpsert.qa_all_null`](pg_upsert.md#pg_upsert.PgUpsert.qa_all_null)
    2. [`PgUpsert.qa_all_pk`](pg_upsert.md#pg_upsert.PgUpsert.qa_all_pk)
    3. [`PgUpsert.qa_all_fk`](pg_upsert.md#pg_upsert.PgUpsert.qa_all_fk)
    4. [`PgUpsert.qa_all_ck`](pg_upsert.md#pg_upsert.PgUpsert.qa_all_ck)

    **Example:**

    ```python
    PgUpsert(
        uri="postgresql://user@localhost:5432/database",
        tables=("genres", "books", "publishers", "authors", "book_authors"),
        staging_schema="staging",
        base_schema="public",
        do_commit=False,
        interactive=False,
        exclude_cols=("rev_user", "rev_time", "created_at", "updated_at"),
        exclude_null_check_cols=("rev_user", "rev_time", "created_at", "updated_at", "alias"),
    ).qa_all()
    ```
    """  # noqa: E501
    self._validate_control()
    self._control.clear_results()
    self._qa_findings = self._qa.run_all(
        list(self.tables),
        interactive=self.interactive,
        compact=self.compact,
        callback=self._callback,
    )
    self._update_qa_passed()
    return self

def qa_all_null(self: PgUpsert) -> PgUpsert:
    """Performs null checks for non-null columns in selected staging tables."""
    total = len(self.tables)
    for i, table in enumerate(self.tables, 1):
        ctx = CheckContext(table_num=i, total_tables=total)
        self._qa_findings.extend(self._qa.check_nulls(table, ctx=ctx))
    self._update_qa_passed()
    return self

def qa_one_null(self: PgUpsert, table: str) -> PgUpsert:
    """Performs null checks for non-null columns in a single staging table.

    Args:
        table (str): The name of the staging table to check for null values.
    """
    self._validate_table(table)
    self._qa_findings.extend(self._qa.check_nulls(table))
    self._update_qa_passed()
    return self

def qa_all_pk(self: PgUpsert) -> PgUpsert:
    """Performs primary key checks for duplicated primary key values in selected staging tables."""
    total = len(self.tables)
    for i, table in enumerate(self.tables, 1):
        ctx = CheckContext(table_num=i, total_tables=total)
        self._qa_findings.extend(self._qa.check_pks(table, interactive=self.interactive, ctx=ctx))
    self._update_qa_passed()
    return self

def qa_one_pk(self: PgUpsert, table: str) -> PgUpsert:
    """Performs primary key checks for duplicated primary key values in a single staging table.

    Args:
        table (str): The name of the staging table to check for duplicate primary key values.
    """
    self._validate_table(table)
    self._qa_findings.extend(self._qa.check_pks(table, interactive=self.interactive))
    self._update_qa_passed()
    return self

def qa_all_fk(self: PgUpsert) -> PgUpsert:
    """Performs foreign key checks for invalid foreign key values in selected staging tables."""
    total = len(self.tables)
    for i, table in enumerate(self.tables, 1):
        ctx = CheckContext(table_num=i, total_tables=total)
        self._qa_findings.extend(self._qa.check_fks(table, interactive=self.interactive, ctx=ctx))
    self._update_qa_passed()
    return self

def qa_one_fk(self: PgUpsert, table: str) -> PgUpsert:
    """Performs foreign key checks for invalid foreign key values in a single staging table.

    Args:
        table (str): The name of the staging table to check for invalid foreign key values.
    """
    self._validate_table(table)
    self._qa_findings.extend(self._qa.check_fks(table, interactive=self.interactive))
    self._update_qa_passed()
    return self

def qa_all_ck(self: PgUpsert) -> PgUpsert:
    """Performs check constraint checks for invalid check constraint values in selected staging tables."""
    total = len(self.tables)
    for i, table in enumerate(self.tables, 1):
        ctx = CheckContext(table_num=i, total_tables=total)
        self._qa_findings.extend(self._qa.check_cks(table, ctx=ctx))
    self._update_qa_passed()
    return self

def qa_one_ck(self: PgUpsert, table: str) -> PgUpsert:
    """Performs check constraint checks for invalid check constraint values in a single staging table.

    Args:
        table (str): The name of the staging table to check for invalid check constraint values.
    """
    self._validate_table(table)
    self._qa_findings.extend(self._qa.check_cks(table))
    self._update_qa_passed()
    return self

def qa_all_unique(self: PgUpsert) -> PgUpsert:
    """Performs unique constraint checks on all selected staging tables."""
    total = len(self.tables)
    for i, table in enumerate(self.tables, 1):
        ctx = CheckContext(table_num=i, total_tables=total)
        self._qa_findings.extend(self._qa.check_unique(table, interactive=self.interactive, ctx=ctx))
    self._update_qa_passed()
    return self

def qa_one_unique(self: PgUpsert, table: str) -> PgUpsert:
    """Performs unique constraint checks on a single staging table.

    Args:
        table (str): The name of the staging table to check.
    """
    self._validate_table(table)
    self._qa_findings.extend(self._qa.check_unique(table, interactive=self.interactive))
    self._update_qa_passed()
    return self

def qa_column_existence(self: PgUpsert) -> PgUpsert:
    """Checks that all base table columns exist in the staging tables.

    Respects the ``exclude_cols`` setting — excluded columns are not flagged.
    """
    total = len(self.tables)
    for i, table in enumerate(self.tables, 1):
        ctx = CheckContext(table_num=i, total_tables=total)
        self._qa_findings.extend(self._qa.check_column_existence(table, ctx=ctx))
    self._update_qa_passed()
    return self

def qa_type_mismatch(self: PgUpsert) -> PgUpsert:
    """Checks for hard type incompatibilities between staging and base columns.

    Only flags mismatches where PostgreSQL has no implicit or assignment cast.
    """
    total = len(self.tables)
    for i, table in enumerate(self.tables, 1):
        ctx = CheckContext(table_num=i, total_tables=total)
        self._qa_findings.extend(self._qa.check_type_mismatch(table, ctx=ctx))
    self._update_qa_passed()
    return self

def upsert_all(self: PgUpsert) -> PgUpsert:
    """Performs upsert operations on all selected tables in the base schema.

    **Objects created:**

    | table / view | description |
    |--------------|-------------|
    | `ups_dependencies` | Temporary table containing the dependencies of the base schema. |
    | `ups_ordered_tables` | Temporary table containing the selected tables ordered by dependency. |
    """  # noqa: E501
    self._validate_control()
    if not self.qa_passed:
        display.console.print(
            "  [bold red]QA checks have not passed — refusing to upsert.[/bold red]",
        )
        _file_logger.error("QA checks have not passed — refusing to upsert.")
        return self
    display.console.print()
    display.console.rule("[bold]Upsert[/bold]", style="cyan")
    commit_label = "[green]ON[/green]" if self.do_commit else "[dim]OFF[/dim]"
    display.console.print(f"  method={self.upsert_method}  commit={commit_label}")
    _file_logger.info(f"=== Upsert (method={self.upsert_method}, commit={self.do_commit}) ===")
    # Sync any runtime change to upsert_method before delegating.
    self._executor.upsert_method = self.upsert_method
    self._executor.upsert_all(list(self.tables), interactive=self.interactive)
    return self

def upsert_one(self: PgUpsert, table: str) -> PgUpsert:
    """Performs an upsert operation on a single table.

    Args:
        table (str): The name of the table to upsert.
    """
    self._validate_table(table)
    # Sync any runtime change to upsert_method before delegating.
    self._executor.upsert_method = self.upsert_method
    self._executor.upsert_one(table, interactive=self.interactive)
    return self

def run(self: PgUpsert) -> UpsertResult:
    """Run all QA checks and upsert operations.

    This method runs `PgUpsert` methods in the following order:

    1. [`PgUpsert.qa_all()`](pg_upsert.md#pg_upsert.PgUpsert.qa_all)
    2. [`PgUpsert.upsert_all()`](pg_upsert.md#pg_upsert.PgUpsert.upsert_all)
    3. [`PgUpsert.commit()`](pg_upsert.md#pg_upsert.PgUpsert.commit)

    Returns:
        UpsertResult: Structured result containing QA outcomes and row counts.
    """
    start_time = datetime.now()
    start_str = start_time.strftime("%Y-%m-%d %H:%M:%S")
    num_tables = len(self.tables)

    # Logfile header — clear separator for appended runs.
    from importlib.metadata import PackageNotFoundError, version

    try:
        _ver = version("pg_upsert")
    except PackageNotFoundError:
        _ver = "unknown"
    try:
        _pg_ver = self.db.execute("SELECT version()").fetchone()[0]
    except (psycopg2.Error, IndexError, TypeError):
        _pg_ver = "unknown"
    _file_logger.info("")
    _file_logger.info("=" * 60)
    _file_logger.info(f"pg-upsert {_ver} — run started at {start_str}")
    _file_logger.info(f"  {self.staging_schema} → {self.base_schema} ({num_tables} tables)")
    _file_logger.info(f"  PostgreSQL: {_pg_ver}")
    _file_logger.info("=" * 60)

    display.console.print()
    display.console.print(f"  [dim]Started at {start_str}[/dim]")
    display.console.print(
        f"  Tables selected for upsert "
        f"[dim]([/dim][bold]{self.staging_schema}[/bold] [dim]→[/dim] "
        f"[bold]{self.base_schema}[/bold][dim], {num_tables} tables)[/dim]",
    )
    if self.interactive:
        btn, _return_value = self._ui.show_table(
            "Upsert Tables",
            "Tables selected for upsert",
            [
                ("Continue", 0, "<Return>"),
                ("Cancel", 1, "<Escape>"),
            ],
            ["Table"],
            [[table] for table in self.tables],
        )
        if btn != 0:
            display.console.print("  [dim]Upsert cancelled[/dim]")
            _file_logger.info("Upsert cancelled")
            return UpsertResult(tables=[], committed=False)
    for table in self.tables:
        display.console.print(f"    [dim]•[/dim] {table}")
        _file_logger.info(f"  {table}")

    # Reset qa_passed and reinitialise the control table for a fresh run.
    self.qa_passed = False
    self._qa_findings = []
    self._init_ups_control()

    committed = False
    table_results: list[TableResult] = []

    try:
        self._control.clear_results()
        self._qa_findings = self._qa.run_all(
            list(self.tables),
            interactive=self.interactive,
            compact=self.compact,
            callback=self._callback,
        )
        self._update_qa_passed()
        if self.qa_passed:
            table_results = self._executor.upsert_all(
                list(self.tables),
                interactive=self.interactive,
                callback=self._callback,
            )
            committed = self._do_commit()
    except UserCancelledError:
        display.console.print("  [bold yellow]Rolling back changes due to user cancellation[/bold yellow]")
        _file_logger.info("Rolling back changes due to user cancellation")
        self.db.rollback()
    except psycopg2.Error as e:
        display.console.print(f"  [bold red]Database error — rolling back:[/bold red] {e}")
        _file_logger.error(f"Database error — rolling back: {e}")
        self.db.rollback()

    end_time = datetime.now()
    end_str = end_time.strftime("%Y-%m-%d %H:%M:%S")
    duration = elapsed_time(start_time)
    display.console.print(f"\n  [dim]Finished at {end_str} ({duration})[/dim]")

    # Logfile footer.
    _file_logger.info("-" * 60)
    _file_logger.info(f"pg-upsert run finished at {end_str} ({duration})")
    _file_logger.info("-" * 60)

    # Merge QA errors into table results.
    error_map: dict[str, list[QAError]] = {}
    for err in self._qa_findings:
        error_map.setdefault(err.table, []).append(err)

    # Build a TableResult for every table (including QA-only runs).
    result_map: dict[str, TableResult] = {r.table_name: r for r in table_results}
    for table in self.tables:
        if table not in result_map:
            result_map[table] = TableResult(table_name=table)
        result_map[table]._qa_findings = error_map.get(table, [])

    duration_seconds = (end_time - start_time).total_seconds()
    return UpsertResult(
        tables=[result_map[t] for t in self.tables if t in result_map],
        committed=committed,
        staging_schema=self.staging_schema,
        base_schema=self.base_schema,
        upsert_method=self.upsert_method,
        started_at=start_str,
        finished_at=end_str,
        duration_seconds=round(duration_seconds, 3),
    )

def commit(self: PgUpsert) -> PgUpsert:
    """Commits the transaction to the database and show a summary of changes.

    Changes are committed if the following criteria are met:

    - The `do_commit` flag is set to `True`.
    - All QA checks have passed (i.e., the `qa_passed` flag is set to `True`).
    - The summary of changes shows that rows have been updated or inserted.
    - If the `interactive` flag is set to `True` and the `do_commit` flag is is set to `False`, the user is prompted to commit the changes and the user selects "Continue".

    If ``qa_passed`` is ``False``, the transaction is rolled back and no
    commit is attempted.  This guards against accidentally persisting data
    that failed QA checks when using the step-by-step API.
    """  # noqa: E501
    if not self.qa_passed:
        display.console.print(
            "  [bold yellow]QA checks have not passed — refusing to commit. Rolling back.[/bold yellow]",
        )
        _file_logger.warning("QA checks have not passed — refusing to commit. Rolling back.")
        self.db.rollback()
        return self
    self._do_commit()
    return self

def to_dict(self) -> dict:
    """Serialize to a dictionary for JSON output."""
    return {
        "staging_schema": self.staging_schema,
        "base_schema": self.base_schema,
        "upsert_method": self.upsert_method,
        "qa_passed": self.qa_passed,
        "committed": self.committed,
        "total_updated": self.total_updated,
        "total_inserted": self.total_inserted,
        "started_at": self.started_at,
        "finished_at": self.finished_at,
        "duration_seconds": self.duration_seconds,
        "tables": [t.to_dict() for t in self.tables],
    }

def to_json(self, indent: int = 2) -> str:
    """Serialize to a JSON string."""
    return json.dumps(self.to_dict(), indent=indent)

def export_failures(self, directory: str | Path, fmt: str = "csv") -> Path | None:
    """Export QA violations as a "fix sheet" grouped by table.

    Writes one row per unique violating staging row to *directory*,
    with an ``_issues`` column summarising every problem found on
    that row.  Format is selected by *fmt* (``csv``, ``json``, or
    ``xlsx``); see :func:`pg_upsert.export.export_failures` for the
    exact file layout per format.

    Returns the directory written, or ``None`` if there are no
    exportable violations.
    """
    all_errors = [e for t in self.tables for e in t.qa_findings]
    from .export import export_failures

    return export_failures(all_errors, directory, fmt=fmt)