Tracking Ingestion Metadata for Audit Trails

Why Do We Need an Audit Log?

Before we start building, let's establish why an audit log is so crucial. A data pipeline without logging is a black box. If data is missing or incorrect in the final tables, you have no way to trace the problem back to its source. An audit log provides an immutable record of every execution, answering key questions like:

• Which data source was loaded and when?
• Did the ingestion job complete successfully or fail?
• If it failed, what was the error?
• How many new records were inserted versus how many existing records were updated?
• Is the performance of the job degrading over time?

This log becomes the foundation for alerting, dashboards, and debugging efforts that are essential for maintaining a healthy data platform.

Patterns for Auditing

The concept of an "audit table" can mean different things. It's useful to understand the common patterns to see where our specific needs fit. The video below provides an excellent overview of three common approaches for auditing data in a database.

The video primarily discusses auditing changes to data in an application database (an OLTP context), like tracking who updated a customer's address. The "Shadow History" pattern is excellent for this.

Our requirement is slightly different but related. We aren't tracking changes to individual records row-by-row; instead, we want to track the metadata of a process: the ingestion job itself. The principle of using a separate table to log events, however, is exactly what we need. We'll be implementing a "Process Audit" table, which is conceptually similar to the "Shadow History" approach but tailored for logging pipeline executions rather than individual data record changes.

Designing Our Ingestion Audit Table

Let's design a schema for our audit table. We need to capture the essential metadata about each ingestion run. Real-world projects provide great blueprints for this. The PyImageSearch article on building an Airflow pipeline describes a pipeline_runs table that serves a very similar purpose.

Inspired by that example and our specific needs from the previous lesson (tracking upsert counts), we can define a clear schema. We will create this table in a dedicated schema named pipeline_meta to keep our operational tables separate from our data tables (bronze, silver, gold).

Here is the DDL for our table, ingestion_audit_log:

CREATE SCHEMA IF NOT EXISTS pipeline_meta;

CREATE TABLE pipeline_meta.ingestion_audit_log (
    log_id SERIAL PRIMARY KEY,
    source_name VARCHAR(255) NOT NULL,
    target_table VARCHAR(255) NOT NULL,
    run_start_ts TIMESTAMP WITH TIME ZONE NOT NULL,
    run_end_ts TIMESTAMP WITH TIME ZONE,
    status VARCHAR(50) NOT NULL, -- e.g., 'STARTED', 'SUCCESS', 'FAILED'
    rows_processed INT,
    rows_inserted INT,
    rows_updated INT,
    error_message TEXT,
    airflow_run_id VARCHAR(255) -- for future use with our orchestrator
);

Let's break down the columns:

• log_id: A unique identifier for each log entry.
• source_name: A human-readable name for the data source (e.g., 'products_api', 'users_csv').
• target_table: The table the data was loaded into (e.g., 'bronze.products').
• run_start_ts / run_end_ts: Timestamps to measure the duration of the run.
• status: The final outcome of the run.
• rows_processed, rows_inserted, rows_updated: Key performance metrics. Tracking inserts vs. updates is particularly useful given our upsert logic.
• error_message: Captures the exception details if the run fails.
• airflow_run_id: A placeholder for when we integrate with Airflow, allowing us to link a specific log entry back to a specific DAG run.

This structure provides a comprehensive, at-a-glance view of every ingestion job. The OpenETL GitHub project, another real-world example, uses a similar approach by defining an airflow_etl_result table in a dedicated infra_monitor schema, confirming this is a robust and common pattern.

Implementation: Instrumenting Our Python Code

Now, let's integrate this logging into our Python code. The best way to do this is to create a wrapper or use a context manager that handles the audit logging around our main ingestion logic.

The flow will be:

1. Start: Before the ingestion begins, insert a new row into ingestion_audit_log with a status of 'STARTED'. We commit this immediately so we can see jobs that are in-progress.
2. Execute: Run the main upsert_df_to_postgres function from our previous lesson.
3. Success: If the upsert completes without errors, update the corresponding log entry with 'SUCCESS', the end timestamp, and the row counts.
4. Failure: If any exception occurs, catch it, update the log entry with 'FAILED' and the error message, and then re-raise the exception so the calling process (like Airflow) knows the task failed.

Here is a Python function that encapsulates this logic. It acts as a decorator or a high-level wrapper you would call from your Airflow task.

import psycopg2
import logging
from datetime import datetime, timezone

