Guides
Guides
Start typing to search…

Python SDK Ingest API Reference

Overview

This document provides detailed reference information for the client.ingest resource in the Pay-i Python SDK. For usage examples and best practices, see the Python SDK Ingest Guide.

Method Reference

MethodDescriptionReturn TypeAccess PatternREST API Endpoint
units()Submit a single event with usage dataIngestResponseresponse.xproxy_result.fieldREST Ingest API

Note: For bulk event ingestion options, please contact [email protected].

Response Types

IngestResponse

Returned by the units() method.

IngestResponse:
    event_timestamp: datetime     # Timestamp of the event being tracked
    ingest_timestamp: datetime    # Timestamp when Pay-i received/processed the event
    request_id: str               # Unique identifier for the ingest request
    xproxy_result: XproxyResult   # Detailed result information

XProxyResult

Contains detailed information about cost, limit status, and other metadata.

XProxyResult:
    blocked_limit_ids: Optional[List[str]]              # IDs of limits that blocked the request
    cost: Optional[Cost]                                # Cost breakdown for the request
    experience_id: Optional[str]                        # DEPRECATED: Associated experience ID
    limits: Optional[Dict[str, Limits]]                 # Status of affected limits
    request_id: Optional[str]                           # Unique ID for the request
    request_tags: Optional[List[str]]                   # Tags associated with the request
    resource_id: Optional[str]                          # Resource identifier
    unknown_units: Optional[Dict[str, IngestUnits]]     # Any units that couldn't be processed
    use_case_id: Optional[str]                          # Associated use case ID
    use_case_step: Optional[str]                        # Step within the use case
    user_id: Optional[str]                              # Associated user ID

Cost

Provides a breakdown of costs for the request.

Cost:
    currency: Optional[Literal["usd"]]                                 # Currency (currently only USD)
    input: Optional[PayICommonModelsBudgetManagementCostDetailsBase]   # Input cost details
    output: Optional[PayICommonModelsBudgetManagementCostDetailsBase]  # Output cost details
    total: Optional[CostDetails]                                       # Total cost information

CostDetails

Provides detailed cost information.

CostDetails:
    base: float                           # Base cost of the request
    overage_base: Optional[float]         # Additional cost from exceeding quota
    overrun_base: Optional[float]         # Additional cost from overrun

Limits (LimitStatus)

Tracks the status of any limits applied to the request.

Limits:
    state: Optional[Literal["ok", "blocked", "blocked_external", "exceeded", "overrun", "failed"]]

The possible states are:

  • ok: The limit was checked and not exceeded
  • blocked: The request was blocked due to this limit
  • blocked_external: The request was blocked by an external system
  • exceeded: The limit was exceeded but the request was allowed
  • overrun: The limit was overrun (beyond the exceeded threshold)
  • failed: There was an error checking the limit

IngestUnits

Represents the units of usage being tracked.

IngestUnits:
    input: Optional[int]    # Number of input units (e.g., tokens)
    output: Optional[int]   # Number of output units (e.g., tokens)

Method Details

units()

client.ingest.units(
    category: str,                        # Required
    resource: str,                        # Required
    units: Dict[str, IngestUnits],        # Required
    end_to_end_latency_ms: Optional[int] = None,
    event_timestamp: Union[str, datetime, None] = None,
    experience_properties: Optional[Dict[str, str]] = None,
    http_status_code: Optional[int] = None,
    properties: Optional[Dict[str, str]] = None,
    provider_request_headers: Optional[Iterable[PayICommonModelsAPIRouterHeaderInfoParam]] = None,
    provider_request_json: Optional[str] = None,
    provider_request_reasoning_json: Optional[str] = None,
    provider_response_function_calls: Optional[Iterable[ProviderResponseFunctionCall]] = None,
    provider_response_headers: Optional[Iterable[PayICommonModelsAPIRouterHeaderInfoParam]] = None,
    provider_response_id: Optional[str] = None,
    provider_response_json: Union[str, List[str], None] = None,
    provider_uri: Optional[str] = None,
    time_to_first_completion_token_ms: Optional[int] = None,
    time_to_first_token_ms: Optional[int] = None,
    use_case_properties: Optional[Dict[str, str]] = None,
    limit_ids: Optional[list[str]] = None,
    request_tags: Optional[list[str]] = None,
    experience_id: Optional[str] = None,                    # DEPRECATED
    experience_name: Optional[str] = None,                  # DEPRECATED
    use_case_id: Optional[str] = None,
    use_case_name: Optional[str] = None,
    use_case_step: Optional[str] = None,
    use_case_version: Optional[int] = None,
    user_id: Optional[str] = None,
    resource_scope: Optional[str] = None,
    extra_headers: Dict[str, str] = None,
    extra_query: Dict[str, str] = None,
    extra_body: Dict[str, Any] = None,
    timeout: Optional[float] = None
) -> IngestResponse

