forcepy 0.1.7

A faster way to work with Salesforce - Modern Python client with advanced features

forcepy ⚡

A faster way to work with Salesforce data.

Forcepy transforms the way you interact with Salesforce APIs - turning complex operations into simple, Pythonic code. Build integrations, automate workflows, or migrate data with minimal effort and maximum power.

Installation • Quickstart • Features • Documentation • Examples

---

⚡ Core Features

• Transform your Salesforce workflow with ZERO BOILERPLATE (almost)! 🚀
• Build with ANY Salesforce use case (REST API, Bulk API 2.0, Composite API, Chatter, Metadata); or even WITHOUT complexity (just import and go). You name it! 🔥
• 🎯 Simple and Pythonic - Write beautiful, easy-to-read code that feels native to Python
• ⚡ Lightning fast - Smart caching, connection pooling, and auto-retry logic built-in
• 🔐 Multiple auth methods - SOAP, OAuth2, and JWT Bearer Flow all supported
• 🚀 Advanced query building - Q objects, client-side filtering, and SOQL helpers
• 📦 Bulk operations - Handle millions of records with Bulk API 2.0
• 📊 Developer-friendly - Full type hints, comprehensive docs, and 349 passing tests

Installation

# Basic installation
pip install forcepy

# With JWT support
pip install forcepy[jwt]

# Using uv (recommended)
uv add forcepy

Quickstart

A little example

Create a file my_script.py:

from forcepy import Salesforce

# Production org (default)
sf = Salesforce(
    username='[e-mail]',
    password='password'
)

# With security token (if required)
sf = Salesforce(
    username='[e-mail]',
    password='password',
    security_token='yourSecurityToken123'
)

# Sandbox org
sf = Salesforce(
    username='[e-mail]',
    password='password',
    sandbox=True
)

# Query data with dot notation
accounts = sf.query("SELECT Id, Name, Industry FROM Account LIMIT 5")
for account in accounts.records:
    print(f"{account.Name} - {account.Industry}")

Run it:

python my_script.py

That's it! You're already querying Salesforce data. 🎉

Get more power!

Forcepy comes packed with advanced features:

🏭 Production-Ready Patterns

Update Records with .patch()

# Update via direct endpoint
sf.Account[account_id].patch(
    Phone='555-1234',
    Industry='Technology',
    Website='https://acme.com',
    timeout=60  # Custom timeout
)

# Update via object instance
account = sf.Account.get(Name='Acme Corp')
account.patch(sf, Phone='555-5678')

# Bulk updates from query results
cases = sf.query("SELECT Id FROM Case WHERE Status = 'New' AND Priority = 'High'")
for case in cases:
    sf.Case[case.Id].patch(Status='In Progress', OwnerId=user_id)

SOQL Helper Functions

from forcepy import DATE, IN, BOOL
from datetime import datetime, timedelta

# Date formatting for queries
last_week = DATE(datetime.now() - timedelta(days=7))
recent_cases = sf.query(f"SELECT Id FROM Case WHERE CreatedDate >= {last_week}")

# IN clause for lists
regions = ['US', 'EU', 'APAC']
accounts = sf.query(f"SELECT Id, Name FROM Account WHERE Region__c IN {IN(regions)}")

# Boolean formatting
active = sf.query(f"SELECT Id FROM Account WHERE IsActive__c = {BOOL(True)}")

Child Relationship Queries (Subqueries)

# Query parent with multiple child relationships
accounts = sf.query("""
    SELECT Id, Name, Industry,
           (SELECT Id, FirstName, LastName, Email FROM Contacts),
           (SELECT Id, Name, StageName, Amount FROM Opportunities WHERE StageName = 'Closed Won')
    FROM Account
    WHERE Industry = 'Technology'
""")

# Access child records safely
for account in accounts:
    print(f"Account: {account.Name}")
    
    # Pattern 1: Safe check before iteration
    if account.Contacts:
        for contact in account.Contacts.records:
            print(f"  Contact: {contact.FirstName} {contact.LastName}")
    
    # Pattern 2: Default to empty list
    for opp in account.Opportunities and account.Opportunities.records or []:
        print(f"  Opportunity: {opp.Name} - ${opp.Amount}")

Parent Relationship Traversal

