### Install aiosalesforce

Source: https://github.com/georgebv/aiosalesforce/blob/main/README.md

Standard installation command for the library.

```shell
pip install aiosalesforce
```

```shell
pip install aiosalesforce[jwt]
```

--------------------------------

### Install aiosalesforce

Source: https://context7.com/georgebv/aiosalesforce/llms.txt

Install the aiosalesforce library using pip.

```bash
pip install aiosalesforce
```

--------------------------------

### Output of Record Creation

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/index.md

Example output after successfully creating a record.

```shell
Created Contact with ID: 0035e00000Bv2tPAAR
```

--------------------------------

### SOAP Login Authentication Setup

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/authentication.md

Instantiate SoapLogin with your Salesforce username, password, and security token. Obtain your security token from Salesforce user settings.

```python
from aiosalesforce import SoapLogin

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

--------------------------------

### SOAP Login Authentication Example

Source: https://context7.com/georgebv/aiosalesforce/llms.txt

Demonstrates how to authenticate with Salesforce using SOAP Login with username, password, and security token. Ensure the AsyncClient and Salesforce instances are properly managed within an asynchronous context.

```python
import asyncio
from aiosalesforce import Salesforce, SoapLogin
from httpx import AsyncClient

# Create authentication instance (reuse across multiple clients)
auth = SoapLogin(
    username="your-username",
    password="your-password",
    security_token="your-security-token",
)

async def main():
    async with AsyncClient() as client:
        salesforce = Salesforce(
            client=client,
            base_url="https://your-instance.my.salesforce.com",
            auth=auth,
        )
        # Now you can use salesforce client for API calls
        contact = await salesforce.sobject.get("Contact", "0035e00000Bv2tPAAR")
        print(contact)

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

--------------------------------

### Install aiosalesforce with JWT support

Source: https://context7.com/georgebv/aiosalesforce/llms.txt

Install aiosalesforce with the necessary dependencies for JWT Bearer Flow authentication.

```bash
pip install aiosalesforce[jwt]
```

--------------------------------

### Client Credentials Flow Authentication Example

Source: https://context7.com/georgebv/aiosalesforce/llms.txt

Authenticates using Client Credentials Flow for server-to-server integrations. An optional timeout can be set for automatic token refresh.

```python
from aiosalesforce import Salesforce, ClientCredentialsFlow
from httpx import AsyncClient

# Create authentication with optional timeout for automatic token refresh
auth = ClientCredentialsFlow(
    client_id="your-connected-app-consumer-key",
    client_secret="your-connected-app-consumer-secret",
    timeout=900,  # Optional: refresh token after 15 minutes
)

async def main():
    async with AsyncClient() as client:
        salesforce = Salesforce(
            client=client,
            base_url="https://your-instance.my.salesforce.com",
            auth=auth,
        )
        # Perform operations
        accounts = []
        async for record in salesforce.query("SELECT Id, Name FROM Account LIMIT 5"):
            accounts.append(record)
        print(accounts)
```

--------------------------------

### JWT Bearer Flow Authentication Example

Source: https://context7.com/georgebv/aiosalesforce/llms.txt

Authenticates using JWT Bearer Flow with a connected app and an RSA private key. Ensure the 'aiosalesforce[jwt]' extra is installed.

```python
from aiosalesforce import Salesforce, JwtBearerFlow
from httpx import AsyncClient

# Requires: pip install aiosalesforce[jwt]
auth = JwtBearerFlow(
    client_id="your-connected-app-consumer-key",
    username="user@example.com",
    private_key="path/to/private-key.pem",
    private_key_passphrase="optional-passphrase",  # If key is encrypted
)

async def main():
    async with AsyncClient() as client:
        salesforce = Salesforce(
            client=client,
            base_url="https://your-instance.my.salesforce.com",
            auth=auth,
        )
        # Perform operations
        result = await salesforce.sobject.create(
            "Contact",
            {"FirstName": "John", "LastName": "Doe"}
        )
        print(f"Created contact: {result}")
```

--------------------------------

### Client Credentials Flow Authentication Setup

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/authentication.md

Configure ClientCredentials authentication using your connected app's client ID and client secret. This method is suitable for server-to-server integrations.

```python
from aiosalesforce import ClientCredentials

auth = ClientCredentials(
    client_id="client-id",
    client_secret="client-secret",
)
```

--------------------------------

### Get a Salesforce record

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/sobject.md

Retrieve a specific record by its Salesforce ID.

```python
account = await salesforce.sobject.get('Account', '0012v00002Q8f4QAAR')
```

--------------------------------

### Reference Query Results in Subrequest

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/composite/composite.md

Reference data from a previous SOQL query subrequest within a subsequent sObject creation subrequest. This example shows how to use the `records[0].Id` from a queried Account to set the `AccountId` for a new Contact.

```python
async with salesforce.composite() as composite:
    query = composite.query("SELECT Id, Name FROM Account LIMIT 1")
    contact = composite.sobject.create(
        "Contact",
        {
            "FirstName": "Jon",
            "LastName": "Doe",
            "AccountId": query.reference.records[0].Id,
        },
    )
print("Created contact:", contact.id)
```

--------------------------------

### Perform sObject CRUD Operations in Composite Request

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/composite/composite.md

