Getting Started

⚡ aiosalesforce ⚡

Asynchronous Python client for Salesforce APIs

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

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 injest and query operations

Requirements¶

aiosalesforce depends on:

Python 3.11+
httpx
orjson

Optional dependencies:

Installation¶

pip install aiosalesforce

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

pip install aiosalesforce[jwt]

Demo¶

Follow the steps below to create a simple script that authenticates against Salesforce and performs basic operations such as creating a record, reading a record, and executing a SOQL query.

Authenticate¶

First, we need to authenticate against Salesforce. For this example, we will use the SoapLogin authentication method.

import asyncio

from aiosalesforce import Salesforce, SoapLogin
from httpx import AsyncClient

auth = SoapLogin(
    username="your-username",
    password="your-password",
    security_token="your-security-token",
)

Create Salesforce client¶

Next, we create a new Salesforce client using the Salesforce class. Notice two additional parameters:

client - an instance of httpx.AsyncClient used to make HTTP requests
base_url - the base URL of your Salesforce instance

Since we are writing an asynchronous application, we need to wrap everything in an async function. Subsequent sections are written as a continuation of the main function.

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

Create a record¶

Let's create a new Contact in Salesforce. To do this, we will use the create method of the sobject api:

contact_id = await salesforce.sobject.create(
    "Contact",
    {
        "FirstName": "John",
        "LastName": "Doe",
        "Email": "john.doe@example.com",
    },
)
print(f"Created Contact with ID: {contact_id}")

This will output something like:

Created Contact with ID: 0035e00000Bv2tPAAR

Read a record¶

To read a record by ID, we will use the get method of the sobject api:

contact = await salesforce.sobject.get("Contact", contact_id)
print(contact)

This will return a dictionary with the Contact details (truncated for brevity):

{
    "Id": "0035e00000Bv2tPAAR",
    "FirstName": "John",
    "LastName": "Doe",
    "Email": "john.doe@example.com",
    ...
}

Execute a SOQL query¶

Finally, let's execute a SOQL query to retrieve all Contacts:

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

This will create an asynchronous generator yielding records as a dictionaries:

{"Id": "0035e00000Bv2tPAAR", "Name": "John Doe"}
{"Id": "0035e00000Bv2tPAAQ", "Name": "Jane Doe"}
{"Id": "0035e00000Bv2tPAAP", "Name": "Alice Smith"}
...

Putting it all together¶

Putting everything you learned above together, a simple script may look like this:

import asyncio

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

# Reuse authentication session across multiple clients (refreshes automagically)
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": "john.doe@example.com",
            },
        )
        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())

License¶

This project is licensed under the terms of the MIT license.