db.py

import os
import duckdb


def get_connection(read_only: bool = False) -> duckdb.DuckDBPyConnection:
    default = os.path.join(os.path.dirname(__file__), "translation_usage.duckdb")
    db_path = os.environ.get("DEEPL_DB_PATH", default)
    return duckdb.connect(db_path, read_only=read_only)


def init_db(conn: duckdb.DuckDBPyConnection) -> None:
    schema_path = os.path.join(os.path.dirname(__file__), "schema.sql")
    with open(schema_path) as f:
        conn.execute(f.read())


def log_translation(conn: duckdb.DuckDBPyConnection, row: dict) -> None:
    conn.execute(
        """
        INSERT INTO translation_usage
            (logged_at, logged_date, source_lang, target_lang, billed_characters, api_key_alias, request_id, reporting_tag,
             is_success, error_code, error_http_status, error_message, translation_type, document_type)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        [
            row["logged_at"],
            row["logged_date"],
            row["source_lang"],
            row["target_lang"],
            row["billed_characters"],
            row["api_key_alias"],
            row["request_id"],
            row["reporting_tag"],
            row.get("is_success", True),
            row.get("error_code", None),
            row.get("error_http_status", None),
            row.get("error_message", None),
            row.get("translation_type", "Text"),
            row.get("document_type", None),
        ],
    )