Python SQLite and PostgreSQL Database Integration

import sqlite3

# Connect to on-disk database (creates file if absent)
conn = sqlite3.connect("inventory.db")
cursor = conn.cursor()

# Enforce relational schema
cursor.execute("""
CREATE TABLE IF NOT EXISTS products (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    sku TEXT UNIQUE NOT NULL,
    title TEXT NOT NULL,
    unit_cost REAL NOT NULL,
    stock INTEGER DEFAULT 0
)
""")

# Batch insert using parameterized placeholders to prevent SQL injection
new_products = [
    ("SKU-1001", "Mechanical Keycap Set", 24.50, 80),
    ("SKU-1002", "Braided USB-C Cable", 9.99, 250),
    ("SKU-1003", "Desk Mat XL", 18.00, 45)
]

cursor.executemany(
    "INSERT OR IGNORE INTO products (sku, title, unit_cost, stock) VALUES (?, ?, ?, ?)",
    new_products
)

conn.commit()
conn.close()
print("SQLite database provisioned and seeded successfully.")

import sqlite3

conn = sqlite3.connect("inventory.db")
# Enable named column access instead of standard numeric tuple indices
conn.row_factory = sqlite3.Row
cursor = conn.cursor()

threshold = 50
cursor.execute("SELECT sku, title, stock FROM products WHERE stock >= ? ORDER BY stock DESC", (threshold,))

rows = cursor.fetchall()
for row in rows:
    print(f"[{row['sku']}] {row['title']} -> In Stock: {row['stock']}")

conn.close()

import psycopg
from psycopg.rows import dict_row

# Connect using standard PostgreSQL connection URI
connection_uri = "postgresql://postgres:secretpassword@localhost:5432/app_db"

# Open connection and context-managed transaction
with psycopg.connect(connection_uri, row_factory=dict_row) as conn:
    with conn.cursor() as cur:
        # Provision target table
        cur.execute("""
        CREATE TABLE IF NOT EXISTS audit_logs (
            id BIGSERIAL PRIMARY KEY,
            event_type VARCHAR(50) NOT NULL,
            payload JSONB NOT NULL,
            created_at TIMESTAMPTZ DEFAULT CURRENT_TIMESTAMP
        );
        """)

        # Insert record with native JSON handling
        cur.execute(
            """
            INSERT INTO audit_logs (event_type, payload)
            VALUES (%s, %s)
            RETURNING id, created_at;
            """,
            ("USER_LOGIN", '{"user_id": 42, "ip": "192.168.1.1", "success": true}')
        )
        result = cur.fetchone()
        print(f"Inserted Log ID: {result['id']} at {result['created_at']}")

    # Transaction auto-commits at the exit of the connection block
print("PostgreSQL transaction committed.")

from psycopg_pool import ConnectionPool

DATABASE_URL = "postgresql://postgres:secretpassword@localhost:5432/app_db"

# Initialize pool with controlled connection boundaries
with ConnectionPool(conninfo=DATABASE_URL, min_size=2, max_size=10) as pool:
    # Checkout connection for unit of work
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT COUNT(*) AS total_logs FROM audit_logs;")
            count = cur.fetchone()[0]
            print(f"Total logged events: {count}")
    # Connection automatically returns to pool

import sqlite3

conn = sqlite3.connect(":memory:")
cur = conn.cursor()

cur.execute("CREATE TABLE accounts (id INT PRIMARY KEY, balance REAL);")
cur.execute("INSERT INTO accounts VALUES (1, 1000.0), (2, 500.0);")
conn.commit()

def transfer_funds(sender_id: int, recipient_id: int, amount: float):
    try:
        # Start atomic operation
        cur.execute("UPDATE accounts SET balance = balance - ? WHERE id = ?", (amount, sender_id))
        
        # Simulated mid-transaction validation failure
        if amount > 800:
            raise ValueError("Transfer exceeds single-transaction risk limit")
            
        cur.execute("UPDATE accounts SET balance = balance + ? WHERE id = ?", (amount, recipient_id))
        conn.commit()
        print("Transfer completed successfully.")
    except Exception as error:
        conn.rollback()
        print(f"Transaction aborted and rolled back. Error: {error}")

transfer_funds(sender_id=1, recipient_id=2, amount=950.0)

# Verify balances remain untouched after rollback
cur.execute("SELECT * FROM accounts;")
print("Current Balances:", cur.fetchall())
conn.close()

import sqlite3
from typing import List, Dict, Any

def extract_from_sqlite(db_file: str) -> List[Dict[str, Any]]:
    """Extract data rows from SQLite source."""
    with sqlite3.connect(db_file) as conn:
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()
        cursor.execute("SELECT sku, title, unit_cost, stock FROM products")
        return [dict(row) for row in cursor.fetchall()]

def transform_and_validate(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Clean records and enforce business constraints before loading."""
    cleaned = []
    for item in records:
        cleaned.append({
            "sku": item["sku"].strip().upper(),
            "title": item["title"].strip(),
            "unit_cost": round(float(item["unit_cost"]), 2),
            "stock": max(0, int(item["stock"]))
        })
    return cleaned

def load_to_postgresql_mock(target_data: List[Dict[str, Any]]) -> None:
    """Simulate targeted batch insertion into PostgreSQL."""
    print(f"Ready to insert {len(target_data)} records into target database.")
    for record in target_data:
        print(f"  -> STAGED: {record['sku']} | {record['title']} | ${record['unit_cost']}")

# Execution pipeline
if __name__ == "__main__":
    # Setup minimal source database
    source_db = "temp_migration.db"
    with sqlite3.connect(source_db) as init_conn:
        init_conn.execute("CREATE TABLE IF NOT EXISTS products (sku TEXT, title TEXT, unit_cost REAL, stock INT)")
        init_conn.execute("INSERT INTO products VALUES ('sku-99', 'Ergonomic Desk Chair', 229.50, 12)")
        init_conn.commit()

    # Run ETL cycle
    raw_items = extract_from_sqlite(source_db)
    processed_items = transform_and_validate(raw_items)
    load_to_postgresql_mock(processed_items)
    print("\nMigration cycle executed without errors.")