# This file was auto-generated by Fern from our API Definition. import datetime as dt import typing from ..commons.types.trace_with_full_details import TraceWithFullDetails from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.request_options import RequestOptions from .raw_client import AsyncRawTraceClient, RawTraceClient from .types.delete_trace_response import DeleteTraceResponse from .types.traces import Traces # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class TraceClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._raw_client = RawTraceClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> RawTraceClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- RawTraceClient """ return self._raw_client def get( self, trace_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> TraceWithFullDetails: """ Get a specific trace Parameters ---------- trace_id : str The unique langfuse identifier of a trace request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- TraceWithFullDetails Examples -------- from langfuse import LangfuseAPI client = LangfuseAPI( x_langfuse_sdk_name="YOUR_X_LANGFUSE_SDK_NAME", x_langfuse_sdk_version="YOUR_X_LANGFUSE_SDK_VERSION", x_langfuse_public_key="YOUR_X_LANGFUSE_PUBLIC_KEY", username="YOUR_USERNAME", password="YOUR_PASSWORD", base_url="https://yourhost.com/path/to/api", ) client.trace.get( trace_id="traceId", ) """ _response = self._raw_client.get(trace_id, request_options=request_options) return _response.data def delete( self, trace_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> DeleteTraceResponse: """ Delete a specific trace Parameters ---------- trace_id : str The unique langfuse identifier of the trace to delete request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- DeleteTraceResponse Examples -------- from langfuse import LangfuseAPI client = LangfuseAPI( x_langfuse_sdk_name="YOUR_X_LANGFUSE_SDK_NAME", x_langfuse_sdk_version="YOUR_X_LANGFUSE_SDK_VERSION", x_langfuse_public_key="YOUR_X_LANGFUSE_PUBLIC_KEY", username="YOUR_USERNAME", password="YOUR_PASSWORD", base_url="https://yourhost.com/path/to/api", ) client.trace.delete( trace_id="traceId", ) """ _response = self._raw_client.delete(trace_id, request_options=request_options) return _response.data def list( self, *, page: typing.Optional[int] = None, limit: typing.Optional[int] = None, user_id: typing.Optional[str] = None, name: typing.Optional[str] = None, session_id: typing.Optional[str] = None, from_timestamp: typing.Optional[dt.datetime] = None, to_timestamp: typing.Optional[dt.datetime] = None, order_by: typing.Optional[str] = None, tags: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, version: typing.Optional[str] = None, release: typing.Optional[str] = None, environment: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, fields: typing.Optional[str] = None, filter: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> Traces: """ Get list of traces Parameters ---------- page : typing.Optional[int] Page number, starts at 1 limit : typing.Optional[int] Limit of items per page. If you encounter api issues due to too large page sizes, try to reduce the limit. user_id : typing.Optional[str] name : typing.Optional[str] session_id : typing.Optional[str] from_timestamp : typing.Optional[dt.datetime] Optional filter to only include traces with a trace.timestamp on or after a certain datetime (ISO 8601) to_timestamp : typing.Optional[dt.datetime] Optional filter to only include traces with a trace.timestamp before a certain datetime (ISO 8601) order_by : typing.Optional[str] Format of the string [field].[asc/desc]. Fields: id, timestamp, name, userId, release, version, public, bookmarked, sessionId. Example: timestamp.asc tags : typing.Optional[typing.Union[str, typing.Sequence[str]]] Only traces that include all of these tags will be returned. version : typing.Optional[str] Optional filter to only include traces with a certain version. release : typing.Optional[str] Optional filter to only include traces with a certain release. environment : typing.Optional[typing.Union[str, typing.Sequence[str]]] Optional filter for traces where the environment is one of the provided values. fields : typing.Optional[str] Comma-separated list of fields to include in the response. Available field groups: 'core' (always included), 'io' (input, output, metadata), 'scores', 'observations', 'metrics'. If not specified, all fields are returned. Example: 'core,scores,metrics'. Note: Excluded 'observations' or 'scores' fields return empty arrays; excluded 'metrics' returns -1 for 'totalCost' and 'latency'. filter : typing.Optional[str] JSON string containing an array of filter conditions. When provided, this takes precedence over query parameter filters (userId, name, sessionId, tags, version, release, environment, fromTimestamp, toTimestamp). ## Filter Structure Each filter condition has the following structure: ```json [ { "type": string, // Required. One of: "datetime", "string", "number", "stringOptions", "categoryOptions", "arrayOptions", "stringObject", "numberObject", "boolean", "null" "column": string, // Required. Column to filter on (see available columns below) "operator": string, // Required. Operator based on type: // - datetime: ">", "<", ">=", "<=" // - string: "=", "contains", "does not contain", "starts with", "ends with" // - stringOptions: "any of", "none of" // - categoryOptions: "any of", "none of" // - arrayOptions: "any of", "none of", "all of" // - number: "=", ">", "<", ">=", "<=" // - stringObject: "=", "contains", "does not contain", "starts with", "ends with" // - numberObject: "=", ">", "<", ">=", "<=" // - boolean: "=", "<>" // - null: "is null", "is not null" "value": any, // Required (except for null type). Value to compare against. Type depends on filter type "key": string // Required only for stringObject, numberObject, and categoryOptions types when filtering on nested fields like metadata } ] ``` ## Available Columns ### Core Trace Fields - `id` (string) - Trace ID - `name` (string) - Trace name - `timestamp` (datetime) - Trace timestamp - `userId` (string) - User ID - `sessionId` (string) - Session ID - `environment` (string) - Environment tag - `version` (string) - Version tag - `release` (string) - Release tag - `tags` (arrayOptions) - Array of tags - `bookmarked` (boolean) - Bookmark status ### Structured Data - `metadata` (stringObject/numberObject/categoryOptions) - Metadata key-value pairs. Use `key` parameter to filter on specific metadata keys. ### Aggregated Metrics (from observations) These metrics are aggregated from all observations within the trace: - `latency` (number) - Latency in seconds (time from first observation start to last observation end) - `inputTokens` (number) - Total input tokens across all observations - `outputTokens` (number) - Total output tokens across all observations - `totalTokens` (number) - Total tokens (alias: `tokens`) - `inputCost` (number) - Total input cost in USD - `outputCost` (number) - Total output cost in USD - `totalCost` (number) - Total cost in USD ### Observation Level Aggregations These fields aggregate observation levels within the trace: - `level` (string) - Highest severity level (ERROR > WARNING > DEFAULT > DEBUG) - `warningCount` (number) - Count of WARNING level observations - `errorCount` (number) - Count of ERROR level observations - `defaultCount` (number) - Count of DEFAULT level observations - `debugCount` (number) - Count of DEBUG level observations ### Scores (requires join with scores table) - `scores_avg` (number) - Average of numeric scores (alias: `scores`) - `score_categories` (categoryOptions) - Categorical score values ## Filter Examples ```json [ { "type": "datetime", "column": "timestamp", "operator": ">=", "value": "2024-01-01T00:00:00Z" }, { "type": "string", "column": "userId", "operator": "=", "value": "user-123" }, { "type": "number", "column": "totalCost", "operator": ">=", "value": 0.01 }, { "type": "arrayOptions", "column": "tags", "operator": "all of", "value": ["production", "critical"] }, { "type": "stringObject", "column": "metadata", "key": "customer_tier", "operator": "=", "value": "enterprise" } ] ``` ## Performance Notes - Filtering on `userId`, `sessionId`, or `metadata` may enable skip indexes for better query performance - Score filters require a join with the scores table and may impact query performance request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Traces Examples -------- from langfuse import LangfuseAPI client = LangfuseAPI( x_langfuse_sdk_name="YOUR_X_LANGFUSE_SDK_NAME", x_langfuse_sdk_version="YOUR_X_LANGFUSE_SDK_VERSION", x_langfuse_public_key="YOUR_X_LANGFUSE_PUBLIC_KEY", username="YOUR_USERNAME", password="YOUR_PASSWORD", base_url="https://yourhost.com/path/to/api", ) client.trace.list() """ _response = self._raw_client.list( page=page, limit=limit, user_id=user_id, name=name, session_id=session_id, from_timestamp=from_timestamp, to_timestamp=to_timestamp, order_by=order_by, tags=tags, version=version, release=release, environment=environment, fields=fields, filter=filter, request_options=request_options, ) return _response.data def delete_multiple( self, *, trace_ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None, ) -> DeleteTraceResponse: """ Delete multiple traces Parameters ---------- trace_ids : typing.Sequence[str] List of trace IDs to delete request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- DeleteTraceResponse Examples -------- from langfuse import LangfuseAPI client = LangfuseAPI( x_langfuse_sdk_name="YOUR_X_LANGFUSE_SDK_NAME", x_langfuse_sdk_version="YOUR_X_LANGFUSE_SDK_VERSION", x_langfuse_public_key="YOUR_X_LANGFUSE_PUBLIC_KEY", username="YOUR_USERNAME", password="YOUR_PASSWORD", base_url="https://yourhost.com/path/to/api", ) client.trace.delete_multiple( trace_ids=["traceIds", "traceIds"], ) """ _response = self._raw_client.delete_multiple( trace_ids=trace_ids, request_options=request_options ) return _response.data class AsyncTraceClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._raw_client = AsyncRawTraceClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> AsyncRawTraceClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- AsyncRawTraceClient """ return self._raw_client async def get( self, trace_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> TraceWithFullDetails: """ Get a specific trace Parameters ---------- trace_id : str The unique langfuse identifier of a trace request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- TraceWithFullDetails Examples -------- import asyncio from langfuse import AsyncLangfuseAPI client = AsyncLangfuseAPI( x_langfuse_sdk_name="YOUR_X_LANGFUSE_SDK_NAME", x_langfuse_sdk_version="YOUR_X_LANGFUSE_SDK_VERSION", x_langfuse_public_key="YOUR_X_LANGFUSE_PUBLIC_KEY", username="YOUR_USERNAME", password="YOUR_PASSWORD", base_url="https://yourhost.com/path/to/api", ) async def main() -> None: await client.trace.get( trace_id="traceId", ) asyncio.run(main()) """ _response = await self._raw_client.get( trace_id, request_options=request_options ) return _response.data async def delete( self, trace_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> DeleteTraceResponse: """ Delete a specific trace Parameters ---------- trace_id : str The unique langfuse identifier of the trace to delete request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- DeleteTraceResponse Examples -------- import asyncio from langfuse import AsyncLangfuseAPI client = AsyncLangfuseAPI( x_langfuse_sdk_name="YOUR_X_LANGFUSE_SDK_NAME", x_langfuse_sdk_version="YOUR_X_LANGFUSE_SDK_VERSION", x_langfuse_public_key="YOUR_X_LANGFUSE_PUBLIC_KEY", username="YOUR_USERNAME", password="YOUR_PASSWORD", base_url="https://yourhost.com/path/to/api", ) async def main() -> None: await client.trace.delete( trace_id="traceId", ) asyncio.run(main()) """ _response = await self._raw_client.delete( trace_id, request_options=request_options ) return _response.data async def list( self, *, page: typing.Optional[int] = None, limit: typing.Optional[int] = None, user_id: typing.Optional[str] = None, name: typing.Optional[str] = None, session_id: typing.Optional[str] = None, from_timestamp: typing.Optional[dt.datetime] = None, to_timestamp: typing.Optional[dt.datetime] = None, order_by: typing.Optional[str] = None, tags: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, version: typing.Optional[str] = None, release: typing.Optional[str] = None, environment: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None, fields: typing.Optional[str] = None, filter: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> Traces: """ Get list of traces Parameters ---------- page : typing.Optional[int] Page number, starts at 1 limit : typing.Optional[int] Limit of items per page. If you encounter api issues due to too large page sizes, try to reduce the limit. user_id : typing.Optional[str] name : typing.Optional[str] session_id : typing.Optional[str] from_timestamp : typing.Optional[dt.datetime] Optional filter to only include traces with a trace.timestamp on or after a certain datetime (ISO 8601) to_timestamp : typing.Optional[dt.datetime] Optional filter to only include traces with a trace.timestamp before a certain datetime (ISO 8601) order_by : typing.Optional[str] Format of the string [field].[asc/desc]. Fields: id, timestamp, name, userId, release, version, public, bookmarked, sessionId. Example: timestamp.asc tags : typing.Optional[typing.Union[str, typing.Sequence[str]]] Only traces that include all of these tags will be returned. version : typing.Optional[str] Optional filter to only include traces with a certain version. release : typing.Optional[str] Optional filter to only include traces with a certain release. environment : typing.Optional[typing.Union[str, typing.Sequence[str]]] Optional filter for traces where the environment is one of the provided values. fields : typing.Optional[str] Comma-separated list of fields to include in the response. Available field groups: 'core' (always included), 'io' (input, output, metadata), 'scores', 'observations', 'metrics'. If not specified, all fields are returned. Example: 'core,scores,metrics'. Note: Excluded 'observations' or 'scores' fields return empty arrays; excluded 'metrics' returns -1 for 'totalCost' and 'latency'. filter : typing.Optional[str] JSON string containing an array of filter conditions. When provided, this takes precedence over query parameter filters (userId, name, sessionId, tags, version, release, environment, fromTimestamp, toTimestamp). ## Filter Structure Each filter condition has the following structure: ```json [ { "type": string, // Required. One of: "datetime", "string", "number", "stringOptions", "categoryOptions", "arrayOptions", "stringObject", "numberObject", "boolean", "null" "column": string, // Required. Column to filter on (see available columns below) "operator": string, // Required. Operator based on type: // - datetime: ">", "<", ">=", "<=" // - string: "=", "contains", "does not contain", "starts with", "ends with" // - stringOptions: "any of", "none of" // - categoryOptions: "any of", "none of" // - arrayOptions: "any of", "none of", "all of" // - number: "=", ">", "<", ">=", "<=" // - stringObject: "=", "contains", "does not contain", "starts with", "ends with" // - numberObject: "=", ">", "<", ">=", "<=" // - boolean: "=", "<>" // - null: "is null", "is not null" "value": any, // Required (except for null type). Value to compare against. Type depends on filter type "key": string // Required only for stringObject, numberObject, and categoryOptions types when filtering on nested fields like metadata } ] ``` ## Available Columns ### Core Trace Fields - `id` (string) - Trace ID - `name` (string) - Trace name - `timestamp` (datetime) - Trace timestamp - `userId` (string) - User ID - `sessionId` (string) - Session ID - `environment` (string) - Environment tag - `version` (string) - Version tag - `release` (string) - Release tag - `tags` (arrayOptions) - Array of tags - `bookmarked` (boolean) - Bookmark status ### Structured Data - `metadata` (stringObject/numberObject/categoryOptions) - Metadata key-value pairs. Use `key` parameter to filter on specific metadata keys. ### Aggregated Metrics (from observations) These metrics are aggregated from all observations within the trace: - `latency` (number) - Latency in seconds (time from first observation start to last observation end) - `inputTokens` (number) - Total input tokens across all observations - `outputTokens` (number) - Total output tokens across all observations - `totalTokens` (number) - Total tokens (alias: `tokens`) - `inputCost` (number) - Total input cost in USD - `outputCost` (number) - Total output cost in USD - `totalCost` (number) - Total cost in USD ### Observation Level Aggregations These fields aggregate observation levels within the trace: - `level` (string) - Highest severity level (ERROR > WARNING > DEFAULT > DEBUG) - `warningCount` (number) - Count of WARNING level observations - `errorCount` (number) - Count of ERROR level observations - `defaultCount` (number) - Count of DEFAULT level observations - `debugCount` (number) - Count of DEBUG level observations ### Scores (requires join with scores table) - `scores_avg` (number) - Average of numeric scores (alias: `scores`) - `score_categories` (categoryOptions) - Categorical score values ## Filter Examples ```json [ { "type": "datetime", "column": "timestamp", "operator": ">=", "value": "2024-01-01T00:00:00Z" }, { "type": "string", "column": "userId", "operator": "=", "value": "user-123" }, { "type": "number", "column": "totalCost", "operator": ">=", "value": 0.01 }, { "type": "arrayOptions", "column": "tags", "operator": "all of", "value": ["production", "critical"] }, { "type": "stringObject", "column": "metadata", "key": "customer_tier", "operator": "=", "value": "enterprise" } ] ``` ## Performance Notes - Filtering on `userId`, `sessionId`, or `metadata` may enable skip indexes for better query performance - Score filters require a join with the scores table and may impact query performance request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- Traces Examples -------- import asyncio from langfuse import AsyncLangfuseAPI client = AsyncLangfuseAPI( x_langfuse_sdk_name="YOUR_X_LANGFUSE_SDK_NAME", x_langfuse_sdk_version="YOUR_X_LANGFUSE_SDK_VERSION", x_langfuse_public_key="YOUR_X_LANGFUSE_PUBLIC_KEY", username="YOUR_USERNAME", password="YOUR_PASSWORD", base_url="https://yourhost.com/path/to/api", ) async def main() -> None: await client.trace.list() asyncio.run(main()) """ _response = await self._raw_client.list( page=page, limit=limit, user_id=user_id, name=name, session_id=session_id, from_timestamp=from_timestamp, to_timestamp=to_timestamp, order_by=order_by, tags=tags, version=version, release=release, environment=environment, fields=fields, filter=filter, request_options=request_options, ) return _response.data async def delete_multiple( self, *, trace_ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None, ) -> DeleteTraceResponse: """ Delete multiple traces Parameters ---------- trace_ids : typing.Sequence[str] List of trace IDs to delete request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- DeleteTraceResponse Examples -------- import asyncio from langfuse import AsyncLangfuseAPI client = AsyncLangfuseAPI( x_langfuse_sdk_name="YOUR_X_LANGFUSE_SDK_NAME", x_langfuse_sdk_version="YOUR_X_LANGFUSE_SDK_VERSION", x_langfuse_public_key="YOUR_X_LANGFUSE_PUBLIC_KEY", username="YOUR_USERNAME", password="YOUR_PASSWORD", base_url="https://yourhost.com/path/to/api", ) async def main() -> None: await client.trace.delete_multiple( trace_ids=["traceIds", "traceIds"], ) asyncio.run(main()) """ _response = await self._raw_client.delete_multiple( trace_ids=trace_ids, request_options=request_options ) return _response.data