# Traverse up parent relationships
contacts = sf.query("""
    SELECT Id, FirstName, LastName,
           Account.Name,
           Account.Owner.Name,
           Account.Owner.Email
    FROM Contact
    WHERE Account.Industry = 'Technology'
""")

for contact in contacts:
    print(f"{contact.FirstName} works at {contact.Account.Name}")
    print(f"  Account owner: {contact.Account.Owner.Name}")

Query Timeouts

# Set custom timeout for long-running queries
large_dataset = sf.query(
    "SELECT Id, Name, (SELECT Id FROM Contacts) FROM Account",
    timeout=120  # 120 seconds
)

Convenience Methods

# Get earliest/latest records
cases = sf.query("SELECT Id, CaseNumber, CreatedDate FROM Case WHERE Status = 'Open'")

oldest_case = cases.earliest('CreatedDate')  # Or just .earliest()
newest_case = cases.latest('CreatedDate')

# Group and count
by_status = cases.group_by('Status').count()

🔍 Advanced Query Building

from forcepy import Q

# Build complex queries with Q objects
high_value = Q(AnnualRevenue__gt=1000000) | Q(NumberOfEmployees__gt=500)
tech_companies = Q(Industry='Technology') & high_value

accounts = sf.query(f"SELECT Id, Name FROM Account WHERE {tech_companies.compile()}")

🎛️ Client-Side Filtering

# Query once, filter many times
cases = sf.query("SELECT Id, Status, Priority FROM Case LIMIT 1000")

# Filter client-side
urgent = cases.records.filter(Priority='High', Status='New')

# Group and aggregate
by_status = cases.records.group_by('Status').count()
# {'New': 450, 'In Progress': 300, 'Closed': 250}

⚙️ Composite API (Batch Operations)

# Execute up to 25 operations in a single API call
with sf as batch:
    batch.sobjects.Account.post(Name='Acme Corp', Industry='Technology')
    batch.sobjects.Account.post(Name='Global Inc', Industry='Finance')
    batch.sobjects.Contact['003xx000004TmiQQAS'].patch(LastName='Smith')
# All operations execute atomically on exit

💬 Chatter Integration

from forcepy import Chatter

chatter = Chatter(sf)

# Post with @mentions and $record links (entity tagging)
chatter.post("Hey @[005xx0000012345]! Please review $[001xx0000012345]")

# HTML formatting
chatter.post("<b>Important:</b> <i>Q4 goals achieved!</i>")

# Post to groups
chatter.post_to_group("0F9xx000000abcd", "Team meeting at 2pm!")

Entity Tagging:

• @[userId] - Mention users (e.g., @[005xx0000012345])
• $[recordId] - Link to records (e.g., $[001xx0000012345])

🎁 Developer Experience Features

# Convenience methods - cleaner code
accounts = sf.query("SELECT Id, Name FROM Account LIMIT 10")
first = accounts.records.first()  # Instead of [0]
last = accounts.records.last()    # Instead of [-1]

# Case-insensitive filtering - more flexible
cases = sf.query("SELECT Id, Subject, Status FROM Case")
urgent = cases.records.filter(Subject__icontains='urgent')  # Matches "URGENT", "Urgent", etc.
new_cases = cases.records.filter(Status__iexact='new')      # Case-insensitive exact match

# CSV export/import - easy data exchange
accounts.records.to_csv('accounts.csv')
from forcepy.results import ResultSet
imported = ResultSet.from_csv('accounts.csv')

🗄️ Bulk API 2.0

# Handle millions of records efficiently
records = [{'Name': f'Account {i}', 'Industry': 'Technology'} for i in range(10000)]
job = sf.bulk.Account.insert(records)

# Wait for completion and get results
results = job.wait()
print(f"Processed: {results['numberRecordsProcessed']}")
print(f"Failed: {results['numberRecordsFailed']}")

# Or query large datasets
job = sf.bulk.Account.query("SELECT Id, Name FROM Account")
for batch in job.get_results():
    for record in batch:
        print(record['Name'])

Key Features

🎯 Query & Filtering

• Q Objects - Build complex WHERE clauses with boolean logic
• Query helpers - IN(), DATE(), BOOL() for clean SOQL
• SELECT * expansion - Automatically expand to all fields
• Client-side filtering - filter(), group_by(), order_by() on results
• Iterquery - Efficient pagination with optional threading

🔐 Authentication & Security

