Skip to content

GeekStewie/salesforce-py

BranchesTags

Repository files navigation

salesforce-py logo

salesforce-py

Health Check

Disclaimer: This is an independent, community-maintained project and is not affiliated with, endorsed by, or supported by Salesforce, Inc. Salesforce, the Salesforce CLI, and related marks are trademarks of Salesforce, Inc.

A Python wrapper for Salesforce CLIs and APIs. Ships six independent clients:

  • SF CLI wrapper — sync/async subprocess wrapper for the sf CLI
  • REST API — async client for /services/data/vXX.X/ (SOQL/SOSL, sObject CRUD, composite / batch / graph / tree, quick actions, invocable actions, tooling, UI API, analytics, process / approvals, and the rest)
  • Connect REST API — async client for /services/data/vXX.X/connect/ (Chatter, Files, Communities, Commerce, Einstein, and more)
  • Data 360 Connect REST API — async client for /services/data/vXX.X/ssot/ (Data Cloud / Data 360)
  • Models REST API — async client for the Einstein Models generative AI surface (chat-generations, generations, embeddings, feedback)
  • Bulk API 2.0 — async client for /services/data/vXX.X/jobs/ with automatic ORDER BY handling via DuckDB

Requirements

Installation

From PyPI

pip install salesforce-py

With optional extras:

# REST API (SOQL/SOSL, sObject CRUD, composite, quick/invocable actions, tooling, UI API, etc.)
pip install "salesforce-py[rest]"

# Connect REST API (Chatter, Files, Communities, Commerce, Einstein, etc.)
pip install "salesforce-py[connect]"

# Data 360 Connect REST API (Data Cloud / CDP)
pip install "salesforce-py[data360]"

# Models REST API (Einstein generative AI)
pip install "salesforce-py[models]"

# Bulk API 2.0 (ingest + query, with DuckDB for client-side ORDER BY)
pip install "salesforce-py[bulk]"

# Salesforce Code Analyzer plugin support
pip install "salesforce-py[code-analyzer]"

# Everything
pip install "salesforce-py[all]"

Directly from GitHub

Latest main branch:

pip install "salesforce-py @ git+https://github.com/geekstewie/salesforce-py.git"

With extras:

pip install "salesforce-py[all] @ git+https://github.com/geekstewie/salesforce-py.git"

Pinned to a specific tag or commit:

pip install "salesforce-py @ git+https://github.com/geekstewie/salesforce-py.git@v0.1.0"

With uv

uv add salesforce-py
uv add "salesforce-py[all]"

# From repo
uv add "salesforce-py @ git+https://github.com/geekstewie/salesforce-py.git"

Quick start

from salesforce_py.sf import SFOrgTask

# Connect to an org by alias or username
task = SFOrgTask("my-scratch-org")

# Run Apex tests
results = task.apex.run_tests(class_names=["MyAccountTests"], code_coverage=True)

# Deploy metadata
task.project.deploy(source_dir="force-app", wait=33)

# Query data
records = task.data.query("SELECT Id, Name FROM Account LIMIT 10")

# Use the Agentforce operations
task.agent.test_run(api_name="MyAgentTest", wait=5)

SF CLI operations

SFOrgTask exposes all SF CLI command groups as attributes:

Attribute Commands
task.agent Agent lifecycle, spec/bundle generation, preview sessions, test management
task.alias Local alias management
task.apex Anonymous execution, test runs, log management
task.api Raw REST and GraphQL API requests
task.cmdt Custom metadata type generation
task.code_analyzer Code Analyzer scan and results
task.community Community/Experience Cloud publish
task.config SF CLI config get/set/unset/list
task.data SOQL query, record upsert/delete, bulk import/export
task.dev Dev hub, scratch org, sandbox management
task.doctor SF CLI diagnostics
task.flow Flow run and interview management
task.lightning LWC and Aura component generation
task.org Org create, delete, display, open, list
task.package Managed and unlocked package operations
task.plugins SF CLI plugin management
task.project Deploy, retrieve, convert, manifest generation
task.schema Object and field schema inspection
task.sobject sObject CRUD helpers
task.template Project template generation
task.ui_bundle UI Bundle deployment

REST API

