prompt.txt

prompt.txt

I am building a Python 3.13 module to log all SQL queries I run, across both Vertica (via vertica_python) and Microsoft SQL Server (via SQLAlchemy 2 Core). Please generate a complete implementation and usage examples that meet ALL of these requirements:

Logging backend: Use loguru (not the standard library logging).

Log each query execution as one JSON line (json.dumps) into a rotating log file.

Rotate daily at midnight UTC.

Never delete old logs (keep all history).

Compress rotated logs automatically with gzip (.gz).

Use a helper like setup_loguru_jsonl() for initialization, with options for log directory and base filename.

Ensure logs are safe for ingestion (valid JSON Lines, one per line).

Captured metadata per query:

UTC timestamp (isoformat).

SQL string as executed.

Parameters (serialize safely, converting non-JSON types).

Query success/failure (ok flag, exception message if error).

Duration in ms.

Rowcount, if available.

Process ID.

A user-supplied tag string (optional).

Source connection fingerprint:

For Vertica: extract from connection.options (host, port, user, database, session_label).

For SQLAlchemy (MSSQL): extract from the Engine.url or Connection.engine.url (server, port, database, username, drivername).

Always include client hostname.

Sink (write target):

Automatically detect the target table (server/database/schema/table) using sqlglot for INSERT, UPDATE, MERGE, COPY (Vertica), and SELECT INTO (SQL Server).

Allow a sink_override parameter to manually specify or correct the sink table info.

For cases where results are written to files (e.g., Parquet), provide a separate log_external_write() function to log target_kind="file", file path, rowcount, etc.

Functions:

setup_loguru_jsonl(log_dir: str = "./logs", base_name: str = "queries.jsonl") -> None: configure the loguru sink with rotation daily, no deletion, gzip compression.

exec_and_log_vertica(cursor, sql: str, params: dict | Iterable | None = None, *, sink_override: dict | None = None, tag: str | None = None) -> Any: executes Vertica query, logs JSON.

exec_and_log_sqlalchemy(connection, sql: str, params: dict | Iterable | None = None, *, sink_override: dict | None = None, tag: str | None = None) -> Any: executes SQL Server query via SQLAlchemy 2, logs JSON.

logged_transaction(connection, *, tag: str | None = None): context manager that logs begin/commit/rollback events (for Vertica or SQLAlchemy connection).

log_external_write(*, target_kind: str, path: str | None = None, server: str | None = None, database: str | None = None, schema: str | None = None, table: str | None = None, rows_written: int | None = None, extra: dict | None = None, tag: str | None = None) -> None: log when writing results outside SQL (files, other DBs).

Typing rules:

Use Python 3.13 PEP-585 built-in generics (dict[str, Any], list[str], str | None).

Only apply type hints to function parameters and return values.

Do not annotate local variables.

Examples:

Show usage with Vertica (vertica_python) for an INSERT ... SELECT.

Show usage with SQL Server (sqlalchemy.create_engine(..., "mssql+pyodbc://...")) for an INSERT INTO ... SELECT.

Show SELECT INTO in SQL Server with sink_override.

Show reading results into Python and then logging an external write to Parquet.

Demonstrate logged_transaction around multiple Vertica statements.

The output should be a single, self-contained Python module (named db_query_logging.py), plus example usage snippets for both Vertica and SQL Server, written in Python 3.13 style.