• SOAP login - Username/password authentication (auto-detects sandbox)
• JWT Bearer Flow - Certificate-based authentication for production
• OAuth2 - Full OAuth2 support
• Token caching - Automatic caching (memory/Redis) for performance
• Session tracking - Monitor user_id, session expiry, last request time

⚡ Performance & Reliability

• Auto-retry - Configurable retries on 503, 502, 500, UNABLE_TO_LOCK_ROW, 429
• Connection pooling - Efficient HTTP session management
• Smart caching - Describe/metadata results cached automatically
• Manual pagination - Full control with query_more() and next_records_url

🛠️ Developer Experience

• Full type hints - Better IDE autocomplete and type checking
• Dot notation - Access nested fields intuitively
• Dynamic endpoints - Chain attributes to build API paths
• Workbench URLs - Generate shareable query links
• Pretty print - Format SOQL for readability
• Object discovery - List and explore Salesforce objects
• ID utilities - Compare 15/18 char IDs, determine object type from ID
• Convenience methods - .first(), .last() for cleaner code
• Case-insensitive filters - __icontains, __iexact for flexible searching
• CSV export/import - Easy data exchange with .to_csv() and .from_csv()

📦 Batch Operations

• Composite API - Batch up to 25 operations with reference support
• Context manager - Pythonic with statement for batching
• All-or-none - Atomic transactions option

📊 Metadata & Schema

• Cached describe - Fast metadata access with automatic caching
• Field information - Required fields, picklist values, field properties
• Dependent picklists - Filter picklist values by controlling field
• Org limits - Check API usage and limits

Get Inspired

Here's what you can build with forcepy:

🔄 Data Migration

Migrate millions of records between orgs with bulk operations and smart error handling.

🤖 Automation & Integration

Build workflows that sync Salesforce with external systems - CRMs, ERPs, databases, and more.

📊 Analytics & Reporting

Extract Salesforce data for custom analytics, dashboards, and business intelligence.

🧪 Testing & CI/CD

Populate test orgs, validate deployments, and automate quality assurance.

📱 Custom Applications

Build custom apps that extend Salesforce capabilities beyond the platform.

Documentation

• 🏭 Production Patterns - Battle-tested patterns for enterprise use
• 🔄 Migration from simple-salesforce - Step-by-step migration guide
• 📖 Token Caching Guide - Production caching strategies
• 🔐 Authentication Guide - All auth methods explained
• 💬 Chatter Features - Complete Chatter reference
• 💡 Examples - 13 ready-to-run code examples (including bulk operations!)

Resources

• Documentation:
  • Production Patterns - Child queries, updates, helpers, multi-org
  • Migration from simple-salesforce - Step-by-step guide
  • Authentication Guide - All auth methods explained
  • Token Caching Guide - Production caching strategies
  • Chatter Features - Complete Chatter reference
  • Bulk API Guide - Handle large-scale operations
  • Library Comparison - vs simple-salesforce
• Examples: 13 ready-to-run code examples
  • Basic queries and CRUD operations
  • Advanced filtering with Q objects
  • Bulk API 2.0 for large-scale operations
  • JWT and OAuth authentication
  • Chatter integration
  • Composite API batch operations
  • Token caching strategies
  • And more!
• Issues: Report bugs or request features
• Discussions: Get help from the community

Comparison with simple-salesforce

Forcepy builds on the foundation of simple-salesforce with many powerful additions:

Feature	forcepy	simple-salesforce
Q Objects for complex queries	✅	❌
Query building helpers	✅	❌
Client-side filtering/grouping	✅	❌
JWT authentication	✅	❌
Token caching (automatic)	✅	❌
Redis cache support	✅	❌
Composite API	✅	❌
Dependent picklists	✅	❌
SELECT * expansion	✅	❌
Workbench URL generation	✅	❌
ID comparison utilities	✅	❌
Auto-retry on errors	✅	❌
Threaded iterquery	✅	❌
Session info properties	✅	❌
Composite context manager	✅	❌
Chatter integration	✅	Limited
Type hints	✅	Limited
Dot notation	✅	✅
SOQL queries	✅	✅
Bulk API 2.0	✅	✅

Examples

Basic Operations

from forcepy import Salesforce

sf = Salesforce(username='[e-mail]', password='password')