salesforce_py.rest is a fully async client for the Salesforce REST API — the family of endpoints served under /services/data/vXX.X/. It covers SOQL / SOSL queries, every /sobjects/... resource (describe, CRUD, upsert, list views, layouts, relationships, deleted/updated, event schemas, user password), the composite / batch / graph / tree families, quick actions, invocable actions, tooling, UI API, analytics, process / approvals, support, and more. HTTP/2 is negotiated by default.

Option 1 — environment variables (recommended for CI/CD):

import asyncio
from salesforce_py.rest import RestClient

async def main():
    async with await RestClient.from_env() as client:
        rows = await client.query.query("SELECT Id, Name FROM Account LIMIT 10")

asyncio.run(main())

Option 2 — SF CLI session (interactive / dev machines):

import asyncio
from salesforce_py.rest import RestClient

async def main():
    async with await RestClient.from_env("my-org-alias") as client:
        created = await client.sobjects.create("Account", {"Name": "Acme Corp"})

asyncio.run(main())

Option 3 — SF CLI org object:

import asyncio
from salesforce_py.rest import RestClient
from salesforce_py.sf import SFOrgTask

async def main():
    task = SFOrgTask("my-org-alias")
    async with RestClient.from_org(task._org) as client:
        me = await client.sobjects.get("User", "005xx000001SvD8AAK")

asyncio.run(main())

Option 4 — direct token:

import asyncio
from salesforce_py.rest import RestClient

async def main():
    async with RestClient(instance_url="...", access_token="...") as client:
        limits = await client.limits.get_limits()

asyncio.run(main())

Environment variables:

Variable Purpose
SF_REST_CLIENT_ID External Client App consumer key
SF_REST_CLIENT_SECRET External Client App consumer secret
SF_REST_INSTANCE_URL My Domain URL (falls back to SF_INSTANCE_URL)

Install with the rest extra (pulls in httpx[http2]):

pip install "salesforce-py[rest]"

Operation namespaces:

Namespace Endpoint family
client.versions /services/data and /services/data/vXX.X — version + resource listings
client.limits /limits, /limits/recordCount
client.query /query, /queryAll, async SOQL
client.search /search (SOSL), /parameterizedSearch, scope / suggestions / layouts
client.sobjects Everything under /sobjects/ — describe, CRUD, upsert, list views, layouts, relationships, deleted/updated, event schemas, user password, quick actions
client.composite /composite, /composite/batch, /composite/graph, /composite/tree/{sobject}, /composite/sobjects
client.quick_actions Global /quickActions (list, describe, default values, invoke)
client.actions /actions/standard, /actions/custom — invocable actions
client.tabs, client.theme, client.recent /tabs, /theme, /recent
client.app_menu /appMenu (AppSwitcher, Salesforce1)
client.lightning_usage Lightning Adoption metric sObjects
client.process /process/approvals, /process/rules
client.support Data category groups, embedded service, field service, knowledge articles
client.tooling /tooling/ — query, describe, execute-anonymous, run-tests
client.ui_api /ui-api/ — records, record-ui, object-info, layouts, picklist-values
client.metadata /metadata/ passthrough
client.analytics / client.wave / client.folders / client.smart_data_discovery / client.eclair / client.jsonxform Analytics family
client.streaming /sobjects/StreamingChannel/{id}/push
client.financial_services / client.health_cloud / client.manufacturing / client.consumer_goods Industry cloud /connect/ passthroughs
client.asset_management / client.chatter / client.commerce / client.connect / client.consent / client.contact_tracing / client.dedupe / client.jobs / client.knowledge_management / client.licensing / client.localized_value / client.payments / client.scheduling Small-surface passthroughs

See src/salesforce_py/rest/README.md for the full REST API reference — operation namespaces, authentication patterns, error handling, HTTP/2 configuration, and testing guidance.

Connect REST API

salesforce_py.connect is a fully async client for the Salesforce Connect REST API. It covers Chatter, Files, Communities, Commerce, Einstein, Managed Topics, Named Credentials, Search, and most other endpoints served under /services/data/vXX.X/connect/ (and sibling namespaces). HTTP/2 is negotiated by default, with transparent fallback to HTTP/1.1.

Option 1 — environment variables (recommended for CI/CD):

