forcepy 0.1.0

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.

What is forcepy?

Forcepy is a modern Python client for Salesforce that makes working with the Salesforce API feel natural and intuitive. Whether you're querying data, updating records, or building complex integrations, forcepy handles the heavy lifting so you can focus on solving business problems.

Why choose forcepy?

• 🎯 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 features: Q objects, client-side filtering, composite API, bulk operations
• 📊 Developer-friendly: Full type hints, comprehensive docs, and helpful error messages
• 🎓 Beginner-friendly: Object discovery, ID lookup, and intuitive query building

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:

🔍 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 formatting
chatter.post("Hey @[005xx0000012345]! Check this out: <b>Q4 goals achieved!</b>")

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

🎁 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 (Coming Soon)

# Handle millions of records efficiently (Phase 4 feature)
results = sf.bulk.Account.insert(large_dataset)

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

• 📖 Token Caching Guide - Production caching strategies
• 🔐 Authentication Guide - All auth methods explained
• 💬 Chatter Features - Complete Chatter reference
• 💡 Examples - 12 ready-to-run code examples

Resources

• Documentation:
  • 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: 12 ready-to-run code examples
• 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	🔜 Coming soon	✅

Examples

Basic Operations

from forcepy import Salesforce

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

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

# Read
account = sf.sobjects.Account[account_id].get()
print(account['Name'])

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

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

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
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.