openai_interface/rest/

post.rs

use std::{future::Future, str::FromStr};

use eventsource_stream::Eventsource;
use futures_util::{Stream, StreamExt, TryStreamExt};
use serde::{Serialize, de::DeserializeOwned};

use crate::errors::OapiError;

pub trait Post {
    fn is_streaming(&self) -> bool;
    /// Builds the URL for the request.
    ///
    /// `base_url` should be like <https://api.openai.com/v1>
    fn build_url(&self, base_url: &str) -> Result<String, OapiError>;
}

pub trait PostNoStream: Post + Serialize + Sync + Send {
    type Response: DeserializeOwned + FromStr<Err = OapiError> + Send + Sync;

    /// Sends a POST request and returns the raw response body.
    ///
    /// The `client` is supplied by the caller; see [`crate::rest::default_client`]
    /// for a sensible default.
    fn get_response_string(
        &self,
        client: &reqwest::Client,
        base_url: &str,
        key: &str,
    ) -> impl Future<Output = Result<String, OapiError>> + Send + Sync {
        async move {
            if self.is_streaming() {
                return Err(OapiError::NonStreamingViolation);
            }

            let response = client
                .post(self.build_url(base_url)?)
                .header("Content-Type", "application/json")
                .header("Accept", "application/json")
                .bearer_auth(key)
                .json(self)
                .send()
                .await?;

            crate::rest::response_text_checked(response).await
        }
    }

    /// Sends a POST request and deserializes the response.
    fn get_response(
        &self,
        client: &reqwest::Client,
        url: &str,
        key: &str,
    ) -> impl Future<Output = Result<Self::Response, OapiError>> + Send + Sync {
        async move {
            let text = self.get_response_string(client, url, key).await?;
            let result = Self::Response::from_str(&text)?;
            Ok(result)
        }
    }
}

pub trait PostStream: Post + Serialize + Sync + Send {
    type Response: DeserializeOwned + FromStr<Err = OapiError> + Send + Sync + 'static;

    /// Sends a streaming POST request and returns the raw SSE data items as
    /// strings.
    ///
    /// The returned stream stops after the `data: [DONE]` sentinel. Errors
    /// terminate the stream.
    ///
    /// The `client` is supplied by the caller; see [`crate::rest::default_client`]
    /// for a sensible default.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// use std::sync::LazyLock;
    /// use futures_util::StreamExt;
    /// use openai_interface::chat::create::request::{Message, RequestBody};
    /// use openai_interface::rest::{default_client, post::PostStream};
    ///
    /// const DEEPSEEK_API_KEY: &str = "YOUR_API_KEY";
    /// const DEEPSEEK_CHAT_URL: &'static str = "https://api.deepseek.com";
    /// const DEEPSEEK_MODEL: &'static str = "deepseek-chat";
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let request = RequestBody {
    ///     messages: vec![
    ///         Message::System {
    ///             content: "This is a request of test purpose. Reply briefly".to_string(),
    ///             name: None,
    ///         },
    ///         Message::User {
    ///             content: "What's your name?".into(),
    ///             name: None,
    ///         },
    ///     ],
    ///     model: DEEPSEEK_MODEL.to_string(),
    ///     stream: Some(true),
    ///     ..Default::default()
    /// };
    ///
    /// let mut response = request
    ///     .get_stream_response_string(&default_client(), DEEPSEEK_CHAT_URL, DEEPSEEK_API_KEY)
    ///     .await?;
    ///
    /// while let Some(chunk) = response.next().await {
    ///     println!("{}", chunk?);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    fn get_stream_response_string(
        &self,
        client: &reqwest::Client,
        base_url: &str,
        api_key: &str,
    ) -> impl Future<
        Output = Result<
            impl Stream<Item = Result<String, OapiError>> + Send + Unpin + 'static,
            OapiError,
        >,
    > + Send
    + Sync {
        async move {
            if !self.is_streaming() {
                return Err(OapiError::StreamingViolation);
            }

            let response = client
                .post(self.build_url(base_url)?)
                .header("Content-Type", "application/json")
                .header("Accept", "text/event-stream")
                .bearer_auth(api_key)
                .json(self)
                .send()
                .await?;

            // Parse the body as an SSE stream. Errors end the stream.
            let stream = crate::rest::check_status(response)
                .await?
                .bytes_stream()
                .eventsource()
                .map(|event| match event {
                    Ok(event) => Ok(event.data),
                    Err(e) => Err(OapiError::SseParseError(format!("SSE parse error: {e}"))),
                });

            Ok(stream)
        }
    }