import asyncio
from salesforce_py.connect import ConnectClient

async def main():
    async with await ConnectClient.from_env() as client:
        me = await client.users.get_user("me")

asyncio.run(main())

Option 2 — SF CLI session (interactive / dev machines):

import asyncio
from salesforce_py.connect import ConnectClient

async def main():
    async with await ConnectClient.from_env("my-org-alias") as client:
        await client.chatter.post_feed_item(body="Shipped.", subject_id="me")

asyncio.run(main())

Option 3 — SF CLI org object:

import asyncio
from salesforce_py.connect import ConnectClient
from salesforce_py.sf import SFOrgTask

async def main():
    task = SFOrgTask("my-org-alias")
    async with ConnectClient.from_org(task._org) as client:
        me = await client.users.get_user("me")

asyncio.run(main())

Option 4 — direct token:

import asyncio
from salesforce_py.connect import ConnectClient

async def main():
    async with ConnectClient(instance_url="...", access_token="...") as client:
        me = await client.users.get_user("me")

asyncio.run(main())

Environment variables:

Variable Purpose
SF_CONNECT_CLIENT_ID External Client App consumer key
SF_CONNECT_CLIENT_SECRET External Client App consumer secret
SF_CONNECT_INSTANCE_URL My Domain URL (falls back to SF_INSTANCE_URL)

Install with the connect extra (pulls in httpx[http2]):

pip install "salesforce-py[connect]"

See src/salesforce_py/connect/README.md for the full Connect API reference — operation namespaces, authentication patterns, error handling, ID normalisation, HTTP/2 configuration, and testing guidance.

Data 360 REST API

salesforce_py.data360 is a fully async client for the Salesforce Data 360 Connect REST API — the endpoint family under /services/data/vXX.X/ssot/ (formerly Data Cloud / CDP). HTTP/2 is negotiated by default.

Option 1 — environment variables (recommended for CI/CD):

import asyncio
from salesforce_py.data360 import Data360Client

async def main():
    async with await Data360Client.from_env() as client:
        segments = await client.segments.get_segments()

asyncio.run(main())

Option 2 — SF CLI session (interactive / dev machines):

import asyncio
from salesforce_py.data360 import Data360Client

async def main():
    async with await Data360Client.from_env("my-org-alias") as client:
        submitted = await client.query.submit_sql_query(
            {"sql": "SELECT Id__c, Email__c FROM UnifiedIndividual__dlm LIMIT 10"},
            dataspace="default",
        )

asyncio.run(main())

Option 3 — SF CLI org object:

import asyncio
from salesforce_py.data360 import Data360Client
from salesforce_py.sf import SFOrgTask

async def main():
    task = SFOrgTask("my-org-alias")
    async with Data360Client.from_org(task._org) as client:
        segments = await client.segments.get_segments()

asyncio.run(main())

Option 4 — direct token:

import asyncio
from salesforce_py.data360 import Data360Client

async def main():
    async with Data360Client(instance_url="...", access_token="...") as client:
        segments = await client.segments.get_segments()

asyncio.run(main())

Environment variables:

Variable Purpose
SF_DATA360_CLIENT_ID External Client App consumer key
SF_DATA360_CLIENT_SECRET External Client App consumer secret
SF_DATA360_INSTANCE_URL My Domain URL (falls back to SF_INSTANCE_URL)

Install with the data360 extra (pulls in httpx[http2]):

pip install "salesforce-py[data360]"

Operation namespaces — 26 grouped wrappers covering the full /ssot/ surface:

