1pip install simple-salesforce

1from simple_salesforce import Salesforce
2
3# Connect to Salesforce
4sf = Salesforce(
5    username='your-email@company.com',
6    password='your-password',
7    security_token='your-security-token'
8)
9
10print("✅ Connected to Salesforce")

1from simple_salesforce import Salesforce
2
3sf = Salesforce(
4    instance_url='https://your-instance.salesforce.com',
5    session_id='oauth-session-id'
6)

1import os
2from simple_salesforce import Salesforce
3
4sf = Salesforce(
5    username=os.getenv('SF_USERNAME'),
6    password=os.getenv('SF_PASSWORD'),
7    security_token=os.getenv('SF_SECURITY_TOKEN')
8)

1export SF_USERNAME="your-email@company.com"
2export SF_PASSWORD="your-password"
3export SF_SECURITY_TOKEN="your-security-token"

1def create_lead(first_name, last_name, company, email, phone=None):
2    """
3    Create a new lead in Salesforce.
4    
5    Args:
6        first_name: Lead's first name
7        last_name: Lead's last name
8        company: Company name
9        email: Email address
10        phone: Phone number (optional)
11    
12    Returns:
13        Lead ID if successful
14    """
15    
16    lead_data = {
17        'FirstName': first_name,
18        'LastName': last_name,
19        'Company': company,
20        'Email': email,
21        'Status': 'Open - Not Contacted',
22        'LeadSource': 'Web'
23    }
24    
25    if phone:
26        lead_data['Phone'] = phone
27    
28    try:
29        result = sf.Lead.create(lead_data)
30        
31        if result['success']:
32            lead_id = result['id']
33            print(f"✅ Lead created: {lead_id}")
34            return lead_id
35        else:
36            print(f"❌ Failed to create lead: {result}")
37            return None
38            
39    except Exception as e:
40        print(f"❌ Error creating lead: {e}")
41        return None
42
43
44# Usage
45create_lead(
46    first_name="Sarah",
47    last_name="Johnson",
48    company="Acme Corp",
49    email="sarah@acme.com",
50    phone="555-0123"
51)

1def create_contact(first_name, last_name, account_id, email, phone=None, title=None):
2    """
3    Create a new contact in Salesforce.
4    
5    Args:
6        first_name: Contact's first name
7        last_name: Contact's last name
8        account_id: Associated Account ID
9        email: Email address
10        phone: Phone number (optional)
11        title: Job title (optional)
12    
13    Returns:
14        Contact ID if successful
15    """
16    
17    contact_data = {
18        'FirstName': first_name,
19        'LastName': last_name,
20        'AccountId': account_id,
21        'Email': email
22    }
23    
24    if phone:
25        contact_data['Phone'] = phone
26    if title:
27        contact_data['Title'] = title
28    
29    try:
30        result = sf.Contact.create(contact_data)
31        
32        if result['success']:
33            print(f"✅ Contact created: {result['id']}")
34            return result['id']
35        else:
36            print(f"❌ Failed to create contact: {result}")
37            return None
38            
39    except Exception as e:
40        print(f"❌ Error creating contact: {e}")
41        return None

1def create_opportunity(name, account_id, amount, close_date, stage='Prospecting'):
2    """
3    Create a new opportunity in Salesforce.
4    
5    Args:
6        name: Opportunity name
7        account_id: Associated Account ID
8        amount: Deal value
9        close_date: Expected close date (YYYY-MM-DD)
10        stage: Sales stage (default: Prospecting)
11    
12    Returns:
13        Opportunity ID if successful
14    """
15    
16    opp_data = {
17        'Name': name,
18        'AccountId': account_id,
19        'Amount': amount,
20        'CloseDate': close_date,
21        'StageName': stage
22    }
23    
24    try:
25        result = sf.Opportunity.create(opp_data)
26        
27        if result['success']:
28            print(f"✅ Opportunity created: {result['id']}")
29            return result['id']
30        else:
31            print(f"❌ Failed to create opportunity: {result}")
32            return None
33            
34    except Exception as e:
35        print(f"❌ Error creating opportunity: {e}")
36        return None
37
38
39# Usage
40create_opportunity(
41    name="Q4 2025 Deal - Acme Corp",
42    account_id="001...",  # Account ID
43    amount=50000,
44    close_date="2025-12-31",
45    stage="Qualification"
46)

