⚡ aiosalesforce ⚡

Asynchronous Python client for Salesforce APIs

Documentation: https://georgebv.github.io/aiosalesforce

License: MIT

Support: ask a question or create an issue, any input is appreciated and would help develop the project

About

aiosalesforce is a modern, production-ready asynchronous Python client for Salesforce APIs. It is built on top of the httpx library and provides a simple and intuitive API for interacting with Salesforce's APIs (such as REST and Bulk).

• Fast: designed from the ground up to be fully asynchronous 🚀
• Resilient: flexible and robust retrying configuration ⚙️
• Fully typed: every part of the library is fully typed and annotated 🏷️
• Intuitive: API follows naming conventions of Salesforce's APIs while staying idiomatic to Python 🐍
• Salesforce first: built with years of experience working with the Salesforce API it is configured to work out of the box and incorporates best practices and latest Salesforce API features ☁️
• Track your API usage: built-in support for tracking Salesforce API usage :chart_with_upwards_trend:

Supported APIs:

API	Status	Description
SOQL Query	✅	Execute SOQL queries
sObject	✅	CRUD operations on Salesforce objects
Composite Batch	✅	Execute multiple independent subrequests
Composite	✅	Execute multiple dependent subrequests
Composite Graph	🕒	Execute multiple composite graphs
sObject Tree	🕒	Create one or more sObject trees
sObject Collections	🕒	Execute multiple sObject operations
Bulk API 1.0	❌	Not planned (use Bulk API 2.0)
Bulk API 2.0	✅	Bulk ingest and query operations

Requirements

aiosalesforce depends on:

• Python 3.11+
• httpx
• orjson

Optional dependencies:

• PyJWT
• cryptography

Installation

pip install aiosalesforce

To use the JWT Bearer Flow authentication install with the jwt extra:

pip install aiosalesforce[jwt]

Demo

Example below shows how to:

• Authenticate against Salesforce using the SOAP login method
• Create a Salesforce client
• Create a new Contact
• Read a Contact by ID
• Execute a SOQL query

import asyncio

from aiosalesforce import Salesforce
from aiosalesforce.auth import SoapLogin
from httpx import AsyncClient

# Reuse authentication session across multiple clients (refreshes automatically)
auth = SoapLogin(
    username="your-username",
    password="your-password",
    security_token="your-security-token",
)

async def main():
    async with AsyncClient() as client:
        # Create a Salesforce client
        salesforce = Salesforce(
            client,
            base_url="https://your-instance.my.salesforce.com",
            auth=auth,
        )

        # Create a new Contact
        contact_id = await salesforce.sobject.create(
            "Contact",
            {
                "FirstName": "John",
                "LastName": "Doe",
                "Email": "[email protected]",
            },
        )
        print(f"Created Contact with ID: {contact_id}")

        # Read Contact by ID
        contact = await salesforce.sobject.get("Contact", contact_id)
        print(contact)

        # Execute a SOQL query
        async for record in salesforce.query("SELECT Id, Name FROM Contact"):
            print(record)


if __name__ == "__main__":
    asyncio.run(main())