Python SDK

import os
from salesos import SalesOS

client = SalesOS(api_key=os.environ["SALESOS_API_KEY"])

lead = client.leads.create({"firstName": "Ada", "lastName": "Lovelace", "company": "Analytical"})
client.leads.update(lead["id"], {"status": "CONTACTED"})

for deal in client.opportunities.iterate(owner="me", open=True):
    print(deal["name"], deal.get("amount"))

from salesos import AsyncSalesOS

async with AsyncSalesOS(api_key=key) as client:
    async for lead in client.leads.iterate(status="NEW"):
        ...

from salesos.operations import call_operation

stages = call_operation(client, "pipelines_findAll")

from salesos import SalesOSError

try:
    client.opportunities.update(deal_id, {"stage": "CLOSED_WON"})
except SalesOSError as e:
    if e.code == "STAGE_REQUIREMENTS_NOT_MET":
        print(e.body["missingFields"])
    elif e.is_rate_limited:
        ...
    else:
        raise

from salesos import ClientCredentialsAuth, OAuthClient, SalesOS

oauth = OAuthClient(os.environ["SALESOS_CLIENT_ID"], os.environ["SALESOS_CLIENT_SECRET"])
client = SalesOS(auth=ClientCredentialsAuth(oauth, scope=["READ_LEADS", "WRITE_LEADS"]))

oauth = OAuthClient("sos_ci_...")
req = oauth.create_authorization_request(
    redirect_uri="https://app.example.com/oauth/callback",
    scope=["READ_LEADS", "offline_access"],
)
# Store req.state and req.code_verifier, then redirect the user to req.url.

code = oauth.parse_callback(request_url, stored_state)   # checks state, error and iss (RFC 9207)
tokens = oauth.exchange_code(code=code, redirect_uri=req.redirect_uri, code_verifier=stored_verifier)

from salesos import OAuthTokenAuth, StoredTokens

auth = OAuthTokenAuth(
    oauth,
    StoredTokens(tokens["access_token"], tokens.get("refresh_token"), tokens["expires_at"]),
    on_tokens=lambda t, raw: save(t.access_token, t.refresh_token, t.expires_at),
)
client = SalesOS(auth=auth)

job = client.bulk.create_job("lead", "upsert", file="leads.csv", external_id_field="email")
job = client.bulk.wait_for_job(job["id"], on_progress=lambda j: print(j["processedRows"], "rows"))
if job["failedRows"]:
    client.bulk.download_failures(job["id"], "failures.csv")

from salesos import SalesOSError, construct_event

# Flask
@app.post("/salesos/webhooks")
def salesos_webhook():
    try:
        event = construct_event(request.get_data(), request.headers, os.environ["SALESOS_WEBHOOK_SECRET"])
    except SalesOSError:
        return "", 400
    if already_processed(event["id"]):   # ids are stable across retries and replays
        return "", 200
    handle(event["type"], event["data"])
    return "", 200

from salesos import installation_secret, verify_context_token

# Each org's components are signed with a secret derived from your app secret and the org id.
secret = installation_secret(os.environ["SALESOS_APP_SECRET"], event["organizationId"])

# The context token an extension frame forwards to your backend.
claims = verify_context_token(token, secret, app_slug="acme-enrich")
print(claims["sub"], claims["org"], claims.get("rec"))

cd packages/sdk-python
python scripts/generate.py --fetch    # refresh spec/ from production and regenerate
python scripts/generate.py --check    # CI: exit 1 when the generated files are stale
python scripts/make_async.py --check  # the async resources are derived from the sync ones

pkg = client.metadata.retrieve()
deploy = client.metadata.deploy(pkg["files"], check_only=True)
result = client.metadata.wait_for_deploy(deploy["id"])
run = client.metadata.run_tests(pkg["files"], run_tests="ALL")   # check-only, runs the tests
tested = client.metadata.wait_for_deploy(run["id"])              # tested["testResults"]["results"]
client.metadata.get_test_policy()                               # {"requireTestsFor": [...]}
client.functions.invoke("score_lead", {"leadId": lead_id})
usage = client.api_usage.summary(days=30)
client.event_stream.trigger("lead.created")   # to `salesos listen` sessions