Execute multiple sObject operations (update, upsert, create) within a single composite request. This example demonstrates updating an Account, upserting a Contact using an external ID, and creating an Appointment related to the Contact, referencing the Contact's ID.

```python
async with salesforce.composite(all_or_none=True) as composite:
    account = composite.sobject.update(
        "Account",
        "0011R00001K1H2IQAV",
        {"Name": "New Name"},
    )
    contact = composite.sobject.upsert(
        "Contact",
        "ExternalId__c",
        {
            "ExternalId__c": "123",
            "FirstName": "Jon",
            "LastName": "Doe",
            "AccountId": account.reference.id,
        },
    )
    appointment = composite.sobject.create(
        "Appointment__c",
        {
            "Name__c": "Treatment 123",
            "Contact__c": contact.reference.id,
        },
    )
print("Updated account:", account.id)
print("Upserted contact:", contact.id)
print("Created appointment:", appointment.id)
```

--------------------------------

### Define Exception Rule

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/client.md

Configure an ExceptionRule to retry requests when specific exceptions occur. This example retries on httpx.TransportError but excludes TimeoutException, with a maximum of 3 retries.

```python
from aiosalesforce import ExceptionRule

exception_rule = ExceptionRule(
    httpx.TransportError,
    lambda exc: not isinstance(exc, httpx.TimeoutException),
    max_retries=3,
)
```

--------------------------------

### Initialize Salesforce Client

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/client.md

Create a Salesforce client instance using an httpx AsyncClient and SoapLogin credentials.

```python
from aiosalesforce import Salesforce, SoapLogin
from httpx import AsyncClient

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

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

        # Your code here
        ...
```

--------------------------------

### Initialize Salesforce Client

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/introduction.md

Demonstrates authenticating with SoapLogin and initializing the Salesforce client within an asynchronous context.

```python
import asyncio

from aiosalesforce import Salesforce, SoapLogin
from httpx import AsyncClient

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

async def main():
    async with AsyncClient() as client:
        salesforce = Salesforce(
            client=client,
            base_url="https://your-instance.my.salesforce.com",
            auth=auth,
        )
        contact = salesforce.sobject.get("Contact", "0033h00000KzZ3AAAV")
        print(contact)


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

--------------------------------

### Salesforce Client Initialization with SOAP Login

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/authentication.md

Demonstrates initializing the Salesforce client with SOAP Login authentication. Ensure your credentials are not hardcoded and are managed securely.

```python
import asyncio

from aiosalesforce import Salesforce, SoapLogin
from httpx import AsyncClient

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


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


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

--------------------------------

### Initialize Salesforce Client

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/index.md

Create the Salesforce client instance within an asynchronous context using an httpx AsyncClient.

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

--------------------------------

### Authenticate and interact with Salesforce

Source: https://github.com/georgebv/aiosalesforce/blob/main/README.md

Demonstrates SOAP authentication, client initialization, sObject creation, retrieval, and SOQL query execution.

```python
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": "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())
```

--------------------------------

### Complete aiosalesforce integration script

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/index.md

Demonstrates full lifecycle including authentication, record creation, retrieval, and querying.

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

--------------------------------

### Execute Composite Batch

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/composite/batch.md

Demonstrates batch execution using a context manager versus manual execution.

```python
async with salesforce.composite.batch(halt_on_error=True) as batch:
    query = batch.query("SELECT Id, Name FROM Account LIMIT 10")
    contact = batch.sobject.create(
        "Contact",
        {"FirstName": "Jon", "LastName": "Doe"},
    )
print(query.records)
print(contact.id)
```

```python
batch = salesforce.composite.batch(halt_on_error=True)
query = batch.query("SELECT Id, Name FROM Account LIMIT 10")
contact = batch.sobject.create(
    "Contact",
    {"FirstName": "Jon", "LastName": "Doe"},
)
await batch.execute()
print(query.records)
print(contact.id)
```

--------------------------------

### Manually execute a bulk job

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/bulk/v2.md

Demonstrates the manual workflow of creating a job, uploading data, polling for completion, and retrieving successful results.

```python
from aiosalesforce.bulk.v2._csv import (
    deserialize_ingest_results,
    serialize_ingest_data,
)

data = [
    {"FirstName": "Jon", "LastName": "Doe"},
    {"FirstName": "Jane", "LastName": "Doe"},
]
job = await salesforce.bulk_v2.ingest.create_job("Contact", "insert")
job = await salesforce.bulk_v2.ingest.upload_job_data(
    job.id,
    serialize_ingest_data(data),
)
while job.state in {"Open", "UploadComplete", "InProgress"}:
    await asyncio.sleep(5)
    job = await self.get_job(job.id)
response = await salesforce.request(
    "GET",
    f"{salesforce.bulk_v2.ingest.base_url}/{job.id}/successfulResults",
)
successful_results = deserialize_ingest_results(response.content)
```

--------------------------------

### Authenticate with SoapLogin

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/index.md

Initialize the authentication object using SoapLogin credentials.

```python
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 a Salesforce Record

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/index.md

Perform a create operation on a specific sObject.

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

--------------------------------

### Configure Salesforce Client API Version and Concurrency

Source: https://context7.com/georgebv/aiosalesforce/llms.txt

Initialize the Salesforce client with custom API versions and concurrency limits to manage request throughput.

```python
from aiosalesforce import Salesforce, SoapLogin
from httpx import AsyncClient

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

