# This file was auto-generated by Fern from our API Definition. import typing from json.decoder import JSONDecodeError from ...commons.errors.access_denied_error import AccessDeniedError from ...commons.errors.error import Error from ...commons.errors.method_not_allowed_error import MethodNotAllowedError from ...commons.errors.not_found_error import NotFoundError from ...commons.errors.unauthorized_error import UnauthorizedError from ...core.api_error import ApiError from ...core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ...core.http_response import AsyncHttpResponse, HttpResponse from ...core.pydantic_utilities import parse_obj_as from ...core.request_options import RequestOptions from .types.metrics_response import MetricsResponse class RawMetricsV1Client: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def metrics( self, *, query: str, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[MetricsResponse]: """ Get metrics from the Langfuse project using a query object. Consider using the [v2 metrics endpoint](/api-reference#tag/metricsv2/GET/api/public/v2/metrics) for better performance. For more details, see the [Metrics API documentation](https://langfuse.com/docs/metrics/features/metrics-api). Parameters ---------- query : str JSON string containing the query parameters with the following structure: ```json { "view": string, // Required. One of "traces", "observations", "scores-numeric", "scores-categorical" "dimensions": [ // Optional. Default: [] { "field": string // Field to group by, e.g. "name", "userId", "sessionId" } ], "metrics": [ // Required. At least one metric must be provided { "measure": string, // What to measure, e.g. "count", "latency", "value" "aggregation": string // How to aggregate, e.g. "count", "sum", "avg", "p95", "histogram" } ], "filters": [ // Optional. Default: [] { "column": string, // Column to filter on "operator": string, // Operator, e.g. "=", ">", "<", "contains" "value": any, // Value to compare against "type": string, // Data type, e.g. "string", "number", "stringObject" "key": string // Required only when filtering on metadata } ], "timeDimension": { // Optional. Default: null. If provided, results will be grouped by time "granularity": string // One of "minute", "hour", "day", "week", "month", "auto" }, "fromTimestamp": string, // Required. ISO datetime string for start of time range "toTimestamp": string, // Required. ISO datetime string for end of time range "orderBy": [ // Optional. Default: null { "field": string, // Field to order by "direction": string // "asc" or "desc" } ], "config": { // Optional. Query-specific configuration "bins": number, // Optional. Number of bins for histogram (1-100), default: 10 "row_limit": number // Optional. Row limit for results (1-1000) } } ``` request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[MetricsResponse] """ _response = self._client_wrapper.httpx_client.request( "api/public/metrics", method="GET", params={ "query": query, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( MetricsResponse, parse_obj_as( type_=MetricsResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.text, ) raise ApiError( status_code=_response.status_code, headers=dict(_response.headers), body=_response_json, ) class AsyncRawMetricsV1Client: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def metrics( self, *, query: str, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[MetricsResponse]: """ Get metrics from the Langfuse project using a query object. Consider using the [v2 metrics endpoint](/api-reference#tag/metricsv2/GET/api/public/v2/metrics) for better performance. For more details, see the [Metrics API documentation](https://langfuse.com/docs/metrics/features/metrics-api). Parameters ---------- query : str JSON string containing the query parameters with the following structure: ```json { "view": string, // Required. One of "traces", "observations", "scores-numeric", "scores-categorical" "dimensions": [ // Optional. Default: [] { "field": string // Field to group by, e.g. "name", "userId", "sessionId" } ], "metrics": [ // Required. At least one metric must be provided { "measure": string, // What to measure, e.g. "count", "latency", "value" "aggregation": string // How to aggregate, e.g. "count", "sum", "avg", "p95", "histogram" } ], "filters": [ // Optional. Default: [] { "column": string, // Column to filter on "operator": string, // Operator, e.g. "=", ">", "<", "contains" "value": any, // Value to compare against "type": string, // Data type, e.g. "string", "number", "stringObject" "key": string // Required only when filtering on metadata } ], "timeDimension": { // Optional. Default: null. If provided, results will be grouped by time "granularity": string // One of "minute", "hour", "day", "week", "month", "auto" }, "fromTimestamp": string, // Required. ISO datetime string for start of time range "toTimestamp": string, // Required. ISO datetime string for end of time range "orderBy": [ // Optional. Default: null { "field": string, // Field to order by "direction": string // "asc" or "desc" } ], "config": { // Optional. Query-specific configuration "bins": number, // Optional. Number of bins for histogram (1-100), default: 10 "row_limit": number // Optional. Row limit for results (1-1000) } } ``` request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[MetricsResponse] """ _response = await self._client_wrapper.httpx_client.request( "api/public/metrics", method="GET", params={ "query": query, }, request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( MetricsResponse, parse_obj_as( type_=MetricsResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 400: raise Error( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 401: raise UnauthorizedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 403: raise AccessDeniedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 405: raise MethodNotAllowedError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) if _response.status_code == 404: raise NotFoundError( headers=dict(_response.headers), body=typing.cast( typing.Any, parse_obj_as( type_=typing.Any, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError( status_code=_response.status_code, headers=dict(_response.headers), body=_response.text, ) raise ApiError( status_code=_response.status_code, headers=dict(_response.headers), body=_response_json, )