diff --git a/crates/sqlload/src/http_source/_helpers/fetch.rs b/crates/sqlload/src/http_source/_helpers/fetch.rs index c094127f..1e66f1b6 100644 --- a/crates/sqlload/src/http_source/_helpers/fetch.rs +++ b/crates/sqlload/src/http_source/_helpers/fetch.rs @@ -1,8 +1,7 @@ -//! One GET with the shared rate limit, retries with backoff, `Retry-After`, and `on_status`. +//! One request with the shared rate limit, retries with backoff, `Retry-After`, and `on_status`. use std::time::{Duration, SystemTime, UNIX_EPOCH}; -use reqwest::Url; use reqwest::header::{LINK, RETRY_AFTER}; use crate::http::constants::{FIRST_SERVER_ERROR, TOO_MANY_REQUESTS, UNAUTHORIZED}; @@ -12,11 +11,11 @@ use crate::http_source::constants::{ ERROR_EXCERPT_CHARS, FIRST_RETRY_DELAY, MAX_REDIRECTS, MAX_RETRY_AFTER, MAX_RETRY_DELAY, SKIPPED_REFUSAL_RENEWAL_AGE, }; -use crate::http_source::models::{Attempt, Calls, Credential, Fetched, StatusAction}; +use crate::http_source::models::{Attempt, Calls, Credential, Fetched, PageCall, StatusAction}; use crate::python_sources::errors::ReadError; -/// The body at `url`, or `None` when `on_status` skips its status; transport errors, 429 and 5xx are retried up to `calls.retries` times, and are still worth retrying later once those run out. -pub(crate) async fn fetch(calls: &Calls, url: &Url) -> Result, ReadError> { +/// The response to `call`, its JSON body resent on every retry, or `None` when `on_status` skips its status; transport errors, 429 and 5xx are retried up to `calls.retries` times, and are still worth retrying later once those run out. +pub(crate) async fn fetch(calls: &Calls, call: &PageCall) -> Result, ReadError> { let policy = RetryPolicy { first_delay: FIRST_RETRY_DELAY, max_delay: MAX_RETRY_DELAY, @@ -37,7 +36,7 @@ pub(crate) async fn fetch(calls: &Calls, url: &Url) -> Result, R ), _ => None, }; - match once(calls, url, access.as_deref(), !renewed).await? { + match once(calls, call, access.as_deref(), !renewed).await? { Attempt::Done(fetched) => return Ok(fetched), Attempt::Refused => { refused = access; @@ -58,18 +57,27 @@ pub(crate) async fn fetch(calls: &Calls, url: &Url) -> Result, R } } -/// One try at `url`, with the OAuth2 `access` token when the connection uses one; a 401 from the connection's origin comes back as `Refused` while `may_renew`, unless `on_status` skips 401 and the token is younger than `SKIPPED_REFUSAL_RENEWAL_AGE`. +/// One try at `call`, with the OAuth2 `access` token when the connection uses one; a 401 from the connection's origin comes back as `Refused` while `may_renew`, unless `on_status` skips 401 and the token is younger than `SKIPPED_REFUSAL_RENEWAL_AGE`. async fn once( calls: &Calls, - url: &Url, + call: &PageCall, access: Option<&str>, may_renew: bool, ) -> Result { let endpoint = &calls.endpoint; let secrets = endpoint.current_secrets(); - let shown = redacted(&format!("GET {url}"), &secrets); + let url = &call.url; + let shown = redacted(&format!("{} {url}", call.method.as_str()), &secrets); let own = same_origin(&endpoint.base_url, url); - let mut request = calls.http.get(url.clone()); + if !own && call.body.is_some() { + return Err(format!( + "{shown}: the request is on another origin than the connection's base_url, and the job's json_body never leaves the connection's origin" + )); + } + let mut request = match &call.body { + Some(_) => calls.http.post(url.clone()), + None => calls.http.get(url.clone()), + }; for (name, value) in endpoint.headers.iter().filter(|_| own) { request = request.header(name, value); } @@ -79,6 +87,9 @@ async fn once( (Credential::OAuth2(_), true) => request.bearer_auth(access.unwrap_or_default()), _ => request, }; + if let Some(body) = &call.body { + request = request.json(body); + } let response = match request.send().await { Ok(response) => response, Err(error) if error.is_builder() => { @@ -138,7 +149,7 @@ async fn once( _ if skipped => Attempt::Done(None), 300..=399 => { return Err(format!( - "{shown}: HTTP {status}: the redirect was not followed: it leads to another origin, has no Location, or goes past {MAX_REDIRECTS} hops; credentials never leave the connection's origin, so use the final URL as base_url" + "{shown}: HTTP {status}: the redirect was not followed: it leads to another origin, has no Location, goes past {MAX_REDIRECTS} hops, or is a 301, 302 or 303 after a POST, which would drop its body; credentials never leave the connection's origin, so use the final URL as base_url" )); } TOO_MANY_REQUESTS | FIRST_SERVER_ERROR.. => Attempt::Retry { diff --git a/crates/sqlload/src/http_source/_helpers/job.rs b/crates/sqlload/src/http_source/_helpers/job.rs index 5d03c394..5e278b9a 100644 --- a/crates/sqlload/src/http_source/_helpers/job.rs +++ b/crates/sqlload/src/http_source/_helpers/job.rs @@ -10,7 +10,7 @@ use tokio::task::JoinSet; 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; +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::python_sources::errors::ReadError; @@ -20,10 +20,12 @@ use crate::row_batches::models::{Projection, ProjectionRules}; use std::time::Duration; use crate::http::main::install_tls_provider::install_tls_provider; -use crate::http_source::constants::{JOB_THREADS, MAX_REDIRECTS, PAGES_AHEAD}; +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, Over, Paginator, Request, Selector, - StatusAction, + Auth, Calls, FanOut, FieldSpec, HttpJob, HttpPage, KeyFeed, Location, Method, Over, Paginator, + Request, Selector, StatusAction, }; /// Every problem the job's definition has, found before any request, including projected rows that break the table's `rules`. @@ -84,6 +86,7 @@ fn request_problems(request: &Request) -> Vec { let mut problems: Vec = Vec::new(); problems.extend(selector(request.select.as_deref()).err()); problems.extend(paginator_problem(&request.paginator)); + problems.extend(body_problems(request)); problems.extend(match &request.client.auth { Auth::Oauth2ClientCredentials(spec) | Auth::Oauth2RefreshToken(spec) => { token_url(spec).err() @@ -135,6 +138,64 @@ fn paginator_problem(paginator: &Paginator) -> Option { } } +/// Why the job's body settings cannot work: a body or body paging on a `GET`, a body path that is not a plain path, or body paging into a body that is not an object. +fn body_problems(request: &Request) -> Vec { + let (paths, location) = match &request.paginator { + Paginator::Offset { + param, + limit_param, + location, + .. + } => (vec![param, limit_param], *location), + Paginator::PageNumber { + param, location, .. + } + | Paginator::Cursor { + param, location, .. + } => (vec![param], *location), + Paginator::Single | Paginator::NextLink { .. } | Paginator::LinkHeader => { + (Vec::new(), Location::Query) + } + }; + let mut problems: Vec = Vec::new(); + if request.method == Method::Get && request.json_body.is_some() { + problems.push("json_body is sent only with method=\"POST\"".to_owned()); + } + if location == Location::Query { + return problems; + } + if request.method == Method::Get { + problems.push("a paginator with location=\"body\" needs method=\"POST\"".to_owned()); + } + if request + .json_body + .as_ref() + .is_some_and(|body| !body.is_object()) + { + problems.push( + "a paginator with location=\"body\" sets fields in json_body, so json_body must be an object".to_owned(), + ); + } + problems.extend(paths.iter().filter_map(|path| match selector(Some(path)) { + Ok(parsed) if parsed.field.is_none() && !parsed.rows.is_empty() => None, + Ok(_) | Err(_) => Some(format!( + "body path {path:?} is not a path such as \"$.page.cursor\"" + )), + })); + if problems.is_empty() { + let mut body = request + .json_body + .clone() + .unwrap_or_else(|| Value::Object(Map::new())); + problems.extend( + paths + .iter() + .filter_map(|path| set_at(&mut body, path, Value::Null).err()), + ); + } + problems +} + /// Why a header cannot be sent, if it cannot. fn header_problem(name: &str, value: &str) -> Option { match ( @@ -248,7 +309,14 @@ async fn run_job( connections: Map, pages: Sender, ) -> Result<(), ReadError> { - let http = job_client()?; + 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) + } + }; + let http = job_client(posts)?; match job { HttpJob::Paginate(request) => { let calls = request_calls(&request, &connections, &http)?; @@ -258,11 +326,16 @@ async fn run_job( } } -/// A pooled client that follows redirects only within the first request's origin, so connection headers and credentials never reach another host. -fn job_client() -> Result { +/// 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(); reqwest::Client::builder() - .redirect(reqwest::redirect::Policy::custom(same_origin_redirect)) + .redirect(reqwest::redirect::Policy::custom(move |attempt| { + match posts && BODY_DROPPING_REDIRECTS.contains(&attempt.status().as_u16()) { + true => attempt.stop(), + false => same_origin_redirect(attempt), + } + })) .pool_idle_timeout(Duration::from_secs(90)) .connect_timeout(Duration::from_secs(15)) .timeout(Duration::from_secs(300)) diff --git a/crates/sqlload/src/http_source/_helpers/pages.rs b/crates/sqlload/src/http_source/_helpers/pages.rs index 2ef1f26f..d0270f1e 100644 --- a/crates/sqlload/src/http_source/_helpers/pages.rs +++ b/crates/sqlload/src/http_source/_helpers/pages.rs @@ -10,9 +10,13 @@ use tokio::sync::mpsc::Sender; use crate::http_source::_helpers::fetch::fetch; use crate::http_source::_helpers::job::request_projection; -use crate::http_source::_helpers::selector::{pointer_text, selected_node, selector}; +use crate::http_source::_helpers::selector::{ + pointer_text, pointer_value, selected_node, selector, +}; use crate::http_source::_helpers::urls::{redacted, request_url}; -use crate::http_source::models::{Calls, Fetched, HttpPage, Paginator, Request, Selector}; +use crate::http_source::models::{ + Calls, Fetched, HttpPage, Location, Method, PageCall, Paginator, Request, Selector, +}; use crate::python_sources::errors::ReadError; /// Send every page of `request` at `path` (its placeholders filled) to `pages`, with where its rows are; a page whose selected node is missing or null has no rows and is not sent. Stops quietly once `pages` closes; a page read twice, or a response repeating the previous one, is an error rather than a loop. @@ -26,18 +30,22 @@ pub(crate) async fn paginate( let projection = request_projection(request)?; let rows_path = Some(rows.rows.join(".")).filter(|path| !path.is_empty()); let first = request_url(&calls.endpoint.base_url, path)?; - let mut url = with_params(&first, &request.params, &start_params(&request.paginator)); - let mut seen: HashSet = HashSet::new(); + let mut call = page_call(request, &first, start_values(&request.paginator))?; + let mut seen: HashSet<(Url, Option)> = HashSet::new(); let mut previous: Option = None; for step in 1_u64.. { - let shown = redacted(&format!("GET {url}"), &calls.endpoint.current_secrets()); - if !seen.insert(url.clone()) { + let shown = redacted( + &format!("{} {}", call.method.as_str(), call.url), + &calls.endpoint.current_secrets(), + ); + let body_text = call.body.as_ref().map(Value::to_string); + if !seen.insert((call.url.clone(), body_text)) { return Err(format!( "{shown}: the paginator came back to a page it already read; check its pointer or parameter" ) .into()); } - let Some(fetched) = fetch(calls, &url).await? else { + let Some(fetched) = fetch(calls, &call).await? else { return Ok(()); }; let digest = counted(&request.paginator).then(|| digest(&fetched.bytes)); @@ -50,8 +58,12 @@ pub(crate) async fn paginate( previous = digest; let document = parsed(request, &rows, &fetched.bytes) .map_err(|error| format!("{shown}: the response is not JSON: {error}"))?; - let next = next_url(request, &first, (&fetched, document.as_ref(), step)) - .map_err(|error| format!("{shown}: {error}"))?; + let next = next_call( + request, + (&first, &call), + (&fetched, document.as_ref(), step), + ) + .map_err(|error| format!("{shown}: {error}"))?; let has_rows = document.as_ref().is_none_or(|document| { selected_node(document, &rows).is_some_and(|node| !node.is_null()) }); @@ -65,7 +77,7 @@ pub(crate) async fn paginate( return Ok(()); } match next { - Some(next) => url = next, + Some(next) => call = next, None => return Ok(()), } } @@ -106,18 +118,19 @@ fn digest(bytes: &[u8]) -> u64 { hasher.finish() } -/// The paginator's parameters on the first request. -fn start_params(paginator: &Paginator) -> Vec<(String, String)> { +/// The paginator's values on the first request. +fn start_values(paginator: &Paginator) -> Vec<(String, Value)> { match paginator { Paginator::Offset { param, limit_param, limit, + .. } => vec![ - (param.clone(), "0".to_owned()), - (limit_param.clone(), limit.to_string()), + (param.clone(), Value::from(0_u64)), + (limit_param.clone(), Value::from(*limit)), ], - Paginator::PageNumber { param, start } => vec![(param.clone(), start.to_string())], + Paginator::PageNumber { param, start, .. } => vec![(param.clone(), Value::from(*start))], Paginator::Single | Paginator::NextLink { .. } | Paginator::LinkHeader @@ -125,58 +138,126 @@ fn start_params(paginator: &Paginator) -> Vec<(String, String)> { } } -/// The page after `fetched` (the `step`th), or `None` at the end; links resolve against the URL the response came from, after any redirect; `document` is the parsed response whenever the paginator reads the body. -fn next_url( +/// Where the paginator's values go. +fn location(paginator: &Paginator) -> Location { + match paginator { + Paginator::Offset { location, .. } + | Paginator::PageNumber { location, .. } + | Paginator::Cursor { location, .. } => *location, + Paginator::Single | Paginator::NextLink { .. } | Paginator::LinkHeader => Location::Query, + } +} + +/// The request at `url` with the paginator's `values`: in the query after `params`, or at their paths in a fresh copy of `json_body`, which is never changed itself. A `POST` without `json_body` starts from `{}`. +fn page_call( request: &Request, - first: &Url, + url: &Url, + values: Vec<(String, Value)>, +) -> Result { + let (query, in_body) = match location(&request.paginator) { + Location::Query => (values, Vec::new()), + Location::Body => (Vec::new(), values), + }; + let query: Vec<(String, String)> = query + .into_iter() + .map(|(name, value)| (name, param_text(&value))) + .collect(); + let body = match request.method { + Method::Get => None, + Method::Post => { + let mut body = request + .json_body + .clone() + .unwrap_or_else(|| Value::Object(Map::new())); + for (path, value) in in_body { + set_at(&mut body, &path, value)?; + } + Some(body) + } + }; + Ok(PageCall { + method: request.method, + url: with_params(url, &request.params, &query), + body, + }) +} + +/// `body` with `value` at `path` (such as `$.page.cursor`), adding objects on the way. +pub(crate) fn set_at(body: &mut Value, path: &str, value: Value) -> Result<(), String> { + let names = selector(Some(path))?.rows; + let Some((last, parents)) = names.split_last() else { + return Err(format!("body path {path:?} names no field")); + }; + let mut node = body; + for name in parents { + node = node + .as_object_mut() + .ok_or_else(|| format!("json_body has no object to set {path} in"))? + .entry(name.clone()) + .or_insert_with(|| Value::Object(Map::new())); + } + node.as_object_mut() + .ok_or_else(|| format!("json_body has no object to set {path} in"))? + .insert(last.clone(), value); + Ok(()) +} + +/// A value as a query parameter: text as itself, anything else as JSON. +fn param_text(value: &Value) -> String { + match value { + Value::String(text) => text.clone(), + other => other.to_string(), + } +} + +/// The request after `fetched` (the `step`th page, sent as `current`), or `None` at the end; links resolve against the URL the response came from, after any redirect, and keep the current body; `document` is the parsed response whenever the paginator reads the body. +fn next_call( + request: &Request, + (first, current): (&Url, &PageCall), (fetched, document, step): (&Fetched, Option<&Value>, u64), -) -> Result, String> { +) -> Result, String> { let body = || document.ok_or_else(|| "the paginator needs a JSON response".to_owned()); let joined = |link: String| { fetched .url .join(&link) - .map(Some) + .map(|url| { + Some(PageCall { + url, + ..current.clone() + }) + }) .map_err(|error| format!("next page {link} is not a URL: {error}")) }; - let params = &request.params; + let next = |values: Vec<(String, Value)>| page_call(request, first, values).map(Some); match &request.paginator { Paginator::Single => Ok(None), Paginator::LinkHeader => fetched.next_link.clone().map_or(Ok(None), joined), Paginator::NextLink { pointer } => { pointer_text(body()?, &selector(Some(pointer))?).map_or(Ok(None), joined) } - Paginator::Cursor { pointer, param } => { - Ok(pointer_text(body()?, &selector(Some(pointer))?) - .map(|token| with_params(first, params, &[(param.clone(), token)]))) + Paginator::Cursor { pointer, param, .. } => { + match pointer_value(body()?, &selector(Some(pointer))?) { + Some(token) => next(vec![(param.clone(), token)]), + None => Ok(None), + } } Paginator::Offset { param, limit_param, limit, - } => { - let full = page_rows(body()?, request)? >= *limit; - Ok(full.then(|| { - with_params( - first, - params, - &[ - (param.clone(), (step * limit).to_string()), - (limit_param.clone(), limit.to_string()), - ], - ) - })) - } - Paginator::PageNumber { param, start } => { - let more = page_rows(body()?, request)? > 0; - Ok(more.then(|| { - with_params( - first, - params, - &[(param.clone(), (start + step).to_string())], - ) - })) - } + .. + } => match page_rows(body()?, request)? >= *limit { + true => next(vec![ + (param.clone(), Value::from(step * limit)), + (limit_param.clone(), Value::from(*limit)), + ]), + false => Ok(None), + }, + Paginator::PageNumber { param, start, .. } => match page_rows(body()?, request)? > 0 { + true => next(vec![(param.clone(), Value::from(start + step))]), + false => Ok(None), + }, } } diff --git a/crates/sqlload/src/http_source/_helpers/selector.rs b/crates/sqlload/src/http_source/_helpers/selector.rs index 83a8925a..3eb1dee4 100644 --- a/crates/sqlload/src/http_source/_helpers/selector.rs +++ b/crates/sqlload/src/http_source/_helpers/selector.rs @@ -60,15 +60,23 @@ pub(crate) fn selected_rows<'a>(document: &'a Value, selector: &Selector) -> Vec } } -/// The text at `pointer` in `document` (a number as its digits); `None` when missing, null or empty. -pub(crate) fn pointer_text(document: &Value, pointer: &Selector) -> Option { +/// The value at `pointer` in `document` when it is a non-empty string or a number; `None` when missing, null, empty or another kind. +pub(crate) fn pointer_value(document: &Value, pointer: &Selector) -> Option { let node = pointer .rows .iter() .try_fold(document, |node, name| node.get(name))?; match node { - Value::String(text) if !text.is_empty() => Some(text.clone()), - Value::Number(number) => Some(number.to_string()), + Value::String(text) if !text.is_empty() => Some(node.clone()), + Value::Number(_) => Some(node.clone()), _ => None, } } + +/// The text at `pointer` in `document` (a number as its digits); `None` when missing, null or empty. +pub(crate) fn pointer_text(document: &Value, pointer: &Selector) -> Option { + match pointer_value(document, pointer)? { + Value::String(text) => Some(text), + other => Some(other.to_string()), + } +} diff --git a/crates/sqlload/src/http_source/constants.rs b/crates/sqlload/src/http_source/constants.rs index d9c7bb16..510cdcbc 100644 --- a/crates/sqlload/src/http_source/constants.rs +++ b/crates/sqlload/src/http_source/constants.rs @@ -33,6 +33,9 @@ pub const RELATION_PARAMETER: &str = "rel"; /// Redirects followed within the connection's origin before a request fails. pub const MAX_REDIRECTS: usize = 10; +/// Redirects a client follows by resending a `POST` as a `GET` without its body. +pub const BODY_DROPPING_REDIRECTS: [u16; 3] = [301, 302, 303]; + /// The `Link` relation of the next page. pub const NEXT_RELATION: &str = "next"; diff --git a/crates/sqlload/src/http_source/models.rs b/crates/sqlload/src/http_source/models.rs index fca1afda..0c455710 100644 --- a/crates/sqlload/src/http_source/models.rs +++ b/crates/sqlload/src/http_source/models.rs @@ -24,6 +24,11 @@ pub struct Request { /// Query parameters; a null value is left out. #[serde(default)] pub params: Map, + #[serde(default)] + pub method: Method, + /// The JSON body every `POST` starts from; each page's copy gets the paginator's values when they go in the body. + #[serde(default)] + pub json_body: Option, pub paginator: Paginator, /// Where the rows are in each response, such as `$.items`; `$` (or none) is the whole response. #[serde(default)] @@ -39,6 +44,41 @@ pub struct Request { pub keep_record: Option, } +/// The HTTP method of a job's requests. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "UPPERCASE")] +pub enum Method { + #[default] + Get, + Post, +} + +impl Method { + pub fn as_str(self) -> &'static str { + match self { + Self::Get => "GET", + Self::Post => "POST", + } + } +} + +/// Where a paginator's values go: the query string, or a path in the JSON body. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum Location { + #[default] + Query, + Body, +} + +/// One request of a page: its method, URL and JSON body. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct PageCall { + pub method: Method, + pub url: reqwest::Url, + pub body: Option, +} + /// One projected column of a job's rows. #[derive(Debug, Clone, Deserialize)] pub struct FieldSpec { @@ -182,16 +222,28 @@ pub enum Paginator { NextLink { pointer: String }, /// The `Link` header's `rel="next"`. LinkHeader, - /// `param` counts rows from 0 by `limit`; ends at a page with fewer rows. + /// `param` counts rows from 0 by `limit`; ends at a page with fewer rows. In the body, `param` and `limit_param` are paths such as `$.page.offset`. Offset { param: String, limit_param: String, limit: u64, + #[serde(default)] + location: Location, }, /// `param` counts pages from `start`; ends at an empty page. - PageNumber { param: String, start: u64 }, - /// The token at `pointer` in the body goes back as `param`; ends when it is missing, null or empty. - Cursor { pointer: String, param: String }, + PageNumber { + param: String, + start: u64, + #[serde(default)] + location: Location, + }, + /// The token at `pointer` in the response goes back as `param`; ends when it is missing, null or empty. + Cursor { + pointer: String, + param: String, + #[serde(default)] + location: Location, + }, } /// What a listed status does. diff --git a/packages/sqlload/python/sqlload/_http.py b/packages/sqlload/python/sqlload/_http.py index f6cc18ed..31a9ff15 100644 --- a/packages/sqlload/python/sqlload/_http.py +++ b/packages/sqlload/python/sqlload/_http.py @@ -51,19 +51,29 @@ def link_header() -> Paginator: return Paginator({"kind": "link_header"}) -def offset(param: str = "offset", *, limit: int, limit_param: str = "limit") -> Paginator: - """`param` counts rows from 0 by `limit`; ends at a page with fewer than `limit` rows.""" - return Paginator({"kind": "offset", "param": param, "limit_param": limit_param, "limit": limit}) +def offset(param: str = "offset", *, limit: int, limit_param: str = "limit", location: str = "query") -> Paginator: + """`param` counts rows from 0 by `limit`; ends at a page with fewer than `limit` rows. + With `location="body"`, `param` and `limit_param` are paths in the POST's JSON body, such as `"$.page.offset"`.""" + return Paginator({"kind": "offset", "param": param, "limit_param": limit_param, "limit": limit, "location": _location(location)}) -def page_number(param: str = "page", *, start: int = 1) -> Paginator: - """`param` counts pages from `start`; ends at an empty page.""" - return Paginator({"kind": "page_number", "param": param, "start": start}) +def page_number(param: str = "page", *, start: int = 1, location: str = "query") -> Paginator: + """`param` counts pages from `start`; ends at an empty page. With `location="body"`, `param` is a path in the JSON body.""" + return Paginator({"kind": "page_number", "param": param, "start": start, "location": _location(location)}) -def cursor(pointer: str, param: str) -> Paginator: - """The token at `pointer` in the body goes back as query parameter `param`; ends when missing, null or empty.""" - return Paginator({"kind": "cursor", "pointer": pointer, "param": param}) + +def cursor(pointer: str, param: str, *, location: str = "query") -> Paginator: + """The token at `pointer` in the response goes back as query parameter `param`; ends when missing, null or empty. + + With `location="body"`, `param` is a path in the POST's JSON body, such as `"$.page.cursor"`.""" + return Paginator({"kind": "cursor", "pointer": pointer, "param": param, "location": _location(location)}) + + +def _location(location: str) -> str: + if location not in ("query", "body"): + raise ValueError(f'location must be "query" or "body", not {location!r}') + return location def bearer() -> Auth: @@ -170,12 +180,16 @@ def paginate( on_status: Mapping[int, str] | None = None, fields: Fields | None = None, keep_record: str | None = None, + method: str = "GET", + json_body: Mapping[str, Any] | 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.""" - return Job({"kind": "paginate", **self._request(path, paginator, params, select, on_status), **_projection(fields, keep_record)}) + 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"`.""" + request = self._request(path, paginator, params, select, on_status, (method, json_body)) + return Job({"kind": "paginate", **request, **_projection(fields, keep_record)}) def fan_out( self, @@ -189,6 +203,8 @@ def fan_out( on_status: Mapping[int, str] | None = None, fields: Fields | None = None, keep_record: str | None = None, + method: str = "GET", + json_body: Mapping[str, Any] | 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; @@ -202,7 +218,7 @@ def fan_out( if isinstance(over, Job) else {"kind": "values", "values": list(over)} ) - request = self._request(path, paginator or single(), params, select, on_status) + 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}) @@ -213,11 +229,17 @@ def _request( params: Mapping[str, Any] | None, select: str | None, on_status: Mapping[int, str] | None, + body: tuple[str, Mapping[str, Any] | None], ) -> dict[str, Any]: + method, json_body = body + if method.upper() not in ("GET", "POST"): + raise ValueError(f'method must be "GET" or "POST", not {method!r}') return { "client": self._client, "path": path, "params": dict(params or {}), + "method": method.upper(), + "json_body": None if json_body is None else json.loads(json.dumps(json_body, default=_param_text)), "paginator": dict(paginator.spec), "select": select, "on_status": {str(status): action for status, action in (on_status or {}).items()}, diff --git a/packages/sqlload/tests/test_http_jobs.py b/packages/sqlload/tests/test_http_jobs.py index 1ac92bd0..9f452cc7 100644 --- a/packages/sqlload/tests/test_http_jobs.py +++ b/packages/sqlload/tests/test_http_jobs.py @@ -74,6 +74,38 @@ def do_GET(self) -> None: return self.reply(404, {"message": "no route"}) def do_POST(self) -> None: + """JSON search endpoints under `/v1/`, recording each body in `self.server.bodies`; any other path is the OAuth2 token endpoint.""" + if self.path.startswith("/v1/"): + return self.search() + self.token() + + def search(self) -> None: + """Orders after `$.filter.after`, paged by `$.page.offset`/`$.page.limit` or by `$.page.cursor` (four at a time).""" + body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) + url = urlparse(self.path) + self.server.bodies.append((url.path, ", ".join(self.headers.get_all("Content-Type") or []), body)) # type: ignore[attr-defined] + if self.headers.get("Authorization") != f"Bearer {TOKEN}": + return self.reply(401, {"message": "bad token"}) + moved = { + "/v1/search-302": (302, "/v1/wrapped"), + "/v1/search-307": (307, "/v1/search-cursor"), + "/v1/search-away": (200, None), + }.get(url.path) + if moved == (200, None): + return self.reply(200, {"items": ORDERS[:1], "next": f"{self.server.other}/steal"}) # type: ignore[attr-defined] + if moved is not None: + return self.reply(moved[0], {}, {"Location": moved[1]}) + after = body.get("filter", {}).get("after") or 0 + rows = [row for row in ORDERS if row["order_id"] > after] + page = body.get("page", {}) + if url.path == "/v1/search": + start, limit = page["offset"], page["limit"] + return self.reply(200, {"items": rows[start : start + limit]}) + start = int(page.get("cursor") or 0) + token: Any = start + 4 if url.path == "/v1/search-number" else str(start + 4) + self.reply(200, {"items": rows[start : start + 4], "meta": {"next": token if start + 4 < len(rows) else None}}) + + def token(self) -> None: """The OAuth2 token endpoint: `self.server.oauth` sets the lifetime, uses before revocation and rotation of tokens it issues.""" form = {key: values[0] for key, values in parse_qs(self.rfile.read(int(self.headers["Content-Length"])).decode()).items()} basic = self.headers.get("Authorization", "") @@ -186,6 +218,8 @@ def do_GET(self) -> None: self.end_headers() self.wfile.write(data) + do_POST = do_GET + def serve(handler: type[BaseHTTPRequestHandler]) -> ThreadingHTTPServer: server = ThreadingHTTPServer(("127.0.0.1", 0), handler) @@ -205,6 +239,7 @@ def elsewhere() -> Iterator[ThreadingHTTPServer]: def api(elsewhere: ThreadingHTTPServer) -> Iterator[ThreadingHTTPServer]: server = serve(Api) server.log = [] # type: ignore[attr-defined] + server.bodies = [] # type: ignore[attr-defined] server.flaky = 0 # type: ignore[attr-defined] server.refused = 0 # type: ignore[attr-defined] server.oauth = {} # type: ignore[attr-defined] @@ -609,3 +644,124 @@ def test_given_a_401_from_another_origin_when_loading_then_no_token_is_sent_ther assert "/denied: HTTP 401" in done.stdout + done.stderr, done.stdout + done.stderr assert len(api.token_requests) == 1 # type: ignore[attr-defined] assert not any("authorization" in {name.lower() for name in headers} for headers in elsewhere.seen) # type: ignore[attr-defined] + + +SEARCH_FILTER = {"filter": {"status": "any"}} + + +@pytest.mark.parametrize( + ("job", "expected_bodies"), + [ + ( + "api.paginate('/search', method='POST', json_body=FILTER, paginator=sl.offset('$.page.offset', limit=3, limit_param='$.page.limit', location='body'), select='$.items')", + [{**SEARCH_FILTER, "page": {"offset": offset, "limit": 3}} for offset in (0, 3, 6)], + ), + ( + "api.paginate('/search-cursor', method='post', json_body=FILTER, paginator=sl.cursor('$.meta.next', '$.page.cursor', location='body'), select='$.items')", + [SEARCH_FILTER, {**SEARCH_FILTER, "page": {"cursor": "4"}}], + ), + ( + "api.paginate('/search-number', method='POST', json_body=FILTER, paginator=sl.cursor('$.meta.next', '$.page.cursor', location='body'), select='$.items')", + [SEARCH_FILTER, {**SEARCH_FILTER, "page": {"cursor": 4}}], + ), + ], + ids=["offset-in-body", "cursor-token-in-body", "numeric-cursor-token-in-body"], +) +def test_given_a_post_job_paging_in_its_body_when_loading_then_each_page_sends_the_template_with_its_values( + tmp_path: Path, api: ThreadingHTTPServer, job: str, expected_bodies: list[dict[str, Any]] +) -> None: + tables = f"FILTER = {SEARCH_FILTER!r}\n\n@shop.table(description='Orders.')\ndef orders(ctx):\n job = {job}\n assert FILTER == {SEARCH_FILTER!r}\n return job\n" + project = make_project(tmp_path, api, tables) + done = sqlload(project, "load") + assert done.returncode == 0, done.stdout + done.stderr + assert query(project, 'SELECT order_id, status FROM "RAW".shop__orders ORDER BY order_id') == ALL_ORDERS + assert [body for _, _, body in api.bodies] == expected_bodies # type: ignore[attr-defined] + assert {content_type for _, content_type, _ in api.bodies} == {"application/json"} # type: ignore[attr-defined] + assert api.log == [] # type: ignore[attr-defined] + + +@pytest.mark.parametrize( + ("path", "paginator", "expected_code", "expected_text", "expected_paths"), + [ + ("/search-307", "sl.single()", 0, "", ["/v1/search-307", "/v1/search-cursor"]), + ("/search-302", "sl.single()", 1, "or is a 301, 302 or 303 after a POST", ["/v1/search-302"]), + ("/search-away", "sl.next_link('$.next')", 1, "json_body never leaves the connection's origin", ["/v1/search-away"]), + ], + ids=["307-resends-the-body", "302-is-not-followed-as-a-get", "next-link-to-another-host"], +) +def test_given_a_post_answered_with_a_redirect_or_link_when_loading_then_its_body_is_resent_or_the_load_fails( + tmp_path: Path, + api: ThreadingHTTPServer, + elsewhere: ThreadingHTTPServer, + path: str, + paginator: str, + expected_code: int, + expected_text: str, + expected_paths: list[str], +) -> None: + tables = f"@shop.table(description='Orders.')\ndef orders(ctx):\n return api.paginate('{path}', method='POST', json_body={SEARCH_FILTER!r}, paginator={paginator}, select='$.items')\n" + project = make_project(tmp_path, api, tables) + done = sqlload(project, "load") + assert done.returncode == expected_code, done.stdout + done.stderr + assert expected_text in done.stdout + done.stderr + assert [(path, body) for path, _, body in api.bodies] == [(path, SEARCH_FILTER) for path in expected_paths] # type: ignore[attr-defined] + assert api.log == [] and elsewhere.seen == [] # type: ignore[attr-defined] + + +def test_given_a_connection_content_type_when_posting_then_it_is_the_only_one_sent(tmp_path: Path, api: ThreadingHTTPServer) -> None: + tables = ( + "typed = sl.http(shop, auth=sl.bearer(), headers={'Content-Type': 'application/json; charset=utf-8'})\n\n" + "@shop.table(description='Orders.')\ndef orders(ctx):\n" + " return typed.paginate('/search-cursor', method='POST', paginator=sl.single(), select='$.items')\n" + ) + project = make_project(tmp_path, api, tables) + done = sqlload(project, "load") + assert done.returncode == 0, done.stdout + done.stderr + assert [content_type for _, content_type, _ in api.bodies] == ["application/json; charset=utf-8"] # type: ignore[attr-defined] + + +def test_given_an_incremental_cursor_in_the_body_when_loading_twice_then_the_second_run_sends_the_stored_cursor( + tmp_path: Path, api: ThreadingHTTPServer +) -> None: + tables = ( + "@shop.table(description='Orders.', write_strategy='append', cursor_column='order_id')\n" + "def orders(ctx):\n" + " return api.paginate('/search', method='POST', json_body={'filter': {'after': ctx.cursor_value}}, " + "paginator=sl.offset('$.page.offset', limit=10, limit_param='$.page.limit', location='body'), select='$.items')\n" + ) + project = make_project(tmp_path, api, tables) + for _ in range(2): + done = sqlload(project, "load") + assert done.returncode == 0, done.stdout + done.stderr + assert [body["filter"]["after"] for _, _, body in api.bodies] == [None, 7] # type: ignore[attr-defined] + assert query(project, 'SELECT count(*), count(DISTINCT order_id) FROM "RAW".shop__orders') == [(7, 7)] + + +def test_given_a_get_job_when_loading_then_its_requests_are_unchanged_by_body_paging(tmp_path: Path, api: ThreadingHTTPServer) -> None: + project = make_project(tmp_path, api, "@shop.table(description='Orders.')\ndef orders(ctx):\n return api.paginate('/offset', paginator=sl.offset(limit=3), params={'q': 'x'})\n") + done = sqlload(project, "load") + assert done.returncode == 0, done.stdout + done.stderr + expected = [("/v1/offset", {"q": "x", "offset": str(offset), "limit": "3"}, f"Bearer {TOKEN}") for offset in (0, 3, 6)] + assert api.log == expected # type: ignore[attr-defined] + assert api.bodies == [] # type: ignore[attr-defined] + + +@pytest.mark.parametrize( + ("job", "expected_text"), + [ + ("api.paginate('/search', json_body={'a': 1}, paginator=sl.single())", 'json_body is sent only with method="POST"'), + ("api.paginate('/search', paginator=sl.cursor('$.next', '$.page.cursor', location='body'))", 'a paginator with location="body" needs method="POST"'), + ("api.paginate('/search', method='POST', paginator=sl.cursor('$.next', 'page[cursor]', location='body'))", 'body path "page[cursor]" is not a path such as "$.page.cursor"'), + ( + "api.paginate('/search-cursor', method='POST', json_body={'page': 1}, paginator=sl.cursor('$.meta.next', '$.page.cursor', location='body'))", + "json_body has no object to set $.page.cursor in", + ), + ], + ids=["body-on-get", "body-paging-on-get", "bad-body-path", "body-path-through-a-number"], +) +def test_given_a_bad_body_setting_when_loading_then_no_request_is_sent(tmp_path: Path, api: ThreadingHTTPServer, job: str, expected_text: str) -> None: + project = make_project(tmp_path, api, f"@shop.table(description='Orders.')\ndef orders(ctx):\n return {job}\n") + done = sqlload(project, "load") + 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] diff --git a/website/src/content/docs/docs/sources/python.mdx b/website/src/content/docs/docs/sources/python.mdx index 57f3ba79..328cfa31 100644 --- a/website/src/content/docs/docs/sources/python.mdx +++ b/website/src/content/docs/docs/sources/python.mdx @@ -220,17 +220,17 @@ thread-safe. Set `parallel_tables=True` on the source when its client is. | Function | Meaning | |----------|---------| | `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)` | 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). | -| `api.fan_out(path, *, over, concurrency=8, paginator=None, params=None, select=None, on_status=None, fields=None, keep_record=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. | +| `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`. | | Paginator | Next page | |-----------|-----------| | `sl.single()` | None: one request. | | `sl.next_link(pointer)` | The URL at `pointer` in the body, such as `"$.next"`. | | `sl.link_header()` | The `Link` header's `rel="next"`. | -| `sl.offset(param="offset", *, limit, limit_param="limit")` | Counts rows by `limit`; ends at a short page. | -| `sl.page_number(param="page", *, start=1)` | Counts pages; ends at an empty page. | -| `sl.cursor(pointer, param)` | The token at `pointer` goes back as `param`. | +| `sl.offset(param="offset", *, limit, limit_param="limit", location="query")` | Counts rows by `limit`; ends at a short page. `location` is from 0.36.0 (not released yet). | +| `sl.page_number(param="page", *, start=1, location="query")` | Counts pages; ends at an empty page. `location` is from 0.36.0 (not released yet). | +| `sl.cursor(pointer, param, *, location="query")` | The token at `pointer` goes back as `param`. `location` is from 0.36.0 (not released yet). | | Auth | Sends | |------|-------| @@ -250,6 +250,38 @@ 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. +### POST with a JSON body + +From 0.36.0 (not released yet). `method="POST"` sends `json_body` with every request, as +`Content-Type: application/json`; `GET` stays the default, and its requests don't change. With +`location="body"`, `sl.offset`, `sl.page_number` and `sl.cursor` put their values in the body +instead of the query string, and `param` (and `limit_param`) are paths such as `"$.page.cursor"`. +Each request starts from a fresh copy of `json_body`, so one page's values never leak into the +next, and a retry resends the same body. A POST without `json_body` starts from `{}`. + +To send the incremental cursor in the body, put `ctx.cursor_value` in `json_body`, as you would in +`params`. Dates, times and decimals go as text (ISO text for dates and times); wrap the value in +`int()` or `float()` if the API wants a JSON number. + +```python +@shop.table(description="Orders.", write_strategy="append", cursor_column="updated_at") +def orders(ctx): + return api.paginate( + "/orders/search", + method="POST", + json_body={"filter": {"updated_after": ctx.cursor_value}}, + paginator=sl.cursor("$.meta.next", "$.page.cursor", location="body"), + select="$.items", + ) +``` + +`json_body` on a `GET`, `location="body"` without `method="POST"`, or a body path that +`json_body` already holds as a non-object (such as `$.page.cursor` with `{"page": 1}`) fails before +any request. A POST follows only 307 and 308 redirects, which keep the method and body; a 301, 302 +or 303 would resend it as a `GET` without its body, so it fails instead. `sl.next_link` and +`sl.link_header` resend the same body to the next page, and fail if that page is on another +origin, so `json_body` never leaves the connection's origin. + ### OAuth2 From 0.27.0. `sl.oauth2_client_credentials` and `sl.oauth2_refresh_token` fetch short-lived