async with AsyncClient() as client:
    salesforce = Salesforce(
        client=client,
        base_url="https://your-instance.my.salesforce.com",
        auth=auth,
        version="58.0",           # Specify Salesforce API version
        concurrency_limit=25,     # Limit simultaneous requests (default: 100)
    )

    # Use the configured client
    async for record in salesforce.query("SELECT Id, Name FROM Account LIMIT 10"):
        print(record)
```

--------------------------------

### Manage Event Callbacks Dynamically

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/client.md

Subscribe or unsubscribe from event hooks after the client has been initialized.

```python
salesforce.event_bus.subscribe_callback(callback)
salesforce.event_bus.unsubscribe_callback(callback)
```

--------------------------------

### Composite Batch Execution

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/composite/batch.md

Demonstrates how to execute multiple subrequests using the Composite Batch API, both with and without a context manager.

```APIDOC
## Composite Batch Execution Examples

### Using Context Manager

This example shows how to create a batch, add subrequests (a query and an sObject create operation), and execute them within an `async with` block.

```python
async with salesforce.composite.batch(halt_on_error=True) as batch:
    query = batch.query("SELECT Id, Name FROM Account LIMIT 10")
    contact = batch.sobject.create(
        "Contact",
        {"FirstName": "Jon", "LastName": "Doe"},
    )
print(query.records)
print(contact.id)
```

### Without Context Manager

This example demonstrates executing a batch by explicitly creating, populating, and awaiting its execution.

```python
batch = salesforce.composite.batch(halt_on_error=True)
query = batch.query("SELECT Id, Name FROM Account LIMIT 10")
contact = batch.sobject.create(
    "Contact",
    {"FirstName": "Jon", "LastName": "Doe"},
)
await batch.execute()
print(query.records)
print(contact.id)
```

### Batch Parameters

Control the behavior of the batch execution with the following parameters:

- `halt_on_error` (bool): If `True`, subsequent subrequests are skipped if any subrequest fails. Defaults to `False`.
- `autoraise` (bool): If `True`, an exception is raised if any subrequest fails. Defaults to `False`.
- `group_errors` (bool): If `True` and `autoraise` is `True`, raises an `ExceptionGroup` for multiple errors. Defaults to `False`.

### Subrequest Attributes

After execution, each subrequest object provides the following attributes:

- `response` (dict): The raw response dictionary from the subrequest.
- `done` (bool): Indicates if the subrequest has been executed.
- `status_code` (int): The HTTP status code of the subrequest.
- `result` (any): The processed result or body of the subrequest.
- `is_success` (bool): A boolean indicating if the subrequest was successful.

### Raising Exceptions for Failed Subrequests

Use the `raise_for_status()` method on a subrequest object to raise an exception if it failed.

```python
subrequest.raise_for_status()
```
```

--------------------------------

### Extract Private Key from Keystore

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/authentication.md

Commands to convert a Salesforce-exported certificate into a PEM-formatted RSA private key.

```bash
keytool -importkeystore -srckeystore certificate.crt -destkeystore \
certificate.p12 -deststoretype PKCS12
```

```bash
openssl pkcs12 -in certificate.p12 -nocerts -nodes -out private-key.pem
```

--------------------------------

### Create a new record

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/sobject.md

Create a new record by providing the object name and a dictionary of field data.

```python
data = {
    "FirstName": "Jon",
    "LastName": "Doe",
    "Phone": "1234567890",
    "Email": "jon.doe@example.com",
}
contact_id = await salesforce.sobject.create('Contact', data)
```

--------------------------------

### Configure Salesforce Client with Concurrency Limit

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/client.md

Instantiate the Salesforce client, setting a custom concurrency limit to control the number of simultaneous API requests. The default is 100.

```python
salesforce = Salesforce(
    client=client,
    base_url="https://your-instance.my.salesforce.com",
    auth=auth,
    concurrency_limit=25,
)
```

--------------------------------

### Client Credentials Flow with Token Timeout

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/authentication.md

Set up ClientCredentials authentication with an optional timeout for automatic access token refresh. This prevents 401 errors due to expired tokens.

```python
auth = ClientCredentials(
    client_id="client-id",
    client_secret="client-secret",
    timeout=900,  # 15 minutes
)
```

--------------------------------

### Execute SOQL Query

Source: https://context7.com/georgebv/aiosalesforce/llms.txt

Retrieves records using SOQL, returning an async generator that supports automatic pagination.

```python
from aiosalesforce import format_soql

# Basic query - returns async generator
records = []
async for record in salesforce.query("SELECT Id, Name FROM Account"):
    records.append(record)
print(records)
# Output: [{"Id": "0011R00001K1H2IQAV", "Name": "Acme Corp"}, ...]

# Include archived and deleted records
async for record in salesforce.query(
    "SELECT Id, Name FROM Account",
    include_all_records=True,
):
    print(record)
```

--------------------------------

### Perform a basic SOQL query

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/soql.md

Use the query method with an async for loop to iterate over records returned by a SOQL query.

```python
records = []
async for record in client.query("SELECT Id, Name FROM Account"):
    records.append(record)
```

--------------------------------

### Initialize RetryPolicy

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/client.md

Instantiate a RetryPolicy with response and exception rules, maximum retries, and a timeout. This policy governs how the client handles failed requests.

```python
retry_policy = RetryPolicy(
    response_rules=[...],
    exception_rules=[...],
    max_retries=10,
    timeout=60,
)
```

--------------------------------

### Authentication Module

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/api-reference/auth.md

Overview of authentication functionalities provided by the aiosalesforce.auth module.

```APIDOC
## aiosalesforce.auth

