Salesforce API

Provides patterns for interacting with the Salesforce REST API.

Authentication

import requests

def get_salesforce_access_token(
    client_id: str,
    client_secret: str,
    username: str,
    password: str,
    security_token: str,
    login_url: str = "https://login.salesforce.com"
) -> dict:
    """
    Authenticate with Salesforce using username-password flow.

    Returns dict with access_token and instance_url.

    Usage:
        auth = get_salesforce_access_token(
            client_id="your_client_id",
            client_secret="your_client_secret",
            username="user@company.com",
            password="password",
            security_token="token"
        )
        access_token = auth["access_token"]
        instance_url = auth["instance_url"]
    """
    response = requests.post(
        f"{login_url}/services/oauth2/token",
        data={
            "grant_type": "password",
            "client_id": client_id,
            "client_secret": client_secret,
            "username": username,
            "password": password + security_token
        }
    )
    response.raise_for_status()
    return response.json()

def get_headers(access_token: str) -> dict:
    """Get headers for API requests."""
    return {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json"
    }

SOQL Queries

def query(instance_url: str, access_token: str, soql: str) -> list[dict]:
    """
    Execute SOQL query.

    Usage:
        contacts = query(
            instance_url,
            access_token,
            "SELECT Id, FirstName, LastName, Email FROM Contact LIMIT 100"
        )
    """
    import urllib.parse

    url = f"{instance_url}/services/data/v58.0/query"
    params = {"q": soql}

    response = requests.get(url, headers=get_headers(access_token), params=params)
    response.raise_for_status()
    return response.json().get("records", [])

def query_all(instance_url: str, access_token: str, soql: str) -> list[dict]:
    """Execute SOQL query with pagination."""
    records = []
    url = f"{instance_url}/services/data/v58.0/query"
    params = {"q": soql}

    while url:
        response = requests.get(url, headers=get_headers(access_token), params=params)
        response.raise_for_status()
        data = response.json()

        records.extend(data.get("records", []))

        next_url = data.get("nextRecordsUrl")
        url = f"{instance_url}{next_url}" if next_url else None
        params = None  # Only needed for first request

    return records

CRUD Operations

def get_record(instance_url: str, access_token: str, object_type: str, record_id: str) -> dict:
    """
    Get a single record by ID.

    Usage:
        contact = get_record(instance_url, access_token, "Contact", "0031234567890")
    """
    url = f"{instance_url}/services/data/v58.0/sobjects/{object_type}/{record_id}"
    response = requests.get(url, headers=get_headers(access_token))
    response.raise_for_status()
    return response.json()

def create_record(instance_url: str, access_token: str, object_type: str, data: dict) -> str:
    """
    Create a new record.

    Returns the new record ID.

    Usage:
        contact_id = create_record(instance_url, access_token, "Contact", {
            "FirstName": "John",
            "LastName": "Doe",
            "Email": "john@example.com"
        })
    """
    url = f"{instance_url}/services/data/v58.0/sobjects/{object_type}"
    response = requests.post(url, headers=get_headers(access_token), json=data)
    response.raise_for_status()
    return response.json()["id"]

def update_record(instance_url: str, access_token: str, object_type: str, record_id: str, data: dict) -> bool:
    """
    Update an existing record.

    Usage:
        update_record(instance_url, access_token, "Contact", "0031234567890", {
            "Phone": "555-1234"
        })
    """
    url = f"{instance_url}/services/data/v58.0/sobjects/{object_type}/{record_id}"
    response = requests.patch(url, headers=get_headers(access_token), json=data)
    return response.status_code == 204

def delete_record(instance_url: str, access_token: str, object_type: str, record_id: str) -> bool:
    """Delete a record."""
    url = f"{instance_url}/services/data/v58.0/sobjects/{object_type}/{record_id}"
    response = requests.delete(url, headers=get_headers(access_token))
    return response.status_code == 204

def upsert_record(
    instance_url: str,
    access_token: str,
    object_type: str,
    external_id_field: str,
    external_id: str,
    data: dict
) -> dict:
    """
    Upsert record using external ID.

    Usage:
        result = upsert_record(
            instance_url, access_token,
            "Contact", "External_ID__c", "ext-123",
            {"FirstName": "John", "LastName": "Doe"}
        )
    """
    url = f"{instance_url}/services/data/v58.0/sobjects/{object_type}/{external_id_field}/{external_id}"
    response = requests.patch(url, headers=get_headers(access_token), json=data)
    response.raise_for_status()
    return {"created": response.status_code == 201}

Bulk Operations

def bulk_create(instance_url: str, access_token: str, object_type: str, records: list[dict]) -> list[dict]:
    """
    Create multiple records using Composite API.

    Usage:
        results = bulk_create(instance_url, access_token, "Contact", [
            {"FirstName": "John", "LastName": "Doe"},
            {"FirstName": "Jane", "LastName": "Smith"}
        ])
    """
    url = f"{instance_url}/services/data/v58.0/composite/sobjects"

    payload = {
        "allOrNone": False,
        "records": [
            {"attributes": {"type": object_type}, **record}
            for record in records
        ]
    }

    response = requests.post(url, headers=get_headers(access_token), json=payload)
    response.raise_for_status()
    return response.json()

Common Queries

def get_all_contacts(instance_url: str, access_token: str) -> list[dict]:
    """Get all contacts."""
    return query_all(
        instance_url, access_token,
        "SELECT Id, FirstName, LastName, Email, Phone, AccountId FROM Contact"
    )

def get_contacts_by_account(instance_url: str, access_token: str, account_id: str) -> list[dict]:
    """Get contacts for an account."""
    return query(
        instance_url, access_token,
        f"SELECT Id, FirstName, LastName, Email FROM Contact WHERE AccountId = '{account_id}'"
    )

def search_contacts(instance_url: str, access_token: str, email: str) -> list[dict]:
    """Search contacts by email."""
    return query(
        instance_url, access_token,
        f"SELECT Id, FirstName, LastName, Email FROM Contact WHERE Email = '{email}'"
    )

Helper Script

Use helper.py for the SalesforceClient class with comprehensive API features.