    /// Sends a streaming POST request and deserializes every SSE data item
    /// into [`Self::Response`].
    ///
    /// The stream ends after the `data: [DONE]` sentinel or on the first
    /// error, whichever comes first.
    fn get_stream_response(
        &self,
        client: &reqwest::Client,
        base_url: &str,
        api_key: &str,
    ) -> impl Future<
        Output = Result<
            impl Stream<Item = Result<Self::Response, OapiError>> + Send + Unpin + 'static,
            OapiError,
        >,
    > + Send
    + Sync {
        async move {
            let stream = self
                .get_stream_response_string(client, base_url, api_key)
                .await?;

            let parsed_stream = stream
                // The `data: [DONE]` sentinel is not JSON; end the stream there.
                // `future::ready` keeps the combinators `Unpin` (async blocks
                // are not).
                .take_while(|result| {
                    std::future::ready(matches!(result, Ok(data) if data != "[DONE]"))
                })
                .and_then(|data| std::future::ready(Self::Response::from_str(&data)));

            Ok(parsed_stream)
        }
    }
}

/// Trait for POST requests whose response body is binary rather than JSON.
///
/// Endpoints such as `POST /audio/speech` return the raw audio bytes instead
/// of a JSON object, so the body is returned as raw bytes.
pub trait PostBinary: Post + Serialize + Sync + Send {
    /// Sends a POST request with a JSON body and returns the raw response
    /// body bytes.
    ///
    /// The `client` is supplied by the caller; see [`crate::rest::default_client`]
    /// for a sensible default.
    fn get_response_bytes(
        &self,
        client: &reqwest::Client,
        base_url: &str,
        api_key: &str,
    ) -> impl Future<Output = Result<Vec<u8>, OapiError>> + Send + Sync {
        async move {
            let response = client
                .post(self.build_url(base_url)?)
                .header("Content-Type", "application/json")
                .header("Accept", "application/octet-stream")
                .bearer_auth(api_key)
                .json(self)
                .send()
                .await?;

            crate::rest::response_bytes_checked(response).await
        }
    }

    /// Sends a POST request with a JSON body and streams the raw response
    /// body as byte chunks.
    ///
    /// Unlike [`Self::get_response_bytes`], the body is not buffered: each
    /// item is one transport-level chunk. This matches the official SDK's
    /// streaming binary responses (e.g. `POST /audio/speech` with
    /// `stream_format`, where the chunks carry either raw audio or SSE
    /// frames). Parsing any framing inside the byte stream is up to the
    /// caller.
    ///
    /// The `client` is supplied by the caller; see [`crate::rest::default_client`]
    /// for a sensible default.
    fn get_stream_response_bytes(
        &self,
        client: &reqwest::Client,
        base_url: &str,
        api_key: &str,
    ) -> impl Future<
        Output = Result<
            impl Stream<Item = Result<Vec<u8>, OapiError>> + Send + Unpin + 'static,
            OapiError,
        >,
    > + Send
    + Sync {
        async move {
            let response = client
                .post(self.build_url(base_url)?)
                .header("Content-Type", "application/json")
                .header("Accept", "application/octet-stream")
                .bearer_auth(api_key)
                .json(self)
                .send()
                .await?;

            let response = crate::rest::check_status(response).await?;

            Ok(response.bytes_stream().map(|chunk| {
                chunk
                    .map(|bytes| bytes.to_vec())
                    .map_err(OapiError::SendError)
            }))
        }
    }
}