# Get by ID (three ways!)
account = sf.Account.get('001xx0000012345')
# or: sf.Account['001xx0000012345'].get()
# or: sf.sobjects.Account['001xx0000012345'].get()

# Get by field name 🎉
case = sf.Case.get(CaseNumber='00001234')
account = sf.Account.get(Name='Acme Corp')

# Filter - NEW! 🔥 (simple-salesforce doesn't have this!)
tech_accounts = sf.Account.filter(Industry='Technology')
open_cases = sf.Case.filter(Status='Open', Priority='High')
contacts = sf.Contact.filter(Email__contains='@acme.com')

# Filter with custom fields
accounts = sf.Account.filter(
    fields=['Id', 'Name', 'Industry', 'AnnualRevenue'],
    Industry='Technology',
    AnnualRevenue__gte=1000000
)

# Create
result = sf.Account.post(Name='Acme Corp', Industry='Technology')
account_id = result['id']

# Update
sf.Account[account_id].patch(Industry='Manufacturing')

# Delete
sf.Account[account_id].delete()

💡 Tip: .get() returns one record (or errors). .filter() returns multiple records. Both support field queries!

Advanced Query with Q Objects

from forcepy import Q

# Complex boolean logic
tech_or_finance = Q(Industry='Technology') | Q(Industry='Finance')
high_revenue = Q(AnnualRevenue__gte=1000000)
california = Q(BillingState='CA')

query = tech_or_finance & high_revenue & california

accounts = sf.query(f"SELECT Id, Name FROM Account WHERE {query.compile()}")

Client-Side Data Manipulation

# Query all opportunities
opps = sf.query("SELECT Id, Name, Amount, StageName FROM Opportunity LIMIT 1000")

# Filter to closed-won
won = opps.records.filter(StageName='Closed Won')

# Group by stage and sum amounts
pipeline = opps.records.group_by('StageName').count()

# Sort by amount
top_deals = won.order_by('Amount', asc=False)[:10]

# Extract field values
amounts = opps.records.values_list('Amount', flat=True)
total = sum(amounts)

JWT Authentication (Production)

sf = Salesforce()
sf.login_with_jwt(
    client_id='your-connected-app-id',
    private_key='/path/to/private.key',
    username='[e-mail]'
)

Token Caching for Performance

# Redis cache for Kubernetes/multi-pod deployments
sf = Salesforce(
    username='[e-mail]',
    password='password',
    cache_backend='redis',
    redis_url='redis://redis-service:6379'
)
# Second authentication reuses cached token - no API call!

Object Discovery (Perfect for Beginners!)

# List all custom objects
custom = sf.list_objects(custom_only=True)
for obj in custom:
    print(f"{obj['name']}: {obj['label']}")

# Find what object a record ID belongs to
obj_type = sf.get_object_type_from_id('006xx0000012345')
print(f"This is a {obj_type} record")

Metadata & Describe

# Get object metadata (cached)
describe = sf.describe('Account')

# Required fields
for field in describe.required_fields:
    print(field['name'])

# Picklist values
industries = describe.get_picklist_values('Industry')

# Dependent picklists
subcategories = describe.get_dependent_picklist_values(
    field_name='Sub_Category__c',
    controlling_value='Hardware'
)

Contributing

We welcome contributions! Here's how you can help:

1. Report bugs - Open an issue with details and reproduction steps
2. Suggest features - Share your ideas for improvements
3. Submit PRs - See CONTRIBUTING.md for guidelines
4. Improve docs - Help make our documentation better
5. Share examples - Contribute real-world usage examples

Development

# Clone repository
git clone https://github.com/sanjan/forcepy.git
cd forcepy

# Install just (modern task runner)
brew install just  # Mac
cargo install just  # Any platform with Rust
# See docs/INSTALLING_JUST.md for other platforms

# Install dependencies
just install-dev

# Run tests
just test

# Run tests with coverage
just test-cov

# Lint and format
just format
just lint

# Run all quality checks
just quality

# Build package
just build

# See all commands
just --list

Without just: You can also use uv run directly:

uv sync --all-extras --dev
uv run pytest
uv run ruff check src/ tests/
uv run mypy src/forcepy/

License

MIT License - see LICENSE file for details.

Support

• 🐛 Bug reports: GitHub Issues
• 💬 Questions: GitHub Discussions
• 📧 Email: [e-mail]

---

Made with ❤️ by developers, for developers.