Skip to content

georgebv/aiosalesforce

Repository files navigation

⚡ aiosalesforce ⚡

Asynchronous Python client for Salesforce APIs

Test Coverage PyPI Package

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:

Optional dependencies:

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())