Replicating from Aurora for PostgreSQL via Debezium

CREATE TABLE lineitem (
    lineitem_id BIGINT PRIMARY KEY, -- Unique identifier for each line item
    transaction_id BIGINT NOT NULL, -- Links the line item to a specific transaction (e.g., order_id or invoice_id)
    product_id BIGINT NOT NULL, -- ID of the product or service being transacted
    quantity NUMERIC(10, 2) NOT NULL, -- Quantity of the product being ordered
    unit_price NUMERIC(15, 4) NOT NULL, -- Price per unit of the product
    discount NUMERIC(5, 2) DEFAULT 0.00, -- Discount applied to the line item, if any
    tax_rate NUMERIC(5, 2) DEFAULT 0.00, -- Applicable tax rate
    total_amount NUMERIC(20, 4) GENERATED ALWAYS AS (quantity * unit_price * (1 - discount / 100) * (1 + tax_rate / 100)) STORED, -- Calculated total amount for the line item
    created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP, -- Timestamp when the line item was created
    transaction_date DATE NOT NULL, -- Date of the transaction
    status VARCHAR(50) DEFAULT 'active', -- Status of the line item (e.g., 'active', 'void', 'pending')
    notes TEXT -- Optional notes or comments about the line item
);


INSERT INTO lineitem (lineitem_id,  transaction_id, product_id, quantity, unit_price, discount, tax_rate, transaction_date, status, notes
) VALUES
    (1, 1001, 2001, 10.00, 50.00, 5.00, 10.00, '2024-10-15', 'active', 'First line item for this transaction'),
    (2, 1001, 2002, 5.00, 100.00, 0.00, 15.00, '2024-10-15', 'active', 'Second line item with no discount'),
    (3, 1002, 2003, 2.00, 250.00, 10.00, 12.00, '2024-10-16', 'pending', 'Third line item in a different transaction'),
    (4, 1003, 2004, 7.00, 75.00, 3.00, 8.00, '2024-10-16', 'void', 'Voided line item for test purposes');

select * from lineitem;

import psycopg2
from psycopg2 import sql
import random
import time

# PostgreSQL connection parameters
conn = psycopg2.connect(
    dbname="postgres",
    user="postgresuser",
    password="postgrespw",
    host="[Your Amazon Aurora PostgreSQL hostname]",
    port="5432"
)

# Create a cursor object
cur = conn.cursor()


# Function to insert dummy lineitem records
def insert_dummy_lineitem(lineitem_id):
    transaction_id = random.randint(1000, 1100)  # Random transaction_id between 1000 and 1100
    product_id = random.randint(2000, 2100)      # Random product_id between 2000 and 2100
    quantity = round(random.uniform(1, 20), 2)   # Random quantity between 1 and 20
    unit_price = round(random.uniform(10, 200), 2)  # Random unit price between 10 and 200
    discount = round(random.uniform(0, 10), 2)   # Random discount between 0% and 10%
    tax_rate = round(random.uniform(5, 15), 2)   # Random tax rate between 5% and 15%
    transaction_date = time.strftime('%Y-%m-%d')  # Current date
    status = random.choice(['active', 'pending', 'void'])  # Random status
    notes = f"Random note {random.randint(1, 100)}"  # Random notes

    # Insert statement
    insert_query = sql.SQL("""
        INSERT INTO lineitem (
            lineitem_id, transaction_id, product_id, quantity, unit_price, discount, tax_rate, transaction_date, status, notes
        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    """)

    # Execute insert statement
    cur.execute(insert_query, (
        lineitem_id, transaction_id, product_id, quantity, unit_price, discount, tax_rate, transaction_date, status, notes
    ))

    # Commit transaction
    conn.commit()

# Insert dummy data every few milliseconds
try:
    lineitem_id = 5
    while True:
        insert_dummy_lineitem(lineitem_id)
        print("Inserted a new lineitem.")
        lineitem_id = lineitem_id + 1
except KeyboardInterrupt:
    print("Insertion process stopped.")
finally:
    # Close the cursor and connection
    cur.close()
    conn.close()

SELECT 
    product_id, 
    AVG(tax_rate) AS avg_tax_rate
FROM 
    lineitem
GROUP BY 
    product_id
ORDER BY 
    avg_tax_rate DESC;