The Salesforce client your web app deserves. Async or sync, same API.
{#- Below 76em only one window shows; the radios drive the shadcn-style tabs in its bar. -#}import even_simpler_salesforce as esf
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.responses import JSONResponse
sf = esf.AsyncSalesforce(auth, budget=esf.Budget(block_all_above=0.9))
@asynccontextmanager
async def lifespan(app: FastAPI):
async with sf: # one pooled client, closed on shutdown
yield
app = FastAPI(lifespan=lifespan)
@app.get("/leads/by-email")
async def lead_by_email(email: str):
# values are quoted for you: no SOQL injection
lead = await sf.query_one(
"SELECT Id, Name, Owner.Name FROM Lead WHERE Email = {}", email)
return esf.flatten(lead) # {"Id", "Name", "Owner.Name"}
@app.get("/accounts/{account_id}/contacts")
async def contacts(account_id: str):
soql = "SELECT Name, Email FROM Contact WHERE AccountId = {}"
return [esf.flatten(c) async for c in sf.query_iter(soql, account_id)]
@app.get("/limits")
async def limits():
api = (await sf.limits())["DailyApiRequests"]
return {"used": api.used, "max": api.max}
@app.exception_handler(esf.NotFound)
async def not_found(request, exc: esf.NotFound):
return JSONResponse({"error": "not found"}, status_code=404)
import atexit
import even_simpler_salesforce as esf
from flask import Flask, request
sf = esf.Salesforce(auth, budget=esf.Budget(block_all_above=0.9))
atexit.register(sf.close) # one pooled client, closed on exit
app = Flask(__name__)
@app.get("/leads/by-email")
def lead_by_email():
# values are quoted for you: no SOQL injection
email = request.args["email"]
lead = sf.query_one(
"SELECT Id, Name, Owner.Name FROM Lead WHERE Email = {}", email)
return esf.flatten(lead) # {"Id", "Name", "Owner.Name"}
@app.get("/accounts/<account_id>/contacts")
def contacts(account_id: str):
soql = "SELECT Name, Email FROM Contact WHERE AccountId = {}"
return [esf.flatten(c) for c in sf.query_iter(soql, account_id)]
@app.get("/limits")
def limits():
api = sf.limits()["DailyApiRequests"]
return {"used": api.used, "max": api.max}
@app.errorhandler(esf.NotFound)
def not_found(exc: esf.NotFound):
return {"error": "not found"}, 404