Salesforce connection

Connect a task to a Salesforce org to run SOQL queries and read/write objects over a managed Leoflow Connection. SalesforceHook (built on simple-salesforce) authenticates with username + password + security token, or with a connected-app flow configured in Extra.

Declare the provider

# leoflow.yaml
dag_id: salesforce_query
connectors:
  - salesforce

URI shape

salesforce://<login>:<password>@?__extra__=<json>

There is no host — Salesforce is reached at the org’s instance URL, which lives in Extra. The username is in Login, the password in Password (percent-escaped), and the security token, instance URL, and API version ride in Extra under __extra__.

Fields the UI asks for

Example DAG

# dag.py
from airflow.sdk import DAG, task


@task
def count_accounts():
    from airflow.providers.salesforce.hooks.salesforce import SalesforceHook

    hook = SalesforceHook(salesforce_conn_id="salesforce_default")
    result = hook.make_query("SELECT count() FROM Account")
    print("account count:", result["totalSize"])


with DAG("salesforce_query", schedule=None, catchup=False, tags=["example"]):
    count_accounts()

# leoflow.yaml
schema_version: "1.0"
dag_id: salesforce_query
python_version: "3.12"
connectors:
  - salesforce
connections:
  - salesforce_default

Security notes

• Security token rotation: the token is invalidated when the password changes; update the Connection’s Extra on every reset.
• Connected apps: prefer OAuth/JWT for production; set the consumer key and secret in Extra rather than storing a raw password.
• Never log AIRFLOW_CONN_SALESFORCE_DEFAULT; it carries the password and security token.

Related

• ADR 0019 — secret encryption at rest.
• ADR 0021 — agent secret delivery (AIRFLOW_CONN_<CONN_ID>).
• TestSalesforceConnectionURIShapeIntegration — chain-of-custody delivery test.