1# Get all open leads
2leads = sf.query("SELECT Id, Name, Email, Company FROM Lead WHERE Status='Open - Not Contacted'")
3
4print(f"Found {leads['totalSize']} open leads")
5
6for lead in leads['records']:
7    print(f"- {lead['Name']} from {lead['Company']} ({lead['Email']})")

1def get_high_value_opportunities(min_amount=10000):
2    """Get opportunities above a certain value."""
3    
4    query = f"""
5        SELECT Id, Name, Amount, StageName, CloseDate, Account.Name
6        FROM Opportunity
7        WHERE Amount >= {min_amount}
8        AND IsClosed = false
9        ORDER BY Amount DESC
10    """
11    
12    results = sf.query(query)
13    
14    opportunities = []
15    for record in results['records']:
16        opportunities.append({
17            'id': record['Id'],
18            'name': record['Name'],
19            'amount': record['Amount'],
20            'stage': record['StageName'],
21            'close_date': record['CloseDate'],
22            'account': record['Account']['Name']
23        })
24    
25    return opportunities
26
27
28# Usage
29high_value_opps = get_high_value_opportunities(min_amount=50000)
30
31for opp in high_value_opps:
32    print(f"{opp['name']}: ${opp['amount']:,.2f} - {opp['stage']}")

1from datetime import datetime, timedelta
2
3def get_recent_leads(days=7):
4    """Get leads created in the last N days."""
5    
6    # Calculate date N days ago
7    date_filter = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%dT%H:%M:%SZ')
8    
9    query = f"""
10        SELECT Id, Name, Email, Company, CreatedDate
11        FROM Lead
12        WHERE CreatedDate >= {date_filter}
13        ORDER BY CreatedDate DESC
14    """
15    
16    results = sf.query(query)
17    return results['records']
18
19
20# Get leads from last 7 days
21recent_leads = get_recent_leads(7)
22print(f"Found {len(recent_leads)} leads in the last 7 days")

1def update_lead_status(lead_id, new_status, notes=None):
2    """
3    Update lead status.
4    
5    Args:
6        lead_id: Salesforce Lead ID
7        new_status: New status value
8        notes: Optional notes to add
9    
10    Returns:
11        True if successful
12    """
13    
14    update_data = {
15        'Status': new_status
16    }
17    
18    if notes:
19        update_data['Description'] = notes
20    
21    try:
22        sf.Lead.update(lead_id, update_data)
23        print(f"✅ Lead {lead_id} updated to {new_status}")
24        return True
25        
26    except Exception as e:
27        print(f"❌ Error updating lead: {e}")
28        return False
29
30
31# Usage
32update_lead_status(
33    lead_id="00Q...",
34    new_status="Working - Contacted",
35    notes="Had initial phone call, scheduling demo"
36)

1def bulk_update_leads(lead_ids, field, value):
2    """
3    Update multiple leads at once.
4    
5    Args:
6        lead_ids: List of Lead IDs
7        field: Field name to update
8        value: New value
9    
10    Returns:
11        Number of successfully updated records
12    """
13    
14    success_count = 0
15    
16    for lead_id in lead_ids:
17        try:
18            sf.Lead.update(lead_id, {field: value})
19            success_count += 1
20        except Exception as e:
21            print(f"❌ Failed to update {lead_id}: {e}")
22    
23    print(f"✅ Updated {success_count}/{len(lead_ids)} leads")
24    return success_count
25
26
27# Usage: Mark multiple leads as contacted
28lead_ids = ["00Q...", "00Q...", "00Q..."]
29bulk_update_leads(lead_ids, 'Status', 'Working - Contacted')