### Description
This module provides classes for handling authentication with Salesforce.

### Classes
- **Auth**: Base class for authentication mechanisms.
- **SoapLogin**: Handles authentication using SOAP login.
- **ClientCredentialsFlow**: Implements the OAuth 2.0 client credentials flow.
```

--------------------------------

### Format dynamic SOQL queries

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/soql.md

Use format_soql to safely inject dynamic parameters into query strings. Do not manually add quotes around parameter values.

```python
query = format_soql(
    "SELECT Id, Name FROM Account WHERE Name = {name}",
    name="My Account",
)
async for record in client.query(query):
    ...
```

```sql
SELECT Id, Name FROM Account WHERE Name = 'My Account'
```

--------------------------------

### Execute Composite Request with Context Manager

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/composite/composite.md

Use a context manager to automatically execute a composite request. This is suitable for creating related records like an Account and its associated Contact, referencing the Account's ID for the Contact.

```python
async with salesforce.composite(all_or_none=True) as composite:
    account = composite.sobject.create(
        "Account",
        {"Name": "Acme Corporation"},
    )
    contact = composite.sobject.create(
        "Contact",
        {"FirstName": "Jon", "LastName": "Doe", "AccountId": account.reference.id},
    )
print(account.id)
print(contact.id)
```

--------------------------------

### Execute Composite Request Without Context Manager

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/composite/composite.md

Manually execute a composite request by creating a composite instance, adding subrequests, and then calling `await composite.execute()`. This method requires explicit execution.

```python
composite = salesforce.composite(all_or_none=True)
account = composite.sobject.create(
    "Account",
    {"Name": "Acme Corporation"},
)
contact = composite.sobject.create(
    "Contact",
    {
        "FirstName": "Jon",
        "LastName": "Doe",
        "AccountId": account.reference.id,
    },
)
await composite.execute()
print(account.id)
print(contact.id)
```

--------------------------------

### Create sObject Record

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/sobject.md

Creates a new record in Salesforce for a specified object type.

```APIDOC
## POST /sobject/create

### Description
Creates a new record in Salesforce.

### Parameters
#### Request Body
- **object_name** (string) - Required - The name of the Salesforce object.
- **data** (dict/string/bytes) - Required - The record data to create.
```

--------------------------------

### Specify Salesforce API Version

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/client.md

Define a specific Salesforce API version during client initialization.

```python
salesforce = Salesforce(
    client=client,
    base_url="https://your-instance.my.salesforce.com",
    auth=auth,
    version="52.0",
)
```

--------------------------------

### Define and Register Event Hooks

Source: https://context7.com/georgebv/aiosalesforce/llms.txt

Define asynchronous callback functions to monitor request, response, and retry events, and register them with the Salesforce client.

```python
# Define callback functions
async def log_requests(event: Event):
    if isinstance(event, RequestEvent):
        print(f"Request: {event.request.method} {event.request.url}")

async def log_responses(event: Event):
    if isinstance(event, ResponseEvent):
        print(f"Response: {event.response.status_code}")
        print(f"API calls remaining: {event.remaining}")

async def log_retries(event: Event):
    if isinstance(event, RetryEvent):
        print(f"Retry attempt {event.attempt} for {event.request.url}")

def track_api_usage(event: Event):
    match event:
        case RestApiCallConsumptionEvent():
            print(f"REST API calls consumed: {event.count}")
        case BulkApiBatchConsumptionEvent():
            print(f"Bulk API batches consumed: {event.count}")

# Create client with event hooks
salesforce = Salesforce(
    client=client,
    base_url="https://your-instance.my.salesforce.com",
    auth=auth,
    event_hooks=[log_requests, log_responses, log_retries, track_api_usage],
)

# Or add/remove callbacks after creation
salesforce.event_bus.subscribe_callback(log_requests)
salesforce.event_bus.unsubscribe_callback(log_requests)
```

--------------------------------

### Format SOQL queries with collections

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/soql.md

Pass a list, tuple, or set to format_soql when using the IN operator.

```python
query = format_soql(
    "SELECT Id, Name FROM Account WHERE Name IN {names}",
    names=["My Account", "Another Account"],
)
```

```sql
SELECT Id, Name FROM Account WHERE Name IN ('My Account', 'Another Account')
```

--------------------------------

### Safe SOQL Parameter Formatting

Source: https://context7.com/georgebv/aiosalesforce/llms.txt

Uses format_soql to safely inject dynamic parameters into queries, preventing injection vulnerabilities.

```python
from aiosalesforce import format_soql

# Format query with single value
query = format_soql(
    "SELECT Id, Name FROM Account WHERE Name = {name}",
    name="Acme Corporation",
)
# Result: SELECT Id, Name FROM Account WHERE Name = 'Acme Corporation'

# Format query with list of values
query = format_soql(
    "SELECT Id, Name FROM Account WHERE Name IN {names}",
    names=["Acme Corp", "Global Industries", "Tech Solutions"],
)
# Result: SELECT Id, Name FROM Account WHERE Name IN ('Acme Corp', 'Global Industries', 'Tech Solutions')

