langfuse_client_base/apis/
ingestion_api.rs

1/*
2 * langfuse
3 *
4 * ## Authentication  Authenticate with the API using [Basic Auth](https://en.wikipedia.org/wiki/Basic_access_authentication), get API keys in the project settings:  - username: Langfuse Public Key - password: Langfuse Secret Key  ## Exports  - OpenAPI spec: https://cloud.langfuse.com/generated/api/openapi.yml
5 *
6 * The version of the OpenAPI document:
7 *
8 * Generated by: https://openapi-generator.tech
9 */
10
11use super::{configuration, ContentType, Error};
12use crate::{apis::ResponseContent, models};
13use reqwest;
14use serde::{de::Error as _, Deserialize, Serialize};
15
16/// struct for typed errors of method [`ingestion_batch`]
17#[derive(Debug, Clone, Serialize, Deserialize)]
18#[serde(untagged)]
19pub enum IngestionBatchError {
20    Status400(serde_json::Value),
21    Status401(serde_json::Value),
22    Status403(serde_json::Value),
23    Status404(serde_json::Value),
24    Status405(serde_json::Value),
25    UnknownValue(serde_json::Value),
26}
27
28/// **Legacy endpoint for batch ingestion for Langfuse Observability.**  -> Please use the OpenTelemetry endpoint (`/api/public/otel/v1/traces`). Learn more: https://langfuse.com/integrations/native/opentelemetry  Within each batch, there can be multiple events. Each event has a type, an id, a timestamp, metadata and a body. Internally, we refer to this as the \"event envelope\" as it tells us something about the event but not the trace. We use the event id within this envelope to deduplicate messages to avoid processing the same event twice, i.e. the event id should be unique per request. The event.body.id is the ID of the actual trace and will be used for updates and will be visible within the Langfuse App. I.e. if you want to update a trace, you'd use the same body id, but separate event IDs.  Notes: - Introduction to data model: https://langfuse.com/docs/observability/data-model - Batch sizes are limited to 3.5 MB in total. You need to adjust the number of events per batch accordingly. - The API does not return a 4xx status code for input errors. Instead, it responds with a 207 status code, which includes a list of the encountered errors.
29#[bon::builder]
30pub async fn ingestion_batch(
31    configuration: &configuration::Configuration,
32    ingestion_batch_request: models::IngestionBatchRequest,
33) -> Result<models::IngestionResponse, Error<IngestionBatchError>> {
34    // add a prefix to parameters to efficiently prevent name collisions
35    let p_body_ingestion_batch_request = ingestion_batch_request;
36
37    let uri_str = format!("{}/api/public/ingestion", configuration.base_path);
38    let mut req_builder = configuration
39        .client
40        .request(reqwest::Method::POST, &uri_str);
41
42    if let Some(ref user_agent) = configuration.user_agent {
43        req_builder = req_builder.header(reqwest::header::USER_AGENT, user_agent.clone());
44    }
45    if let Some(ref auth_conf) = configuration.basic_auth {
46        req_builder = req_builder.basic_auth(auth_conf.0.to_owned(), auth_conf.1.to_owned());
47    };
48    req_builder = req_builder.json(&p_body_ingestion_batch_request);
49
50    let req = req_builder.build()?;
51    let resp = configuration.client.execute(req).await?;
52
53    let status = resp.status();
54    let content_type = resp
55        .headers()
56        .get("content-type")
57        .and_then(|v| v.to_str().ok())
58        .unwrap_or("application/octet-stream");
59    let content_type = super::ContentType::from(content_type);
60
61    if !status.is_client_error() && !status.is_server_error() {
62        let content = resp.text().await?;
63        match content_type {
64            ContentType::Json => serde_json::from_str(&content).map_err(Error::from),
65            ContentType::Text => return Err(Error::from(serde_json::Error::custom("Received `text/plain` content type response that cannot be converted to `models::IngestionResponse`"))),
66            ContentType::Unsupported(unknown_type) => return Err(Error::from(serde_json::Error::custom(format!("Received `{unknown_type}` content type response that cannot be converted to `models::IngestionResponse`")))),
67        }
68    } else {
69        let content = resp.text().await?;
70        let entity: Option<IngestionBatchError> = serde_json::from_str(&content).ok();
71        Err(Error::ResponseError(ResponseContent {
72            status,
73            content,
74            entity,
75        }))
76    }
77}
langfuse_client_base/apis/ingestion_api.rs

langfuse_client_base/apis/
ingestion_api.rs
