Code Editor for app.py

import streamlit as st
import duckdb
from simple_salesforce import Salesforce
import pandas as pd

# Page config
st.set_page_config(page_title="Salesforce to DuckDB ETL", layout="wide")
st.title("Salesforce to DuckDB ETL")

# Initialize connection state
if 'connected' not in st.session_state:
    st.session_state.connected = False

if 'sf' not in st.session_state:
    st.session_state.sf = None

# Sidebar for connection details
with st.sidebar:
    st.header("Salesforce Credentials")
    username = st.text_input("Username", type="default")
    password = st.text_input("Password", type="password")
    security_token = st.text_input("Security Token", type="password")
    
    if st.button("Connect to Salesforce"):
        try:
            sf = Salesforce(
                username=username,
                password=password,
                security_token=security_token
            )
            st.session_state.sf = sf
            st.session_state.connected = True
            st.success("Connected successfully!")
        except Exception as e:
            st.error(f"Connection failed: {str(e)}")

# Main content
if st.session_state.connected:
    st.header("Salesforce Tables")
    
    # Initialize DuckDB connection
    conn = duckdb.connect('salesforce_data.ddb')
    
    # Get list of objects
    desc = st.session_state.sf.describe()
    sobjects = [obj['name'] for obj in desc['sobjects'] if not obj['custom']]
    
    # Create multiselect for tables
    selected_tables = st.multiselect(
        "Select tables to download",
        sobjects,
        default=["Account", "Contact", "Opportunity"]
    )
    
    if st.button("Download and Load Selected Tables"):
        progress_bar = st.progress(0)
        status_text = st.empty()
        
        for i, table in enumerate(selected_tables):
            try:
                # Update status
                status_text.text(f"Processing {table}...")
                
                # Query Salesforce
                query = f"SELECT * FROM {table}"
                result = st.session_state.sf.query_all(query)
                
                # Convert to DataFrame
                df = pd.DataFrame(result['records']).drop(columns=['attributes'])
                
                # Load into DuckDB
                conn.execute(f"CREATE TABLE IF NOT EXISTS {table} AS SELECT * FROM df")
                
                # Update progress
                progress = (i + 1) / len(selected_tables)
                progress_bar.progress(progress)
                
            except Exception as e:
                st.error(f"Error processing {table}: {str(e)}")
        
        status_text.text("Processing complete!")
        
        # Show tables in DuckDB
        st.header("Tables in DuckDB")
        tables = conn.execute("SHOW TABLES").fetchall()
        for table in tables:
            st.subheader(f"Table: {table[0]}")
            preview = conn.execute(f"SELECT * FROM {table[0]} LIMIT 5").df()
            st.dataframe(preview)
        
        # Option to export
        if st.button("Export DuckDB Database"):
            conn.close()
            with open("salesforce_data.ddb", "rb") as fp:
                st.download_button(
                    label="Download DuckDB file",
                    data=fp,
                    file_name="salesforce_data.ddb",
                    mime="application/octet-stream"
                )
else:
    st.info("Please connect to Salesforce using the sidebar to begin.")