JAYASREESS/duckdb

1import duckdb2import os3from bronze import setup_bronze_layer4 5def setup_silver_layer():6    """7    Connects to DuckDB, reads from the bronze layer,8    and applies transformations to create the silver table.9    """10    # Ensure the bronze layer exists before proceeding11    setup_bronze_layer()12 13    db_path = os.path.join('..', 'data', 'fraud_detection.duckdb')14    con = duckdb.connect(database=db_path, read_only=False)15 16    # Create schema for silver data17    con.execute("CREATE SCHEMA IF NOT EXISTS silver;")18 19    print("Transforming data for the silver layer...")20 21    # Perform transformations and create the silver table22    con.execute("""23        CREATE OR REPLACE TABLE silver.silver_transactions AS24        SELECT25            *,26            -- The column is already a timestamp, so just alias it27            trans_date_trans_time AS trans_date_time,28            -- Calculate age of the cardholder at the time of transaction29            date_part('year', trans_date_trans_time) - date_part('year', dob) AS age,30            -- Extract hour of day from transaction time31            date_part('hour', trans_date_trans_time) AS trans_hour32        FROM bronze.bronze_transactions;33    """)34 35    print("Silver layer setup complete.")36 37    # Verify the new columns in the silver table38    print("Columns in silver.silver_transactions:")39    print(con.execute("DESCRIBE silver.silver_transactions;").fetchall())40 41    record_count = con.execute("SELECT COUNT(*) FROM silver.silver_transactions;").fetchone()[0]42    print(f"Total records in silver_transactions: {record_count}")43 44 45    con.close()46 47if __name__ == "__main__":48    # For direct execution, this will now run the full pipeline up to silver49    print("Setting up silver layer (which includes bronze)...")50    setup_silver_layer()51    print("Silver layer setup finished.")