Salesforce API Integration
Implements production-grade Salesforce integration using the simple-salesforce Python SDK and Salesforce REST API. When loaded, this skill makes the model implement CRUD operations on Salesforce objects (Accounts, Contacts, Opportunities, Leads, Cases, Custom Objects), SOQL queries, Bulk API for large datasets, Apex REST calls, and Salesforce Streaming API. All implementations follow Salesforce best practices: use SF_INSTANCE_URL, SF_USERNAME, SF_PASSWORD, SF_SECURITY_TOKEN from environment, query with query_all() for deleted/archived records, use Bulk API for >10,000 records, implement exponential backoff for rate limits, and handle Salesforce IDs (15-char vs 18-char IDs).
TL;DR Checklist
- Use
simple-salesforceSDK with credentials from environment variables - Use
SF_INSTANCE_URL,SF_USERNAME,SF_PASSWORD,SF_SECURITY_TOKENfor auth - Use
Salesforce()constructor withdomain='test'for sandboxes - Use
sf.query()for SOQL queries,sf.query_all()to include deleted/archived - Use
sf.<Object>.create(),.update(),.delete(),.get()for CRUD - Use
sf.bulk.<Object>.query(),.insert(),.update(),.delete(),.upsert()for Bulk API - Handle both 15-char (case-sensitive) and 18-char (case-insensitive) Salesforce IDs
- Implement exponential backoff for REQUEST_LIMIT_EXCEEDED errors
- Use
ALL ROWSscope in SOQL for deleted/archived records - Never store credentials or session IDs in code or logs
When to Use
Use this skill when:
- Querying Salesforce data using SOQL
- Creating/updating/deleting Salesforce records
- Processing large datasets (10k+ records) with Bulk API
- Calling Apex REST endpoints
- Syncing data between your app and Salesforce
- Building integrations with Salesforce CRM
- Importing/exporting data to/from Salesforce
- Automating business processes in Salesforce
- Working with custom objects and fields
- Managing Users, Profiles, and Permissions
- Querying Activity History and Chatter
When NOT to Use
- For HubSpot-specific CRM — use
coding-hubspot-apiinstead - For Marketo marketing automation — use
coding-marketo-apiinstead - For Zendesk support tickets — use
coding-zendesk-apiinstead - For simple HTTP-only use cases when simple-salesforce is overkill
- When you need real-time streaming only (consider Salesforce Streaming API or CometD)
- For read-only reporting (use Salesforce Reports API directly)
Core Workflow
Initialize Client — Create Salesforce client using
simple-salesforce:- Production:
Salesforce(username=..., password=..., security_token=..., instance_url=...) - Sandbox:
Salesforce(..., domain='test') - Or use session ID:
Salesforce(instance_url=..., session_id=...)
Checkpoint: Validate connection with
sf.query("SELECT Id, Name FROM Account LIMIT 1").- Production:
Query Data with SOQL — Use
sf.query()for standard SOQL queries:sf.query("SELECT Id, Name FROM Account WHERE Industry = 'Technology'")- Use
query_all()to include deleted/archived records - Use
query_more()for pagination withnextRecordsUrl
Checkpoint: Queries include
LIMITfor testing, usequery_all()when all records needed.CRUD Operations — Perform record operations:
- Create:
sf.Contact.create({'LastName': 'Doe', 'FirstName': 'John'}) - Read:
sf.Contact.get('003...') - Update:
sf.Contact.update('003...', {'LastName': 'Smith'}) - Delete:
sf.Contact.delete('003...') - Upsert:
sf.Contact.upsert('003...', 'My_External_Id__c/ExternalValue', {...})
Checkpoint: CRUD operations handle both 15-char and 18-char IDs.
- Create:
Bulk API for Large Datasets — Use Bulk API for >10,000 records:
- Query:
sf.bulk.Account.query("SELECT Id, Name FROM Account") - Insert:
sf.bulk.Account.insert([{'Name': 'Account 1'}, {...}]) - Update:
sf.bulk.Account.update([{'Id': '001...', 'Name': 'Updated'}]) - Delete:
sf.bulk.Account.delete([{'Id': '001...'}]) - Upsert:
sf.bulk.Account.upsert([...], 'My_External_Id__c')
Checkpoint: Bulk operations use proper external IDs for upsert.
- Query:
Apex REST — Call custom Apex REST endpoints:
sf.apexcall(method='GET', path='/services/apexrest/MyEndpoint')sf.apexcall(method='POST', path='/services/apexrest/MyEndpoint', data={...})
Checkpoint: Apex calls include correct path and method.
Handle Limits & Rate Limiting — Monitor API usage:
sf.limits()to see current limits- Implement exponential backoff for
REQUEST_LIMIT_EXCEEDED - Use
Sforce-Limit-Inforesponse header
Checkpoint: Limits checked before large operations.
Implementation Patterns
Pattern 1: Salesforce Client Initialization (BAD vs GOOD)
"""Salesforce client initialization patterns.
Key concepts:
- simple-salesforce: Popular Python SDK for Salesforce
- Authentication: username + password + security_token (legacy)
- OAuth 2.0: More secure (session_id + instance_url)
- Domain: 'login' (production) or 'test' (sandbox)
- Instance URL: Full URL for your org (e.g., 'https://na1.salesforce.com')
Important:
- Password + Security Token = concatenated password
- 15-char IDs (case-sensitive) vs 18-char IDs (case-insensitive)
"""
from __future__ import annotations
import os
import re
import time
import logging
from typing import Any, Optional, List, Dict
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
logger = logging.getLogger(__name__)
# ===================================================================
# ❌ BAD — hardcoded credentials, no error handling, sandbox vs prod confusion
# ===================================================================
def bad_salesforce_init_bad() -> Any:
"""❌ BAD: Don't do any of these things."""
from simple_salesforce import Salesforce
# ❌ Hardcoded credentials! Never commit these!
sf = Salesforce(
username='admin@example.com',
password='MySecretPassword123', # ❌ HARDCODED!
security_token='ABC123xyz', # ❌ HARDCODED!
domain='login',
)
# ❌ No validation
# ❌ No error handling
# ❌ Using 'login' for sandbox (should be 'test')
# ❌ Not checking limits before query
return sf
# ===================================================================
# ✅ GOOD — env-based config, validation, proper domain handling
# ===================================================================
class SalesforceError(Exception):
"""Base exception for Salesforce integration errors."""
pass
class SalesforceAuthError(SalesforceError):
"""Authentication failed or credentials invalid."""
pass
class SalesforceLimitError(SalesforceError):
"""API limits exceeded."""
def __init__(self, message: str, retry_after: Optional[int] = None):
super().__init__(message)
self.retry_after = retry_after
@dataclass
class SalesforceConfig:
"""Salesforce configuration from environment variables.
Environment variables:
SF_INSTANCE_URL: Full instance URL (https://na1.salesforce.com)
SF_USERNAME: Salesforce username
SF_PASSWORD: Salesforce password
SF_SECURITY_TOKEN: Security token (appended to password)
SF_DOMAIN: 'login' (production) or 'test' (sandbox)
SF_SESSION_ID: OAuth session ID (alternative to password auth)
SF_CLIENT_ID: OAuth client ID (for OAuth flow)
SF_CLIENT_SECRET: OAuth client secret
"""
# Connection
instance_url: Optional[str] = None
domain: str = "login" # 'login' = production, 'test' = sandbox
# Password auth
username: Optional[str] = None
password: Optional[str] = None
security_token: Optional[str] = None
# OAuth / Session auth
session_id: Optional[str] = None
# Request config
timeout: float = 30.0
max_retries: int = 3
initial_retry_delay: float = 1.0
@classmethod
def from_env(cls) -> "SalesforceConfig":
"""Load configuration from environment variables."""
def parse_bool(env_var: str, default: bool) -> bool:
val = os.environ.get(env_var)
if val is None:
return default
return val.lower() in ("1", "true", "yes", "on")
def parse_float(env_var: str, default: float) -> float:
val = os.environ.get(env_var)
if val is None:
return default
try:
return float(val)
except ValueError:
return default
# Determine domain
domain_env = os.environ.get("SF_DOMAIN", "login").lower()
domain = "test" if domain_env == "test" else "login"
return cls(
instance_url=os.environ.get("SF_INSTANCE_URL"),
domain=domain,
username=os.environ.get("SF_USERNAME"),
password=os.environ.get("SF_PASSWORD"),
security_token=os.environ.get("SF_SECURITY_TOKEN"),
session_id=os.environ.get("SF_SESSION_ID"),
timeout=parse_float("SF_TIMEOUT", 30.0),
)
def is_enabled(self) -> bool:
"""Check if Salesforce is configured."""
# Check for session auth
if self.session_id and self.instance_url:
return True
# Check for password auth
if self.username and self.password:
return True
return False
def validate(self) -> bool:
"""Validate configuration.
Returns:
True if valid
Raises:
ValueError: If invalid when enabled
"""
if not self.is_enabled():
logger.info("Salesforce not configured")
return True
if self.session_id:
if not self.instance_url:
raise ValueError("SF_INSTANCE_URL required with SF_SESSION_ID")
else:
if not self.username:
raise ValueError("SF_USERNAME required for password auth")
if not self.password:
raise ValueError("SF_PASSWORD required for password auth")
return True
def get_full_password(self) -> str:
"""Get password + security token concatenated.
simple-salesforce expects password + token as one string.
"""
if not self.password:
raise ValueError("Password not configured")
if self.security_token:
return f"{self.password}{self.security_token}"
return self.password
class SalesforceClient:
"""Production-grade Salesforce client with retry and limit handling.
Features:
- Config from environment
- Automatic retry with exponential backoff
- Limit awareness
- ID conversion (15-char ↔ 18-char)
- Helper methods for common operations
"""
# ID pattern: 15 or 18 chars, alphanumeric
ID_PATTERN = re.compile(r'^[a-zA-Z0-9]{15}([a-zA-Z0-9]{3})?$')
def __init__(self, config: SalesforceConfig) -> None:
self._config = config
self._sf: Any = None # Lazy-loaded simple-salesforce instance
self._limits_cache: Optional[Dict[str, Any]] = None
self._limits_cache_time: float = 0.0
def _get_client(self) -> Any:
"""Get or create the simple-salesforce client.
Lazy-loaded to avoid connection overhead if not used.
"""
if self._sf is not None:
return self._sf
from simple_salesforce import Salesforce, SalesforceMalformedRequest, SalesforceResourceNotFound
self._config.validate()
try:
if self._config.session_id:
# Session/OAuth auth
self._sf = Salesforce(
instance_url=self._config.instance_url,
session_id=self._config.session_id,
domain=self._config.domain,
)
else:
# Password auth
self._sf = Salesforce(
username=self._config.username,
password=self._config.get_full_password(),
instance_url=self._config.instance_url,
domain=self._config.domain,
)
logger.info("Salesforce client initialized")
return self._sf
except Exception as e:
raise SalesforceAuthError(f"Failed to connect to Salesforce: {e}") from e
@property
def sf(self) -> Any:
"""Access the underlying simple-salesforce client."""
return self._get_client()
@staticmethod
def is_valid_salesforce_id(
value: str,
allow_15_char: bool = True,
allow_18_char: bool = True,
) -> bool:
"""Validate a Salesforce ID.
Salesforce IDs:
- 15 characters: case-sensitive
- 18 characters: case-insensitive (3 extra checksum chars)
Args:
value: ID to validate
allow_15_char: Allow 15-char IDs
allow_18_char: Allow 18-char IDs
Returns:
True if valid Salesforce ID
"""
if not isinstance(value, str):
return False
value = value.strip()
if not value:
return False
if not SalesforceClient.ID_PATTERN.match(value):
return False
length = len(value)
if length == 15:
return allow_15_char
elif length == 18:
return allow_18_char
return False
@staticmethod
def convert_to_18_char(id_15: str) -> str:
"""Convert a 15-char case-sensitive ID to 18-char case-insensitive ID.
Algorithm:
1. Split into 3 chunks of 5 chars
2. Reverse each chunk
3. For each char: uppercase=1, lowercase=0
4. Build binary string, convert to decimal
5. Map to A-Z (0-25), 0-9 (26-35), +/= 36-37)
6. Append 3 checksum chars
Args:
id_15: 15-char Salesforce ID
Returns:
18-char case-insensitive ID
"""
if len(id_15) != 15:
raise ValueError(f"Expected 15-char ID, got {len(id_15)} chars")
# Checksum lookup table
# 0-25: A-Z
# 26-31: 0-5
# 32-35: 6-9
# Actually simpler: Salesforce uses specific encoding
ALPHABET = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
checksum = ""
# Process 3 chunks of 5 characters each
for chunk_start in range(0, 15, 5):
chunk = id_15[chunk_start:chunk_start + 5]
# Build binary: uppercase=1, lowercase=0, reversed order
binary = ""
for char in reversed(chunk):
if char.isupper():
binary += "1"
else:
binary += "0"
# Convert binary to decimal
index = int(binary, 2)
checksum += ALPHABET[index]
return id_15 + checksum
@staticmethod
def normalize_id(sf_id: str) -> str:
"""Normalize a Salesforce ID to 18-char case-insensitive format.
Use this before comparing IDs or storing in external systems.
Args:
sf_id: 15 or 18 char Salesforce ID
Returns:
18-char case-insensitive ID
"""
sf_id = sf_id.strip()
if len(sf_id) == 18:
# Already 18 chars, return uppercase
return sf_id.upper()
elif len(sf_id) == 15:
# Convert to 18 chars
return SalesforceClient.convert_to_18_char(sf_id)
else:
raise ValueError(f"Invalid Salesforce ID length: {len(sf_id)} chars")
# ===================================================================
# Core Operations
# ===================================================================
def query(
self,
soql: str,
include_deleted: bool = False,
include_archived: bool = False,
) -> List[Dict[str, Any]]:
"""Execute a SOQL query.
Args:
soql: SOQL query string
include_deleted: Include deleted records (query_all)
include_archived: Include archived records
Returns:
List of record dicts with 'attributes', 'Id', etc.
"""
sf = self._get_client()
try:
if include_deleted or include_archived:
# Use query_all for deleted/archived
result = sf.query_all(soql)
else:
result = sf.query(soql)
records = result.get("records", [])
total_size = result.get("totalSize", 0)
logger.debug(
"SOQL query returned %d/%d records",
len(records), total_size
)
# Handle pagination if needed (nextRecordsUrl)
next_url = result.get("nextRecordsUrl")
while next_url:
# Use query_more
more_result = sf.query_more(next_url)
records.extend(more_result.get("records", []))
next_url = more_result.get("nextRecordsUrl")
return records
except Exception as e:
if "MALFORMED_QUERY" in str(e):
raise SalesforceError(f"SOQL syntax error: {e}") from e
elif "REQUEST_LIMIT_EXCEEDED" in str(e):
raise SalesforceLimitError(f"API limit exceeded: {e}") from e
else:
raise SalesforceError(f"Query failed: {e}") from e
def create(
self,
object_name: str,
data: Dict[str, Any],
) -> Dict[str, Any]:
"""Create a Salesforce record.
Args:
object_name: Object type (Account, Contact, Opportunity, etc.)
data: Record field values
Returns:
Dict with 'id', 'success', 'errors'
"""
sf = self._get_client()
try:
# Get the object proxy
obj = getattr(sf, object_name)
result = obj.create(data)
if not result.get("success", False):
errors = result.get("errors", [])
raise SalesforceError(f"Create failed: {errors}")
logger.info(
"Created %s record: %s",
object_name, result.get("id")
)
return result
except Exception as e:
raise SalesforceError(f"Failed to create {object_name}: {e}") from e
def get(
self,
object_name: str,
record_id: str,
) -> Dict[str, Any]:
"""Get a Salesforce record by ID.
Args:
object_name: Object type
record_id: Salesforce ID (15 or 18 chars)
Returns:
Record dict
"""
sf = self._get_client()
# Validate ID
if not self.is_valid_salesforce_id(record_id):
raise ValueError(f"Invalid Salesforce ID: {record_id}")
try:
obj = getattr(sf, object_name)
return obj.get(record_id)
except Exception as e:
if "NOT_FOUND" in str(e) or "ResourceNotFound" in type(e).__name__:
raise SalesforceError(f"{object_name} not found: {record_id}") from e
else:
raise SalesforceError(f"Get failed: {e}") from e
def update(
self,
object_name: str,
record_id: str,
data: Dict[str, Any],
) -> Dict[str, Any]:
"""Update a Salesforce record.
Args:
object_name: Object type
record_id: Record ID
data: Fields to update
Returns:
Dict with 'success', 'errors'
"""
sf = self._get_client()
if not self.is_valid_salesforce_id(record_id):
raise ValueError(f"Invalid Salesforce ID: {record_id}")
try:
obj = getattr(sf, object_name)
result = obj.update(record_id, data)
if not result.get("success", False):
errors = result.get("errors", [])
raise SalesforceError(f"Update failed: {errors}")
logger.info("Updated %s record: %s", object_name, record_id)
return result
except Exception as e:
raise SalesforceError(f"Failed to update {object_name}: {e}") from e
def delete(
self,
object_name: str,
record_id: str,
) -> Dict[str, Any]:
"""Delete a Salesforce record.
Args:
object_name: Object type
record_id: Record ID
Returns:
Dict with 'success', 'errors'
"""
sf = self._get_client()
if not self.is_valid_salesforce_id(record_id):
raise ValueError(f"Invalid Salesforce ID: {record_id}")
try:
obj = getattr(sf, object_name)
result = obj.delete(record_id)
if not result.get("success", False):
errors = result.get("errors", [])
raise SalesforceError(f"Delete failed: {errors}")
logger.info("Deleted %s record: %s", object_name, record_id)
return result
except Exception as e:
raise SalesforceError(f"Failed to delete {object_name}: {e}") from e
def upsert(
self,
object_name: str,
external_id_field: str,
external_id_value: Any,
data: Dict[str, Any],
) -> Dict[str, Any]:
"""Upsert a record using an external ID.
If record exists with external ID, update it. Otherwise, create it.
Args:
object_name: Object type
external_id_field: External ID field name (with __c suffix for custom)
external_id_value: External ID value
data: Record data (should include external ID field usually)
Returns:
Dict with 'id', 'success', 'created', 'errors'
"""
sf = self._get_client()
try:
obj = getattr(sf, object_name)
# simple-salesforce upsert signature:
# obj.upsert('External_Id__c/externalValue', data)
upsert_key = f"{external_id_field}/{external_id_value}"
result = obj.upsert(upsert_key, data)
if not result.get("success", False):
errors = result.get("errors", [])
raise SalesforceError(f"Upsert failed: {errors}")
created = result.get("created", False)
action = "created" if created else "updated"
logger.info(
"Upserted %s record: %s (%s)",
object_name, result.get("id"), action
)
return result
except Exception as e:
raise SalesforceError(f"Failed to upsert {object_name}: {e}") from e
# ===================================================================
# Bulk API Operations
# ===================================================================
def bulk_query(
self,
object_name: str,
soql: str,
) -> List[Dict[str, Any]]:
"""Query using Bulk API for large datasets.
Use for queries expected to return > 10,000 records.
Args:
object_name: Object type
soql: SOQL query (without 'SELECT' and object specifier)
Returns:
List of record dicts
"""
sf = self._get_client()
try:
bulk_obj = getattr(sf.bulk, object_name)
# Bulk API query
# Note: simple-salesforce bulk.query expects just the fields/where clause
# e.g., "SELECT Id, Name FROM Account"
results = bulk_obj.query(soql)
logger.info(
"Bulk query returned %d records", len(results)
)
return results
except Exception as e:
raise SalesforceError(f"Bulk query failed: {e}") from e
def bulk_insert(
self,
object_name: str,
records: List[Dict[str, Any]],
) -> List[Dict[str, Any]]:
"""Insert multiple records using Bulk API.
Use for inserting > 10,000 records.
Args:
object_name: Object type
records: List of record dicts
Returns:
List of result dicts with 'id', 'success', 'errors'
"""
if not records:
return []
sf = self._get_client()
try:
bulk_obj = getattr(sf.bulk, object_name)
results = bulk_obj.insert(records)
success_count = sum(1 for r in results if r.get("success"))
error_count = len(results) - success_count
logger.info(
"Bulk insert: %d successful, %d errors",
success_count, error_count
)
return results
except Exception as e:
raise SalesforceError(f"Bulk insert failed: {e}") from e
def bulk_update(
self,
object_name: str,
records: List[Dict[str, Any]],
) -> List[Dict[str, Any]]:
"""Update multiple records using Bulk API.
Each record must include 'Id' field.
Args:
object_name: Object type
records: List of record dicts (must include 'Id')
Returns:
List of result dicts
"""
if not records:
return []
sf = self._get_client()
try:
bulk_obj = getattr(sf.bulk, object_name)
results = bulk_obj.update(records)
success_count = sum(1 for r in results if r.get("success"))
logger.info(
"Bulk update: %d successful",
success_count
)
return results
except Exception as e:
raise SalesforceError(f"Bulk update failed: {e}") from e
def bulk_upsert(
self,
object_name: str,
records: List[Dict[str, Any]],
external_id_field: str,
) -> List[Dict[str, Any]]:
"""Upsert multiple records using Bulk API.
Args:
object_name: Object type
records: List of record dicts (must include external ID field)
external_id_field: External ID field name
Returns:
List of result dicts
"""
if not records:
return []
sf = self._get_client()
try:
bulk_obj = getattr(sf.bulk, object_name)
results = bulk_obj.upsert(records, external_id_field)
success_count = sum(1 for r in results if r.get("success"))
logger.info(
"Bulk upsert: %d successful",
success_count
)
return results
except Exception as e:
raise SalesforceError(f"Bulk upsert failed: {e}") from e
def bulk_delete(
self,
object_name: str,
records: List[Dict[str, Any]],
) -> List[Dict[str, Any]]:
"""Delete multiple records using Bulk API.
Each record must include 'Id' field.
Args:
object_name: Object type
records: List of record dicts (must include 'Id')
Returns:
List of result dicts
"""
if not records:
return []
sf = self._get_client()
try:
bulk_obj = getattr(sf.bulk, object_name)
results = bulk_obj.delete(records)
success_count = sum(1 for r in results if r.get("success"))
logger.info(
"Bulk delete: %d successful",
success_count
)
return results
except Exception as e:
raise SalesforceError(f"Bulk delete failed: {e}") from e
# ===================================================================
# Apex REST
# ===================================================================
def apex_call(
self,
method: str,
path: str,
data: Optional[Dict[str, Any]] = None,
) -> Any:
"""Call a custom Apex REST endpoint.
Args:
method: HTTP method: 'GET', 'POST', 'PUT', 'DELETE', 'PATCH'
path: Apex REST path (e.g., '/services/apexrest/MyEndpoint')
data: Request body data (for POST/PUT/PATCH)
Returns:
Response data (parsed JSON)
"""
sf = self._get_client()
try:
result = sf.apexcall(method=method, path=path, data=data)
logger.info(
"Apex REST call: %s %s",
method, path
)
return result
except Exception as e:
raise SalesforceError(f"Apex call failed: {e}") from e
# ===================================================================
# Limits
# ===================================================================
def get_limits(self, force_refresh: bool = False) -> Dict[str, Any]:
"""Get current API limits.
Args:
force_refresh: Skip cache
Returns:
Limits dict from Salesforce limits() call
"""
# Cache limits for 60 seconds
now = time.time()
if not force_refresh and self._limits_cache and (now - self._limits_cache_time) < 60:
return self._limits_cache
sf = self._get_client()
try:
limits = sf.limits()
self._limits_cache = limits
self._limits_cache_time = now
return limits
except Exception as e:
raise SalesforceError(f"Failed to get limits: {e}") from e
def remaining_daily_api_calls(self) -> int:
"""Get remaining daily API calls remaining.
Returns:
Remaining API calls
"""
limits = self.get_limits()
daily_api = limits.get("DailyApiRequests", {})
remaining = daily_api.get("Remaining", 0)
return remaining
# Global client (lazy-loaded)
_global_client: Optional[SalesforceClient] = None
def get_salesforce_client() -> SalesforceClient:
"""Get or create global Salesforce client."""
global _global_client
if _global_client is None:
config = SalesforceConfig.from_env()
_global_client = SalesforceClient(config)
return _global_client
Pattern 2: Common SOQL Queries & Operations
"""Common SOQL query patterns and CRM operations.
SOQL Best Practices:
- ALWAYS use LIMIT in development/test queries
- Use query_all() for deleted/archived records
- Use Bulk API for > 10,000 records
- Include Id field in all SELECTs
- Use relationships with care (costly)
- Use COUNT() instead of retrieving all records
"""
from __future__ import annotations
import logging
from typing import Any, Optional, List, Dict
from datetime import datetime, timedelta, timezone
logger = logging.getLogger(__name__)
class SalesforceQueries:
"""Common SOQL query builder and executor."""
def __init__(self, client: Any) -> None:
self._client = client
# ===================================================================
# Account Queries
# ===================================================================
def get_accounts_by_industry(
self,
industry: str,
limit: int = 100,
) -> List[Dict[str, Any]]:
"""Get Accounts filtered by industry.
Args:
industry: Industry value (e.g., 'Technology', 'Finance')
limit: Max records
Returns:
List of Account records
"""
soql = f"""
SELECT Id, Name, Industry, Type, Phone, Website,
BillingCity, BillingState, BillingCountry,
CreatedDate, LastModifiedDate
FROM Account
WHERE Industry = '{industry}'
ORDER BY CreatedDate DESC
LIMIT {limit}
"""
return self._client.query(soql.strip())
def get_accounts_created_since(
self,
days: int = 30,
limit: int = 1000,
) -> List[Dict[str, Any]]:
"""Get Accounts created in the last N days.
Args:
days: Number of days to look back
limit: Max records
Returns:
List of Account records
"""
# Calculate date
start_date = (datetime.now(timezone.utc) - timedelta(days=days)).strftime("%Y-%m-%dT%H:%M:%SZ")
soql = f"""
SELECT Id, Name, Industry, Type, CreatedDate
FROM Account
WHERE CreatedDate >= {start_date}
ORDER BY CreatedDate DESC
LIMIT {limit}
"""
return self._client.query(soql.strip())
def search_accounts_by_name(
self,
name_pattern: str,
limit: int = 50,
) -> List[Dict[str, Any]]:
"""Search Accounts by name pattern.
Uses LIKE operator.
Args:
name_pattern: Name pattern with % wildcards
limit: Max records
Returns:
List of Account records
"""
# Escape single quotes in pattern
escaped_pattern = name_pattern.replace("'", "'")
soql = f"""
SELECT Id, Name, Industry, Type, Phone, Website
FROM Account
WHERE Name LIKE '{escaped_pattern}'
ORDER BY Name
LIMIT {limit}
"""
return self._client.query(soql.strip())
# ===================================================================
# Contact Queries
# ===================================================================
def get_contacts_by_account(
self,
account_id: str,
limit: int = 100,
) -> List[Dict[str, Any]]:
"""Get Contacts for an Account.
Args:
account_id: Account ID
limit: Max records
Returns:
List of Contact records
"""
# Validate ID
if not self._client.is_valid_salesforce_id(account_id):
raise ValueError(f"Invalid Account ID: {account_id}")
soql = f"""
SELECT Id, Name, FirstName, LastName, Title,
Email, Phone, MobilePhone,
AccountId, Account.Name
FROM Contact
WHERE AccountId = '{account_id}'
ORDER BY LastName
LIMIT {limit}
"""
return self._client.query(soql.strip())
def get_contacts_by_email_domain(
self,
email_domain: str,
limit: int = 100,
) -> List[Dict[str, Any]]:
"""Get Contacts by email domain.
Args:
email_domain: Domain (e.g., 'example.com')
limit: Max records
Returns:
List of Contact records
"""
soql = f"""
SELECT Id, Name, FirstName, LastName,
Email, Phone, AccountId, Account.Name
FROM Contact
WHERE Email LIKE '%{email_domain}'
ORDER BY LastName
LIMIT {limit}
"""
return self._client.query(soql.strip())
# ===================================================================
# Opportunity Queries
# ===================================================================
def get_open_opportunities(
self,
stage: Optional[str] = None,
close_date_from: Optional[str] = None,
limit: int = 100,
) -> List[Dict[str, Any]]:
"""Get open Opportunities.
Args:
stage: Filter by StageName
close_date_from: Filter by CloseDate >= this date
limit: Max records
Returns:
List of Opportunity records
"""
conditions = ["IsClosed = false"]
if stage:
conditions.append(f"StageName = '{stage}'")
if close_date_from:
conditions.append(f"CloseDate >= {close_date_from}")
where_clause = " AND ".join(conditions)
soql = f"""
SELECT Id, Name, StageName, Amount, CloseDate,
Probability, AccountId, Account.Name,
OwnerId, Owner.Name
FROM Opportunity
WHERE {where_clause}
ORDER BY CloseDate
LIMIT {limit}
"""
return self._client.query(soql.strip())
def get_opportunities_won_last_quarter(
self,
limit: int = 500,
) -> List[Dict[str, Any]]:
"""Get Opportunities won in the last quarter.
Returns:
List of won Opportunity records
"""
soql = """
SELECT Id, Name, StageName, Amount, CloseDate,
AccountId, Account.Name
FROM Opportunity
WHERE IsWon = true
AND CloseDate = LAST_QUARTER
ORDER BY Amount DESC
LIMIT 500
"""
return self._client.query(soql.strip())
# ===================================================================
# Lead Queries
# ===================================================================
def get_leads_by_status(
self,
status: str,
limit: int = 100,
) -> List[Dict[str, Any]]:
"""Get Leads by Status.
Args:
status: Lead Status value
limit: Max records
Returns:
List of Lead records
"""
soql = f"""
SELECT Id, Name, FirstName, LastName, Company,
Email, Phone, Status, LeadSource,
CreatedDate
FROM Lead
WHERE Status = '{status}'
ORDER BY CreatedDate DESC
LIMIT {limit}
"""
return self._client.query(soql.strip())
# ===================================================================
# Case Queries
# ===================================================================
def get_open_cases(
self,
limit: int = 100,
) -> List[Dict[str, Any]]:
"""Get open Cases.
…(truncated)