Submits a single event with usage data to Pay-i.

Calls: REST Ingest API REST API endpoint

Parameters:

Required Parameters:

  • category (str): The category the resource belongs to (e.g., "system.openai")
  • resource (str): Specific resource being used (e.g., "gpt-4")
  • units (Dict[str, IngestUnits]): Dictionary containing unit types with input and output values (e.g., {"text": {"input": 100, "output": 200}})

Optional Parameters:

  • end_to_end_latency_ms (int): End-to-end latency in milliseconds
  • event_timestamp (str|datetime|None): Timestamp of the event (ISO 8601 format if string)
  • experience_properties (Dict[str, str]): DEPRECATED: Properties for the experience
  • http_status_code (int): HTTP status code of the provider response
  • properties (Dict[str, str]): Key-value properties for the event
  • provider_request_headers (Iterable[PayICommonModelsAPIRouterHeaderInfoParam]): Headers sent to the provider
  • provider_request_json (str): JSON string of the request sent to the provider
  • provider_request_reasoning_json (str): JSON string of reasoning info sent to provider
  • provider_response_function_calls (Iterable[ProviderResponseFunctionCall]): Function calls in the response
  • provider_response_headers (Iterable[PayICommonModelsAPIRouterHeaderInfoParam]): Headers from the provider response
  • provider_response_id (str): Provider's ID for the response
  • provider_response_json (str|List[str]|None): JSON string or list of response from provider
  • provider_uri (str): URI of the provider endpoint
  • time_to_first_completion_token_ms (int): Time to first completion token in milliseconds
  • time_to_first_token_ms (int): Time to first token in milliseconds
  • use_case_properties (Dict[str, str]): Properties for the use case
  • limit_ids (list[str]): List of limit IDs to associate with the request
  • request_tags (list[str]): Tags to associate with the request
  • experience_id (str): DEPRECATED: ID of the experience (use use_case_id instead)
  • experience_name (str): DEPRECATED: Name of the experience (use use_case_name instead)
  • use_case_id (str): ID of the use case instance
  • use_case_name (str): Name of the use case
  • use_case_step (str): Current step within the use case
  • use_case_version (int): Version of the use case
  • user_id (str): ID of the user associated with the request
  • resource_scope (str): Scope of the resource

Additional Request Configuration:

  • extra_headers (Dict[str, str]): Additional headers for the request
  • extra_query (Dict[str, str]): Additional query parameters
  • extra_body (Dict[str, Any]): Additional body parameters
  • timeout (float): Request timeout in seconds

Returns: An IngestResponse object containing:

  • request_id: Unique identifier for the ingest request
  • event_timestamp: Timestamp of the event being tracked
  • ingest_timestamp: Timestamp when Pay-i received/processed the event
  • xproxy_result: Detailed information including cost, limits status, and request metadata

Example:

from payi import Payi
from payi.lib.helpers import PayiCategories

client = Payi()  # API key loaded from PAYI_API_KEY environment variable

response = client.ingest.units(
    category=PayiCategories.openai,
    resource="gpt-4o",
    units={
        "text": {
            "input": 156,    # Number of input tokens
            "output": 1746   # Number of output tokens
        }
    },
    user_id="user-123",
    request_tags=["customer-support", "documentation"],
    use_case_name="content-generation",
    use_case_step="draft-creation",
    properties={
        "model_version": "4o-20240502",
        "application": "content-management-system"
    }
)

# Check tracking status and cost information
print(f"Request ID: {response.request_id}")
print(f"Event timestamp: {response.event_timestamp}")
print(f"Base cost: ${response.xproxy_result.cost.total.base}")

Implementation Notes

The following implementation details may be helpful when using the Python SDK:

  1. The SDK automatically converts parameter lists like limit_ids and request_tags into comma-separated strings for header transmission.

  2. For timestamp handling, the event_timestamp parameter accepts:

    • A string in ISO 8601 format (e.g., "2024-04-15T14:22:36Z")
    • A Python datetime object
    • None (defaults to the current time in UTC)

  3. The units parameter requires a specific structure:

    • A dictionary with unit type keys (typically "text")
    • Each unit type maps to an object with "input" and/or "output" fields
    • Both fields should be integers representing the count (e.g., tokens)

  4. Experience-related parameters (experience_id, experience_name) are deprecated in favor of use case parameters (use_case_id, use_case_name). The SDK handles backward compatibility, but new code should use the use case parameters.

  5. Parameters can be passed as direct method arguments or through dictionary unpacking, which is useful for dynamic construction:

    params = {
    "category": PayiCategories.openai,
    "resource": "gpt-4o",
    "units": {"text": {"input": 100, "output": 200}},
    "user_id": "user-123"
}
response = client.ingest.units(**params)

  6. The SDK handles proper conversion of use_case_version from an integer to a string when sending to the API.

Updated about 2 months ago