sfq 0.0.42

Python wrapper for the Salesforce's Query API.

sfq (Salesforce Query)

sfq is a lightweight Python wrapper library designed to simplify querying Salesforce, reducing repetitive code for accessing Salesforce data.

For more varied workflows, consider using an alternative like Simple Salesforce. This library was even referenced on the Salesforce Developers Blog.

Features

• Simplified query execution for Salesforce instances.
• Integration with Salesforce authentication via refresh tokens.
• Option to interact with Salesforce Tooling API for more advanced queries.
• Platform Events support (list available & publish single/batch).

Installation

You can install the sfq library using pip:

pip install sfq

Usage

Library Querying

from sfq import SFAuth

# Initialize the SFAuth class with authentication details
sf = SFAuth(
    instance_url="https://example-dev-ed.trailblaze.my.salesforce.com",
    client_id="your-client-id-here",
    client_secret="your-client-secret-here",
    refresh_token="your-refresh-token-here"
)

# Execute a query to fetch account records
print(sf.query("SELECT Id FROM Account LIMIT 5"))

# Execute a query to fetch Tooling API data
print(sf.tooling_query("SELECT Id, FullName, Metadata FROM SandboxSettings LIMIT 5"))

Composite Batch Queries

multiple_queries = {
    "Recent Users": """
        SELECT Id, Name,CreatedDate
        FROM User
        ORDER BY CreatedDate DESC
        LIMIT 10""",
    "Recent Accounts": "SELECT Id, Name, CreatedDate FROM Account ORDER BY CreatedDate DESC LIMIT 10",
    "Frozen Users": "SELECT Id, UserId FROM UserLogin WHERE IsFrozen = true",  # If exceeds 2000 records, will paginate
}

batched_response = sf.cquery(multiple_queries)

for subrequest_identifer, subrequest_response in batched_response.items():
    print(f'"{subrequest_identifer}" returned {subrequest_response["totalSize"]} records')
>>> "Recent Users" returned 10 records
>>> "Recent Accounts" returned 10 records
>>> "Frozen Users" returned 4082 records

Collection Deletions

response = sf.cdelete(['07La0000000bYgj', '07La0000000bYgk', '07La0000000bYgl'])
>>> [{'id': '07La0000000bYgj', 'success': True, 'errors': []}, {'id': '07La0000000bYgk', 'success': True, 'errors': []}, {'id': '07La0000000bYgl', 'success': True, 'errors': []}]

Static Resources

page = sf.read_static_resource_id('081aj000009jUMXAA2')
print(f'Initial resource: {page}')
>>> Initial resource: <h1>It works!</h1>
sf.update_static_resource_name('HelloWorld', '<h1>Hello World</h1>')
page = sf.read_static_resource_name('HelloWorld')
print(f'Updated resource: {page}')
>>> Updated resource: <h1>Hello World</h1>
sf.update_static_resource_id('081aj000009jUMXAA2', '<h1>It works!</h1>')

sObject Key Prefixes

# Key prefix via IDs
print(sf.get_sobject_prefixes())
>>> {'0Pp': 'AIApplication', '6S9': 'AIApplicationConfig', '9qd': 'AIInsightAction', '9bq': 'AIInsightFeedback', '0T2': 'AIInsightReason', '9qc': 'AIInsightValue', ...}

# Key prefix via names
print(sf.get_sobject_prefixes(key_type="name"))
>>> {'AIApplication': '0Pp', 'AIApplicationConfig': '6S9', 'AIInsightAction': '9qd', 'AIInsightFeedback': '9bq', 'AIInsightReason': '0T2', 'AIInsightValue': '9qc', ...}

Platform Events

Platform Events allow publishing and subscribing to real-time events. Requires a custom Platform Event (e.g., 'sfq__e' with fields like 'text__c').

from sfq import SFAuth

sf = SFAuth(
    instance_url="https://example-dev-ed.trailblaze.my.salesforce.com",
    client_id="your-client-id-here",
    client_secret="your-client-secret-here",
    refresh_token="your-refresh-token-here"
)

# List available events
events = sf.list_events()
print(events)  # e.g., ['sfq__e']

# Publish single event
result = sf.publish('sfq__e', {'text__c': 'Hello Event!'})
print(result)  # {'success': True, 'id': '2Ee...'}

# Publish batch
events_data = [
    {'text__c': 'Batch 1 message'},
    {'text__c': 'Batch 2 message'}
]
batch_result = sf.publish_batch(events_data, 'sfq__e')
print(batch_result['results'])  # List of results

## How to Obtain Salesforce Tokens

To use the `sfq` library, you'll need a **client ID** and **refresh token**. The easiest way to obtain these is by using the Salesforce CLI:

### Steps to Get Tokens

1. **Install the Salesforce CLI**:  
   Follow the instructions on the [Salesforce CLI installation page](https://developer.salesforce.com/tools/salesforcecli).
   
2. **Authenticate with Salesforce**:  
   Login to your Salesforce org using the following command:
   
   ```bash
   sf org login web --alias int --instance-url https://corpa--int.sandbox.my.salesforce.com

3. Display Org Details:
   To get the client ID, client secret, refresh token, and instance URL, run:

   sf org display --target-org int --verbose --json
   

   The output will look like this:

   {
     "status": 0,
     "result": {
       "id": "00Daa0000000000000",
       "apiVersion": "63.0",
       "accessToken": "00Daa0000000000000!evaU3fYZEWGUrqI5rMtaS8KYbHfeqA7YWzMgKToOB43Jk0kj7LtiWCbJaj4owPFQ7CqpXPAGX1RDCHblfW9t8cNOCNRFng3o",
       "instanceUrl": "https://example-dev-ed.trailblaze.my.salesforce.com",
       "username": "[e-mail]",
       "clientId": "PlatformCLI",
       "connectedStatus": "Connected",
       "sfdxAuthUrl": "force://PlatformCLI::nwAeSuiRqvRHrkbMmCKvLHa[e-mail]",
       "alias": "int"
     }
   }
   

4. Extract and Use the Tokens:
   The sfdxAuthUrl is structured as:

   force://<client_id>:<client_secret>:<refresh_token>@<instance_url>
   

   This means with the above output sample, you would use the following information:

   # This is for illustrative purposes; use environment variables instead
   client_id = "PlatformCLI"
   client_secret = ""
   refresh_token = "nwAeSuiRqvRHrkbMmCKvLHasS0vRbh3Cf2RF41WZzmjtThnCwOuDvn9HObcUjKuTExJPqPegIwnLB5aH6GNWYhU"
   instance_url = "https://example-dev-ed.trailblaze.my.salesforce.com"
   
   from sfq import SFAuth
   sf = SFAuth(
       instance_url=instance_url,
       client_id=client_id,
       client_secret=client_secret,
       refresh_token=refresh_token,
   )
   

Important Considerations

• Security: Safeguard your client_id, client_secret, and refresh_token diligently, as they provide access to your Salesforce environment. Avoid sharing or exposing them in unsecured locations.
• Efficient Data Retrieval: The query and cquery function automatically handles pagination, simplifying record retrieval across large datasets. It's recommended to use the LIMIT clause in queries to control the volume of data returned.
• Advanced Tooling Queries: Utilize the tooling_query function to access the Salesforce Tooling API. This option is designed for performing complex operations, enhancing your data management capabilities.