# Format query with LIKE operator (use :like spec)
query = format_soql(
    "SELECT Id, Name FROM User WHERE Name LIKE '%{name:like}%'",
    name="Jon",
)
# Result: SELECT Id, Name FROM User WHERE Name LIKE '%Jon%'

async for record in salesforce.query(query):
    print(record)
```

--------------------------------

### Implement Custom Authentication Class

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/authentication.md

Subclass the Auth class to define custom token acquisition and expiration logic. Ensure super().__init__() is called during initialization.

```python
from aiosalesforce.auth import Auth


class MyAuth(Auth):
    def __init__(
        self,
        # Your custom arguments
        ...
    ):
        super().__init__()
        # Your custom initialization logic
        ...

    async def _acquire_new_access_token(self, client: Salesforce) -> str:
        # Your custom logic to acquire new access token
        ...

    async def _refresh_access_token(self, client: Salesforce) -> str:
        # Your custom logic to refresh access token
        ...

    @property
    def expired(self) -> bool:
        super().expired
        # Your custom logic to check if access token has expired
        ...
```

--------------------------------

### Create Salesforce sObject Record

Source: https://context7.com/georgebv/aiosalesforce/llms.txt

Creates a new record in Salesforce for a specified object type with provided field data. The function returns the ID of the newly created record.

```python
# Create a new Contact record
data = {
    "FirstName": "Jon",
    "LastName": "Doe",
    "Phone": "1234567890",
    "Email": "jon.doe@example.com",
}
contact_id = await salesforce.sobject.create("Contact", data)
print(f"Created Contact with ID: {contact_id}")
# Output: Created Contact with ID: 0035e00000Bv2tPAAR
```

--------------------------------

### Execute a SOQL query

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/index.md

Retrieves records using an asynchronous generator for efficient iteration.

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

--------------------------------

### Log Salesforce API Requests

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/client.md

Set up an event hook to log information about outgoing Salesforce API requests, specifically the HTTP method and URL.

```python
from aiosalesforce import Event, RequestEvent


async def log_requests(event: Event):
    match event:
        case RequestEvent():
            logger.info("%s %s", event.request.method, event.request.url)

salesforce.event_bus.subscribe_callback(log_requests)
```

--------------------------------

### Composite Request with Transaction Rollback

Source: https://context7.com/georgebv/aiosalesforce/llms.txt

Use a context manager with `all_or_none=True` to ensure all operations within the block succeed or the entire transaction is rolled back. This is useful for creating related records atomically.

```python
async with salesforce.composite(all_or_none=True, autoraise=True) as composite:
    # Create parent Account
    account = composite.sobject.create(
        "Account",
        {"Name": "Acme Corporation"},
    )

    # Create Contact linked to Account using reference
    contact = composite.sobject.create(
        "Contact",
        {
            "FirstName": "Jon",
            "LastName": "Doe",
            "AccountId": account.reference.id,  # Reference to account result
        },
    )

    # Query and use result in another subrequest
    query = composite.query("SELECT Id FROM Account WHERE Name = 'Partner Corp' LIMIT 1")
    partner_contact = composite.sobject.create(
        "Contact",
        {
            "FirstName": "Jane",
            "LastName": "Smith",
            "AccountId": query.reference.records[0].Id,  # Reference query result
        },
    )

# Access results
print(f"Created Account: {account.id}")
print(f"Created Contact: {contact.id}")
print(f"Created Partner Contact: {partner_contact.id}")
```

--------------------------------

### POST /composite

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/composite/composite.md

Executes a batch of subrequests. Supports all_or_none rollback, parallel execution, and result referencing.

```APIDOC
## POST /composite

### Description
Executes up to 25 subrequests in a single call. Results from previous subrequests can be referenced in subsequent ones using the reference attribute.

### Parameters
#### Request Body
- **all_or_none** (boolean) - Optional - If True, all subrequests are rolled back if any subrequest fails. Default is False.
- **collate_subrequests** (boolean) - Optional - If True, independent subrequests are executed in parallel. Default is True.
- **autoraise** (boolean) - Optional - Raise an ExceptionGroup if any subrequest fails. Default is False.

### Request Example
```python
async with salesforce.composite(all_or_none=True) as composite:
    account = composite.sobject.create("Account", {"Name": "Acme Corporation"})
    contact = composite.sobject.create("Contact", {"FirstName": "Jon", "LastName": "Doe", "AccountId": account.reference.id})
```

### Response
#### Success Response (200)
- **response** (dict) - The raw response dictionary.
- **response_body** (dict) - The subrequest response body.
- **status_code** (int) - The HTTP status code of the subrequest.
- **is_success** (boolean) - Whether the subrequest was successful.
```

--------------------------------

### Authenticate with JWT Bearer Flow

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/authentication.md

Initializes the authentication flow using a client ID, username, and path to a PEM-formatted RSA private key.

```python
from aiosalesforce import JwtBearerFlow

auth = JwtBearerFlow(
    client_id="client-id",
    username="username",
    private_key="path/to/private-key.pem",
)
```

--------------------------------

### Log Salesforce API Retries

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/client.md

Implement a callback function to log details about retried Salesforce API requests, including the method, URL, reason for retry, and attempt number.

```python
from aiosalesforce import Event, RetryEvent


