From 88efe032de9d02f0251469828a354226e0b254d8 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Thu, 12 Nov 2026 16:51:14 +0000 Subject: [PATCH 1/2] feat(http): multi-level fan-out, with ancestor keys and errors that name them A fan_out whose select picks keys can now be another fan_out's over, as deep as they nest. Each key keeps the fields of the keys above it, so a third level's path can use {order_id} and {item_id}; key= renames a level's field. Every level keeps its own paginator, concurrency, on_status, retries and rate limit. plan checks each level's placeholders against the levels above it, and an error names the keys that led to the failing request. One-level jobs are unchanged. --- .../sqlload/src/http_source/_helpers/job.rs | 307 +++++++++++++----- crates/sqlload/src/http_source/models.rs | 23 +- .../http_source/tests/test_job_problems.rs | 19 ++ packages/sqlload/python/sqlload/_http.py | 30 +- packages/sqlload/tests/test_http_jobs.py | 111 +++++++ .../src/content/docs/docs/sources/python.mdx | 32 +- 6 files changed, 426 insertions(+), 96 deletions(-) diff --git a/crates/sqlload/src/http_source/_helpers/job.rs b/crates/sqlload/src/http_source/_helpers/job.rs index cdb4431b..92600dd9 100644 --- a/crates/sqlload/src/http_source/_helpers/job.rs +++ b/crates/sqlload/src/http_source/_helpers/job.rs @@ -1,6 +1,8 @@ //! Checking a job before any request, and running it: one endpoint, or a fan-out over keys. use std::collections::BTreeMap; +use std::future::Future; +use std::pin::Pin; use std::sync::Arc; use serde_json::{Map, Value}; @@ -12,7 +14,7 @@ use crate::http_source::_helpers::endpoint::{endpoint, interval, shared_limiter} use crate::http_source::_helpers::oauth::token_url; use crate::http_source::_helpers::pages::{paginate, set_at}; use crate::http_source::_helpers::selector::{selected_rows, selector}; -use crate::http_source::_helpers::urls::{filled_path, placeholders}; +use crate::http_source::_helpers::urls::{filled_path, placeholders, redacted}; use crate::python_sources::errors::ReadError; use crate::row_batches::main::projection::projection; use crate::row_batches::main::projection_problem::projection_problem; @@ -24,8 +26,8 @@ use crate::http_source::constants::{ BODY_DROPPING_REDIRECTS, JOB_THREADS, MAX_REDIRECTS, PAGES_AHEAD, }; use crate::http_source::models::{ - Auth, Calls, FanOut, FieldSpec, HttpJob, HttpPage, KeyFeed, Location, Method, Over, Paginator, - Request, Selector, StatusAction, + Auth, Calls, FanKey, FanOut, FieldSpec, HttpJob, HttpPage, KeyFeed, Location, Method, Over, + Paginator, Request, Selector, Sink, StatusAction, }; /// Every problem the job's definition has, found before any request, including projected rows that break the table's `rules`. @@ -53,6 +55,9 @@ pub(crate) fn job_problems(job: &HttpJob, rules: ProjectionRules<'_>) -> Vec fan_out_problems(fan_out), }); + problems.extend(rows.key.as_ref().map(|key| { + format!("key={key:?} names the keys a job gives as a fan_out's over; this job loads rows") + })); problems } @@ -210,50 +215,116 @@ fn header_problem(name: &str, value: &str) -> Option { } fn fan_out_problems(fan_out: &FanOut) -> Vec { + let mut problems = level_problems(fan_out); + problems.extend(field_problem(&fan_out.request)); + problems +} + +/// The problems of a fan-out level and of every level above it, whether it loads rows or feeds keys to the level below. +fn level_problems(fan_out: &FanOut) -> Vec { let path = &fan_out.request.path; let wanted = placeholders(path); let mut problems = request_problems(&fan_out.request); - problems.extend(field_problem(&fan_out.request)); problems.extend( (fan_out.concurrency == 0).then(|| "fan_out concurrency must be at least 1".to_owned()), ); - match &fan_out.over { - Over::Paginate(over) => { - problems.extend(request_problems(over)); + let (picker, kind) = match &fan_out.over { + Over::Values { .. } if wanted.is_empty() => { + problems.push(format!( + "fan-out path {path} has no placeholder, so every key would request the same URL; add one such as {{id}}" + )); + return problems; + } + Over::Values { values } => { problems.extend( - (over.fields.is_some() || over.keep_record.is_some()) - .then(|| "fan_out's over picks keys with its select; fields and keep_record are for the fan-out's own rows".to_owned()), + values + .iter() + .find_map(|value| key_problem(path, &wanted, value)), ); + return problems; + } + Over::Paginate(over) => { + problems.extend(request_problems(over)); problems.extend(placeholders(&over.path).first().map(|name| { format!("fan_out over path {} has the placeholder {{{name}}}; over must be a plain paginate", over.path) })); - let field = match selector(over.select.as_deref()) { - Ok(selected) => selected.field, - Err(_) => None, - }; - match field { - Some(field) => problems.extend( - wanted - .iter() - .filter(|name| **name != field) - .map(|name| format!("fan-out path {path} needs {{{name}}}, but over gives only {field}")), - ), - None => problems.push(format!( - "fan_out over a paginate needs a select naming each key's field, such as \"$.items[*].id\", not {:?}", - over.select.clone().unwrap_or_else(|| "$".to_owned()) - )), + (&**over, "a paginate") + } + Over::FanOut(inner) => { + problems.extend(level_problems(inner)); + (&inner.request, "a fan_out") + } + }; + problems.extend( + (picker.fields.is_some() || picker.keep_record.is_some()) + .then(|| "fan_out's over picks keys with its select; fields and keep_record are for the fan-out's own rows".to_owned()), + ); + let Some(name) = key_name(picker) else { + problems.push(format!( + "fan_out over {kind} needs a select naming each key's field, such as \"$.items[*].id\", not {:?}", + picker.select.clone().unwrap_or_else(|| "$".to_owned()) + )); + return problems; + }; + let above = match &fan_out.over { + Over::FanOut(inner) => provided_keys(&inner.over, &placeholders(&inner.request.path)), + Over::Values { .. } | Over::Paginate(_) => Vec::new(), + }; + problems.extend(above.contains(&name).then(|| { + format!( + "fan_out over picks {name}, which a level above already gives; name it apart with key=" + ) + })); + let given: Vec = above.into_iter().chain([name]).collect(); + problems.extend( + wanted + .iter() + .filter(|name| !given.contains(name)) + .map(|name| { + format!( + "fan-out path {path} needs {{{name}}}, but over gives only {}", + given.join(", ") + ) + }), + ); + problems +} + +/// The name the keys `picker` selects fill: its `key`, else the selected field; `None` when its select names no field. +fn key_name(picker: &Request) -> Option { + let field = match selector(picker.select.as_deref()) { + Ok(selected) => selected.field, + Err(_) => None, + }?; + Some(picker.key.clone().unwrap_or(field)) +} + +/// Whether every object of `objects` has the field `name`. +fn in_every(objects: &[&Map], name: &str) -> bool { + objects.iter().all(|object| object.contains_key(name)) +} + +/// The placeholder names every key `over` sends fills, for a level whose path has the placeholders `wanted`. +fn provided_keys(over: &Over, wanted: &[String]) -> Vec { + match over { + Over::Values { values } => { + let objects: Vec<&Map> = + values.iter().filter_map(Value::as_object).collect(); + match objects.split_first() { + Some((first, rest)) if objects.len() == values.len() => first + .keys() + .filter(|name| in_every(rest, name)) + .cloned() + .collect(), + Some(_) | None => wanted.to_vec(), } } - Over::Values { .. } if wanted.is_empty() => problems.push(format!( - "fan-out path {path} has no placeholder, so every key would request the same URL; add one such as {{id}}" - )), - Over::Values { values } => problems.extend( - values - .iter() - .find_map(|value| key_problem(path, &wanted, value)), - ), + Over::Paginate(over) => key_name(over).into_iter().collect(), + Over::FanOut(inner) => provided_keys(&inner.over, &placeholders(&inner.request.path)) + .into_iter() + .chain(key_name(&inner.request)) + .collect(), } - problems } /// Why the listed key `value` cannot fill the placeholders `wanted` of `path`, if it cannot. @@ -311,10 +382,7 @@ async fn run_job( ) -> Result<(), ReadError> { let posts = match &job { HttpJob::Paginate(request) => request.method == Method::Post, - HttpJob::FanOut(fan_out) => { - fan_out.request.method == Method::Post - || matches!(&fan_out.over, Over::Paginate(over) if over.method == Method::Post) - } + HttpJob::FanOut(fan_out) => fan_out_posts(fan_out), }; let http = job_client(posts)?; match job { @@ -322,10 +390,22 @@ async fn run_job( let calls = request_calls(&request, &connections, &http)?; paginate(&calls, &request, &request.path, &pages).await } - HttpJob::FanOut(fan_out) => run_fan_out(*fan_out, &connections, &http, pages).await, + HttpJob::FanOut(fan_out) => { + run_fan_out(*fan_out, &connections, &http, Sink::Pages(pages)).await + } } } +/// Whether any level of `fan_out` sends a `POST`. +fn fan_out_posts(fan_out: &FanOut) -> bool { + fan_out.request.method == Method::Post + || match &fan_out.over { + Over::Values { .. } => false, + Over::Paginate(over) => over.method == Method::Post, + Over::FanOut(inner) => fan_out_posts(inner), + } +} + /// A pooled client that follows redirects only within the first request's origin, so connection headers and credentials never reach another host; when the job `posts`, it also stops at 301, 302 and 303, which would resend the request as a `GET` without its body. fn job_client(posts: bool) -> Result { install_tls_provider(); @@ -386,58 +466,109 @@ fn request_calls( }) } -/// Page through the fan-out's path for each key, up to `concurrency` at once, while the keys still arrive. -async fn run_fan_out( +/// Page through the fan-out's path for each key, up to `concurrency` at once, while the keys still arrive, sending the pages to `sink`. In a job of more than one fan-out level, errors start with the trail of requests that led to the failing one. +fn run_fan_out<'a>( fan_out: FanOut, - connections: &Map, - http: &reqwest::Client, - pages: Sender, + connections: &'a Map, + http: &'a reqwest::Client, + sink: Sink, +) -> LevelRun<'a> { + Box::pin(async move { + let calls = request_calls(&fan_out.request, connections, http)?; + let (key_sender, key_receiver) = channel::(fan_out.concurrency * 2); + let keys = Arc::new(Mutex::new(key_receiver)); + let mut tasks: JoinSet> = JoinSet::new(); + let nested = matches!(sink, Sink::Keys(_)) || matches!(fan_out.over, Over::FanOut(_)); + let feed = match fan_out.over.clone() { + Over::Values { values } => KeyFeed::Values(values), + Over::Paginate(over) => { + KeyFeed::Pages(Box::new((request_calls(&over, connections, http)?, *over))) + } + Over::FanOut(inner) => { + KeyFeed::FanOut(Box::new((*inner, connections.clone(), http.clone()))) + } + }; + let wanted = placeholders(&fan_out.request.path); + tasks.spawn(feed_keys(feed, wanted, key_sender)); + let request = Arc::new(fan_out.request); + for _ in 0..fan_out.concurrency { + let (calls, request, keys, sink) = + (calls.clone(), request.clone(), keys.clone(), sink.clone()); + tasks.spawn(async move { + loop { + let key = keys.lock().await.recv().await; + let Some(key) = key else { return Ok(()) }; + let trail = match nested { + true => redacted(&key.trail, &calls.endpoint.current_secrets()), + false => String::new(), + }; + fan_out_key(&calls, &request, &key, &sink) + .await + .map_err(|error| ReadError { + message: format!("{trail}{}", error.message), + retryable: error.retryable, + })?; + } + }); + } + while let Some(finished) = tasks.join_next().await { + finished.map_err(|error| format!("a fan-out request stopped: {error}"))??; + } + Ok(()) + }) +} + +/// A fan-out level's run, boxed so a level can run the level above it. +type LevelRun<'a> = Pin> + Send + 'a>>; + +/// Every page of `request` for `key`: out as rows, or picked into keys for the level below, each keeping `key`'s fields. +async fn fan_out_key( + calls: &Calls, + request: &Request, + key: &FanKey, + sink: &Sink, ) -> Result<(), ReadError> { - let calls = request_calls(&fan_out.request, connections, http)?; - let (key_sender, key_receiver) = channel::>(fan_out.concurrency * 2); - let keys = Arc::new(Mutex::new(key_receiver)); - let mut tasks: JoinSet> = JoinSet::new(); - let feed = match fan_out.over.clone() { - Over::Values { values } => KeyFeed::Values(values), - Over::Paginate(over) => { - KeyFeed::Pages(Box::new((request_calls(&over, connections, http)?, *over))) + let path = filled_path(&request.path, &key.fields)?; + match sink { + Sink::Pages(pages) => paginate(calls, request, &path, pages).await, + Sink::Keys(keys) => { + let picker = key_picker(request)?; + let (page_sender, page_receiver) = channel::(PAGES_AHEAD); + let shown = path.clone(); + let pager = async move { paginate(calls, request, &path, &page_sender).await }; + let forward = forward_keys(page_receiver, &picker, (key, &shown), keys); + let (paged, forwarded) = tokio::join!(pager, forward); + paged.and(forwarded) } - }; - let wanted = placeholders(&fan_out.request.path); - tasks.spawn(feed_keys(feed, wanted, key_sender)); - let request = Arc::new(fan_out.request); - for _ in 0..fan_out.concurrency { - let (calls, request, keys, pages) = - (calls.clone(), request.clone(), keys.clone(), pages.clone()); - tasks.spawn(async move { - loop { - let key = keys.lock().await.recv().await; - let Some(key) = key else { return Ok(()) }; - let path = filled_path(&request.path, &key)?; - paginate(&calls, &request, &path, &pages).await?; - } - }); } - while let Some(finished) = tasks.join_next().await { - finished.map_err(|error| format!("a fan-out request stopped: {error}"))??; - } - Ok(()) } -/// Send each key: a value fills the path's one placeholder, an object fills them by name, and a paginate's selected field is named after itself. +/// What a key-picking job selects: the rows, the field of each to read, and the name it fills. +fn key_picker(request: &Request) -> Result<(Selector, String, String), String> { + let selected = selector(request.select.as_deref())?; + let field = selected.field.clone().unwrap_or_default(); + let name = request.key.clone().unwrap_or_else(|| field.clone()); + Ok((selected, field, name)) +} + +/// Send each key: a value fills the path's one placeholder, an object fills them by name, and a paginate's or fan-out's selected field is named after itself or its `key`. async fn feed_keys( feed: KeyFeed, wanted: Vec, - keys: Sender>, + keys: Sender, ) -> Result<(), ReadError> { - let single = |value: Value| -> Map { - match value { + let single = |value: Value| -> FanKey { + let fields = match value { Value::Object(key) => key, other => wanted .first() .map(|name| (name.clone(), other)) .into_iter() .collect(), + }; + FanKey { + fields, + trail: String::new(), } }; match feed { @@ -451,22 +582,28 @@ async fn feed_keys( } KeyFeed::Pages(pages) => { let (calls, over) = *pages; - let selected = selector(over.select.as_deref())?; - let field = selected.field.clone().unwrap_or_default(); + let picker = key_picker(&over)?; + let path = over.path.clone(); let (page_sender, page_receiver) = channel::(PAGES_AHEAD); let pager = async move { paginate(&calls, &over, &over.path, &page_sender).await }; - let forward = forward_keys(page_receiver, (&selected, &field), &keys); + let root = FanKey::default(); + let forward = forward_keys(page_receiver, &picker, (&root, &path), &keys); let (paged, forwarded) = tokio::join!(pager, forward); paged.and(forwarded) } + KeyFeed::FanOut(level) => { + let (inner, connections, http) = *level; + run_fan_out(inner, &connections, &http, Sink::Keys(keys)).await + } } } -/// Each page's selected field values as keys named after the field. +/// Each page's picked field values as keys named by `picker`, each with `parent`'s fields and its trail extended by the request at `path` that gave it. async fn forward_keys( mut pages: Receiver, - (selected, field): (&Selector, &str), - keys: &Sender>, + (selected, field, name): &(Selector, String, String), + (parent, path): (&FanKey, &str), + keys: &Sender, ) -> Result<(), ReadError> { while let Some(page) = pages.recv().await { let document: Value = serde_json::from_slice(&page.bytes) @@ -475,8 +612,14 @@ async fn forward_keys( let Some(value) = row.get(field).filter(|value| !value.is_null()) else { continue; }; - let key: Map = [(field.to_owned(), value.clone())].into_iter().collect(); - if keys.send(key).await.is_err() { + let mut fields = parent.fields.clone(); + fields.insert(name.clone(), value.clone()); + let shown = match value { + Value::String(text) => text.clone(), + other => other.to_string(), + }; + let trail = format!("{}{path} {name}={shown} → ", parent.trail); + if keys.send(FanKey { fields, trail }).await.is_err() { return Ok(()); } } diff --git a/crates/sqlload/src/http_source/models.rs b/crates/sqlload/src/http_source/models.rs index 57da2d7f..c1e68b22 100644 --- a/crates/sqlload/src/http_source/models.rs +++ b/crates/sqlload/src/http_source/models.rs @@ -4,6 +4,7 @@ use std::time::{Duration, Instant}; use serde::Deserialize; use serde_json::{Map, Value}; +use tokio::sync::mpsc::Sender; /// A job a table function built with `sl.http(...)`: run entirely in Rust, its pages parsed like `sl.json(...)` bytes. #[derive(Debug, Clone, Deserialize)] @@ -42,6 +43,9 @@ pub struct Request { /// A column holding each row's whole object as JSON. #[serde(default)] pub keep_record: Option, + /// For a job used as a fan-out's `over`: the name its picked values fill as keys, when not the selected field's own (such as `order_id` for `$.items[*].id`). + #[serde(default)] + pub key: Option, } /// The HTTP method of a job's requests. @@ -106,6 +110,8 @@ pub enum Over { Values { values: Vec }, /// The values its `select` (`$.items[*].id`) picks from every page, named after the last field. Paginate(Box), + /// The values another fan-out's `select` picks from each of its pages, each key keeping the fields of the key above it, so a chain of fan-outs goes as many levels deep as it nests. + FanOut(Box), } /// The connection a job uses and how it calls it. @@ -348,9 +354,24 @@ pub struct Fetched { pub url: reqwest::Url, } -/// Where a fan-out's keys come from: listed values, or another endpoint's pages. +/// Where a fan-out's keys come from: listed values, another endpoint's pages, or a fan-out one level up, run with the job's connections and client. #[derive(Debug)] pub enum KeyFeed { Values(Vec), Pages(Box<(Calls, Request)>), + FanOut(Box<(FanOut, Map, reqwest::Client)>), +} + +/// A fan-out key: the placeholder values it fills, its ancestors' included, and the trail of requests that produced it (such as `/orders order_id=7 → `), which prefixes a multi-level job's errors. +#[derive(Debug, Clone, Default)] +pub struct FanKey { + pub fields: Map, + pub trail: String, +} + +/// Where a fan-out level's pages go: out as the job's rows, or into keys for the level below. +#[derive(Debug, Clone)] +pub enum Sink { + Pages(Sender), + Keys(Sender), } diff --git a/crates/sqlload/src/http_source/tests/test_job_problems.rs b/crates/sqlload/src/http_source/tests/test_job_problems.rs index 7a3ce9fb..87aabcfd 100644 --- a/crates/sqlload/src/http_source/tests/test_job_problems.rs +++ b/crates/sqlload/src/http_source/tests/test_job_problems.rs @@ -49,6 +49,25 @@ fn given_a_job_definition_when_checking_it_then_every_problem_is_found_before_an "fan-out path /{region}/orders/{id} needs {id}, which the key {\"region\":\"uk\"} lacks", ], }, + JobProblemsTestCase { + description: "a third level may use the keys of both levels above it", + job: r#"{"kind":"fan_out","client":{"connection":"shop","retries":0},"path":"/o/{order_id}/i/{item_id}/n","paginator":{"kind":"single"},"concurrency":2,"over":{"kind":"fan_out","client":{"connection":"shop","retries":0},"path":"/o/{order_id}/i","paginator":{"kind":"single"},"select":"$.items[*].item_id","concurrency":2,"over":{"kind":"paginate","client":{"connection":"shop","retries":0},"path":"/o","paginator":{"kind":"single"},"select":"$.items[*].order_id"}}}"#, + expected_problems: &[], + }, + JobProblemsTestCase { + description: "a third level's placeholder that no level above gives", + job: r#"{"kind":"fan_out","client":{"connection":"shop","retries":0},"path":"/o/{order_id}/i/{item_id}/n/{customer_id}","paginator":{"kind":"single"},"concurrency":2,"over":{"kind":"fan_out","client":{"connection":"shop","retries":0},"path":"/o/{order_id}/i","paginator":{"kind":"single"},"select":"$.items[*].item_id","concurrency":2,"over":{"kind":"paginate","client":{"connection":"shop","retries":0},"path":"/o","paginator":{"kind":"single"},"select":"$.items[*].order_id"}}}"#, + expected_problems: &[ + "fan-out path /o/{order_id}/i/{item_id}/n/{customer_id} needs {customer_id}, but over gives only order_id, item_id", + ], + }, + JobProblemsTestCase { + description: "a level that picks a key name a level above already gives", + job: r#"{"kind":"fan_out","client":{"connection":"shop","retries":0},"path":"/o/{order_id}/n","paginator":{"kind":"single"},"concurrency":2,"over":{"kind":"fan_out","client":{"connection":"shop","retries":0},"path":"/o/{order_id}/i","paginator":{"kind":"single"},"select":"$.items[*].order_id","concurrency":2,"over":{"kind":"paginate","client":{"connection":"shop","retries":0},"path":"/o","paginator":{"kind":"single"},"select":"$.items[*].order_id"}}}"#, + expected_problems: &[ + "fan_out over picks order_id, which a level above already gives; name it apart with key=", + ], + }, JobProblemsTestCase { description: "a pointer that is not a path, a zero limit and a bad header", job: r#"{"kind":"paginate","client":{"connection":"shop","retries":0,"headers":{"X Bad":"1"}},"path":"/o","paginator":{"kind":"next_link","pointer":"next"}}"#, diff --git a/packages/sqlload/python/sqlload/_http.py b/packages/sqlload/python/sqlload/_http.py index 0d66f564..3c8b1dde 100644 --- a/packages/sqlload/python/sqlload/_http.py +++ b/packages/sqlload/python/sqlload/_http.py @@ -192,14 +192,16 @@ def paginate( keep_record: str | None = None, method: str = "GET", json_body: Mapping[str, Any] | None = None, + key: str | None = None, ) -> Job: """Every page of `path`; `select` (such as `"$.items"`) is where the rows are, `on_status={404: "skip"}` treats a status as no rows. `fields` loads only those values of each row, renamed (`{"account_code": "$.code"}`); `keep_record` adds each row's whole object as a JSON column. `method="POST"` sends `json_body` as JSON with every page, each page's copy - holding the paginator's values when its `location` is `"body"`.""" + holding the paginator's values when its `location` is `"body"`. As a fan-out's `over`, `key` names the + placeholder the selected values fill when it is not the field's own name (`select="$.items[*].id", key="order_id"`).""" request = self._request(path, paginator, params, select, on_status, (method, json_body)) - return Job({"kind": "paginate", **request, **_projection(fields, keep_record)}) + return Job({"kind": "paginate", **request, **_projection(fields, keep_record), **_key(key)}) def fan_out( self, @@ -215,22 +217,21 @@ def fan_out( keep_record: str | None = None, method: str = "GET", json_body: Mapping[str, Any] | None = None, + key: str | None = None, ) -> Job: """Every page of `path` once per key of `over`, up to `concurrency` at a time: `over` is a paginate whose `select` picks the keys (`"$.items[*].id"` fills `{id}`), or values (`["uk", "fr"]` fill the one placeholder; - dicts fill them by name).""" - if isinstance(over, Job) and over._definition["kind"] != "paginate": - raise TypeError("fan_out over= takes a paginate job or values, not another fan_out") + dicts fill them by name). + + `over` can also be another fan_out whose `select` picks keys, as deep as they nest: each key keeps the + fields of the keys above it, so a third level's path can use `{order_id}` and `{item_id}`. `key` names the + placeholder this fan-out's selected values fill when it is itself an `over`.""" if isinstance(over, (str, bytes)): - raise TypeError("fan_out over= takes a paginate job or a list of values, not one string") - over_spec = ( - {"kind": "paginate", **{k: v for k, v in over._definition.items() if k != "kind"}} - if isinstance(over, Job) - else {"kind": "values", "values": list(over)} - ) + raise TypeError("fan_out over= takes a paginate or fan_out job or a list of values, not one string") + over_spec = dict(over._definition) if isinstance(over, Job) else {"kind": "values", "values": list(over)} request = self._request(path, paginator or single(), params, select, on_status, (method, json_body)) projected = _projection(fields, keep_record) - return Job({"kind": "fan_out", **request, **projected, "over": over_spec, "concurrency": concurrency}) + return Job({"kind": "fan_out", **request, **projected, **_key(key), "over": over_spec, "concurrency": concurrency}) def _request( self, @@ -256,6 +257,11 @@ def _request( } +def _key(key: str | None) -> dict[str, Any]: + """`key` as a job field, left out when unset so jobs without it are unchanged.""" + return {} if key is None else {"key": key} + + def _projection(fields: Fields | None, keep_record: str | None) -> dict[str, Any]: listed = field_list(fields) return { diff --git a/packages/sqlload/tests/test_http_jobs.py b/packages/sqlload/tests/test_http_jobs.py index c6c2cf1b..75aa0b5c 100644 --- a/packages/sqlload/tests/test_http_jobs.py +++ b/packages/sqlload/tests/test_http_jobs.py @@ -68,6 +68,8 @@ def do_GET(self) -> None: }.get(url.path) if route is not None: return route(query) + if url.path.startswith("/v1/tree/"): + return self.tree(url.path, query) parts = url.path.split("/") if len(parts) == 5 and parts[2] == "orders" and parts[4] == "items": order = int(parts[3]) @@ -176,6 +178,24 @@ def offset(self, query: dict[str, str]) -> None: start, limit = int(query["offset"]), int(query["limit"]) self.reply(200, ORDERS[start : start + limit]) + def tree(self, path: str, query: dict[str, str]) -> None: + """Three levels: orders after `after`, three a page; order o's items `o*10+k` for k < o % 3 (none for 3 and 6, 404 for 4); an item's notes, + two a page for pages 1 and 2, a 404 when the item is not the order's, and a 500 on item 51's page 2 with `boom`.""" + parts = path.split("/")[3:] + if len(parts) == 1: + page, after = int(query["p"]), int(query.get("after") or 0) + return self.reply(200, {"items": [{"order_id": n} for n in range(after + 1, 8)][(page - 1) * 3 : page * 3]}) + order = int(parts[1]) + if len(parts) == 3: + return self.reply(404, {"message": "no order"}) if order == 4 else self.reply(200, {"items": [{"item_id": order * 10 + k} for k in range(order % 3)]}) + item, page = int(parts[3]), int(query["page"]) + if item // 10 != order: + return self.reply(404, {"message": f"item {item} is not order {order}'s"}) + if query.get("boom") and item == 51 and page == 2: + return self.reply(500, {"message": "notes are down"}) + notes = [{"order_id": order, "item_id": item, "note": f"{item}-{page}-{n}"} for n in range(2)] if page <= 2 else [] + self.reply(200, {"items": notes}) + def page_number(self, query: dict[str, str]) -> None: page = int(query["p"]) self.reply(200, {"items": ORDERS[(page - 1) * 3 : page * 3]}) @@ -832,3 +852,94 @@ def test_given_a_bad_body_setting_when_loading_then_no_request_is_sent(tmp_path: assert done.returncode != 0 assert expected_text in done.stdout + done.stderr, done.stdout + done.stderr assert api.log == [] and api.bodies == [] # type: ignore[attr-defined] + + +TREE = """ +orders = api.paginate('/tree/orders', paginator=sl.page_number('p'), params={'after': AFTER}, select='$.items[*].order_id') +items = api.fan_out('/tree/orders/{order_id}/items', over=orders, concurrency=3, select='$.items[*].item_id', on_status=ON_STATUS) +""" +TREE_ITEMS = [10, 20, 21, 50, 51, 70] + + +def tree_project(tmp_path: Path, api: ThreadingHTTPServer, notes: str, *, after: str = "None", on_status: str = "{404: 'skip'}", table: str = "") -> Path: + body = TREE.replace("AFTER", after).replace("ON_STATUS", on_status) + tables = f"@shop.table(description='Notes.'{table})\ndef notes(ctx):\n" + textwrap.indent(body, " ") + f" return {notes}\n" + return make_project(tmp_path, api, tables) + + +NOTES = "api.fan_out('/tree/orders/{order_id}/items/{item_id}/notes', over=items, concurrency=3, paginator=sl.page_number('page'), select='$.items')" + + +def test_given_three_fan_out_levels_when_loading_then_each_note_is_read_under_its_own_order_and_item(tmp_path: Path, api: ThreadingHTTPServer) -> None: + project = tree_project(tmp_path, api, NOTES) + done = sqlload(project, "load") + assert done.returncode == 0, done.stdout + done.stderr + expected = sorted((item // 10, item, f"{item}-{page}-{n}") for item in TREE_ITEMS for page in (1, 2) for n in range(2)) + assert query(project, 'SELECT order_id, item_id, note FROM "RAW".shop__notes ORDER BY item_id, note') == expected + notes = {(path, query["page"]) for path, query, _ in api.log if path.endswith("/notes")} # type: ignore[attr-defined] + assert notes == {(f"/v1/tree/orders/{item // 10}/items/{item}/notes", str(page)) for item in TREE_ITEMS for page in (1, 2, 3)} + assert sorted(path for path, _, _ in api.log if path.endswith("/items")) == [f"/v1/tree/orders/{order}/items" for order in range(1, 8)] # type: ignore[attr-defined] + assert [query["p"] for path, query, _ in api.log if path == "/v1/tree/orders"] == ["1", "2", "3", "4"] # type: ignore[attr-defined] + + +@pytest.mark.parametrize( + ("on_status", "notes", "expected_code", "expected_text"), + [ + ( + "{}", + NOTES, + 1, + "/tree/orders order_id=4 → GET http://127.0.0.1:PORT/v1/tree/orders/4/items: HTTP 404", + ), + ( + "{404: 'skip'}", + NOTES.replace("api.fan_out", "once.fan_out").replace("sl.page_number('page')", "sl.page_number('page'), params={'boom': 1}"), + 75, + "/tree/orders order_id=5 → /tree/orders/5/items item_id=51 → GET http://127.0.0.1:PORT/v1/tree/orders/5/items/51/notes?boom=1&page=2: HTTP 500", + ), + ], + ids=["a-404-at-level-2-names-its-order", "a-500-at-level-3-names-its-order-and-item-and-retries-per-level"], +) +def test_given_a_failing_level_when_loading_then_the_error_names_the_keys_that_led_to_it( + tmp_path: Path, api: ThreadingHTTPServer, on_status: str, notes: str, expected_code: int, expected_text: str +) -> None: + project = tree_project(tmp_path, api, notes, on_status=on_status) + shop_py = project / "sources" / "shop.py" + shop_py.write_text(shop_py.read_text().replace("\n\n@shop.table", "\nonce = sl.http(shop, auth=sl.bearer(), retries=0)\n\n@shop.table", 1)) + done = sqlload(project, "load") + output = done.stdout + done.stderr + assert done.returncode == expected_code, output + assert expected_text.replace("PORT", str(api.server_address[1])) in output, output + boomed = [path for path, query, _ in api.log if path.endswith("/51/notes") and query["page"] == "2"] # type: ignore[attr-defined] + assert len(boomed) == (1 if expected_code == 75 else 0) + + +def test_given_a_cursor_on_the_top_level_when_loading_twice_then_the_rerun_requests_no_lower_level(tmp_path: Path, api: ThreadingHTTPServer) -> None: + project = tree_project(tmp_path, api, NOTES, after="ctx.cursor_value", table=", write_strategy='append', cursor_column='order_id'") + assert sqlload(project, "load").returncode == 0 + first = len(api.log) # type: ignore[attr-defined] + done = sqlload(project, "load") + assert done.returncode == 0, done.stdout + done.stderr + rerun = api.log[first:] # type: ignore[attr-defined] + assert [(path, query.get("after")) for path, query, _ in rerun] == [("/v1/tree/orders", "7")] + assert query(project, 'SELECT count(*), count(DISTINCT note) FROM "RAW".shop__notes') == [(24, 24)] + + +ONE_LEVEL_JOB = ( + '{"kind": "fan_out", "client": {"connection": "shop", "auth": {"kind": "bearer"}, "rate_limit": null, "retries": 3, "headers": {}}, ' + '"path": "/orders/{order_id}/items", "params": {}, "method": "GET", "json_body": null, "paginator": {"kind": "single"}, "select": "$.items", ' + '"on_status": {"404": "skip"}, "fields": null, "keep_record": null, "over": {"kind": "paginate", "client": {"connection": "shop", ' + '"auth": {"kind": "bearer"}, "rate_limit": null, "retries": 3, "headers": {}}, "path": "/orders", "params": {}, "method": "GET", ' + '"json_body": null, "paginator": {"kind": "page_number", "param": "p", "start": 1, "location": "query"}, "select": "$.items[*].order_id", ' + '"on_status": {}, "fields": null, "keep_record": null}, "concurrency": 3}' +) + + +def test_given_a_one_level_fan_out_when_defined_then_its_job_is_unchanged_by_multi_level_support() -> None: + code = ( + "import sqlload as sl\nshop = sl.source('shop', connection='shop')\napi = sl.http(shop, auth=sl.bearer(), retries=3)\n" + "orders = api.paginate('/orders', paginator=sl.page_number('p'), select='$.items[*].order_id')\n" + "print(api.fan_out('/orders/{order_id}/items', over=orders, concurrency=3, select='$.items', on_status={404: 'skip'}).__sqlload_http_job__())" + ) + done = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, timeout=60) + assert done.stdout.strip() == ONE_LEVEL_JOB, done.stderr diff --git a/website/src/content/docs/docs/sources/python.mdx b/website/src/content/docs/docs/sources/python.mdx index 81ae6c0a..f411d1e3 100644 --- a/website/src/content/docs/docs/sources/python.mdx +++ b/website/src/content/docs/docs/sources/python.mdx @@ -221,7 +221,7 @@ thread-safe. Set `parallel_tables=True` on the source when its client is. |----------|---------| | `sl.http(source, *, auth=None, rate_limit=None, retries=5, headers=None, connection=None)` | Requests to the source's `http` connection. `rate_limit` (`"10/s"`, `"600/m"`, `"1000/h"`) is shared by every job on the connection with the same limit. | | `api.paginate(path, *, paginator, params=None, select=None, on_status=None, fields=None, keep_record=None, method="GET", json_body=None)` | Every page of `path`. `select` is where the rows are: `$`, or a dotted path such as `$.data.items`. `fields` and `keep_record` (from 0.27.0) pick and rename each row's values; see [Fields](#fields). `method` and `json_body` are from 0.36.0 (not released yet); see [POST with a JSON body](#post-with-a-json-body). | -| `api.fan_out(path, *, over, concurrency=8, paginator=None, params=None, select=None, on_status=None, fields=None, keep_record=None, method="GET", json_body=None)` | `path` once per key of `over`, up to `concurrency` at a time, while keys still arrive. `fields` and `keep_record` are from 0.27.0. `over` is a paginate whose `select` picks keys (`$.items[*].id` fills `{id}`), or a list: a value fills the path's one placeholder, and a dict fills placeholders by name. `method` and `json_body` are from 0.36.0 (not released yet), as for `paginate`. | +| `api.fan_out(path, *, over, concurrency=8, paginator=None, params=None, select=None, on_status=None, fields=None, keep_record=None, method="GET", json_body=None)` | `path` once per key of `over`, up to `concurrency` at a time, while keys still arrive. `fields` and `keep_record` are from 0.27.0. `over` is a paginate whose `select` picks keys (`$.items[*].id` fills `{id}`), or a list: a value fills the path's one placeholder, and a dict fills placeholders by name. From 0.37.0 (not released yet), `over` can also be another `fan_out`; see [Multi-level fan-out](#multi-level-fan-out). `method` and `json_body` are from 0.36.0 (not released yet), as for `paginate`. | | Paginator | Next page | |-----------|-----------| @@ -251,6 +251,36 @@ thread-safe. Set `parallel_tables=True` on the source when its client is. - A job's rows never come back into Python. Reading a row from a job fails (rule `python_row_loop`); filter rows downstream. +### Multi-level fan-out + +From 0.37.0 (not released yet). A `fan_out` whose `select` picks keys can be the `over` of another +`fan_out`, as many levels deep as they nest: list → detail → sub-detail. Each key keeps the fields +of the keys above it, so a third level's path can use `{order_id}` as well as `{item_id}`. When a +level's field has the same name as one above it (every level calls it `id`), name it apart with +`key=`, as in `select="$.items[*].id", key="item_id"`. + +```python +@shop.table(description="Order notes.", write_strategy="append", cursor_column="order_id") +def notes(ctx): + orders = api.paginate("/orders", paginator=sl.page_number("p"), + params={"after": ctx.cursor_value}, select="$.items[*].order_id") + items = api.fan_out("/orders/{order_id}/items", over=orders, select="$.items[*].item_id") + return api.fan_out("/orders/{order_id}/items/{item_id}/notes", over=items, + paginator=sl.page_number("page"), select="$.items") +``` + +- Every level has its own paginator, `concurrency`, `on_status` and, through the `sl.http` it + comes from, its own retries and rate limit. Up to the sum of the levels' `concurrency` requests + are in flight at once; a shared `rate_limit` caps them. +- An incremental cursor goes on the top level, as above; lower levels only see the keys it picks. +- Only the last level's rows load. A level that should be a table of its own is a table function + whose job stops at that level, so the levels above it are requested again for that table. +- Rows load as the API returns them: a key's fields are not added to them, so select an ID the + response repeats, such as `order_id` above. +- An error names the keys that led to it, such as `/orders order_id=5 → /orders/5/items item_id=51 → + GET .../notes?page=2: HTTP 500`. +- `plan` checks every level before any request: each placeholder must come from a level above. + ### POST with a JSON body From 0.36.0 (not released yet). `method="POST"` sends `json_body` with every request, as From 8bd987c49be803f2f07d9a07a96aeb99142b8ea6 Mon Sep 17 00:00:00 2001 From: Kevin Longe <273747162+kvlonge@users.noreply.github.com> Date: Thu, 12 Nov 2026 17:14:50 +0000 Subject: [PATCH 2/2] fix(http): keep one-level plan error order and redact each breadcrumb step with its own level's secrets Also run key= renames end to end and correct the in-flight request count in the docs. --- .../sqlload/src/http_source/_helpers/job.rs | 75 +++++++++++++------ .../http_source/tests/test_job_problems.rs | 16 ++++ packages/sqlload/tests/test_http_jobs.py | 13 ++++ .../src/content/docs/docs/sources/python.mdx | 4 +- 4 files changed, 84 insertions(+), 24 deletions(-) diff --git a/crates/sqlload/src/http_source/_helpers/job.rs b/crates/sqlload/src/http_source/_helpers/job.rs index 92600dd9..2bc90a7e 100644 --- a/crates/sqlload/src/http_source/_helpers/job.rs +++ b/crates/sqlload/src/http_source/_helpers/job.rs @@ -215,16 +215,17 @@ fn header_problem(name: &str, value: &str) -> Option { } fn fan_out_problems(fan_out: &FanOut) -> Vec { - let mut problems = level_problems(fan_out); + let mut problems = request_problems(&fan_out.request); problems.extend(field_problem(&fan_out.request)); + problems.extend(level_problems(fan_out)); problems } -/// The problems of a fan-out level and of every level above it, whether it loads rows or feeds keys to the level below. +/// The problems of a fan-out level apart from its own request's, and of every level above it, whether it loads rows or feeds keys to the level below. fn level_problems(fan_out: &FanOut) -> Vec { let path = &fan_out.request.path; let wanted = placeholders(path); - let mut problems = request_problems(&fan_out.request); + let mut problems: Vec = Vec::new(); problems.extend( (fan_out.concurrency == 0).then(|| "fan_out concurrency must be at least 1".to_owned()), ); @@ -245,20 +246,19 @@ fn level_problems(fan_out: &FanOut) -> Vec { } Over::Paginate(over) => { problems.extend(request_problems(over)); + problems.extend(picker_problem(over)); problems.extend(placeholders(&over.path).first().map(|name| { format!("fan_out over path {} has the placeholder {{{name}}}; over must be a plain paginate", over.path) })); (&**over, "a paginate") } Over::FanOut(inner) => { + problems.extend(request_problems(&inner.request)); problems.extend(level_problems(inner)); + problems.extend(picker_problem(&inner.request)); (&inner.request, "a fan_out") } }; - problems.extend( - (picker.fields.is_some() || picker.keep_record.is_some()) - .then(|| "fan_out's over picks keys with its select; fields and keep_record are for the fan-out's own rows".to_owned()), - ); let Some(name) = key_name(picker) else { problems.push(format!( "fan_out over {kind} needs a select naming each key's field, such as \"$.items[*].id\", not {:?}", @@ -290,6 +290,12 @@ fn level_problems(fan_out: &FanOut) -> Vec { problems } +/// Why `picker`, which feeds keys to the level below, cannot have `fields` or `keep_record`, if it has them. +fn picker_problem(picker: &Request) -> Option { + (picker.fields.is_some() || picker.keep_record.is_some()) + .then(|| "fan_out's over picks keys with its select; fields and keep_record are for the fan-out's own rows".to_owned()) +} + /// The name the keys `picker` selects fill: its `key`, else the selected field; `None` when its select names no field. fn key_name(picker: &Request) -> Option { let field = match selector(picker.select.as_deref()) { @@ -489,7 +495,7 @@ fn run_fan_out<'a>( } }; let wanted = placeholders(&fan_out.request.path); - tasks.spawn(feed_keys(feed, wanted, key_sender)); + tasks.spawn(feed_keys(feed, wanted, key_sender, nested)); let request = Arc::new(fan_out.request); for _ in 0..fan_out.concurrency { let (calls, request, keys, sink) = @@ -498,14 +504,17 @@ fn run_fan_out<'a>( loop { let key = keys.lock().await.recv().await; let Some(key) = key else { return Ok(()) }; - let trail = match nested { - true => redacted(&key.trail, &calls.endpoint.current_secrets()), - false => String::new(), - }; fan_out_key(&calls, &request, &key, &sink) .await .map_err(|error| ReadError { - message: format!("{trail}{}", error.message), + message: match nested { + true => format!( + "{}{}", + redacted(&key.trail, &calls.endpoint.current_secrets()), + error.message + ), + false => error.message, + }, retryable: error.retryable, })?; } @@ -536,7 +545,8 @@ async fn fan_out_key( let (page_sender, page_receiver) = channel::(PAGES_AHEAD); let shown = path.clone(); let pager = async move { paginate(calls, request, &path, &page_sender).await }; - let forward = forward_keys(page_receiver, &picker, (key, &shown), keys); + let secrets = calls.endpoint.current_secrets(); + let forward = forward_keys(page_receiver, &picker, (key, &shown, Some(&secrets)), keys); let (paged, forwarded) = tokio::join!(pager, forward); paged.and(forwarded) } @@ -556,6 +566,7 @@ async fn feed_keys( feed: KeyFeed, wanted: Vec, keys: Sender, + nested: bool, ) -> Result<(), ReadError> { let single = |value: Value| -> FanKey { let fields = match value { @@ -584,10 +595,16 @@ async fn feed_keys( let (calls, over) = *pages; let picker = key_picker(&over)?; let path = over.path.clone(); + let secrets = nested.then(|| calls.endpoint.current_secrets()); let (page_sender, page_receiver) = channel::(PAGES_AHEAD); let pager = async move { paginate(&calls, &over, &over.path, &page_sender).await }; let root = FanKey::default(); - let forward = forward_keys(page_receiver, &picker, (&root, &path), &keys); + let forward = forward_keys( + page_receiver, + &picker, + (&root, &path, secrets.as_deref()), + &keys, + ); let (paged, forwarded) = tokio::join!(pager, forward); paged.and(forwarded) } @@ -598,11 +615,11 @@ async fn feed_keys( } } -/// Each page's picked field values as keys named by `picker`, each with `parent`'s fields and its trail extended by the request at `path` that gave it. +/// Each page's picked field values as keys named by `picker`, each with `parent`'s fields and, when `secrets` is given, its trail extended by the request at `path` that gave it, with `secrets` (that request's) redacted. async fn forward_keys( mut pages: Receiver, (selected, field, name): &(Selector, String, String), - (parent, path): (&FanKey, &str), + (parent, path, secrets): (&FanKey, &str, Option<&[String]>), keys: &Sender, ) -> Result<(), ReadError> { while let Some(page) = pages.recv().await { @@ -614,11 +631,9 @@ async fn forward_keys( }; let mut fields = parent.fields.clone(); fields.insert(name.clone(), value.clone()); - let shown = match value { - Value::String(text) => text.clone(), - other => other.to_string(), - }; - let trail = format!("{}{path} {name}={shown} → ", parent.trail); + let trail = secrets.map_or_else(String::new, |secrets| { + trail_step(&parent.trail, (path, name, value), secrets) + }); if keys.send(FanKey { fields, trail }).await.is_err() { return Ok(()); } @@ -626,3 +641,19 @@ async fn forward_keys( } Ok(()) } + +/// `trail` followed by the request at `path` and the key `name` = `value` it gave, with `secrets` redacted from the step. +fn trail_step( + trail: &str, + (path, name, value): (&str, &str, &Value), + secrets: &[String], +) -> String { + let shown = match value { + Value::String(text) => text.clone(), + other => other.to_string(), + }; + format!( + "{trail}{}", + redacted(&format!("{path} {name}={shown} → "), secrets) + ) +} diff --git a/crates/sqlload/src/http_source/tests/test_job_problems.rs b/crates/sqlload/src/http_source/tests/test_job_problems.rs index 87aabcfd..a145ba85 100644 --- a/crates/sqlload/src/http_source/tests/test_job_problems.rs +++ b/crates/sqlload/src/http_source/tests/test_job_problems.rs @@ -109,6 +109,22 @@ fn given_a_job_definition_when_checking_it_then_every_problem_is_found_before_an "fan_out's over picks keys with its select; fields and keep_record are for the fan-out's own rows", ], }, + JobProblemsTestCase { + description: "a one-level fan-out's own select problem comes before its listed key's, as before multi-level fan-out", + job: r#"{"kind":"fan_out","client":{"connection":"shop","retries":0},"path":"/{region}/orders/{id}","paginator":{"kind":"single"},"select":"$.items[*].sku","concurrency":2,"over":{"kind":"values","values":[{"region":"uk"}]}}"#, + expected_problems: &[ + "select \"$.items[*].sku\" names a key field (sku); that form is for fan_out's over, so select the rows instead", + "fan-out path /{region}/orders/{id} needs {id}, which the key {\"region\":\"uk\"} lacks", + ], + }, + JobProblemsTestCase { + description: "a one-level over's fields problem comes before its placeholder problem, as before multi-level fan-out", + job: r#"{"kind":"fan_out","client":{"connection":"shop","retries":0},"path":"/orders/{id}","paginator":{"kind":"single"},"concurrency":2,"over":{"kind":"paginate","client":{"connection":"shop","retries":0},"path":"/shops/{shop}/orders","paginator":{"kind":"single"},"select":"$.items[*].id","keep_record":"raw"}}"#, + expected_problems: &[ + "fan_out's over picks keys with its select; fields and keep_record are for the fan-out's own rows", + "fan_out over path /shops/{shop}/orders has the placeholder {shop}; over must be a plain paginate", + ], + }, ]; for test_case in test_cases { let job: HttpJob = serde_json::from_str(test_case.job).expect(test_case.description); diff --git a/packages/sqlload/tests/test_http_jobs.py b/packages/sqlload/tests/test_http_jobs.py index 75aa0b5c..bb9eccb3 100644 --- a/packages/sqlload/tests/test_http_jobs.py +++ b/packages/sqlload/tests/test_http_jobs.py @@ -882,6 +882,19 @@ def test_given_three_fan_out_levels_when_loading_then_each_note_is_read_under_it assert [query["p"] for path, query, _ in api.log if path == "/v1/tree/orders"] == ["1", "2", "3", "4"] # type: ignore[attr-defined] +def test_given_levels_renaming_their_keys_when_loading_then_each_renamed_key_fills_its_placeholder(tmp_path: Path, api: ThreadingHTTPServer) -> None: + project = tree_project(tmp_path, api, NOTES.replace("{order_id}/items/{item_id}", "{o}/items/{i}")) + shop_py = project / "sources" / "shop.py" + renamed = shop_py.read_text().replace("select='$.items[*].order_id')", "select='$.items[*].order_id', key='o')") + renamed = renamed.replace("/tree/orders/{order_id}/items', over=orders", "/tree/orders/{o}/items', over=orders").replace("select='$.items[*].item_id',", "select='$.items[*].item_id', key='i',") + shop_py.write_text(renamed) + done = sqlload(project, "load") + assert done.returncode == 0, done.stdout + done.stderr + assert query(project, 'SELECT note FROM "RAW".shop__notes ORDER BY note') == sorted((f"{item}-{page}-{n}",) for item in TREE_ITEMS for page in (1, 2) for n in range(2)) + notes = {path for path, _, _ in api.log if path.endswith("/notes")} # type: ignore[attr-defined] + assert notes == {f"/v1/tree/orders/{item // 10}/items/{item}/notes" for item in TREE_ITEMS} + + @pytest.mark.parametrize( ("on_status", "notes", "expected_code", "expected_text"), [ diff --git a/website/src/content/docs/docs/sources/python.mdx b/website/src/content/docs/docs/sources/python.mdx index f411d1e3..88c7fe99 100644 --- a/website/src/content/docs/docs/sources/python.mdx +++ b/website/src/content/docs/docs/sources/python.mdx @@ -270,8 +270,8 @@ def notes(ctx): ``` - Every level has its own paginator, `concurrency`, `on_status` and, through the `sl.http` it - comes from, its own retries and rate limit. Up to the sum of the levels' `concurrency` requests - are in flight at once; a shared `rate_limit` caps them. + comes from, its own retries and rate limit. Up to one more request than the sum of the fan-out + levels' `concurrency` is in flight at once (the top paginate's); a shared `rate_limit` caps them. - An incremental cursor goes on the top level, as above; lower levels only see the keys it picks. - Only the last level's rows load. A level that should be a table of its own is a table function whose job stops at that level, so the levels above it are requested again for that table.