1import requests
2from simple_salesforce import Salesforce
3
4def process_web_form_submission(form_data):
5    """
6    Create Salesforce lead from web form.
7    
8    Args:
9        form_data: Dictionary with form fields
10    
11    Returns:
12        Lead ID or None
13    """
14    
15    # Connect to Salesforce
16    sf = Salesforce(
17        username=os.getenv('SF_USERNAME'),
18        password=os.getenv('SF_PASSWORD'),
19        security_token=os.getenv('SF_SECURITY_TOKEN')
20    )
21    
22    # Map form fields to Salesforce fields
23    lead_data = {
24        'FirstName': form_data.get('first_name'),
25        'LastName': form_data.get('last_name'),
26        'Company': form_data.get('company'),
27        'Email': form_data.get('email'),
28        'Phone': form_data.get('phone'),
29        'LeadSource': 'Website',
30        'Status': 'Open - Not Contacted',
31        'Description': form_data.get('message', '')
32    }
33    
34    # Add custom fields if present
35    if 'industry' in form_data:
36        lead_data['Industry'] = form_data['industry']
37    
38    try:
39        result = sf.Lead.create(lead_data)
40        
41        if result['success']:
42            lead_id = result['id']
43            print(f"✅ Lead created from web form: {lead_id}")
44            
45            # Send confirmation email (optional)
46            send_confirmation_email(form_data['email'], lead_id)
47            
48            return lead_id
49        else:
50            print(f"❌ Failed to create lead: {result}")
51            return None
52            
53    except Exception as e:
54        print(f"❌ Error processing form: {e}")
55        return None
56
57
58# Usage
59form_data = {
60    'first_name': 'John',
61    'last_name': 'Smith',
62    'company': 'Tech Startup Inc',
63    'email': 'john@techstartup.com',
64    'phone': '555-9999',
65    'industry': 'Technology',
66    'message': 'Interested in enterprise plan'
67}
68
69process_web_form_submission(form_data)

1import imaplib
2import email
3from email.header import decode_header
4
5def sync_email_to_salesforce(email_address):
6    """
7    Find contact by email and log email activity.
8    
9    Args:
10        email_address: Contact's email address
11    
12    Returns:
13        True if logged successfully
14    """
15    
16    # Find contact by email
17    query = f"SELECT Id, Name FROM Contact WHERE Email='{email_address}' LIMIT 1"
18    results = sf.query(query)
19    
20    if results['totalSize'] == 0:
21        print(f"❌ No contact found for {email_address}")
22        return False
23    
24    contact_id = results['records'][0]['Id']
25    contact_name = results['records'][0]['Name']
26    
27    # Create task to log the email
28    task_data = {
29        'WhoId': contact_id,
30        'Subject': 'Email Received',
31        'Status': 'Completed',
32        'Description': f'Email conversation with {contact_name}',
33        'ActivityDate': datetime.now().strftime('%Y-%m-%d')
34    }
35    
36    try:
37        result = sf.Task.create(task_data)
38        
39        if result['success']:
40            print(f"✅ Email activity logged for {contact_name}")
41            return True
42        else:
43            print(f"❌ Failed to log activity: {result}")
44            return False
45            
46    except Exception as e:
47        print(f"❌ Error logging email: {e}")
48        return False

1def score_lead(lead):
2    """
3    Calculate lead score based on various factors.
4    
5    Args:
6        lead: Lead record from Salesforce
7    
8    Returns:
9        Integer score (0-100)
10    """
11    
12    score = 0
13    
14    # Company size scoring
15    if lead.get('NumberOfEmployees'):
16        employees = lead['NumberOfEmployees']
17        if employees > 1000:
18            score += 30
19        elif employees > 100:
20            score += 20
21        elif employees > 10:
22            score += 10
23    
24    # Industry scoring
25    high_value_industries = ['Technology', 'Financial Services', 'Healthcare']
26    if lead.get('Industry') in high_value_industries:
27        score += 20
28    
29    # Engagement scoring
30    if lead.get('Website'):
31        score += 10
32    if lead.get('Phone'):
33        score += 10
34    
35    # Job title scoring
36    decision_maker_titles = ['CEO', 'CTO', 'VP', 'Director', 'Manager']
37    title = lead.get('Title', '').lower()
38    if any(role.lower() in title for role in decision_maker_titles):
39        score += 20
40    
41    return min(score, 100)  # Cap at 100
42
43
44def auto_assign_leads():
45    """
46    Score and assign unassigned leads to sales reps.
47    """
48    
49    # Get unassigned leads
50    query = """
51        SELECT Id, FirstName, LastName, Company, Email, Phone,
52               Industry, Title, NumberOfEmployees
53        FROM Lead
54        WHERE OwnerId='00G...'  # Replace with your queue/default owner ID
55        AND Status='Open - Not Contacted'
56    """
57    
58    results = sf.query(query)
59    
60    # Sales rep assignments (simplified)
61    sales_reps = {
62        'high': '005...',  # Senior rep for high-score leads
63        'medium': '005...',  # Mid-level rep
64        'low': '005...'  # Junior rep
65    }
66    
67    for lead in results['records']:
68        score = score_lead(lead)
69        
70        # Determine assignment based on score
71        if score >= 70:
72            owner_id = sales_reps['high']
73            priority = 'High'
74        elif score >= 40:
75            owner_id = sales_reps['medium']
76            priority = 'Medium'
77        else:
78            owner_id = sales_reps['low']
79            priority = 'Low'
80        
81        # Update lead
82        sf.Lead.update(lead['Id'], {
83            'OwnerId': owner_id,
84            'Rating': priority,
85            'Lead_Score__c': score  # Custom field
86        })
87        
88        print(f"✅ Assigned {lead['FirstName']} {lead['LastName']} (score: {score}) to rep")
89
90
91# Run auto-assignment
92auto_assign_leads()