async def log_retries(event: Event):
    match event:
        case RetryEvent():
            print(
                f"Retrying {event.request.method} request to {event.request.url} "
                f"due to: {event.response or type(event.exception).__name__}. "
                f"This is attempt {event.attempt}"
            )

salesforce.event_bus.subscribe_callback(log_retries)
```

--------------------------------

### Supported Subrequests

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/composite/batch.md

Details on the types of subrequests that can be included in a Composite Batch.

```APIDOC
## Supported Subrequests

### Query Subrequest

Execute a SOQL query and retrieve its results as part of a batch.

```python
async with salesforce.composite.batch() as batch:
    query_accounts = batch.query("SELECT Id, Name FROM Account LIMIT 10")
    query_contacts = batch.query("SELECT Id, Name FROM Contact LIMIT 10")
print(query_accounts.records)
print(query_contacts.records)
```

- **Result Attribute**: Subrequest results are available via the `records` attribute.

### sObject Subrequest

Perform CRUD operations (create, update, upsert) on sObjects within a batch. This uses the same interface as the `salesforce.sobject` resource.

```python
async with salesforce.composite.batch() as batch:
    contact = batch.sobject.create(
        "Contact",
        {"FirstName": "Jon", "LastName": "Doe"},
    )
    account = batch.sobject.update(
        "Account",
        "0011R00001K1H2IQAV",
        {"Name": "New Name"},
    )
    opportunity = batch.sobject.upsert(
        "Opportunity",
        "ExternalId__c",
        {"ExternalId__c": "123", "Name": "New Opportunity"},
    )
print("Created contact:", contact.id)
print("Updated account:", account.id)
print("Upserted opportunity:", opportunity.id)
```

- **Result Attributes**: Depending on the operation, the subrequest result may contain:
  - `create`: `id`
  - `get`: `record`
  - `upsert`: `id`, `created`
```

--------------------------------

### Execute Batch Queries

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/composite/batch.md

Executes multiple SOQL queries within a single batch.

```python
async with salesforce.composite.batch() as batch:
    query_accounts = batch.query("SELECT Id, Name FROM Account LIMIT 10")
    query_contacts = batch.query("SELECT Id, Name FROM Contact LIMIT 10")
print(query_accounts.records)
print(query_contacts.records)
```

--------------------------------

### Register Event Hooks

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/client.md

Pass callback functions to the Salesforce client using the event_hooks parameter.

```python
salesforce = Salesforce(
    ...,
    event_hooks=[callback, ...],
)
```

--------------------------------

### Define Event Hook Callback

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/client.md

Create a callback function to handle events emitted by the Salesforce client.

```python
async def callback(event):
    # Do something with the event
    ...
```

--------------------------------

### Read sObject Record

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/sobject.md

Retrieves a record from Salesforce by ID or external ID.

```APIDOC
## GET /sobject/get

### Description
Retrieves a record as a dictionary.

### Parameters
#### Path Parameters
- **object_name** (string) - Required - The name of the Salesforce object.
- **id** (string) - Required - The record ID or external ID value.
- **external_id_field** (string) - Optional - The field name if using an external ID.

#### Query Parameters
- **fields** (list) - Optional - Specific fields to retrieve.
```

--------------------------------

### Update records in bulk

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/bulk/v2.md

Updates existing records by providing a list of dictionaries that must include the 'Id' field.

```python
data = [
    {"Id": "0031R00001K1H2IQAV", "FirstName": "Jon", "LastName": "Doe"},
    {"Id": "0031R00001K1H2JQAV", "FirstName": "Jane", "LastName": "Doe"},
]
result = await salesforce.bulk_v2.update("Contact", data)
```

--------------------------------

### Format SOQL queries with LIKE operator

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/soql.md

Use the {:like} format specifier for expressions containing the LIKE operator to handle escaping correctly.

```python
query = format_soql(
    "SELECT Id, Name FROM User WHERE Name LIKE '%{name:like}",
    name="Jon%",
)
```

```sql
SELECT Id, Name FROM User WHERE Name LIKE '%Jon\%'
```

--------------------------------

### Read Salesforce sObject Record

Source: https://context7.com/georgebv/aiosalesforce/llms.txt

Retrieves a Salesforce record by its ID. You can fetch all fields or specify a list of fields to return.

```python
# Read all fields for a Contact
contact = await salesforce.sobject.get("Contact", "0035e00000Bv2tPAAR")
print(contact)
# Output: {"Id": "0035e00000Bv2tPAAR", "FirstName": "Jon", "LastName": "Doe", ...}

# Read specific fields only
contact = await salesforce.sobject.get(
    "Contact",
    "0035e00000Bv2tPAAR",
    fields=["FirstName", "LastName", "Email"],
)
print(contact)
# Output: {"Id": "0035e00000Bv2tPAAR", "FirstName": "Jon", "LastName": "Doe", "Email": "jon.doe@example.com"}
```

--------------------------------

### Query including archived or deleted records

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/soql.md

Set the include_all_records parameter to True to retrieve archived or deleted records in addition to active ones.

```python
records = []
async for record in client.query(
    "SELECT Id, Name FROM Account",
    include_all_records=True,
):
    records.append(record)
```

--------------------------------

### Configure Custom Salesforce Retry Policy

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/client.md

Define a custom retry policy for the Salesforce client using ResponseRule and ExceptionRule to handle specific error conditions and limit retries.

```python
from aiosalesforce import RetryPolicy, ExceptionRule, ResponseRule