Namespace Endpoint family
client.activation_targets /ssot/activation-targets
client.activations /ssot/activations, /ssot/activation-external-platforms
client.calculated_insights /ssot/calculated-insights
client.connections /ssot/connections, schema/sitemap/actions
client.connectors /ssot/connectors
client.data_action_targets /ssot/data-action-targets
client.data_actions /ssot/data-actions
client.data_clean_room /ssot/data-clean-room/{collaborations,providers,templates,specifications}
client.data_graphs /ssot/data-graphs
client.data_kits /ssot/data-kits
client.data_lake_objects /ssot/data-lake-objects
client.data_model_objects /ssot/data-model-objects, /ssot/data-model-object-mappings
client.data_spaces /ssot/data-spaces
client.data_streams /ssot/data-streams
client.data_transforms /ssot/data-transforms
client.document_ai /ssot/document-processing/…
client.identity_resolutions /ssot/identity-resolutions
client.insights /ssot/insight/metadata, /ssot/insight/calculated-insights/{name}
client.machine_learning /ssot/machine-learning/…
client.metadata /ssot/metadata, /ssot/metadata-entities
client.private_network_routes /ssot/private-network-routes
client.profile /ssot/profile/…
client.query /ssot/query, /ssot/queryv2, /ssot/query-sql*
client.search_index /ssot/search-index
client.segments /ssot/segments
client.universal_id_lookup /ssot/universalIdLookup/…

See src/salesforce_py/data360/README.md for full usage, authentication patterns, error handling, and testing guidance.

Models REST API

salesforce_py.models is a fully async client for the Salesforce Einstein Models REST API at https://api.salesforce.com/einstein/platform/v1/. It covers chat generation, text generation, embeddings, and feedback. HTTP/2 is negotiated by default.

The Models API uses a client-credentials OAuth flow. Client credentials are always required — standard SF CLI session tokens do not carry the sfap_api/einstein_gpt_api scopes.

Option 1 — environment variables (recommended for CI/CD):

import asyncio
from salesforce_py.models import ModelsClient
from salesforce_py.models.supported_models import BEDROCK_ANTHROPIC_CLAUDE_46_SONNET

async def main():
    async with await ModelsClient.from_env() as client:
        reply = await client.chat_generations.generate(
            BEDROCK_ANTHROPIC_CLAUDE_46_SONNET,
            messages=[{"role": "user", "content": "What is a DMO?"}],
        )

asyncio.run(main())

Option 2 — SF CLI org object (supply client creds, domain resolved from CLI):

import asyncio
from salesforce_py.models import ModelsClient
from salesforce_py.sf import SFOrgTask

async def main():
    task = SFOrgTask("my-org-alias")
    async with await ModelsClient.from_org(
        task._org, consumer_key="<KEY>", consumer_secret="<SECRET>",
    ) as client:
        vectors = await client.embeddings.embed(
            "sfdc_ai__DefaultOpenAITextEmbeddingAda_002",
            ["customer loyalty program"],
        )

asyncio.run(main())

Option 3 — direct token:

import asyncio
from salesforce_py.models import ModelsClient

async def main():
    async with ModelsClient(access_token="...") as client:
        reply = await client.generations.generate(
            "sfdc_ai__DefaultOpenAIGPT4OmniMini",
            "Summarise this case note in one sentence.",
        )

asyncio.run(main())

Environment variables:

Variable Purpose
SF_MODELS_CLIENT_ID External Client App consumer key (sfap_api einstein_gpt_api api scopes)
SF_MODELS_CLIENT_SECRET External Client App consumer secret
SF_MODELS_INSTANCE_URL My Domain URL (falls back to SF_INSTANCE_URL)

Install with the models extra (pulls in httpx[http2]):

pip install "salesforce-py[models]"

Operation namespaces:

Namespace Endpoint
client.generations /models/{modelName}/generations
client.chat_generations /models/{modelName}/chat-generations
client.embeddings /models/{modelName}/embeddings
client.feedback /feedback

Supported model API names are exported as constants from salesforce_py.models.supported_models (BEDROCK_ANTHROPIC_CLAUDE_46_SONNET, GPT_4_OMNI, VERTEX_GEMINI_25_PRO, etc.). BYOLLM model names work too — pass any string.

See src/salesforce_py/models/README.md for full usage, rate limits, error handling, and testing guidance.

Bulk API 2.0

salesforce_py.bulk is a fully async client for the Salesforce Bulk API 2.0 — the endpoint family under /services/data/vXX.X/jobs/. It covers the full ingest (CSV write) and query (SOQL read) lifecycle documented in the Bulk API 2.0 Developer Guide, with HTTP/2 negotiated by default.

