b2c_tooling_sdk.operations.metrics
Metrics operations for B2C Commerce (Observability).
Mirrors src/operations/metrics/index.ts. Provides typed, high-level functions for retrieving operational time-series metrics from the SCAPI Observability Metrics API (observability/metrics/v1). Each metric category has its own function; all return the same metrics-data-response envelope (as raw JSON dict).
Categories:
get_overall_metrics— overall application metricsget_sales_metrics— sales metricsget_ecdn_metrics— embedded CDN metricsget_third_party_metrics— third-party service call metricsget_scapi_metrics— SCAPI request metricsget_scapi_hooks_metrics— SCAPI hook execution metricsget_mrt_metrics— Managed Runtime metricsget_controller_metrics— controller/pipeline metricsget_ocapi_metrics— OCAPI request metrics
get_metrics_by_category dispatches to the correct function by category name.
The Metrics API wire format is epoch seconds; these operations convert millisecond inputs to seconds on the way out and normalize response timestamps back to milliseconds on the way in.
Authentication requires OAuth client-credentials with the sfcc.metrics admin scope, attached automatically by create_metrics_client.
Classes
MetricsQueryOptions
class MetricsQueryOptions(MetricsTimeWindow)Union of every option shape accepted by get_metrics_by_category.
Fields
| Name | Type | Default |
|---|---|---|
third_party_service_id | str | None | None |
api_family | str | None | None |
api_name | str | None | None |
ocapi_category | str | None | None |
ocapi_api | str | None | None |
MetricsTagContext
class MetricsTagContextThe request identity and applied filters used to derive authoritative tags.
realm/environment are parsed from tenant_id. The optional filter fields mirror the Metrics API's category filters; when a filter was sent, that dimension is known from the request and is stamped onto every series as an authoritative tag — rather than being (mis)parsed from a drilled-down series id.
Fields
| Name | Type | Default |
|---|---|---|
tenant_id | str | |
api_family | str | None | None |
api_name | str | None | None |
ocapi_category | str | None | None |
ocapi_api | str | None | None |
third_party_service_id | str | None | None |
MetricsTimeWindow
class MetricsTimeWindowTime-window options common to every metrics operation.
Accepts either a datetime or a number of epoch milliseconds (the same unit as Date.now()) for both from_ and to. Both optional; when omitted the API applies its default window.
Fields
| Name | Type | Default |
|---|---|---|
from_ | datetime | int | float | None | None |
to | datetime | int | float | None | None |
MetricsWindowInput
class MetricsWindowInputRaw from/to/window inputs before resolution. Any subset may be provided.
Fields
| Name | Type | Default |
|---|---|---|
from_ | MetricsBoundInput | None | None |
to | MetricsBoundInput | None | None |
window | int | str | None | None |
OcapiMetricsOptions
class OcapiMetricsOptions(MetricsTimeWindow)Options for get_ocapi_metrics: time window plus optional OCAPI filters.
Fields
| Name | Type | Default |
|---|---|---|
ocapi_category | str | None | None |
ocapi_api | str | None | None |
ResolvedMetricsWindow
class ResolvedMetricsWindowA resolved metrics window. Both from_ and to are always present.
Fields
| Name | Type |
|---|---|
from_ | datetime |
to | datetime |
from_iso | str |
to_iso | str |
from_epoch_seconds | int |
to_epoch_seconds | int |
clamped_from | bool |
defaulted_window | bool |
ScapiMetricsOptions
class ScapiMetricsOptions(MetricsTimeWindow)Options for get_scapi_metrics: time window plus optional SCAPI filters.
Fields
| Name | Type | Default |
|---|---|---|
api_family | str | None | None |
api_name | str | None | None |
ThirdPartyMetricsOptions
class ThirdPartyMetricsOptions(MetricsTimeWindow)Options for get_third_party_metrics: time window plus a service filter.
Fields
| Name | Type | Default |
|---|---|---|
third_party_service_id | str | None | None |
Functions
enrich_metrics_tags
def enrich_metrics_tags(response: dict[str, Any], category: MetricCategory, context: MetricsTagContext) -> MetricsTaggedResponseEnrich a metrics response by adding a structured tags map to every series.
Returns a new response (the input is not mutated). Walks every metric and series and attaches series["tags"] derived from the series id, the metric id, the category, and the request context. Existing fields (id, name, data) are preserved exactly, so the enriched response is a structural superset. The category must be supplied because a metrics response does not carry it (it is implied by the endpoint that produced the response).
get_controller_metrics
async def get_controller_metrics(client: MetricsClient, tenant_id: str, options: MetricsTimeWindow | None = None) -> dict[str, Any]Retrieve controller/pipeline metrics for an organization.
Raises
RuntimeError— if the request fails.
get_ecdn_metrics
async def get_ecdn_metrics(client: MetricsClient, tenant_id: str, options: MetricsTimeWindow | None = None) -> dict[str, Any]Retrieve embedded CDN (eCDN) metrics for an organization.
Raises
RuntimeError— if the request fails.
get_metrics_by_category
async def get_metrics_by_category(client: MetricsClient, tenant_id: str, category: MetricCategory, options: MetricsQueryOptions | None = None) -> dict[str, Any]Retrieve metrics for a category by name, dispatching to the category-specific function.
Category-specific filters are applied only for the categories that support them and ignored otherwise.
Raises
RuntimeError— if the request fails.ValueError— if the category is unknown.
get_mrt_metrics
async def get_mrt_metrics(client: MetricsClient, tenant_id: str, options: MetricsTimeWindow | None = None) -> dict[str, Any]Retrieve Managed Runtime (MRT) metrics for an organization.
Raises
RuntimeError— if the request fails.
get_ocapi_metrics
async def get_ocapi_metrics(client: MetricsClient, tenant_id: str, options: OcapiMetricsOptions | None = None) -> dict[str, Any]Retrieve OCAPI request metrics for an organization.
Raises
RuntimeError— if the request fails.
get_overall_metrics
async def get_overall_metrics(client: MetricsClient, tenant_id: str, options: MetricsTimeWindow | None = None) -> dict[str, Any]Retrieve overall application metrics for an organization.
Raises
RuntimeError— if the request fails.
get_sales_metrics
async def get_sales_metrics(client: MetricsClient, tenant_id: str, options: MetricsTimeWindow | None = None) -> dict[str, Any]Retrieve sales metrics for an organization.
Raises
RuntimeError— if the request fails.
get_scapi_hooks_metrics
async def get_scapi_hooks_metrics(client: MetricsClient, tenant_id: str, options: MetricsTimeWindow | None = None) -> dict[str, Any]Retrieve SCAPI hook execution metrics for an organization.
Raises
RuntimeError— if the request fails.
get_scapi_metrics
async def get_scapi_metrics(client: MetricsClient, tenant_id: str, options: ScapiMetricsOptions | None = None) -> dict[str, Any]Retrieve SCAPI request metrics for an organization.
Raises
RuntimeError— if the request fails.
get_third_party_metrics
async def get_third_party_metrics(client: MetricsClient, tenant_id: str, options: ThirdPartyMetricsOptions | None = None) -> dict[str, Any]Retrieve third-party service call metrics for an organization.
Raises
RuntimeError— if the request fails.
parse_metrics_bound
def parse_metrics_bound(value: MetricsBoundInput, now: datetime | None = None) -> datetimeParse a single metrics time bound (from or to) into a datetime.
Resolves a single bound in isolation; deriving the companion bound and applying the 24-hour default window is the job of resolve_metrics_window.
Parameters
| Name | Type | Description |
|---|---|---|
value | MetricsBoundInput | The bound: a datetime, epoch milliseconds, a relative duration (5m/1h/2d, relative to now), or an ISO 8601 timestamp. |
now | datetime | None | Reference time for relative durations (defaults to the current time; injectable for deterministic tests). |
Raises
ValueError— if a string value is neither a valid relative duration nor a parseable ISO 8601 timestamp.
parse_series_tags
def parse_series_tags(*, category: MetricCategory, metric_id: str, series_id: str, context: MetricsTagContext) -> MetricSeriesTagsExtract the dimension tags for a single series id.
Combines three tiers, most-authoritative last:
- Request identity —
realm/environmentfrom the tenant id. - String heuristics — category/metric-specific dimensions parsed from the packed id, or the raw remainder under
serieswhen no rule matches. - Applied filters — any filter that was sent with the request (
MetricsTagContext) is stamped last, overriding a heuristic guess.
The result is always a superset of the request context and never throws.
resolve_metrics_window
def resolve_metrics_window(input: MetricsWindowInput | None = None, now: datetime | None = None) -> ResolvedMetricsWindowResolve from/to/window inputs into concrete bounds for the Metrics API.
Always produces an explicit from_+to pair. Resolution rules:
from+to— used as given;windowmust NOT also be set.from+window—to = from + window.to+window—from = to - window.windowonly — the lastwindow:to = now,from = now - window.fromonly — a 24-hour window forward:to = min(from + 24h, now).toonly — a 24-hour window back:from = to - 24h.- nothing — the last 24 hours.
A from at or beyond the 30-day retention floor is clamped forward to now - 30 days + margin so requests like --from 30d are not rejected by the server's slightly-later clock.
Parameters
| Name | Type | Description |
|---|---|---|
input | MetricsWindowInput | None | The raw from/to/window inputs. |
now | datetime | None | Reference time for relative bounds, default/window modes, and the retention clamp (defaults to the current time; injectable for tests). |
Raises
ValueError— if a bound or the window string is unparseable, if all three are supplied, or if the resolvedfromis afterto.
Attributes
METRICS_DEFAULT_WINDOW_MS
METRICS_DEFAULT_WINDOW_MS = 24 * 60 * 60 * 1000METRICS_RETENTION_MS
METRICS_RETENTION_MS = 30 * 24 * 60 * 60 * 1000METRICS_RETENTION_SAFETY_MARGIN_MS
METRICS_RETENTION_SAFETY_MARGIN_MS = 5 * 60 * 1000METRIC_CATEGORIES
METRIC_CATEGORIES: tuple[str, ...] = ('overall', 'sales', 'ecdn', 'third-party', 'scapi', 'scapi-hooks', 'mrt', 'controller', 'ocapi')MetricCategory
MetricCategory = Literal['overall', 'sales', 'ecdn', 'third-party', 'scapi', 'scapi-hooks', 'mrt', 'controller', 'ocapi']MetricSeriesTags
MetricSeriesTags = dict[str, str]MetricsBoundInput
MetricsBoundInput = datetime | int | float | strMetricsTaggedResponse
MetricsTaggedResponse = dict[str, Any]