Querying Records

Salesforce Toolkit provides a powerful query builder for creating and executing SOQL queries.

Basic Queries

The simplest way to create a query is using the query() method on your SObject class:

from sf_toolkit.data import SObject
from sf_toolkit.data.fields import IdField, TextField

class Account(SObject):
    Id = IdField()
    Name = TextField()
    Industry = TextField()

# Create a query for all records
query = Account.query()

# Execute the query
results = query.execute()

# Process results
for account in results:
    print(account.Name)

Implicit Query Execution

Iterating over a SOQL query implicitly calls .execute() to generate a QueryResult, which serves as the iterable object.

from sf_toolkit.data import SObject
from sf_toolkit.data.fields import IdField, TextField

class Account(SObject):
    Id = IdField()
    Name = TextField()
    Industry = TextField()

# Create a query for all records
query = Account.query()

# implicit __iter__ executes query
for account in query:
    print(account.Name)

# or, in an async method
# implicit __aiter__ also executes query,
# and creates tasks in the event loop to fetch
# all subsequent batches in the query locator
async def foo():
    query = Account.query() # ...
    async for account in query:
        print(account.Name)

Multiple Iterations of `QueryResult`s

Iterating over a QueryResult (the product of an SoqlQuery.execute()) multiple times generates multiple QueryResult instances with the same list of QueryResultBatch`es. The records for a particular QueryResult will be the same record instances as a copied the records. This means that modifications to these instances will appear to propogate across any query result instances derived from the original instance generated by the first `SoqlQuery.execute() call

query_result = Account.query().execute()
result_copy = query_result.copy()

record_list = query_result.as_list()
for index, (original, copy) in enumerate(zip(query_result, result_copy)):
    assert record_list[index] is record  # This should always be true
    assert original is copy  # This should also always be true

Filtering Records

You can filter records using the where() method with field conditions:

# Simple equality condition
query = Account.query().where(Industry="Technology")

# Comparison operators using field__operator syntax
query = Account.query().where(
    AnnualRevenue__gt=1000000,    # Greater than
    Name__like="Test%"            # LIKE operator
)

# IN operator
query = Account.query().where(Industry__in=["Technology", "Healthcare"])

Combining Conditions

Use and_where() and or_where() to build complex conditions incrementally:

# Combine conditions with AND logic
query = Account.query()
    .where(Industry="Technology")
    .and_where(AnnualRevenue__gt=1000000)

# Add multiple conditions at once
    .and_where(
    NumberOfEmployees__gt=50,
    BillingCountry="USA"
)

# Combine conditions with OR logic
query = Account.query().where(Industry="Technology")
query = query.or_where(Industry="Healthcare")

# Mixing AND and OR logic
query = Account.query().where(AnnualRevenue__gt=500000)
query = query.and_where(Industry="Technology")
query = query.or_where(
    Industry="Healthcare",
    AnnualRevenue__gt=1000000
)

# Building a query step by step
query = Account.query()
if filter_by_industry:
    query = query.where(Industry__in=["Technology", "Healthcare"])
if filter_by_revenue:
    query = query.and_where(AnnualRevenue__gt=min_revenue)
if search_term:
    query = query.and_where(Name__like=f"%{search_term}%")

Complex Conditions

For more complex conditions, use the logical operators AND and OR:

from sf_toolkit.data.query_builder import AND, OR, EQ, GT

# Complex boolean logic
query = Account.query().where(
    OR(
        EQ("Industry", "Technology"),
        AND(
            GT("AnnualRevenue", 1000000),
            GT("NumberOfEmployees", 100)
        )
    )
)

Raw WHERE Clauses

You can also use raw SOQL WHERE clauses for advanced filtering:

query = Account.query().where(
    "Name LIKE 'Test%' AND CreatedDate = LAST_N_DAYS:30"
)

Grouping and Aggregates

Support for GROUP BY and HAVING clauses:

# Basic GROUP BY
query = Account.query().group_by("Industry")

# GROUP BY with HAVING clause
query = Account.query().group_by("Industry").having(
    COUNT__Id__gt=5
)

# Multiple HAVING conditions
query = Account.query().group_by("Industry").having(
    COUNT__Id__gt=5
).and_having(
    SUM__AnnualRevenue__gt=1000000
).or_having(
    SUM__AnnualRevenue__gt=5000000
)

Sorting Results

Order results using the order_by() method:

from sf_toolkit.data.query_builder import Order

# Using Order objects
query = Account.query().order_by(Order("Name", "DESC"))

# Using field=direction syntax
query = Account.query().order_by(Name="DESC", CreatedDate="ASC")

Pagination

Control result pagination using limit() and offset():

query = Account.query().limit(10).offset(20)

Handling Results

Query results are returned as a QueryResult object which is an iterator over SObject records:

results = query.execute()

# Check if all records were retrieved
if not results.done:
    print("More records are available")

# Get total record count
total = len(results)

# Access all records as a list
all_records = results.as_list()

# Iterate through records automatically handling pagination
for account in results:
    print(account.Name)

# Convert to a list to get all records at once
account_list = list(results)

Counting Records

Execute a COUNT() query to get the total number of matching records:

query = Account.query().where(Industry="Technology")
count = query.count()
print(f"Found {count} Technology accounts")

Tooling API Queries

Query Tooling API objects by setting the tooling=True flag on your SObject class:

class CustomObject(SObject, tooling=True):
    Id = IdField()
    Name = TextField()

# Query will automatically use the Tooling API endpoint
results = CustomObject.query().execute()

Date and DateTime Values

Handle date and datetime values in queries:

from datetime import datetime, date

# Query with datetime
now = datetime.now().astimezone()
query = Account.query().where(CreatedDate__gt=now)

# Query with date
today = date.today()
query = Opportunity.query().where(CloseDate=today)