Key features:

  • Full lifecycle — create_job → upload_data → upload_complete → poll → download / concatenate results → delete_job, plus convenience wrappers (ingest.upsert, query.run_query) for the common end-to-end patterns.
  • Automatic ORDER BY stripping — Bulk 2.0 disables PK chunking when a query has ORDER BY, which can push large queries into timeouts. The client removes the clause before submission, captures the sort keys, and reapplies the order client-side via DuckDB after downloading the paginated CSV pages.
  • Client-side limit validation — delimiters, line endings, operations, and upload sizes are validated before dispatch, failing fast with a clear ValueError instead of an opaque server rejection.
  • Parallel result download — query.get_parallel_results exposes the /resultPages locator cursors (API 58.0+) so you can asyncio.gather the downloads.

Option 1 — environment variables (recommended for CI/CD):

import asyncio
from salesforce_py.bulk import BulkClient

async def main():
    async with await BulkClient.from_env() as client:
        csv_bytes = await client.query.run_query(
            "SELECT Id, Name FROM Account ORDER BY CreatedDate DESC LIMIT 10000"
        )

asyncio.run(main())

Option 2 — SF CLI session (interactive / dev machines):

import asyncio
from salesforce_py.bulk import BulkClient

async def main():
    async with await BulkClient.from_env("my-org-alias") as client:
        await client.ingest.upsert(
            object_name="Account",
            external_id_field="ExtId__c",
            csv_data=b"ExtId__c,Name\nA1,Acme\nB2,Beta\n",
        )

asyncio.run(main())

Option 3 — SF CLI org object:

import asyncio
from salesforce_py.bulk import BulkClient
from salesforce_py.sf import SFOrgTask

async def main():
    task = SFOrgTask("my-org-alias")
    async with BulkClient.from_org(task._org) as client:
        job = await client.ingest.create_job(object_name="Account", operation="insert")

asyncio.run(main())

Option 4 — direct token:

import asyncio
from salesforce_py.bulk import BulkClient

async def main():
    async with BulkClient(instance_url="...", access_token="...") as client:
        ...

asyncio.run(main())

Environment variables:

Variable Purpose
SF_BULK_CLIENT_ID External Client App consumer key
SF_BULK_CLIENT_SECRET External Client App consumer secret
SF_BULK_INSTANCE_URL My Domain URL (falls back to SF_INSTANCE_URL)

Install with the bulk extra (pulls in httpx[http2] and duckdb):

pip install "salesforce-py[bulk]"

Operation namespaces:

Namespace Endpoint family
client.ingest /jobs/ingest/ — CSV write jobs (insert/update/upsert/delete/hardDelete)
client.query /jobs/query/ — SOQL read jobs with automatic ORDER BY handling

See src/salesforce_py/bulk/README.md for the full Bulk API 2.0 reference — lifecycle diagrams, automatic ORDER BY handling, limit validation, error handling, and testing guidance.

SF CLI setup helper

from salesforce_py.sf import SFCLISetup

setup = SFCLISetup()
print(setup.status())          # Check what's installed
setup.ensure_sf_installed()    # Install SF CLI if missing (via npm/Homebrew)

Exceptions

from salesforce_py.exceptions import CLIError, CLINotFoundError, SalesforcePyError

try:
    task.apex.run(file_path="script.apex")
except CLINotFoundError:
    # sf binary not found on PATH
    ...
except CLIError as e:
    print(e.returncode, e.stdout, e.stderr)
except SalesforcePyError:
    # any other library error
    ...

Development

git clone https://github.com/geekstewie/salesforce-py.git
cd salesforce-py
uv sync --extra dev
uv run pytest
uv run ruff check src/
uv run ty check src/

License

This project is a personal open-source project, released under the Apache License, Version 2.0.

You are free to use, modify, and distribute this software in accordance with the terms of the license. It is provided as-is, without warranty of any kind — express or implied — including but not limited to warranties of merchantability or fitness for a particular purpose. Use it at your own risk.

This project is not an official Salesforce product and is not affiliated with or endorsed by Salesforce, Inc. in any way.

See the LICENSE file for the full license text.

About

Python wrapper for Salesforce CLI and other services

github.com/GeekStewie/salesforce-py

Topics

python data salesforce salesforce-developers connect-api salesforcedx data-cloud data-360

Resources

Readme

License

Apache-2.0 license
Activity

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

No releases published

Packages

 
 
 

Contributors

Languages