def execute_with_audit_logging(
    ingestion_function,
    db_params: dict,
    source_name: str,
    target_table: str,
    *args,
    **kwargs
):
    """
    Executes a given ingestion function while handling audit logging.

    Args:
        ingestion_function: The function to execute (e.g., upsert_df_to_postgres).
        db_params (dict): Database connection parameters.
        source_name (str): A descriptive name for the data source.
        target_table (str): The destination table for the data.
        *args, **kwargs: Arguments to pass to the ingestion_function.
    """
    log_id = None
    start_ts = datetime.now(timezone.utc)




    # 1. Create the 'STARTED' audit log entry
    try:
        with psycopg2.connect(**db_params) as conn:
            with conn.cursor() as cursor:
                cursor.execute(
                    """
                    INSERT INTO pipeline_meta.ingestion_audit_log
                    (source_name, target_table, run_start_ts, status)
                    VALUES (%s, %s, %s, 'STARTED')
                    RETURNING log_id;
                    """,
                    (source_name, target_table, start_ts)
                )
                log_id = cursor.fetchone()[0]
                logging.info(f"Started ingestion for {source_name}. Audit log ID: {log_id}")
    except Exception as e:
        logging.critical(f"Failed to create initial audit log entry for {source_name}: {e}")



        # If we can't even log, we should raise immediately.
        raise

    try:



        # 2. Execute the main ingestion logic
        # We assume the function returns a dict with stats like {'inserted': x, 'updated': y}
        result_stats = ingestion_function(*args, **kwargs)
        
        end_ts = datetime.now(timezone.utc)
        



        # 3. Update the audit log with 'SUCCESS'
        with psycopg2.connect(**db_params) as conn:
            with conn.cursor() as cursor:
                cursor.execute(
                    """
                    UPDATE pipeline_meta.ingestion_audit_log
                    SET run_end_ts = %s, status = 'SUCCESS',
                        rows_inserted = %s, rows_updated = %s,
                        rows_processed = %s
                    WHERE log_id = %s;
                    """,
                    (
                        end_ts,
                        result_stats.get('inserted', 0),
                        result_stats.get('updated', 0),
                        result_stats.get('processed', 0),
                        log_id
                    )
                )
        logging.info(f"Successfully completed ingestion for {source_name}. Audit log ID: {log_id}")

    except Exception as e:
        end_ts = datetime.now(timezone.utc)
        logging.error(f"Ingestion failed for {source_name}: {e}", exc_info=True)
        



        # 4. Update the audit log with 'FAILED'
        with psycopg2.connect(**db_params) as conn:
            with conn.cursor() as cursor:
                cursor.execute(
                    """
                    UPDATE pipeline_meta.ingestion_audit_log
                    SET run_end_ts = %s, status = 'FAILED', error_message = %s
                    WHERE log_id = %s;
                    """,
                    (end_ts, str(e), log_id)
                )



        # Re-raise the exception so the orchestrator knows the task failed
        raise

(Note: To use this wrapper, you would need to slightly modify the upsert_df_to_postgres function from the last lesson to return a dictionary of statistics rather than just logging them.)

Connecting Configuration and Auditing

Let's revisit the flowchart from the start of this module.

The "Metadata Table" in this diagram is used to drive the pipeline—it provides configuration before the run starts (e.g., schema definitions, transformation rules). Our ingestion_audit_log table, by contrast, is used to observe the pipeline—it records the results after the run finishes.

A mature pipeline uses both! You retrieve configuration from one set of metadata tables and log execution results to another. This combination of proactive configuration and reactive logging creates a system that is not only robust and reliable but also transparent and manageable.

Conclusion

In this lesson, we've elevated our pipeline from merely reliable to observable. By designing and implementing an audit table, you've created a durable, queryable log of every ingestion event. This is a non-negotiable component of any serious data engineering workflow.

Key Takeaways:

• Audit logs provide observability, which is essential for monitoring, debugging, and maintaining a data pipeline.
• We are implementing a process audit pattern, which logs the execution metadata of pipeline jobs, as distinct from auditing individual data record changes.
• A well-designed audit table schema captures status, timings, performance metrics (like row counts), and error messages.
• Using a dedicated schema like pipeline_meta helps organize operational tables and keep them separate from your core data.
• Instrumenting your Python code with a wrapper function or decorator ensures that audit logging is applied consistently and reliably around your core logic.

In our next module, we'll start putting all these pieces together in a formal orchestration tool. We'll introduce Apache Airflow, explore its core concepts, and see how to schedule and run the Python ingestion scripts we've built as formal, repeatable tasks. Our audit log will be invaluable as we start managing more complex workflows.