retry_policy = RetryPolicy(
    response_rules=[
        ResponseRule(lambda response: response.status_code >= 500, max_retries=5),
        ResponseRule(lambda response: "UNABLE_TO_LOCK_ROW" in response.text),
    ],
    exception_rules=[
        ExceptionRule(httpx.TransportError, max_retries=3),
    ],
    max_retries=10,
)

salesforce = Salesforce(
    ...
    retry_policy=retry_policy,
)
```

--------------------------------

### Event Hooks for API Usage Tracking

Source: https://context7.com/georgebv/aiosalesforce/llms.txt

Subscribe to various events to monitor and track API consumption, including REST API calls and Bulk API batches. This is useful for logging and auditing API usage.

```python
from aiosalesforce import (
    Salesforce,
    Event,
    RequestEvent,
    ResponseEvent,
    RetryEvent,
    RestApiCallConsumptionEvent,
    BulkApiBatchConsumptionEvent,
)
```

--------------------------------

### Perform Batch sObject CRUD Operations

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/composite/batch.md

Executes create, update, and upsert operations on sObjects within a batch.

```python
async with salesforce.composite.batch() as batch:
    contact = batch.sobject.create(
        "Contact",
        {"FirstName": "Jon", "LastName": "Doe"},
    )
    account = batch.sobject.update(
        "Account",
        "0011R00001K1H2IQAV",
        {"Name": "New Name"},
    )
    opportunity = batch.sobject.upsert(
        "Opportunity",
        "ExternalId__c",
        {"ExternalId__c": "123", "Name": "New Opportunity"},
    )
print("Created contact:", contact.id)
print("Updated account:", account.id)
print("Upserted opportunity:", opportunity.id)
```

--------------------------------

### Handle Salesforce Events with Event Type

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/client.md

Use a callback function to process events emitted by the Salesforce client by matching against the event's `type` attribute. This approach is suitable for filtering events based on their string type.

```python
from aiosalesforce import Event

async def callback(event: Event) -> None:
    match event.type:
        case "request":
            # Do something with the request event
            ...
        case "response":
            # Do something with the response event
            ...
        case _:
            # Do nothing for other events
            pass
```

--------------------------------

### Read a Salesforce record by ID

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/index.md

Uses the sobject API to retrieve a record dictionary by its ID.

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

--------------------------------

### Configure Retry Policy for Transient Errors

Source: https://context7.com/georgebv/aiosalesforce/llms.txt

Configure automatic retry behavior for handling transient errors like network issues, server errors, and rate limits. Define custom rules for specific response codes or exception types, and set global limits.

```python
import httpx
from aiosalesforce import Salesforce, SoapLogin, RetryPolicy, ExceptionRule, ResponseRule

# Define retry policy with custom rules
retry_policy = RetryPolicy(
    # Retry on specific HTTP responses
    response_rules=[
        ResponseRule(lambda r: r.status_code >= 500, max_retries=5),  # Server errors
        ResponseRule(lambda r: "UNABLE_TO_LOCK_ROW" in r.text, max_retries=3),  # Row lock errors
        ResponseRule(lambda r: r.status_code == 429, max_retries=10),  # Rate limits
    ],
    # Retry on specific exceptions
    exception_rules=[
        ExceptionRule(
            httpx.TransportError,
            lambda exc: not isinstance(exc, httpx.TimeoutException),
            max_retries=3,
        ),
    ],
    max_retries=10,  # Global maximum retries
    timeout=60,      # Global timeout in seconds
)

salesforce = Salesforce(
    client=client,
    base_url="https://your-instance.my.salesforce.com",
    auth=auth,
    retry_policy=retry_policy,
)
```

--------------------------------

### Track Salesforce API Usage with Event Hooks

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/client.md

Use event hooks to monitor Salesforce API request counts for REST and Bulk API calls. Useful for sending usage metrics to services like AWS CloudWatch.

```python
from aiosalesforce import (
    Event,
    BulkApiBatchConsumptionEvent,
    RestApiCallConsumptionEvent,
)


def track_api_usage(event: Event):
    match event:
        case RestApiCallConsumptionEvent():
            my_metrics_service.put_metric_data(
                Namespace="Salesforce",
                MetricData=[
                    {
                        "MetricName": "Salesforce REST API Call Count",
                        "Value": event.count,
                        "Unit": "Count",
                    },
                ],
            )
        case BulkApiBatchConsumptionEvent():
            my_metrics_service.put_metric_data(
                Namespace="Salesforce",
                MetricData=[
                    {
                        "MetricName": "Salesforce Bulk API Batch Count",
                        "Value": event.count,
                        "Unit": "Count",
                    },
                ],
            )

salesforce.event_bus.subscribe_callback(track_api_usage)
```

--------------------------------

### Bulk Insert Records using Bulk API 2.0

Source: https://context7.com/georgebv/aiosalesforce/llms.txt

Insert large numbers of records efficiently using Bulk API 2.0. This method is recommended for operations involving more than 2,000 records. The `data` parameter should be a list of dictionaries, where each dictionary represents a record.

```python
# Bulk insert contacts
data = [
    {"FirstName": "Jon", "LastName": "Doe", "Email": "jon@example.com"},
    {"FirstName": "Jane", "LastName": "Smith", "Email": "jane@example.com"},
    {"FirstName": "Bob", "LastName": "Wilson", "Email": "bob@example.com"},
    # Can include thousands of records
]
result = await salesforce.bulk_v2.insert("Contact", data)