1def find_duplicate_lead(email, phone=None):
2    """
3    Check if lead already exists.
4    
5    Args:
6        email: Email address to check
7        phone: Phone number to check (optional)
8    
9    Returns:
10        Lead ID if duplicate found, None otherwise
11    """
12    
13    # Search by email
14    query = f"SELECT Id, Name, Email, Status FROM Lead WHERE Email='{email}' LIMIT 1"
15    results = sf.query(query)
16    
17    if results['totalSize'] > 0:
18        lead = results['records'][0]
19        return {
20            'id': lead['Id'],
21            'name': lead['Name'],
22            'status': lead['Status'],
23            'match_type': 'email'
24        }
25    
26    # Search by phone if provided
27    if phone:
28        query = f"SELECT Id, Name, Phone, Status FROM Lead WHERE Phone='{phone}' LIMIT 1"
29        results = sf.query(query)
30        
31        if results['totalSize'] > 0:
32            lead = results['records'][0]
33            return {
34                'id': lead['Id'],
35                'name': lead['Name'],
36                'status': lead['Status'],
37                'match_type': 'phone'
38            }
39    
40    return None
41
42
43def create_lead_with_duplicate_check(lead_data):
44    """
45    Create lead only if no duplicate exists.
46    
47    Args:
48        lead_data: Dictionary with lead information
49    
50    Returns:
51        Lead ID or existing duplicate ID
52    """
53    
54    # Check for duplicates
55    duplicate = find_duplicate_lead(
56        lead_data['Email'],
57        lead_data.get('Phone')
58    )
59    
60    if duplicate:
61        print(f"⚠️  Duplicate found: {duplicate['name']} (matched on {duplicate['match_type']})")
62        print(f"   Existing lead ID: {duplicate['id']}")
63        print(f"   Status: {duplicate['status']}")
64        
65        # Optionally update existing lead instead
66        return duplicate['id']
67    
68    # No duplicate, create new lead
69    result = sf.Lead.create(lead_data)
70    
71    if result['success']:
72        print(f"✅ New lead created: {result['id']}")
73        return result['id']
74    else:
75        print(f"❌ Failed to create lead: {result}")
76        return None

1from simple_salesforce.exceptions import SalesforceError
2import logging
3
4logging.basicConfig(level=logging.INFO)
5logger = logging.getLogger(__name__)
6
7def safe_salesforce_operation(operation_func, *args, **kwargs):
8    """
9    Wrapper for safe Salesforce operations with retry logic.
10    
11    Args:
12        operation_func: Function to execute
13        *args, **kwargs: Arguments to pass to function
14    
15    Returns:
16        Operation result or None if failed
17    """
18    
19    max_retries = 3
20    retry_count = 0
21    
22    while retry_count < max_retries:
23        try:
24            result = operation_func(*args, **kwargs)
25            return result
26            
27        except SalesforceError as e:
28            # API-specific errors
29            logger.error(f"Salesforce API error: {e}")
30            
31            if 'UNABLE_TO_LOCK_ROW' in str(e):
32                # Record locked, retry
33                retry_count += 1
34                logger.info(f"Retrying ({retry_count}/{max_retries})...")
35                time.sleep(2 ** retry_count)  # Exponential backoff
36            else:
37                # Other API errors, don't retry
38                return None
39                
40        except Exception as e:
41            # General errors
42            logger.error(f"Unexpected error: {e}")
43            return None
44    
45    logger.error(f"Max retries ({max_retries}) exceeded")
46    return None
47
48
49# Usage
50result = safe_salesforce_operation(
51    sf.Lead.create,
52    {'FirstName': 'John', 'LastName': 'Doe', 'Company': 'Acme', 'Email': 'john@acme.com'}
53)