basal_client/
batches.rs

1//! Durable Batch/Files API. Resources share the server-configured storage namespace without authentication.
2//! Flat methods retain raw JSONL bytes. Resource helpers add opt-in encoding, polling and result parsing.
3
4use std::hash::{BuildHasher, Hasher};
5
6use http::{
7    header::{HeaderMap, HeaderValue, ACCEPT},
8    Method,
9};
10use serde::{Deserialize, Serialize};
11use serde_json::{Map, Value};
12
13use crate::{Client, DecisionRequest, Error};
14
15/// One stored input or result file. Timestamps are Unix seconds.
16#[derive(Clone, Debug, Serialize, Deserialize)]
17pub struct FileObject {
18    /// Server-generated file id.
19    pub id: String,
20    /// Resource kind (`file`).
21    pub object: String,
22    /// Stored content length.
23    pub bytes: u64,
24    /// Creation time.
25    pub created_at: i64,
26    /// Optional expiry time.
27    pub expires_at: Option<i64>,
28    /// User-supplied name; not a local path.
29    pub filename: String,
30    /// `batch` for inputs, `batch_output` for results.
31    pub purpose: String,
32    /// Compatibility status (`processed`). Validation happens when a batch is created.
33    pub status: String,
34    /// Optional server details.
35    pub status_details: Option<Value>,
36    /// Future response fields.
37    #[serde(flatten)]
38    pub extra: Map<String, Value>,
39}
40
41/// Confirmation of a file deletion.
42#[derive(Clone, Debug, Serialize, Deserialize)]
43pub struct DeletedFile {
44    /// Deleted file id.
45    pub id: String,
46    /// Resource kind (`file`).
47    pub object: String,
48    /// Deletion confirmation.
49    pub deleted: bool,
50    /// Future response fields.
51    #[serde(flatten)]
52    pub extra: Map<String, Value>,
53}
54
55/// One cursor page, ordered by creation time and id descending.
56#[derive(Clone, Debug, Serialize, Deserialize)]
57pub struct BatchPage<T> {
58    /// Resource kind (`list`).
59    pub object: String,
60    /// Resources in this page.
61    pub data: Vec<T>,
62    /// First id, absent on an empty page.
63    pub first_id: Option<String>,
64    /// Pass as `after` to request the next page when `has_more` is true.
65    pub last_id: Option<String>,
66    /// Whether another page exists.
67    pub has_more: bool,
68    /// Future response fields.
69    #[serde(flatten)]
70    pub extra: Map<String, Value>,
71}
72
73/// Limits and cursor of one list request. The SDK never auto-paginates.
74#[derive(Clone, Debug, Default)]
75pub struct BatchListOptions<'a> {
76    /// Cursor of the same resource kind.
77    pub after: Option<&'a str>,
78    /// Page size 1–100 (server default 20).
79    pub limit: Option<u32>,
80}
81
82/// Optional upload retention. The input itself is always uploaded with purpose `batch`.
83#[derive(Clone, Debug, Default)]
84pub struct FileUploadOptions {
85    /// Lifetime from file creation, 3600–2592000 seconds; absent uses the server default.
86    pub expires_after_seconds: Option<u32>,
87}
88
89/// Retention of each output/error file from its creation.
90#[derive(Clone, Debug, Serialize, Deserialize)]
91pub struct BatchExpiresAfter {
92    /// Supported anchor: `created_at`.
93    pub anchor: String,
94    /// Lifetime, 3600–2592000 seconds.
95    pub seconds: u32,
96}
97
98impl BatchExpiresAfter {
99    /// Retain each result file for `seconds` after its creation.
100    pub fn created_at(seconds: u32) -> Self {
101        Self { anchor: "created_at".into(), seconds }
102    }
103}
104
105/// Create a batch from an uploaded JSONL input. Use [`Self::new`] for native defaults.
106#[derive(Clone, Debug, Serialize, Deserialize)]
107pub struct CreateBatchRequest {
108    /// Uploaded input file.
109    pub input_file_id: String,
110    /// Only `/v1/basal` is currently supported.
111    pub endpoint: String,
112    /// Only `24h` is currently supported; this is an expiry window, not a completion guarantee.
113    pub completion_window: String,
114    /// Up to 16 string pairs; server validates key/value limits.
115    #[serde(skip_serializing_if = "Option::is_none")]
116    pub metadata: Option<std::collections::BTreeMap<String, String>>,
117    /// Optional result-file retention.
118    #[serde(skip_serializing_if = "Option::is_none")]
119    pub output_expires_after: Option<BatchExpiresAfter>,
120}
121
122impl CreateBatchRequest {
123    /// Native `/v1/basal` batch with a 24-hour inference window.
124    pub fn new(input_file_id: impl Into<String>) -> Self {
125        Self {
126            input_file_id: input_file_id.into(),
127            endpoint: "/v1/basal".into(),
128            completion_window: "24h".into(),
129            metadata: None,
130            output_expires_after: None,
131        }
132    }
133}
134
135/// Counts after validation. Failed also includes expired records.
136#[derive(Clone, Debug, Serialize, Deserialize)]
137pub struct BatchRequestCounts {
138    /// Fully validated record count (zero until validation finishes).
139    pub total: u64,
140    /// Successful records.
141    pub completed: u64,
142    /// Failed or expired records.
143    pub failed: u64,
144    /// Future response fields.
145    #[serde(flatten)]
146    pub extra: Map<String, Value>,
147}
148
149/// A batch-level validation error.
150#[derive(Clone, Debug, Serialize, Deserialize)]
151pub struct BatchErrorEntry {
152    /// Server error code.
153    pub code: String,
154    /// Error description.
155    pub message: String,
156    /// Affected parameter when known.
157    pub param: Option<String>,
158    /// One-based input line when known.
159    pub line: Option<u64>,
160    /// Future response fields.
161    #[serde(flatten)]
162    pub extra: Map<String, Value>,
163}
164
165/// Bounded list of validation errors.
166#[derive(Clone, Debug, Serialize, Deserialize)]
167pub struct BatchErrors {
168    /// Resource kind (`list`).
169    pub object: String,
170    /// Validation errors.
171    pub data: Vec<BatchErrorEntry>,
172    /// Future response fields.
173    #[serde(flatten)]
174    pub extra: Map<String, Value>,
175}
176
177/// Durable batch state; nullable timestamps are Unix seconds. Future status strings remain readable.
178#[derive(Clone, Debug, Serialize, Deserialize)]
179pub struct Batch {
180    /// Batch id.
181    pub id: String,
182    /// Resource kind (`batch`).
183    pub object: String,
184    /// Native target endpoint.
185    pub endpoint: String,
186    /// Effective model, once resolved.
187    pub model: Option<String>,
188    /// Validation errors, when present.
189    pub errors: Option<BatchErrors>,
190    /// Uploaded JSONL input id.
191    pub input_file_id: String,
192    /// Inference window (`24h`).
193    pub completion_window: String,
194    /// validating/failed/in_progress/finalizing/completed/expired/cancelling/cancelled.
195    pub status: String,
196    /// Successful results file, when published.
197    pub output_file_id: Option<String>,
198    /// Errors file, when published.
199    pub error_file_id: Option<String>,
200    /// Creation time.
201    pub created_at: i64,
202    /// Validation completed and processing became possible.
203    pub in_progress_at: Option<i64>,
204    /// End of inference window.
205    pub expires_at: i64,
206    /// Export started.
207    pub finalizing_at: Option<i64>,
208    /// Completed time.
209    pub completed_at: Option<i64>,
210    /// Failed time.
211    pub failed_at: Option<i64>,
212    /// Expired time.
213    pub expired_at: Option<i64>,
214    /// Cancellation requested time.
215    pub cancelling_at: Option<i64>,
216    /// Cancellation finalized time.
217    pub cancelled_at: Option<i64>,
218    /// Validated/completed/failed counts.
219    pub request_counts: BatchRequestCounts,
220    /// Owner-provided metadata.
221    pub metadata: Option<std::collections::BTreeMap<String, String>>,
222    /// Future response fields.
223    #[serde(flatten)]
224    pub extra: Map<String, Value>,
225}
226
227/// One JSONL input record. Serialize each record followed by LF; upload exact bytes if already encoded.
228#[derive(Clone, Debug, Serialize, Deserialize)]
229pub struct BatchInputLine {
230    /// Caller id, unique within the file.
231    pub custom_id: String,
232    /// `POST`.
233    pub method: String,
234    /// `/v1/basal`.
235    pub url: String,
236    /// Native request body.
237    pub body: DecisionRequest,
238}
239
240impl BatchInputLine {
241    /// Wrap a native request for a JSONL batch input.
242    pub fn new(custom_id: impl Into<String>, body: DecisionRequest) -> Self {
243        Self { custom_id: custom_id.into(), method: "POST".into(), url: "/v1/basal".into(), body }
244    }
245}
246
247/// One result JSONL line; native success and error response bodies remain arbitrary JSON objects.
248#[derive(Clone, Debug, Serialize, Deserialize)]
249pub struct BatchResultLine {
250    /// Generated request id.
251    pub id: String,
252    /// Original caller id.
253    pub custom_id: String,
254    /// An executed native request response.
255    pub response: Option<BatchResultResponse>,
256    /// A record which could not be executed or retained.
257    pub error: Option<BatchResultError>,
258    /// Future response fields.
259    #[serde(flatten)]
260    pub extra: Map<String, Value>,
261}
262
263/// Response portion of a result line.
264#[derive(Clone, Debug, Serialize, Deserialize)]
265pub struct BatchResultResponse {
266    /// Native HTTP status.
267    pub status_code: u16,
268    /// Generated request id.
269    pub request_id: String,
270    /// Native response (including native error bodies).
271    pub body: Value,
272    /// Future response fields.
273    #[serde(flatten)]
274    pub extra: Map<String, Value>,
275}
276
277/// Non-execution error of a result record.
278#[derive(Clone, Debug, Serialize, Deserialize)]
279pub struct BatchResultError {
280    /// Error code, including `batch_expired`.
281    pub code: String,
282    /// Description.
283    pub message: String,
284    /// Future response fields.
285    #[serde(flatten)]
286    pub extra: Map<String, Value>,
287}
288
289pub(super) fn id_component(id: &str, prefix: &str) -> Result<(), Error> {
290    let suffix = id.strip_prefix(prefix).ok_or(Error::InvalidRequest("invalid resource id"))?;
291    if suffix.len() != 32 || !suffix.bytes().all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) {
292        return Err(Error::InvalidRequest("invalid resource id"));
293    }
294    Ok(())
295}
296
297fn list_path(base: &str, options: &BatchListOptions<'_>, prefix: &str, purpose: Option<&str>) -> Result<String, Error> {
298    let mut params = Vec::new();
299    if let Some(after) = options.after {
300        id_component(after, prefix)?;
301        params.push(format!("after={after}"));
302    }
303    if let Some(limit) = options.limit {
304        if !(1..=100).contains(&limit) {
305            return Err(Error::InvalidRequest("list limit must be 1-100"));
306        }
307        params.push(format!("limit={limit}"));
308    }
309    if let Some(purpose) = purpose {
310        if !matches!(purpose, "batch" | "batch_output") {
311            return Err(Error::InvalidRequest("invalid file purpose"));
312        }
313        params.push(format!("purpose={purpose}"));
314    }
315    Ok(if params.is_empty() { base.into() } else { format!("{base}?{}", params.join("&")) })
316}
317
318impl Client {
319    /// Upload caller-owned JSONL bytes (at most 200 MB); memory includes a multipart copy of the input.
320    /// No file paths or URLs are opened. Expiry fields precede the file part.
321    pub async fn upload_file(
322        &self,
323        filename: &str,
324        data: &[u8],
325        options: FileUploadOptions,
326    ) -> Result<FileObject, Error> {
327        if filename.is_empty() || filename.len() > 255 || filename.chars().any(char::is_control) {
328            return Err(Error::InvalidRequest("filename must be 1-255 bytes without control characters"));
329        }
330        if data.len() > 200_000_000 {
331            return Err(Error::InvalidRequest("input file exceeds 200 MB"));
332        }
333        if options.expires_after_seconds.is_some_and(|s| !(3600..=2_592_000).contains(&s)) {
334            return Err(Error::InvalidRequest("file expiry must be 3600-2592000 seconds"));
335        }
336        // A per-call randomized standard-library hasher provides an unpredictable boundary without adding a
337        // dependency. Check the actual input for collisions before constructing the multipart body.
338        let random = std::collections::hash_map::RandomState::new();
339        let mut counter = 0u64;
340        let boundary = loop {
341            let mut h = random.build_hasher();
342            h.write_u64(counter);
343            let candidate = format!("basal-sdk-{:016x}-{counter:016x}", h.finish());
344            if !data.windows(candidate.len()).any(|w| w == candidate.as_bytes()) {
345                break candidate;
346            }
347            counter = counter.checked_add(1).ok_or(Error::InvalidRequest("multipart boundary unavailable"))?;
348        };
349        let mut form = format!("--{boundary}\r\nContent-Disposition: form-data; name=\"purpose\"\r\n\r\nbatch\r\n");
350        if let Some(seconds) = options.expires_after_seconds {
351            for (name, value) in
352                [("expires_after[anchor]", "created_at".to_string()), ("expires_after[seconds]", seconds.to_string())]
353            {
354                form.push_str(&format!(
355                    "--{boundary}\r\nContent-Disposition: form-data; name=\"{name}\"\r\n\r\n{value}\r\n"
356                ));
357            }
358        }
359        let filename = filename.replace('\\', "\\\\").replace('"', "\\\"");
360        form.push_str(&format!("--{boundary}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"{filename}\"\r\nContent-Type: application/jsonl\r\n\r\n"));
361        let mut body = form.into_bytes();
362        body.extend_from_slice(data);
363        body.extend_from_slice(format!("\r\n--{boundary}--\r\n").as_bytes());
364        let content_type = format!("multipart/form-data; boundary={boundary}");
365        let bytes =
366            self.request_bytes(Method::POST, "v1/files", Some(body), Some(&content_type), HeaderMap::new()).await?;
367        serde_json::from_slice(&bytes).map_err(Error::Protocol)
368    }
369
370    /// Read one page of files; optional purpose is `batch` or `batch_output`.
371    pub async fn list_files(
372        &self,
373        options: BatchListOptions<'_>,
374        purpose: Option<&str>,
375    ) -> Result<BatchPage<FileObject>, Error> {
376        self.send(Method::GET, &list_path("v1/files", &options, "file_", purpose)?, None).await
377    }
378    /// Retrieve file metadata.
379    pub async fn get_file(&self, id: &str) -> Result<FileObject, Error> {
380        id_component(id, "file_")?;
381        self.send(Method::GET, &format!("v1/files/{id}"), None).await
382    }
383    /// Delete a file (the server may retain physical bytes until guarded cleanup).
384    pub async fn delete_file(&self, id: &str) -> Result<DeletedFile, Error> {
385        id_component(id, "file_")?;
386        self.send(Method::DELETE, &format!("v1/files/{id}"), None).await
387    }
388    /// Download exact JSONL bytes, bounded by `ClientOptions::max_response_bytes` (default 8 MiB).
389    /// Increase that explicit cap when downloading larger files. Dropping the future cancels I/O.
390    pub async fn download_file(&self, id: &str) -> Result<Vec<u8>, Error> {
391        id_component(id, "file_")?;
392        let mut headers = HeaderMap::new();
393        headers.insert(ACCEPT, HeaderValue::from_static("application/octet-stream"));
394        self.request_bytes(Method::GET, &format!("v1/files/{id}/content"), None, None, headers).await
395    }
396    /// Create a batch; an optional owner-scoped idempotency key is 1–128 visible ASCII bytes.
397    /// Retrying is the caller's decision; reuse the same key only with the same request.
398    pub async fn create_batch(
399        &self,
400        request: &CreateBatchRequest,
401        idempotency_key: Option<&str>,
402    ) -> Result<Batch, Error> {
403        let mut headers = HeaderMap::new();
404        if let Some(key) = idempotency_key {
405            if key.is_empty() || key.len() > 128 || !key.bytes().all(|b| (0x21..=0x7e).contains(&b)) {
406                return Err(Error::InvalidRequest("invalid idempotency key"));
407            }
408            let value = HeaderValue::from_str(key).map_err(|_| Error::InvalidRequest("invalid idempotency key"))?;
409            headers.insert("idempotency-key", value);
410        }
411        let bytes = self
412            .request_bytes(
413                Method::POST,
414                "v1/batches",
415                Some(serde_json::to_vec(request).map_err(Error::Encode)?),
416                Some("application/json"),
417                headers,
418            )
419            .await?;
420        serde_json::from_slice(&bytes).map_err(Error::Protocol)
421    }
422    /// Read one page of batches.
423    pub async fn list_batches(&self, options: BatchListOptions<'_>) -> Result<BatchPage<Batch>, Error> {
424        self.send(Method::GET, &list_path("v1/batches", &options, "batch_", None)?, None).await
425    }
426    /// Retrieve a batch without polling.
427    pub async fn get_batch(&self, id: &str) -> Result<Batch, Error> {
428        id_component(id, "batch_")?;
429        self.send(Method::GET, &format!("v1/batches/{id}"), None).await
430    }
431    /// Request cancellation; partial result files may be published later.
432    pub async fn cancel_batch(&self, id: &str) -> Result<Batch, Error> {
433        id_component(id, "batch_")?;
434        self.send(Method::POST, &format!("v1/batches/{id}/cancel"), None).await
435    }
436    /// Read Batch health. The server exposes this route without authentication.
437    pub async fn batch_health(&self) -> Result<Value, Error> {
438        self.send(Method::GET, "health/batches", None).await
439    }
440}
basal_client/batches.rs

basal_client/
batches.rs