# Get created record IDs
record_ids = [record["sf__Id"] for record in result.successful_results]
print(f"Created {len(record_ids)} records")

# Check for failures
if result.failed_results:
    for failure in result.failed_results:
        print(f"Failed: {failure['sf__Error']}")

# Check for unprocessed records
if result.unprocessed_records:
    print(f"Unprocessed: {len(result.unprocessed_records)} records")
```

--------------------------------

### Handle Salesforce Events with Event Object Type

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/client.md

Use a callback function to process events by matching against the specific event object's class. This method allows for more type-specific handling within the callback.

```python
from aiosalesforce import RequestEvent, ResponseEvent

def callback(event) -> None:
    match event:
        case RequestEvent():
            # Do something with the request event
            ...
        case ResponseEvent():
            # Do something with the response event
            ...
        case _:
            # Do nothing for other events
            pass
```

--------------------------------

### Insert records in bulk

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/bulk/v2.md

Inserts a list of dictionaries into a specified Salesforce object.

```python
data = [
    {"FirstName": "Jon", "LastName": "Doe"},
    {"FirstName": "Jane", "LastName": "Doe"},
]
result = await salesforce.bulk_v2.insert("Contact", data)
record_ids = [record["sf__Id"] for record in result.successful_results]
```

--------------------------------

### Execute Multiple SOQL Queries in Composite Request

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/composite/composite.md

Perform multiple SOQL queries within a single composite request. The results of each query are accessible via the `records` attribute of the respective subrequest object.

```python
async with salesforce.composite() as composite:
    query_accounts = composite.query("SELECT Id, Name FROM Account LIMIT 10")
    query_contacts = composite.query("SELECT Id, Name FROM Contact LIMIT 10")
print(query_accounts.records)
print(query_contacts.records)
```

--------------------------------

### Execute Independent Composite Batch

Source: https://context7.com/georgebv/aiosalesforce/llms.txt

Runs up to 25 independent subrequests in parallel within a single API call using a context manager.

```python
# Using context manager (auto-executes on exit)
async with salesforce.composite.batch(halt_on_error=True, autoraise=True) as batch:
    # Query subrequests
    query_accounts = batch.query("SELECT Id, Name FROM Account LIMIT 10")
    query_contacts = batch.query("SELECT Id, Name FROM Contact LIMIT 10")

    # sObject CRUD subrequests
    new_contact = batch.sobject.create(
        "Contact",
        {"FirstName": "Jon", "LastName": "Doe"},
    )
    update_account = batch.sobject.update(
        "Account",
        "0011R00001K1H2IQAV",
        {"Name": "Updated Name"},
    )

# Access results after batch execution
print("Accounts:", query_accounts.records)
print("Contacts:", query_contacts.records)
print("Created Contact ID:", new_contact.id)

# Check individual subrequest status
print("Query success:", query_accounts.is_success)
print("Status code:", query_accounts.status_code)
```

--------------------------------

### Raise Subrequest Status Exception

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/composite/batch.md

Raises an exception if the specific subrequest failed.

```python
subrequest.raise_for_status()
```

--------------------------------

### Define Response Rule

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/client.md

Create a ResponseRule to retry requests based on specific HTTP status codes. The rule uses a lambda function to check the response status and can specify a maximum number of retries.

```python
from aiosalesforce import ResponseRule

response_rule = ResponseRule(
    lambda response: response.status_code >= 500,
    max_retries=5,
)
```

--------------------------------

### Read and Delete by External ID

Source: https://context7.com/georgebv/aiosalesforce/llms.txt

Performs read and delete operations using an external ID instead of the standard Salesforce ID.

```python
# Read record by external ID
contact = await salesforce.sobject.get(
    "Contact",
    "123456",           # External ID value
    "ExternalId__c",    # External ID field name
)
print(contact)

# Delete record by external ID
await salesforce.sobject.delete(
    "Contact",
    "123456",           # External ID value
    "ExternalId__c",    # External ID field name
)
```

--------------------------------

### Upsert records in bulk

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/bulk/v2.md

Upserts records by providing the object name, the external ID field name, and the data containing the external ID values.

```python
data = [
    {"ExternalId__c": "123", "FirstName": "Jon", "LastName": "Doe"},
    {"ExternalId__c": "456", "FirstName": "Jane", "LastName": "Doe"},
]
result = await salesforce.bulk_v2.upsert("Contact", data, "ExternalId__c")
```

--------------------------------

### Read records with field filtering

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/sobject.md

Retrieve records by ID, optionally specifying a list of fields to return.

```python
contact = await salesforce.sobject.get('Contact', contact_id)
```

```python
contact = await salesforce.sobject.get(
    "Contact",
    contact_id,
    fields=["FirstName", "LastName", "Email"],
)
```

--------------------------------

### Delete records in bulk

Source: https://github.com/georgebv/aiosalesforce/blob/main/docs/documentation/bulk/v2.md

Deletes records by providing a list of dictionaries containing the 'Id' field for each record.

```python
data = [
    {"Id": "0031R00001K1H2IQAV"},
    {"Id": "0031R00001K1H2JQAV"},
]
result = await salesforce.bulk_v2.delete("Contact", data)
```