b2c_tooling_sdk.clients
HTTP clients for B2C Commerce APIs.
Mirrors the clients subpath of the TypeScript SDK. Provides the core HttpClient (openapi-fetch analog with ClientResult), the shared middleware chain, the OCAPI and WebDAV clients, the SCAPI/OCAPI dual-backend utilities, the typed domain clients (SLAS, ODS, MRT, Account Manager, SCAPI Jobs/ Sites/Catalogs/Scripts/Merchant-Users/Merchant-Roles/Schemas, Custom APIs, CDN Zones, Preferences, Metrics, Granular Replications), the middleware registry, TLS/mTLS transport helpers, and API error-message utilities.
Importing this package auto-registers the User-Agent middleware providers (see b2c_tooling_sdk.clients.user_agent).
Classes
AccountManagerClient
class AccountManagerClientUnified Account Manager API client that combines users, roles, API clients, and organizations.
Provides direct access to every Account Manager API operation through a single interface, while internally using separate configured clients for each domain. Role and org mappings (id/enum-name and id/name) are lazily fetched and cached.
get_user method
async def get_user(user_id: str, expand: list[UserExpandOption] | None = None) -> AccountManagerUserGet user by ID.
list_users method
async def list_users(options: ListUsersOptions | None = None) -> UserCollectionList users with pagination.
create_user method
async def create_user(user: UserCreate) -> AccountManagerUserCreate a new user.
update_user method
async def update_user(user_id: str, changes: UserUpdate) -> AccountManagerUserUpdate an existing user.
delete_user method
async def delete_user(user_id: str) -> NoneDisable a user (soft delete).
purge_user method
async def purge_user(user_id: str) -> NonePurge a user (hard delete).
reset_user method
async def reset_user(user_id: str) -> NoneReset a user to INITIAL state.
find_user_by_login method
async def find_user_by_login(login: str, expand: list[UserExpandOption] | None = None) -> AccountManagerUser | NoneFind a user by login (email).
grant_role method
async def grant_role(user_id: str, role: str, scope: str | None = None) -> AccountManagerUserGrant a role to a user, optionally scoped to specific tenants.
revoke_role method
async def revoke_role(user_id: str, role: str, scope: str | None = None) -> AccountManagerUserRevoke a role from a user, optionally removing only specific tenant scopes.
get_role method
async def get_role(role_id: str) -> AccountManagerRoleGet role by ID.
list_roles method
async def list_roles(options: ListRolesOptions | None = None) -> RoleCollectionList roles with pagination.
get_role_mapping method
async def get_role_mapping() -> RoleMappingGet the role mapping (id <-> roleEnumName), lazily cached.
get_org_mapping method
async def get_org_mapping() -> OrgMappingGet the org mapping (id -> name), lazily cached.
list_api_clients method
async def list_api_clients(options: ListApiClientsOptions | None = None) -> APIClientCollectionList API clients with pagination.
get_api_client method
async def get_api_client(api_client_id: str, expand: list[ApiClientExpandOption] | None = None) -> AccountManagerApiClientGet API client by ID.
create_api_client method
async def create_api_client(body: APIClientCreate) -> AccountManagerApiClientCreate a new API client.
update_api_client method
async def update_api_client(api_client_id: str, body: APIClientUpdate) -> AccountManagerApiClientUpdate an existing API client.
delete_api_client method
async def delete_api_client(api_client_id: str) -> NoneDelete an API client (must be disabled 7+ days).
change_api_client_password method
async def change_api_client_password(api_client_id: str, old_password: str, new_password: str) -> NoneChange an API client password.
get_org method
async def get_org(org_id: str) -> AccountManagerOrganizationGet organization by ID.
get_org_by_name method
async def get_org_by_name(name: str) -> AccountManagerOrganizationGet organization by name.
list_orgs method
async def list_orgs(options: ListOrgsOptions | None = None) -> OrganizationCollectionList organizations with pagination.
AccountManagerClientConfig
class AccountManagerClientConfigConfiguration for creating Account Manager clients (users, roles, apiclients, orgs).
Fields
| Name | Type | Default |
|---|---|---|
hostname | str | None | None |
middleware_registry | MiddlewareRegistry | None | None |
AccountManagerOrgsClient
class AccountManagerOrgsClientAccount Manager Organizations API client.
Hand-rolled request logic (mirrors the TS createAccountManagerOrgsClient): a private, non-retrying auth middleware plus registry/logging middleware, and status-code-driven error mapping instead of the ClientResult convention used by the other AM clients.
get_org method
async def get_org(org_id: str) -> AccountManagerOrganizationGet organization by ID.
get_org_by_name method
async def get_org_by_name(name: str) -> AccountManagerOrganizationGet organization by name (searches for exact or partial match).
list_orgs method
async def list_orgs(options: ListOrgsOptions | None = None) -> OrganizationCollectionList organizations with pagination.
BackendBase
class BackendBase(Protocol)Common base for every SCAPI/OCAPI dual backend.
Mirrors the TypeScript BackendBase interface: a backend exposes a name identifying which transport it speaks so the fallback wrapper can report the currently-active backend.
Fields
| Name | Type | Description |
|---|---|---|
name | Literal['ocapi', 'scapi'] | Which transport this backend speaks. |
B2COrgInfo
class B2COrgInfo(BaseModel)Fields
| Name | Type |
|---|---|
is_b2c_customer | bool |
instances | list[Instance] |
B2CTargetInfo
class B2CTargetInfo(BaseModel)Fields
| Name | Type | Default |
|---|---|---|
instance_id | str | |
sites | list[Site] | None | None |
BuildScapiClientOptions
class BuildScapiClientOptionsDomain-specific options for build_scapi_client.
Fields
| Name | Type | Default |
|---|---|---|
path_segment | str | |
domain_key | HttpClientType | |
log_prefix | str | |
scope_cascade | ScopeCascade | None | None |
default_scopes | list[str] | None | None |
CdnZonesClientConfig
class CdnZonesClientConfigConfiguration for creating a CDN Zones client.
Fields
| Name | Type | Default |
|---|---|---|
short_code | str | |
tenant_id | str | |
scopes | list[str] | None | None |
middleware_registry | MiddlewareRegistry | None | field(default=None) |
CdnZonesClientOptions
class CdnZonesClientOptionsOptions for creating a CDN Zones client.
Fields
| Name | Type | Default |
|---|---|---|
read_write | bool | False |
CdnZonesError
class CdnZonesError(BaseModel)Fields
| Name | Type |
|---|---|
title | str |
type | str |
detail | str |
instance | str | None |
ClientResult
class ClientResultResult of a typed client call — mirrors openapi-fetch {data, error, response}.
Never raised: a successful (2xx) response populates data, any other status populates error, and response is always the raw httpx.Response. Only network failures raise (before a result exists).
Fields
| Name | Type | Default |
|---|---|---|
data | Any | None |
error | Any | None |
response | httpx.Response | None | None |
CodeVersion
class CodeVersion(BaseModel)Fields
| Name | Type | Default |
|---|---|---|
id | str | None | Field(None, max_length=256, min_length=1) |
active | bool | None | None |
cartridges | list[Cartridge] | None | None |
compatibilityMode | str | None | Field(None, max_length=100) |
activationTime | AwareDatetime | None | None |
lastModificationTime | AwareDatetime | None | None |
rollback | bool | None | None |
totalSize | int | None | None |
webDavUrl | str | None | Field(None, max_length=4000) |
ContentAssetItemPrivate
class ContentAssetItemPrivate(BaseModel)Details of the published content asset from a private library
Fields
| Name | Type |
|---|---|
contentId | str |
type | Type |
siteId | str |
ContentAssetItemShared
class ContentAssetItemShared(BaseModel)Details of the published content asset from a shared library
Fields
| Name | Type |
|---|---|
contentId | str |
type | Type1 |
libraryId | str |
CustomApisClientConfig
class CustomApisClientConfigConfiguration for creating a Custom APIs client.
Fields
| Name | Type | Default |
|---|---|---|
short_code | str | |
tenant_id | str | |
scopes | list[str] | None | None |
middleware_registry | MiddlewareRegistry | None | field(default=None) |
CustomPreference
class CustomPreference(BaseModel)Preference object
Fields
| Name | Type | Default |
|---|---|---|
groupId | str | Field(..., description='The ID of the preference group.') |
id | str | Field(..., description='The Preference Id.') |
value | Any | Field(..., description='The value for the Preference Id.') |
CustomPreferenceList
class CustomPreferenceList(PaginatedResultBase)Document representing a Custom Preference result.
Fields
| Name | Type |
|---|---|
data | list[CustomPreference] |
limit | int |
offset | int |
total | int |
DualBackendConfig
class DualBackendConfigCommon input shape of every dual-backend factory.
SCAPI coordinates and the scope-flexible auth strategy are sourced from the instance via B2CInstance.scapi_client_config. A backend is "SCAPI-capable" iff that getter returns a value. preference is optional: when None it falls back to the instance's own B2CInstance.api_backend (default "auto").
Fields
| Name | Type | Default |
|---|---|---|
instance | B2CInstance | |
preference | ApiBackendPreference | None | None |
DualBackendCtors
class DualBackendCtors(Generic[T])Constructors needed to build a dual-backend instance.
Each domain plugs in its own SCAPI/OCAPI backend factories; the generic factory wires them together. scapi receives a ScapiBackendCtorConfig; ocapi receives the B2CInstance directly.
Fields
| Name | Type |
|---|---|
domain_name | str |
scapi | Callable[[ScapiBackendCtorConfig], T] |
ocapi | Callable[[B2CInstance], T] |
ExecutionStatus
class ExecutionStatus(Enum)ExitStatus
class ExitStatus(BaseModel)Fields
| Name | Type | Default |
|---|---|---|
code | str | None | Field(None, max_length=256) |
message | str | None | Field(None, max_length=4000) |
status | Status | None | None |
GranularReplicationsClientConfig
class GranularReplicationsClientConfigConfiguration for creating a Granular Replications API client.
Parameters
| Name | Type | Description |
|---|---|---|
short_code | str | The instance short code (e.g. kv7kzm78). |
tenant_id | str | The tenant ID (e.g. zzxy_prd). |
scopes | list[str] | None | Optional custom OAuth scopes. Defaults to sfcc.granular-replications.rw and the tenant-specific scope. |
middleware_registry | MiddlewareRegistry | None | Optional custom middleware registry for request/response interceptors. |
Fields
| Name | Type | Default |
|---|---|---|
short_code | str | |
tenant_id | str | |
scopes | list[str] | None | None |
middleware_registry | MiddlewareRegistry | None | field(default=None) |
GranularReplicationsError
class GranularReplicationsError(BaseModel)Standard error response following RFC 7807
Fields
| Name | Type |
|---|---|
type | str |
title | str |
detail | str | None |
instance | str | None |
HttpClient
class HttpClientAsync HTTP client with a middleware chain, returning ClientResult.
Parameters
| Name | Type | Description |
|---|---|---|
base_url | str | Base URL prefix (e.g. https://host/s/-/dw/data/v25_6). |
middleware | list[Middleware] | None | Middleware in registration order (auth first, logging last). |
client_type | str | The HttpClientType label passed to middleware contexts. |
transport | httpx.AsyncBaseTransport | None | Optional TLS/mTLS transport (mirrors the undici dispatcher). |
Fields
| Name | Type | Default |
|---|---|---|
middleware | list[Middleware] | list(middleware or []) |
use method
def use(middleware: Middleware) -> NoneAppend a middleware to the chain (mirrors openapi-fetch client.use).
aclose method
async def aclose() -> NoneClose the underlying HTTP client and release its connections.
request method
async def request(method: str, path: str, *, params: dict[str, Any] | None = None, body: Any = None, headers: dict[str, str] | None = None) -> ClientResultPerform a request through the middleware chain and return a ClientResult.
Parameters
| Name | Type | Description |
|---|---|---|
method | str | HTTP method (GET, POST, ...). |
path | str | Path appended to base_url; may contain {name} placeholders. |
params | dict[str, Any] | None | Optional {"path": {...}, "query": {...}} parameters. |
body | Any | Optional request body (dict/list → JSON, or raw bytes/str). |
headers | dict[str, str] | None | Optional extra request headers. |
Raises
NetworkError— on transport-level failures (never on 4xx/5xx).
get method
async def get(path: str, options: dict[str, Any] | None = None) -> ClientResultPerform a GET request. options may carry params/headers.
post method
async def post(path: str, options: dict[str, Any] | None = None) -> ClientResultPerform a POST request. options may carry params/body/headers.
put method
async def put(path: str, options: dict[str, Any] | None = None) -> ClientResultPerform a PUT request. options may carry params/body/headers.
patch method
async def patch(path: str, options: dict[str, Any] | None = None) -> ClientResultPerform a PATCH request. options may carry params/body/headers.
delete method
async def delete(path: str, options: dict[str, Any] | None = None) -> ClientResultPerform a DELETE request. options may carry params/headers.
HttpMiddlewareProvider
class HttpMiddlewareProvider(Protocol)Supplies middleware for HTTP clients.
Providers can return different middleware per client type, or None to skip a client type.
Fields
| Name | Type |
|---|---|
name | str |
get_middleware method
def get_middleware(client_type: HttpClientType) -> UnifiedMiddleware | NoneReturn middleware for client_type, or None to skip it.
JobExecution
class JobExecution(BaseModel)Fields
| Name | Type | Default |
|---|---|---|
id | str | Field(..., max_length=256, min_length=1) |
jobId | str | Field(..., max_length=256, min_length=1) |
jobDescription | str | None | Field(None, max_length=4000) |
clientId | str | None | Field(None, max_length=256) |
userLogin | str | None | Field(None, max_length=256) |
executionStatus | ExecutionStatus | None | None |
status | str | Field(..., max_length=256) |
startTime | AwareDatetime | None | None |
endTime | AwareDatetime | None | None |
creationDate | AwareDatetime | None | None |
duration | int | None | None |
effectiveDuration | int | None | None |
modificationTime | AwareDatetime | None | None |
lastModified | AwareDatetime | None | None |
executedServerId | str | None | Field(None, max_length=256) |
exitStatus | ExitStatus | None | None |
statusMetadata | StatusMetadata | None | None |
isLogFileExisting | bool | None | None |
isRestart | bool | None | None |
logFilePath | str | None | Field(None, max_length=4000) |
parameters | list[Parameter] | None | None |
executionScopes | list[ExecutionScope] | None | None |
retryInformation | JobExecutionRetryInformation | None | None |
continueInformation | JobExecutionContinueInformation | None | None |
stepExecutions | list[StepExecution] | None | None |
JobExecutionSearchResult
class JobExecutionSearchResult(PaginatedSearchResult)Fields
| Name | Type |
|---|---|
hits | list[Hit] |
query | Query |
JobParameter
class JobParameter(BaseModel)Fields
| Name | Type |
|---|---|
name | str |
value | str |
JobStepExecution
class JobStepExecution(BaseModel)Fields
| Name | Type | Default |
|---|---|---|
id | str | None | Field(None, max_length=256, min_length=1) |
stepId | str | None | Field(None, max_length=256, min_length=1) |
stepDescription | str | None | Field(None, max_length=4000) |
stepTypeId | str | None | Field(None, max_length=256) |
stepTypeInfo | str | None | Field(None, max_length=4000) |
executionScope | str | None | Field(None, max_length=256) |
executionStatus | ExecutionStatus | None | None |
status | str | None | Field(None, max_length=256) |
startTime | AwareDatetime | None | None |
endTime | AwareDatetime | None | None |
duration | int | None | None |
modificationTime | AwareDatetime | None | None |
statusMetadata | StatusMetadata | None | None |
exitStatus | ExitStatus | None | None |
includeStepsFromJobId | str | None | Field(None, max_length=256) |
isChunkOriented | bool | None | None |
chunkSize | int | None | None |
itemFilterCount | int | None | None |
itemWriteCount | int | None | None |
totalItemCount | int | None | None |
ListApiClientsOptions
class ListApiClientsOptionsOptions for listing API clients.
Fields
| Name | Type | Default |
|---|---|---|
size | int | None | None |
page | int | None | None |
ListOrgsOptions
class ListOrgsOptionsOptions for listing organizations.
Fields
| Name | Type | Default |
|---|---|---|
size | int | None | None |
page | int | None | None |
all | bool | False |
ListRolesOptions
class ListRolesOptionsOptions for listing roles.
Fields
| Name | Type | Default |
|---|---|---|
size | int | None | None |
page | int | None | None |
role_target_type | Literal['ApiClient', 'User'] | None | None |
ListUsersOptions
class ListUsersOptionsOptions for listing users.
Fields
| Name | Type | Default |
|---|---|---|
size | int | None | None |
page | int | None | None |
Metric
class Metric(BaseModel)Fields
| Name | Type | Default |
|---|---|---|
metricId | str | |
title | str | Field(..., max_length=200, min_length=1) |
description | str | Field(..., max_length=500, min_length=1) |
unit | str | None | Field(None, max_length=50, min_length=0) |
dataSeries | list[DataSeries] |
MetricDataPoint
class MetricDataPoint(BaseModel)Fields
| Name | Type | Default |
|---|---|---|
timestamp | int | Field(..., ge=0) |
value | float |
MetricDataSeries
class MetricDataSeries(BaseModel)Fields
| Name | Type | Default |
|---|---|---|
id | str | Field(..., max_length=200, min_length=1) |
name | str | Field(..., max_length=200, min_length=1) |
data | list[DataPoint] |
MetricsClientConfig
class MetricsClientConfigConfiguration for creating a Metrics API client.
Fields
| Name | Type | Default |
|---|---|---|
short_code | str | |
tenant_id | str | |
scopes | list[str] | None | None |
middleware_registry | MiddlewareRegistry | None | field(default=None) |
MetricsDataResponse
class MetricsDataResponse(BaseModel)Fields
| Name | Type |
|---|---|
data | list[Metric] |
MetricsError
class MetricsError(BaseModel)Fields
| Name | Type | Default |
|---|---|---|
title | str | Field(..., max_length=256) |
type | str | Field(..., max_length=2048) |
detail | str | |
instance | str | None | Field(None, max_length=2048) |
Middleware
class Middleware(Protocol)Middleware for HttpClient (analogous to an openapi-fetch middleware).
Both hooks are optional. on_request may mutate ctx.request in place and/or return a replacement request; on_response may return a replacement response (e.g. a retry result). Returning None keeps the current object.
on_request method
async def on_request(ctx: MiddlewareRequestContext) -> httpx.Request | NoneCalled before the request is sent; may mutate or replace it.
on_response method
async def on_response(ctx: MiddlewareResponseContext) -> httpx.Response | NoneCalled after the response is received; may mutate or replace it.
MiddlewareRegistry
class MiddlewareRegistryRegistry for HTTP middleware providers.
Collects middleware from registered providers and returns it in registration order when requested by a client factory.
Fields
| Name | Type | Description |
|---|---|---|
size | int | Number of registered providers. |
register method
def register(provider: HttpMiddlewareProvider) -> NoneRegister a middleware provider (called in registration order).
unregister method
def unregister(name: str) -> boolRemove a provider by name; return True if one was removed.
get_middleware method
def get_middleware(client_type: HttpClientType) -> list[UnifiedMiddleware]Collect middleware from all providers for client_type, in order.
clear method
def clear() -> NoneClear all registered providers (primarily for testing).
get_provider_names method
def get_provider_names() -> list[str]Return the names of all registered providers.
MiddlewareRequestContext
class MiddlewareRequestContextContext passed to a middleware's on_request hook.
Fields
| Name | Type | Default |
|---|---|---|
request | httpx.Request | |
client_type | str | |
schema_path | str | '' |
fetch | FetchFn | None | None |
MiddlewareResponseContext
class MiddlewareResponseContextContext passed to a middleware's on_response hook.
Fields
| Name | Type | Default |
|---|---|---|
request | httpx.Request | |
response | httpx.Response | |
client_type | str | |
schema_path | str | '' |
fetch | FetchFn | None | None |
MrtB2CClientConfig
class MrtB2CClientConfigConfiguration for creating an MRT B2C client.
Fields
| Name | Type | Default |
|---|---|---|
origin | str | None | None |
middleware_registry | MiddlewareRegistry | None | field(default=None) |
MrtClientConfig
class MrtClientConfigConfiguration for creating an MRT client.
Fields
| Name | Type | Default |
|---|---|---|
origin | str | None | None |
middleware_registry | MiddlewareRegistry | None | field(default=None) |
OcapiDeprecatedError
class OcapiDeprecatedError(Exception)Raised when an OCAPI operation fails because OCAPI is deprecated.
Carries actionable SCAPI-setup guidance — including the exact scope the failed operation needs, when supplied — so callers surface a helpful message instead of an opaque "Failed to ..." line. The original error is attached as __cause__ (via raise ... from) at the throw site.
OdsClientConfig
class OdsClientConfigConfiguration for creating an ODS client.
Fields
| Name | Type | Default |
|---|---|---|
host | str | None | None |
extra_params | dict[str, Any] | None | None |
middleware_registry | MiddlewareRegistry | None | field(default=None) |
OpenApiSchema
class OpenApiSchema(BaseModel)An OpenAPI 3.0 schema specification
Fields
| Name | Type | Default |
|---|---|---|
openapi | str | None | |
info | Info | None | None |
paths | dict[str, Any] | None | None |
components | dict[str, Any] | None | None |
OrgMapping
class OrgMappingOrganization mapping built from the Account Manager organizations API. Maps org ID to name.
Fields
| Name | Type |
|---|---|
by_id | dict[str, str] |
OrganizationPreferences
class OrganizationPreferences(BaseModel)Represents custom preferences at the global (organization) level within a preference group. Custom preference attributes are returned with the "c_" prefix.
Fields
| Name | Type |
|---|---|
sitePreferences | list[SitePreferences] | None |
PatchedB2CTargetInfo
class PatchedB2CTargetInfo(BaseModel)Fields
| Name | Type | Default |
|---|---|---|
instance_id | str | None | |
sites | list[Site] | None | None |
PreferenceValue
class PreferenceValue(BaseModel)Represents a single preference value with its attribute definition and site-specific values.
Fields
| Name | Type | Default |
|---|---|---|
id | str | |
description | dict[str, Description3] | None | None |
displayName | dict[str, DisplayName2] | None | None |
attributeDefinition | ObjectAttributeDefinition | None | None |
siteValues | dict[str, Any] | None | |
valueType | ValueType | None | None |
PreferenceValueSearchResult
class PreferenceValueSearchResult(PaginatedSearchResult)Document representing a preference value search result.
Fields
| Name | Type |
|---|---|
hits | list[PreferenceValue] | None |
PreferencesClientConfig
class PreferencesClientConfigConfiguration for creating a Preferences client.
Fields
| Name | Type | Default |
|---|---|---|
short_code | str | |
tenant_id | str | |
scopes | list[str] | None | None |
middleware_registry | MiddlewareRegistry | None | field(default=None) |
PreferencesClientOptions
class PreferencesClientOptionsOptions for creating a Preferences client.
Fields
| Name | Type | Default |
|---|---|---|
read_write | bool | False |
PreferencesError
class PreferencesError(BaseModel)Fields
| Name | Type |
|---|---|
title | str |
type | str |
detail | str |
instance | str | None |
PreferencesSearchRequest
class PreferencesSearchRequest(BaseModel)Document representing a search request for retrieving items within the Data API. The query is a potentially complex set of expressions. The fields and expands that each query supports are defined within the search resource.
Fields
| Name | Type |
|---|---|
limit | int | None |
query | Query |
sorts | list[Sort] | None |
offset | int | None |
PriceTableItem
class PriceTableItem(BaseModel)Details of the published price table (only available if a price table was published)
Fields
| Name | Type |
|---|---|
priceTableId | str |
ProductItem
class ProductItem(BaseModel)Details of the published product (only available if a product was published)
Fields
| Name | Type |
|---|---|
productId | str |
PropfindEntry
class PropfindEntryA single entry returned by a PROPFIND (directory listing).
Fields
| Name | Type | Default |
|---|---|---|
href | str | |
is_collection | bool | |
display_name | str | None | None |
content_length | int | None | None |
last_modified | datetime | None | None |
content_type | str | None | None |
PublishIdResponse
class PublishIdResponse(BaseModel)Item successfully queued for publishing
Fields
| Name | Type |
|---|---|
id | str |
PublishProcessListResponse
class PublishProcessListResponse(ResultBase)Paginated list of publish processes
Fields
| Name | Type |
|---|---|
data | list[PublishProcessResponse] |
offset | int |
PublishProcessResponse
class PublishProcessResponse(BaseModel)Publish process details
Fields
| Name | Type | Default |
|---|---|---|
id | str | |
status | Status | |
startTime | AwareDatetime | |
endTime | AwareDatetime | None | |
initiatedBy | str | |
productItem | ProductItem | None | None |
priceTableItem | PriceTableItem | None | None |
contentAssetItem | ContentAssetItemPrivate | ContentAssetItemShared | None | None |
ResolveBackendOptions
class ResolveBackendOptionsInputs to resolve_scapi_or_ocapi.
Fields
| Name | Type |
|---|---|
preference | ApiBackendPreference |
has_scapi_config | bool |
domain_name | str |
Role
class Role(RootModel[str])Fields
| Name | Type | Default |
|---|---|---|
root | str | Field(..., max_length=256) |
RoleMapping
class RoleMappingRole mapping built from the Account Manager roles API.
Maps between role id (e.g. bm-admin) and roleEnumName (e.g. ECOM_ADMIN).
Fields
| Name | Type |
|---|---|
by_id | dict[str, str] |
by_enum_name | dict[str, str] |
descriptions | dict[str, str] |
RolePermissions
class RolePermissions(BaseModel)Fields
| Name | Type | Default |
|---|---|---|
module | RoleModulePermissions | None | None |
functional | RoleFunctionalPermissions | None | None |
locale | RoleLocalePermissions | None | None |
webdav | RoleWebdavPermissions | None | None |
RoleSearch
class RoleSearch(PaginatedResultBase)Fields
| Name | Type |
|---|---|
data | list[Datum] |
ScapiBackendCtorConfig
class ScapiBackendCtorConfigConfiguration passed to a SCAPI backend constructor.
Domains may read additional data off instance (log/WebDAV access) but always receive short_code + tenant_id + auth.
Fields
| Name | Type |
|---|---|
short_code | str |
tenant_id | str |
auth | AuthStrategy |
instance | B2CInstance |
ScapiCapabilityUnsupportedError
class ScapiCapabilityUnsupportedError(Exception)Raised when a requested operation cannot be expressed on SCAPI.
For example, toggling the disabled flag via the SCAPI Users PATCH, which the SCAPI schema does not include. The fallback wrapper recognizes this and falls back to OCAPI; in explicit scapi mode it propagates so the caller sees the limitation.
ScapiCatalog
class ScapiCatalog(BaseModel)Fields
| Name | Type | Default |
|---|---|---|
id | str | |
name | dict[str, str] | None | None |
description | dict[str, str] | None | None |
online | bool | None | None |
ScapiCatalogs
class ScapiCatalogs(BaseModel)Fields
| Name | Type |
|---|---|
data | list[Catalog] |
limit | int |
offset | int |
total | int |
ScapiClientConfig
class ScapiClientConfigCaller-supplied SCAPI coordinates and overrides.
Fields
| Name | Type | Default |
|---|---|---|
short_code | str | |
tenant_id | str | |
scopes | list[str] | None | None |
middleware_registry | MiddlewareRegistry | None | field(default=None) |
ScapiRequestError
class ScapiRequestError(Exception)A structured SCAPI response failure.
Backends must retain the response status so the shared fallback policy can distinguish a definite rejection from an ambiguous transport/server failure.
ScapiSchemasClientConfig
class ScapiSchemasClientConfigConfiguration for creating a SCAPI Schemas client.
Fields
| Name | Type | Default |
|---|---|---|
short_code | str | |
tenant_id | str | |
scopes | list[str] | None | None |
middleware_registry | MiddlewareRegistry | None | field(default=None) |
ScapiSchemasError
class ScapiSchemasError(BaseModel)Fields
| Name | Type |
|---|---|
title | str |
type | str |
detail | str |
instance | str | None |
ScapiSite
class ScapiSite(BaseModel)Fields
| Name | Type | Default |
|---|---|---|
id | str | Field(..., max_length=32, min_length=1) |
displayName | dict[str, DisplayName] | None | None |
description | dict[str, Description] | None | None |
customerListLink | CustomerListLink | None | None |
inDeletion | bool | None | None |
storefrontStatus | StorefrontStatus | None | None |
siteCatalogId | str | None | Field(None, max_length=256, min_length=1) |
cartridges | str | None | Field(None, max_length=4000) |
customCartridges | str | None | Field(None, max_length=4000) |
creationDate | AwareDatetime | None | None |
lastModified | AwareDatetime | None | None |
ScapiSiteSearchResult
class ScapiSiteSearchResult(PaginatedSearchResult)Fields
| Name | Type |
|---|---|
hits | list[Hit] |
query | Query |
ScapiSites
class ScapiSites(PaginatedResultBase)Fields
| Name | Type |
|---|---|
data | list[Datum] |
ScapiUserAuthUnsupportedError
class ScapiUserAuthUnsupportedError(Exception)Raised when browser-based Account Manager user auth is passed to a SCAPI Admin client.
SchemaListItem
class SchemaListItem(BaseModel)Fields
| Name | Type | Default |
|---|---|---|
schemaVersion | str | None | |
apiFamily | str | None | |
apiName | str | None | |
apiVersion | str | None | |
status | SchemaStatus | None | None |
link | str | None |
SchemaListResult
class SchemaListResult(ResultBase)Fields
| Name | Type | Default |
|---|---|---|
filter | SchemaListFilter | None | None |
data | list[SchemaListItem] | None | None |
ScopeCascade
class ScopeCascadeScope cascade for a SCAPI domain.
The SCAPI auth middleware picks read or write based on the per-operation scope-mode hint and walks the chosen cascade through the auth strategy until one candidate survives at Account Manager. Each candidate is a list of scopes; the auth strategy adds any base (e.g. tenant) scopes itself.
Fields
| Name | Type |
|---|---|
read | list[list[str]] |
write | list[list[str]] |
ScopeTierManager
class ScopeTierManager(Generic[C])Lazy-initialized manager for clients at different scope tiers.
- First read or write call builds the rw client and caches it.
- If the caller detects an
invalid_scopeerror on a read attempt, it callsdowngrade_to_read_onlyand the next read uses the read-only client. - Once downgraded, write requests raise — the API client lacks rw scope.
The same rw client serves both read and write while the rw scope is valid; a separate read-only client is only built after a downgrade.
Fields
| Name | Type | Description |
|---|---|---|
resolved_tier | ScopeTier | None | The currently-resolved tier, or None before first use. |
get_client_for_write method
def get_client_for_write() -> CReturn a client suitable for write operations.
Raises
ScapiCapabilityUnsupportedError— if we've already downgraded to read-only — the API client doesn't have the rw scope. Raising this (rather than a plain error) lets the SCAPI/OCAPI fallback wrapper recognize the capability gap and route the write through OCAPI inautomode.
get_client_for_read method
def get_client_for_read() -> CReturn a client suitable for read operations.
Prefers the rw client if it's already been used successfully (rw scope grants read too).
downgrade_to_read_only method
def downgrade_to_read_only() -> NoneMark the rw scope as unavailable and build a read-only client.
Subsequent get_client_for_write calls will raise; reads use the read-only client.
try_read method
async def try_read(fn: Callable[[C], Awaitable[R]]) -> RRun a read operation, downgrading and retrying once on invalid_scope.
Backends should wrap their reads with this so an API client provisioned with only the read-only scope (e.g. sfcc.scripts) can still read through SCAPI. Writes do not go through this helper — they always require rw, and get_client_for_write already raises after a downgrade.
SiteCustomCartridges
class SiteCustomCartridges(BaseModel)Fields
| Name | Type | Default |
|---|---|---|
customCartridges | str | Field(..., max_length=4000) |
SitePreferences
class SitePreferences(BaseModel)Represents custom preferences at the site level within a preference group. Custom preference attributes are returned with the "c_" prefix.
Fields
| Name | Type | Default |
|---|---|---|
site | Site | None | None |
SlasClientConfig
class SlasClientConfigConfiguration for creating a SLAS client.
Fields
| Name | Type | Default |
|---|---|---|
short_code | str | |
middleware_registry | MiddlewareRegistry | None | field(default=None) |
TlsOptions
class TlsOptionsTLS options for creating a transport.
Attributes
| Name | Type | Description |
|---|---|---|
certificate | str | None | Path to a PKCS12 (.p12/.pfx) certificate file. |
passphrase | str | None | Passphrase for the certificate, if encrypted. |
reject_unauthorized | bool | None | Whether to reject invalid/self-signed server certificates. False disables verification (self-signed mode). |
User
class User(BaseModel)Fields
| Name | Type | Default |
|---|---|---|
login | str | Field(..., max_length=256, min_length=1) |
password | str | None | Field(None, max_length=256) |
email | str | Field(..., max_length=256) |
firstName | str | None | Field(None, max_length=256) |
lastName | str | None | Field(None, max_length=256) |
externalId | str | None | Field(None, max_length=256) |
disabled | bool | None | None |
locked | bool | None | None |
lastLoginDate | date | None | None |
passwordExpirationDate | AwareDatetime | None | None |
passwordModificationDate | AwareDatetime | None | None |
preferredDataLocale | LanguageCountry | LanguageCode | DefaultFallback | None | None |
preferredUiLocale | LanguageCountry | LanguageCode | DefaultFallback | None | None |
roles | list[Role] | None | None |
UserSearch
class UserSearch(PaginatedResultBase)Fields
| Name | Type |
|---|---|
data | list[Datum] |
UserUpdateRequest
class UserUpdateRequest(BaseModel)Fields
| Name | Type | Default |
|---|---|---|
email | str | None | Field(None, max_length=256) |
firstName | str | None | Field(None, max_length=256) |
lastName | str | None | Field(None, max_length=256) |
externalId | str | None | Field(None, max_length=256) |
preferredDataLocale | LanguageCountry | LanguageCode | DefaultFallback | None | None |
preferredUiLocale | LanguageCountry | LanguageCode | DefaultFallback | None | None |
WebDavClient
class WebDavClientWebDAV client for B2C Commerce instance file operations.
Parameters
| Name | Type | Description |
|---|---|---|
hostname | str | WebDAV hostname (may differ from the API hostname). |
auth | AuthStrategy | Authentication strategy used for requests (its fetch injects credentials and handles TLS/mTLS). |
middleware_registry | MiddlewareRegistry | None | Registry supplying webdav middleware (defaults to the global registry). |
transport | httpx.AsyncBaseTransport | None | Optional TLS/mTLS transport passed through to auth.fetch. |
build_url method
def build_url(path: str) -> strBuild the full URL for a WebDAV path (relative to /webdav/Sites/).
request method
async def request(path: str, *, method: str = 'GET', headers: dict[str, str] | None = None, content: bytes | str | None = None) -> httpx.ResponseMake a raw WebDAV request, applying webdav middleware and auth.
Raises
NetworkError— on transport-level failures.
mkcol method
async def mkcol(path: str) -> NoneCreate a directory (collection). Tolerates 405 (already exists).
put method
async def put(path: str, content: bytes | str, content_type: str | None = None) -> NoneUpload a file to path.
get method
async def get(path: str) -> bytesDownload a file, returning its content as bytes.
delete method
async def delete(path: str) -> NoneDelete a file or directory.
propfind method
async def propfind(path: str, depth: str = '1') -> list[PropfindEntry]List directory contents via PROPFIND (depth is "0", "1", or "infinity").
copy method
async def copy(source: str, destination: str, overwrite: bool = True) -> NoneCopy a file or directory from source to destination.
move method
async def move(source: str, destination: str, overwrite: bool = True) -> NoneMove (rename) a file or directory from source to destination.
exists method
async def exists(path: str) -> boolReturn True if path exists (via a HEAD request).
Zone
class Zone(RootModel[str])Fields
| Name | Type |
|---|---|
root | str |
ZonesEnvelope
class ZonesEnvelope(BaseModel)Fields
| Name | Type |
|---|---|
data | list[Zone2] |
Functions
assert_ocapi_compatibility_allowed
def assert_ocapi_compatibility_allowed(preference: ApiBackendPreference | None, capability: str) -> NonePrevent an OCAPI-only compatibility operation from running under explicit SCAPI.
auto remains eligible for the temporary compatibility path, while explicit OCAPI is always allowed.
assert_scapi_admin_auth_supported
def assert_scapi_admin_auth_supported(auth: AuthStrategy) -> NoneReject browser user-auth strategies before a SCAPI request is attempted.
build_scapi_client
def build_scapi_client(options: BuildScapiClientOptions, config: ScapiClientConfig, auth: AuthStrategy) -> HttpClientBuild a typed HTTP client for a SCAPI Admin API.
Parameters
| Name | Type | Description |
|---|---|---|
options | BuildScapiClientOptions | Domain-specific URL/key/scopes/log-prefix. |
config | ScapiClientConfig | Caller-supplied short code, tenant ID, optional overrides. |
auth | AuthStrategy | Auth strategy (scopes are merged via with_scopes). |
Raises
ValueError— if neither or both ofscope_cascade/default_scopesare provided.
build_tenant_scope
def build_tenant_scope(tenant_id: str) -> strBuild the tenant-specific OAuth scope required for SCAPI APIs.
>>> build_tenant_scope("zzxy_prd")
'SALESFORCE_COMMERCE_API:zzxy_prd'
>>> build_tenant_scope("f_ecom_zzxy_prd")
'SALESFORCE_COMMERCE_API:zzxy_prd'change_api_client_password
async def change_api_client_password(client: AccountManagerApiClientsClient, api_client_id: str, old_password: str, new_password: str) -> NoneChange the password for an API client.
create_account_manager_api_clients_client
def create_account_manager_api_clients_client(config: AccountManagerClientConfig, auth: AuthStrategy) -> AccountManagerApiClientsClientCreate a typed Account Manager API Clients API client.
Parameters
| Name | Type | Description |
|---|---|---|
config | AccountManagerClientConfig | Account Manager client configuration. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
Returns: A configured HttpClient.
create_account_manager_client
def create_account_manager_client(config: AccountManagerClientConfig, auth: AuthStrategy) -> AccountManagerClientCreate a unified Account Manager API client (users, roles, API clients, orgs).
Parameters
| Name | Type | Description |
|---|---|---|
config | AccountManagerClientConfig | Account Manager client configuration. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
Returns: A unified AccountManagerClient.
create_account_manager_orgs_client
def create_account_manager_orgs_client(config: AccountManagerClientConfig, auth: AuthStrategy) -> AccountManagerOrgsClientCreate an Account Manager Organizations API client.
Parameters
| Name | Type | Description |
|---|---|---|
config | AccountManagerClientConfig | Account Manager Organizations client configuration. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
Returns: An AccountManagerOrgsClient.
create_account_manager_roles_client
def create_account_manager_roles_client(config: AccountManagerClientConfig, auth: AuthStrategy) -> AccountManagerRolesClientCreate a typed Account Manager Roles API client.
Parameters
| Name | Type | Description |
|---|---|---|
config | AccountManagerClientConfig | Account Manager Roles client configuration. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
Returns: A configured HttpClient.
create_account_manager_users_client
def create_account_manager_users_client(config: AccountManagerClientConfig, auth: AuthStrategy) -> AccountManagerUsersClientCreate a typed Account Manager Users API client.
Parameters
| Name | Type | Description |
|---|---|---|
config | AccountManagerClientConfig | Account Manager Users client configuration. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
Returns: A configured HttpClient.
create_api_client
async def create_api_client(client: AccountManagerApiClientsClient, body: APIClientCreate) -> AccountManagerApiClientCreate a new API client.
Omits active when False so the API uses its default (inactive); some implementations reject or mishandle explicit active: false and return "invalid argument APIClient".
create_auth_middleware
def create_auth_middleware(auth: Any) -> _AuthMiddlewareCreate authentication middleware that injects the auth header and retries on 401.
On a 401 following a prior successful response (indicating token expiry rather than bad credentials), it invalidates the token and retries the request once with a fresh token. Requires the strategy to implement get_authorization_header and invalidate_token.
create_cdn_zones_client
def create_cdn_zones_client(config: CdnZonesClientConfig, auth: AuthStrategy, options: CdnZonesClientOptions | None = None) -> CdnZonesClientCreate a typed CDN Zones API client.
The client automatically handles OAuth scope requirements:
- Domain scope:
sfcc.cdn-zones(read) orsfcc.cdn-zones.rw(read-write) - Tenant scope:
SALESFORCE_COMMERCE_API:{tenant_id}
Parameters
| Name | Type | Description |
|---|---|---|
config | CdnZonesClientConfig | CDN Zones client configuration including short code and tenant ID. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
options | CdnZonesClientOptions | None | Optional settings such as read_write. |
Returns: A configured HttpClient.
create_custom_apis_client
def create_custom_apis_client(config: CustomApisClientConfig, auth: AuthStrategy) -> CustomApisClientCreate a typed Custom APIs DX API client.
The client automatically handles OAuth scope requirements:
- Domain scope:
sfcc.custom-apis(or custom viaconfig.scopes) - Tenant scope:
SALESFORCE_COMMERCE_API:{tenant_id}
Parameters
| Name | Type | Description |
|---|---|---|
config | CustomApisClientConfig | Custom APIs client configuration including short code and tenant ID. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
Returns: A configured HttpClient.
create_dual_backend
def create_dual_backend(config: DualBackendConfig, ctors: DualBackendCtors[T]) -> TResolve a preference + config availability into a concrete backend instance.
Parameters
| Name | Type | Description |
|---|---|---|
config | DualBackendConfig | The instance + optional backend preference. |
ctors | DualBackendCtors[T] | Per-domain SCAPI/OCAPI constructors and domain name. |
Raises
ValueError— when explicit SCAPI is requested without SCAPI config.
create_extra_params_middleware
def create_extra_params_middleware(*, query: dict[str, Any] | None = None, body: dict[str, Any] | None = None, headers: dict[str, str] | None = None) -> _ExtraParamsMiddlewareCreate middleware that adds extra query params, body fields, and/or headers.
Useful for internal/power-user parameters not present in the typed schema.
create_fallback_backend
def create_fallback_backend(scapi: T, ocapi: T, domain_name: str) -> TCreate a fallback wrapper over scapi and ocapi backends.
The returned object presents the same interface as T. Method calls are intercepted: the first call tries SCAPI; on a safe fallback trigger it falls back to OCAPI and re-pins. The name property reflects the resolved backend.
Parameters
| Name | Type | Description |
|---|---|---|
scapi | T | Primary (SCAPI) backend implementation. |
ocapi | T | Fallback (OCAPI) backend implementation. |
domain_name | str | Used in fallback log messages, e.g. "jobs". |
Returns: A wrapper over scapi whose methods route through fallback logic.
create_granular_replications_client
def create_granular_replications_client(config: GranularReplicationsClientConfig, auth: AuthStrategy) -> GranularReplicationsClientCreate a Granular Replications API client for publishing individual items.
The Granular Replications API enables programmatic publishing of individual items (products, price tables, content assets) from staging to production environments.
Parameters
| Name | Type | Description |
|---|---|---|
config | GranularReplicationsClientConfig | Client configuration with short code and tenant ID. |
auth | AuthStrategy | OAuth authentication strategy. |
Returns: A configured HttpClient.
create_logging_middleware
def create_logging_middleware(config: str | dict[str, Any] | None = None) -> _LoggingMiddlewareCreate logging middleware. Pass a prefix string, or a config dict with prefix and mask_body_keys (top-level body keys masked in logs).
create_metrics_client
def create_metrics_client(config: MetricsClientConfig, auth: AuthStrategy) -> MetricsClientCreate a typed Metrics API client.
The client automatically handles OAuth scope requirements:
- Domain scope:
sfcc.metrics(or custom viaconfig.scopes) - Tenant scope:
SALESFORCE_COMMERCE_API:{tenant_id}
Parameters
| Name | Type | Description |
|---|---|---|
config | MetricsClientConfig | Metrics client configuration including short code and tenant ID. |
auth | AuthStrategy | Authentication strategy (typically OAuth client-credentials). |
Returns: A configured HttpClient.
create_mrt_b2c_client
def create_mrt_b2c_client(config: MrtB2CClientConfig, auth: AuthStrategy) -> MrtB2CClientCreate a typed Managed Runtime B2C Commerce API client.
This client handles the B2C Commerce integration endpoints, which manage the connection between MRT targets/environments and B2C Commerce instances. Authentication is handled via the auth middleware (typically backed by an API-key strategy) rather than the SCAPI OAuth scope cascade.
Parameters
| Name | Type | Description |
|---|---|---|
config | MrtB2CClientConfig | MRT B2C client configuration. |
auth | AuthStrategy | Authentication strategy (typically an API-key strategy). |
Returns: A configured HttpClient.
create_mrt_client
def create_mrt_client(config: MrtClientConfig, auth: AuthStrategy) -> MrtClientCreate a typed Managed Runtime API client.
Authentication is handled via the auth middleware (typically backed by an API-key strategy) rather than the SCAPI OAuth scope cascade.
Parameters
| Name | Type | Description |
|---|---|---|
config | MrtClientConfig | MRT client configuration. |
auth | AuthStrategy | Authentication strategy (typically an API-key strategy). |
Returns: A configured HttpClient.
create_ocapi_client
def create_ocapi_client(hostname: str, auth: AuthStrategy, options: dict[str, Any] | str | None = None, *, transport: Any = None) -> OcapiClientCreate an OCAPI Data API client.
Parameters
| Name | Type | Description |
|---|---|---|
hostname | str | B2C instance hostname. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
options | dict[str, Any] | str | None | Optional {"api_version": ..., "middleware_registry": ...} dict, or a bare string for the API version (backwards compatibility). |
transport | Any | Optional TLS/mTLS transport for the underlying client. |
Returns: A configured HttpClient for OCAPI.
create_ods_client
def create_ods_client(config: OdsClientConfig, auth: AuthStrategy) -> OdsClientCreate a typed ODS (On-Demand Sandbox) API client.
Parameters
| Name | Type | Description |
|---|---|---|
config | OdsClientConfig | ODS client configuration. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
Returns: A configured HttpClient.
create_preferences_client
def create_preferences_client(config: PreferencesClientConfig, auth: AuthStrategy, options: PreferencesClientOptions | None = None) -> PreferencesClientCreate a typed Preferences API client.
Authentication is handled by middleware. The client automatically attaches:
- Domain scope:
sfcc.preferences(read) orsfcc.preferences.rw(read-write) - Tenant scope:
SALESFORCE_COMMERCE_API:{tenant_id}
Parameters
| Name | Type | Description |
|---|---|---|
config | PreferencesClientConfig | Preferences client configuration including short code and tenant ID. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
options | PreferencesClientOptions | None | Optional settings such as read_write. |
Returns: A configured HttpClient.
create_rate_limit_middleware
def create_rate_limit_middleware(*, max_retries: int = DEFAULT_RATE_LIMIT_MAX_RETRIES, base_delay_ms: float = DEFAULT_RATE_LIMIT_BASE_DELAY_MS, max_delay_ms: float = DEFAULT_RATE_LIMIT_MAX_DELAY_MS, status_codes: list[int] | None = None, prefix: str | None = None) -> _RateLimitMiddlewareCreate rate-limiting middleware for the typed clients.
Inspects responses for rate-limit status codes (default [429]), uses the Retry-After header when present, otherwise exponential backoff with jitter (base 1s, max 30s), and retries up to max_retries times.
create_scapi_auth_middleware
def create_scapi_auth_middleware(auth: Any, cascade: ScopeCascade) -> _ScapiAuthMiddlewareCreate SCAPI auth middleware that resolves a ScopeCascade per request.
Reads SCOPE_MODE_HEADER, picks the matching cascade, and asks the auth strategy to resolve it (cache-first, then AM with invalid_scope fallback). Strips the header before the request is sent. Falls back to get_authorization_header when the strategy lacks get_access_token_for_cascade or no scope-mode header was supplied. 401 retry matches create_auth_middleware.
create_scapi_catalogs_client
def create_scapi_catalogs_client(config: ScapiCatalogsClientConfig, auth: AuthStrategy) -> ScapiCatalogsClientCreate a typed SCAPI Catalogs Admin API client.
Parameters
| Name | Type | Description |
|---|---|---|
config | ScapiCatalogsClientConfig | SCAPI client configuration including short code and tenant ID. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
Returns: A configured HttpClient.
create_scapi_jobs_client
def create_scapi_jobs_client(config: ScapiJobsClientConfig, auth: AuthStrategy) -> ScapiJobsClientCreate a typed SCAPI Jobs Admin API client.
Parameters
| Name | Type | Description |
|---|---|---|
config | ScapiJobsClientConfig | SCAPI client configuration including short code and tenant ID. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
Returns: A configured HttpClient.
create_scapi_merchant_roles_client
def create_scapi_merchant_roles_client(config: ScapiMerchantRolesClientConfig, auth: AuthStrategy) -> ScapiMerchantRolesClientCreate a typed SCAPI Merchant Roles Admin API client.
Parameters
| Name | Type | Description |
|---|---|---|
config | ScapiMerchantRolesClientConfig | SCAPI client configuration including short code and tenant ID. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
Returns: A configured HttpClient.
create_scapi_merchant_users_client
def create_scapi_merchant_users_client(config: ScapiMerchantUsersClientConfig, auth: AuthStrategy) -> ScapiMerchantUsersClientCreate a typed SCAPI Merchant Users Admin API client.
Parameters
| Name | Type | Description |
|---|---|---|
config | ScapiMerchantUsersClientConfig | SCAPI client configuration including short code and tenant ID. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
Returns: A configured HttpClient.
create_scapi_request_error
def create_scapi_request_error(error: object, response: Any, fallback_message: str) -> ScapiRequestErrorCreate a structured SCAPI error using the repository's common formatter.
Parameters
| Name | Type | Description |
|---|---|---|
response | Any | An object exposing status (and optionally status_text), e.g. an httpx.Response or a lightweight status holder. |
create_scapi_schemas_client
def create_scapi_schemas_client(config: ScapiSchemasClientConfig, auth: AuthStrategy) -> ScapiSchemasClientCreate a typed SCAPI Schemas API client.
The client automatically handles OAuth scope requirements:
- Domain scope:
sfcc.scapi-schemas(or custom viaconfig.scopes) - Tenant scope:
SALESFORCE_COMMERCE_API:{tenant_id}
Parameters
| Name | Type | Description |
|---|---|---|
config | ScapiSchemasClientConfig | SCAPI Schemas client configuration including short code and tenant ID. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
Returns: A configured HttpClient.
create_scapi_scripts_client
def create_scapi_scripts_client(config: ScapiScriptsClientConfig, auth: AuthStrategy) -> ScapiScriptsClientCreate a typed SCAPI Scripts DX API client.
Parameters
| Name | Type | Description |
|---|---|---|
config | ScapiScriptsClientConfig | SCAPI client configuration including short code and tenant ID. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
Returns: A configured HttpClient.
create_scapi_sites_client
def create_scapi_sites_client(config: ScapiSitesClientConfig, auth: AuthStrategy) -> ScapiSitesClientCreate a typed SCAPI Sites Admin API client.
Parameters
| Name | Type | Description |
|---|---|---|
config | ScapiSitesClientConfig | SCAPI client configuration including short code and tenant ID. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
Returns: A configured HttpClient.
create_slas_client
def create_slas_client(config: SlasClientConfig, auth: AuthStrategy) -> SlasClientCreate a typed SLAS Admin API client.
Parameters
| Name | Type | Description |
|---|---|---|
config | SlasClientConfig | SLAS client configuration. |
auth | AuthStrategy | Authentication strategy (typically OAuth). |
Returns: A configured HttpClient.
create_tls_transport
def create_tls_transport(options: TlsOptions) -> httpx.AsyncBaseTransport | NoneCreate an httpx transport with custom TLS options for mTLS / self-signed certs.
Returns None when no TLS customization is needed (no certificate and reject_unauthorized not explicitly False), so callers fall back to the default transport.
Raises
ValueError— on unreadable or invalid certificate files, or a bad passphrase (see_load_pkcs12).
create_user
async def create_user(client: AccountManagerUsersClient, user: UserCreate) -> AccountManagerUserCreate a new user. Raises if the request fails.
create_user_agent_middleware
def create_user_agent_middleware(user_agent: str) -> _UserAgentMiddlewareCreate middleware that sets User-Agent and sfdc_user_agent headers.
delete_api_client
async def delete_api_client(client: AccountManagerApiClientsClient, api_client_id: str) -> NoneDelete an API client. Only clients disabled for at least 7 days can be deleted.
delete_user
async def delete_user(client: AccountManagerUsersClient, user_id: str) -> NoneDisable a user (soft delete -- sets userState to DELETED).
Users must be disabled before they can be purged.
fetch_role_mapping
async def fetch_role_mapping(roles_client: AccountManagerRolesClient) -> RoleMappingFetch all roles and build a mapping between role id and roleEnumName.
find_user_by_login
async def find_user_by_login(client: AccountManagerUsersClient, login: str, expand: list[UserExpandOption] | None = None) -> AccountManagerUser | NoneFind a user by login (email) using the dedicated search endpoint.
Returns: The user if found, None if not found.
get_api_client
async def get_api_client(client: AccountManagerApiClientsClient, api_client_id: str, expand: list[ApiClientExpandOption] | None = None) -> AccountManagerApiClientRetrieve an API client by ID. Raises if not found (404) or the request fails.
get_api_error_message
def get_api_error_message(error: Any, response: StatusLike) -> strExtract a clean error message from an API error response.
Handles multiple API error patterns and falls back to HTTP status so HTML response bodies (like error pages) are never surfaced. Supported patterns:
- ODS/SLAS:
{"error": {"message": "..."}} - OCAPI:
{"fault": {"message": "..."}} - SCAPI/Problem+JSON:
{"detail": "...", "title": "..."} - Standard:
{"message": "..."}
get_role
async def get_role(client: AccountManagerRolesClient, role_id: str) -> AccountManagerRoleRetrieve details of a role by ID. Raises if the role is not found or the request fails.
get_user
async def get_user(client: AccountManagerUsersClient, user_id: str, expand: list[UserExpandOption] | None = None) -> AccountManagerUserRetrieve details of a user by ID. Raises if the user is not found or the request fails.
get_user_agent
def get_user_agent() -> strReturn the current User-Agent string.
is_fallback_trigger
def is_fallback_trigger(error: object) -> boolDetect whether an error should trigger an OCAPI fallback.
Currently:
is_invalid_scope_error: AM rejected the requested scope.ScapiCapabilityUnsupportedError: the SCAPI surface lacks the capability the caller asked for.ScapiRequestError: SCAPI definitively rejected the request with a safe client-error status.
is_invalid_scope_error
def is_invalid_scope_error(error: object) -> boolDetect an Account Manager invalid_scope error.
When a client's API client doesn't have the requested scope configured, Account Manager returns {"error":"invalid_scope", ...} on the token request. The OAuth strategy surfaces that as an exception whose message contains invalid_scope. Used by fallback wrappers to decide whether to downgrade to OCAPI.
is_ocapi_deprecated_fault
def is_ocapi_deprecated_fault(error: Any) -> boolReturn True if an API error object is an OCAPI deprecation fault.
Detection keys off the structured fault.type, not the message text, so it is robust to message wording changes.
is_valid_role_tenant_filter
def is_valid_role_tenant_filter(value: str) -> boolReturn True if value matches the Account Manager role tenant filter format.
Format: ROLE_ENUM:realm_instance(,realm_instance)*(;ROLE_ENUM:...)*. Examples: SALESFORCE_COMMERCE_API:abcd_prd or bm-admin:tenant1,tenant2;ECOM_USER:wxyz_stg.
list_api_clients
async def list_api_clients(client: AccountManagerApiClientsClient, options: ListApiClientsOptions | None = None) -> APIClientCollectionList API clients with pagination.
list_roles
async def list_roles(client: AccountManagerRolesClient, options: ListRolesOptions | None = None) -> RoleCollectionList roles with pagination. Raises if the request fails.
list_users
async def list_users(client: AccountManagerUsersClient, options: ListUsersOptions | None = None) -> UserCollectionList users with pagination. Raises if the request fails.
normalize_tenant_id
def normalize_tenant_id(value: str) -> strNormalize any parseable tenant/organization ID form to canonical underscore format.
Supported input forms (all resolve to abcd_123):
abcd_123— canonical tenant ID (returned as-is)abcd-123— hyphenated tenant IDf_ecom_abcd_123— organization IDf_ecom_abcd-123— org ID with hyphenated tenantabcd-123.dx.commercecloud.salesforce.com— sandbox hostname
>>> normalize_tenant_id("f_ecom_zzxy_prd")
'zzxy_prd'
>>> normalize_tenant_id("zzxy-prd")
'zzxy_prd'
>>> normalize_tenant_id("zzxy-prd.dx.commercecloud.salesforce.com")
'zzxy_prd'ocapi_deprecated_message
def ocapi_deprecated_message(required_scopes: list[str] | None = None) -> strBuild the user-facing guidance shown when an instance has OCAPI deprecated.
Pass the SCAPI scope(s) the failed operation requires to name them in the message. Omit required_scopes for operations that have no SCAPI equivalent — the message then uses the generic sfcc.* phrasing.
purge_user
async def purge_user(client: AccountManagerUsersClient, user_id: str) -> NonePurge a user (hard delete). Users must be in DELETED state before they can be purged.
reset_user
async def reset_user(client: AccountManagerUsersClient, user_id: str) -> NoneReset a user to INITIAL state and send activation instructions.
reset_user_agent
def reset_user_agent() -> NoneReset the User-Agent to the default SDK value (primarily for testing).
resolve_from_internal_role
def resolve_from_internal_role(role_enum_name: str, mapping: RoleMapping) -> strResolve an internal roleEnumName to its external role id using an API-fetched role mapping.
Falls back to a generic transform (lowercase + replace underscores with hyphens) for unknown roles.
resolve_scapi_or_ocapi
def resolve_scapi_or_ocapi(opts: ResolveBackendOptions) -> Literal['ocapi', 'scapi']Resolve a user preference + config availability into a concrete backend choice.
- Explicit
"ocapi"always returns"ocapi". - Explicit
"scapi"requires SCAPI config and raises if missing. "auto"returns"scapi"if SCAPI config is available, otherwise"ocapi".
Raises
ValueError— when explicit SCAPI is requested without the required configuration.
resolve_to_internal_role
def resolve_to_internal_role(role: str, mapping: RoleMapping) -> strResolve a role to its internal roleEnumName using an API-fetched role mapping.
Accepts either the role id (e.g. bm-admin) or roleEnumName (e.g. ECOM_ADMIN). Falls back to a generic transform (uppercase + replace hyphens with underscores) for unknown roles.
scapi_capability_unsupported_message
def scapi_capability_unsupported_message(capability: str) -> strBuild the canonical error message for a capability absent from live SCAPI schemas.
scapi_unavailable_message
def scapi_unavailable_message(domain_name: str) -> strMessage for when explicit SCAPI is requested but the instance can't reach it.
Names both reasons the SCAPI client config can be unavailable — missing coordinates OR an auth flow that can't request scopes.
set_user_agent
def set_user_agent(user_agent: str) -> NoneSet the User-Agent string used for all HTTP + auth requests.
Call early in an application to override the default SDK User-Agent (the CLI uses this to set a combined CLI+SDK User-Agent).
throw_ocapi_error
def throw_ocapi_error(error: Any, response: StatusLike, prefix: str, required_scopes: list[str] | None = None) -> NoneRaise a well-formed error for a failed OCAPI call.
OCAPI deprecation faults become an OcapiDeprecatedError with actionable SCAPI-setup guidance; everything else raises a plain RuntimeError of the form "{prefix}: {message}". Always raises.
to_organization_id
def to_organization_id(tenant_id: str) -> strEnsure a tenant ID has the f_ecom_ prefix for use as a SCAPI organizationId.
If the value already has the prefix, it's returned as-is.
>>> to_organization_id("zzxy_prd")
'f_ecom_zzxy_prd'
>>> to_organization_id("f_ecom_zzxy_prd")
'f_ecom_zzxy_prd'update_api_client
async def update_api_client(client: AccountManagerApiClientsClient, api_client_id: str, body: APIClientUpdate) -> AccountManagerApiClientUpdate an existing API client. Raises if the request fails or the body is invalid.
update_user
async def update_user(client: AccountManagerUsersClient, user_id: str, changes: UserUpdate) -> AccountManagerUserUpdate an existing user. Raises if the request fails.
with_scopes
def with_scopes(auth: AuthStrategy, additional_scopes: list[str]) -> AuthStrategyReturn a copy of auth with additional_scopes merged in.
Falls back to the original auth if the strategy doesn't support scope merging (e.g. basic/api-key auth, or a stored-session strategy where scopes were fixed at acquisition). Centralized so SCAPI client factories don't have to keep extending an isinstance chain as new OAuth strategy types are added.
Attributes
CDN_ZONES_READ_SCOPES
CDN_ZONES_READ_SCOPES = ['sfcc.cdn-zones']CDN_ZONES_RW_SCOPES
CDN_ZONES_RW_SCOPES = ['sfcc.cdn-zones.rw']CUSTOM_APIS_DEFAULT_SCOPES
CUSTOM_APIS_DEFAULT_SCOPES = ['sfcc.custom-apis']DEFAULT_API_VERSION
DEFAULT_API_VERSION = 'v25_6'DEFAULT_MRT_B2C_ORIGIN
DEFAULT_MRT_B2C_ORIGIN = 'https://cloud.mobify.com/api/cc/b2c'DEFAULT_MRT_ORIGIN
DEFAULT_MRT_ORIGIN = 'https://cloud.mobify.com'METRICS_DEFAULT_SCOPES
METRICS_DEFAULT_SCOPES = ['sfcc.metrics']OCAPI_DEPRECATED_MESSAGE
OCAPI_DEPRECATED_MESSAGE = ocapi_deprecated_message()ORGANIZATION_ID_PREFIX
ORGANIZATION_ID_PREFIX = 'f_ecom_'PREFERENCES_READ_SCOPES
PREFERENCES_READ_SCOPES = ['sfcc.preferences']PREFERENCES_RW_SCOPES
PREFERENCES_RW_SCOPES = ['sfcc.preferences.rw']ROLE_TENANT_FILTER_PATTERN
ROLE_TENANT_FILTER_PATTERN = re.compile('^(\\w+:\\w{4,}_\\w{3,}(,\\w{4,}_\\w{3,})*(;)?)*$', re.ASCII)SAFE_SCAPI_FALLBACK_STATUSES
SAFE_SCAPI_FALLBACK_STATUSES: frozenset[int] = frozenset({400, 401, 403, 404, 405, 406, 415})SCAPI_CAPABILITY_BASELINE_RELEASE
SCAPI_CAPABILITY_BASELINE_RELEASE = '26.8'SCAPI_CATALOGS_CASCADE
SCAPI_CATALOGS_CASCADE = ScopeCascade(read=[['sfcc.catalogs.rw'], ['sfcc.catalogs']], write=[['sfcc.catalogs.rw']])SCAPI_JOBS_CASCADE
SCAPI_JOBS_CASCADE = ScopeCascade(read=[['sfcc.jobs.rw'], ['sfcc.jobs']], write=[['sfcc.jobs.rw']])SCAPI_MERCHANT_ROLES_READ_SCOPES
SCAPI_MERCHANT_ROLES_READ_SCOPES = ['sfcc.roles']SCAPI_MERCHANT_ROLES_RW_SCOPES
SCAPI_MERCHANT_ROLES_RW_SCOPES = ['sfcc.roles.rw']SCAPI_MERCHANT_USERS_READ_SCOPES
SCAPI_MERCHANT_USERS_READ_SCOPES = ['sfcc.users']SCAPI_MERCHANT_USERS_RW_SCOPES
SCAPI_MERCHANT_USERS_RW_SCOPES = ['sfcc.users.rw']SCAPI_SCHEMAS_DEFAULT_SCOPES
SCAPI_SCHEMAS_DEFAULT_SCOPES = ['sfcc.scapi-schemas']SCAPI_SCRIPTS_READ_SCOPES
SCAPI_SCRIPTS_READ_SCOPES = ['sfcc.scripts']SCAPI_SCRIPTS_RW_SCOPES
SCAPI_SCRIPTS_RW_SCOPES = ['sfcc.scripts.rw']SCAPI_SITES_CASCADE
SCAPI_SITES_CASCADE = ScopeCascade(read=[['sfcc.sites.rw'], ['sfcc.sites']], write=[['sfcc.sites.rw']])SCAPI_TENANT_SCOPE_PREFIX
SCAPI_TENANT_SCOPE_PREFIX = 'SALESFORCE_COMMERCE_API:'SCOPE_MODE_HEADER
SCOPE_MODE_HEADER = 'x-b2c-scope-mode'APIClientCollection
APIClientCollection = dict[str, Any]APIClientCreate
APIClientCreate = dict[str, Any]APIClientUpdate
APIClientUpdate = dict[str, Any]AccountManagerApiClient
AccountManagerApiClient = dict[str, Any]AccountManagerApiClientsClient
AccountManagerApiClientsClient = HttpClientAccountManagerOrganization
AccountManagerOrganization = dict[str, Any]AccountManagerRole
AccountManagerRole = dict[str, Any]AccountManagerRolesClient
AccountManagerRolesClient = HttpClientAccountManagerUser
AccountManagerUser = dict[str, Any]AccountManagerUsersClient
AccountManagerUsersClient = HttpClientApiBackendPreference
ApiBackendPreference = Literal['ocapi', 'scapi', 'auto']ApiClientExpandOption
ApiClientExpandOption = Literal['organizations', 'roles']CdnZonesClient
CdnZonesClient = HttpClientCustomApisClient
CustomApisClient = HttpClientGranularReplicationsClient
GranularReplicationsClient = HttpClientHttpClientType
HttpClientType = ...MetricsClient
MetricsClient = HttpClientMrtB2CClient
MrtB2CClient = HttpClientMrtClient
MrtClient = HttpClientOcapiClient
OcapiClient = HttpClientOdsClient
OdsClient = HttpClientOrganizationCollection
OrganizationCollection = dict[str, Any]PreferenceInstanceType
PreferenceInstanceType = Literal['staging', 'development', 'sandbox', 'production']PreferencesClient
PreferencesClient = HttpClientRoleCollection
RoleCollection = dict[str, Any]ScapiCatalogsClient
ScapiCatalogsClient = HttpClientScapiCatalogsClientConfig
ScapiCatalogsClientConfig = ScapiClientConfigScapiJobsClient
ScapiJobsClient = HttpClientScapiJobsClientConfig
ScapiJobsClientConfig = ScapiClientConfigScapiMerchantRolesClient
ScapiMerchantRolesClient = HttpClientScapiMerchantRolesClientConfig
ScapiMerchantRolesClientConfig = ScapiClientConfigScapiMerchantUsersClient
ScapiMerchantUsersClient = HttpClientScapiMerchantUsersClientConfig
ScapiMerchantUsersClientConfig = ScapiClientConfigScapiSchemasClient
ScapiSchemasClient = HttpClientScapiScriptsClient
ScapiScriptsClient = HttpClientScapiScriptsClientConfig
ScapiScriptsClientConfig = ScapiClientConfigScapiSitesClient
ScapiSitesClient = HttpClientScapiSitesClientConfig
ScapiSitesClientConfig = ScapiClientConfigScopeTier
ScopeTier = Literal['rw', 'read-only']SlasClient
SlasClient = HttpClientUnifiedMiddleware
UnifiedMiddleware = MiddlewareUserCollection
UserCollection = dict[str, Any]UserCreate
UserCreate = dict[str, Any]UserExpandOption
UserExpandOption = Literal['organizations', 'roles']UserState
UserState = Literal['INITIAL', 'ENABLED', 'DELETED']UserUpdate
UserUpdate = dict[str, Any]global_middleware_registry
global_middleware_registry = MiddlewareRegistry()user_agent_provider
user_agent_provider = _UserAgentProvider()