diff --git a/crates/fmp-py/README.md b/crates/fmp-py/README.md index f131356..90aa503 100644 --- a/crates/fmp-py/README.md +++ b/crates/fmp-py/README.md @@ -236,6 +236,30 @@ The rules: `FMP_BASE_URL` only replaces the base URL; the `token` and `FMP_API_KEY` rules above are unchanged. +To read the proxy's cache status and remaining daily budget on a success, +pass `on_response`. It is called as `on_response(endpoint_id, status, +headers)` after every successful call, from any namespace: + +```python +from fmp import FmpClient + + +def on_response(endpoint_id: str, status: int, headers: dict[str, str]) -> None: + print(endpoint_id, headers.get("x-proxy-cache"), headers.get("x-proxy-daily-remaining")) + + +client = FmpClient(on_response=on_response) +rows = client.quote.short("AAPL") # prints: quote-short HIT 41 +``` + +- `headers` has only the allowlisted headers (`retry-after`, `x-proxy-*`, + `x-ratelimit-*`), lowercase and redacted; a repeated name keeps its first + value, like an error's `headers`, and it is `{}` when none were sent. +- It runs on the calling thread with the GIL, before the method returns, so + it works from several threads at once and each call sees its own headers. +- It is never called for errors: an `FmpError` already carries `headers`. +- If it raises, the method raises that exception and its data is discarded. + ## Custom router or proxy ```python diff --git a/crates/fmp-py/python/fmp/_native/__init__.pyi b/crates/fmp-py/python/fmp/_native/__init__.pyi index b9e6ebd..26e33dc 100644 --- a/crates/fmp-py/python/fmp/_native/__init__.pyi +++ b/crates/fmp-py/python/fmp/_native/__init__.pyi @@ -189,6 +189,18 @@ class FmpClient: non-loopback HTTP requires `danger_allow_insecure_authentication=True`. Redirects are either disabled or restricted to the same origin. + `on_response`, when given, is called as `on_response(endpoint_id, status, + headers)` after every successful response, for every namespace method, for + example `("quote-short", 200, {"x-proxy-cache": "HIT", + "x-proxy-daily-remaining": "42"})`. `headers` holds only the allowlisted + response headers (`retry-after`, `x-proxy-*`, `x-ratelimit-*`), lowercase + and redacted, first value winning for a repeated name, the same as an + error's `headers`, but `{}` rather than `None` when none were sent. It runs + on the calling thread, with the GIL, after the body decoded and before the + method returns, so it is safe to use from several threads. It is never + called for errors, which carry their own `headers`. If it raises, the + method raises that exception and its decoded data is discarded. + `close()` releases the client's pooled connections; afterwards every call, including through a namespace fetched before the close, raises `FmpConfigError`. Closing twice is a no-op, and `with FmpClient(...) as @@ -405,6 +417,12 @@ class FmpClient: max_response_body_bytes: typing.Optional[builtins.int] = None, danger_allow_insecure_authentication: builtins.bool = False, follow_redirects: typing.Optional[builtins.bool] = None, + on_response: typing.Optional[ + typing.Callable[ + [builtins.str, builtins.int, builtins.dict[builtins.str, builtins.str]], + None, + ] + ] = None, ) -> FmpClient: ... def close(self) -> None: r""" diff --git a/crates/fmp-py/src/client.rs b/crates/fmp-py/src/client.rs index c72bfe1..095e13d 100644 --- a/crates/fmp-py/src/client.rs +++ b/crates/fmp-py/src/client.rs @@ -17,6 +17,7 @@ use pyo3_stub_gen::derive::{gen_stub_pyclass, gen_stub_pymethods}; use crate::{ errors::to_py_error, + response_hook, runtime::{self, ClientHandle}, }; @@ -194,6 +195,18 @@ fn positive_duration(value: f64, field: &'static str) -> PyResult { /// non-loopback HTTP requires `danger_allow_insecure_authentication=True`. /// Redirects are either disabled or restricted to the same origin. /// +/// `on_response`, when given, is called as `on_response(endpoint_id, status, +/// headers)` after every successful response, for every namespace method, for +/// example `("quote-short", 200, {"x-proxy-cache": "HIT", +/// "x-proxy-daily-remaining": "42"})`. `headers` holds only the allowlisted +/// response headers (`retry-after`, `x-proxy-*`, `x-ratelimit-*`), lowercase +/// and redacted, first value winning for a repeated name, the same as an +/// error's `headers`, but `{}` rather than `None` when none were sent. It runs +/// on the calling thread, with the GIL, after the body decoded and before the +/// method returns, so it is safe to use from several threads. It is never +/// called for errors, which carry their own `headers`. If it raises, the +/// method raises that exception and its decoded data is discarded. +/// /// `close()` releases the client's pooled connections; afterwards every call, /// including through a namespace fetched before the close, raises /// `FmpConfigError`. Closing twice is a no-op, and `with FmpClient(...) as @@ -208,7 +221,7 @@ pub(crate) struct FmpClient { #[pymethods] impl FmpClient { #[new] - #[pyo3(signature = (*, token=None, base_url=None, path_prefix=None, auth_mode=None, auth_name=None, auth_prefix=None, headers=None, timeout=None, connect_timeout=None, max_response_body_bytes=None, danger_allow_insecure_authentication=false, follow_redirects=None))] + #[pyo3(signature = (*, token=None, base_url=None, path_prefix=None, auth_mode=None, auth_name=None, auth_prefix=None, headers=None, timeout=None, connect_timeout=None, max_response_body_bytes=None, danger_allow_insecure_authentication=false, follow_redirects=None, on_response=None))] #[allow(clippy::too_many_arguments)] fn new( token: Option, @@ -226,6 +239,8 @@ impl FmpClient { max_response_body_bytes: Option>, danger_allow_insecure_authentication: bool, follow_redirects: Option, + #[gen_stub(override_type(type_repr = "typing.Optional[typing.Callable[[builtins.str, builtins.int, builtins.dict[builtins.str, builtins.str]], None]]", imports = ("builtins", "typing")))] + on_response: Option>, ) -> PyResult { let auth = authentication(auth_mode, token, auth_name, auth_prefix)?; let mut builder = ClientBuilder::default().authentication(auth); @@ -262,12 +277,22 @@ impl FmpClient { }); } + let on_response = on_response.filter(|callback| !callback.is_none()); + if let Some(callback) = &on_response { + if !callback.is_callable() { + return Err(invalid_configuration( + "on_response must be callable or None", + )); + } + builder = builder.on_response(response_hook::record); + } + builder .clone() .build() .map_err(|error| build_error(auth_mode, error))?; Ok(Self { - builder: ClientHandle::new(builder), + builder: ClientHandle::new(builder, on_response.map(Bound::unbind)), }) } diff --git a/crates/fmp-py/src/lib.rs b/crates/fmp-py/src/lib.rs index acae223..2338a1b 100644 --- a/crates/fmp-py/src/lib.rs +++ b/crates/fmp-py/src/lib.rs @@ -11,6 +11,7 @@ mod facade_domains; mod models; mod namespaces; mod registration; +mod response_hook; mod runtime; mod secrets; diff --git a/crates/fmp-py/src/response_hook.rs b/crates/fmp-py/src/response_hook.rs new file mode 100644 index 0000000..970d86a --- /dev/null +++ b/crates/fmp-py/src/response_hook.rs @@ -0,0 +1,77 @@ +//! The `on_response` callback of `FmpClient`. +//! +//! The libfmp observer runs inside the request, on whichever thread polls it, +//! without the GIL. It only records what it saw into a per-call task-local +//! slot that [`capture`] scopes around one call; [`deliver`] then runs the +//! Python callback on the calling thread, with the GIL, once the call is done. +//! Concurrent calls each own their slot, so no thread sees another's headers. + +use std::{ + collections::BTreeMap, + future::Future, + mem, + sync::{Arc, Mutex, PoisonError}, +}; + +use libfmp::ResponseInfo; +use pyo3::{prelude::*, types::PyAny}; + +/// One successful response as the callback receives it. +pub(crate) struct Seen { + endpoint: &'static str, + status: u16, + headers: BTreeMap, +} + +type Slot = Arc>>; + +tokio::task_local! { + static SEEN: Slot; +} + +/// The libfmp observer: records one success into the current call's slot. +/// +/// Header names are lowercase and the first value of a repeated name wins, +/// the same rule as the error `headers` attribute. Outside [`capture`] (a +/// client without `on_response`) it does nothing. +pub(crate) fn record(info: &ResponseInfo<'_>) { + let _ = SEEN.try_with(|slot| { + let mut headers = BTreeMap::new(); + for (name, value) in info.headers.iter() { + headers + .entry(name.to_owned()) + .or_insert_with(|| value.to_owned()); + } + slot.lock() + .unwrap_or_else(PoisonError::into_inner) + .push(Seen { + endpoint: info.endpoint, + status: info.status, + headers, + }); + }); +} + +/// Runs `future` with a fresh slot and returns its output with every success +/// recorded during it. +pub(crate) async fn capture(future: Fut) -> (Fut::Output, Vec) { + let slot = Slot::default(); + let output = SEEN.scope(Arc::clone(&slot), future).await; + let seen = mem::take(&mut *slot.lock().unwrap_or_else(PoisonError::into_inner)); + (output, seen) +} + +/// Calls `callback(endpoint_id, status, headers)` for each recorded success, +/// re-acquiring the GIL on the calling thread. The first exception it raises +/// is returned, so the endpoint method raises it instead of returning data. +pub(crate) fn deliver(callback: &Py, seen: Vec) -> PyResult<()> { + if seen.is_empty() { + return Ok(()); + } + Python::attach(|py| { + for success in seen { + callback.call1(py, (success.endpoint, success.status, success.headers))?; + } + Ok(()) + }) +} diff --git a/crates/fmp-py/src/runtime.rs b/crates/fmp-py/src/runtime.rs index 9408b49..f5e8717 100644 --- a/crates/fmp-py/src/runtime.rs +++ b/crates/fmp-py/src/runtime.rs @@ -22,9 +22,12 @@ use std::{ use libfmp::{Client, ClientBuilder}; use tokio::runtime::{Builder as RuntimeBuilder, Handle, Runtime}; -use pyo3::PyResult; +use pyo3::{Py, PyResult, types::PyAny}; -use crate::errors::{FmpError, to_py_error}; +use crate::{ + errors::{FmpError, to_py_error}, + response_hook, +}; fn configuration_error(message: &'static str) -> pyo3::PyErr { to_py_error(libfmp::Error::configuration(message)) @@ -48,13 +51,17 @@ const CLOSED_MESSAGE: &str = "the client is closed"; pub(crate) struct ClientHandle { builder: ClientBuilder, closed: AtomicBool, + on_response: Option>, } impl ClientHandle { - pub(crate) fn new(builder: ClientBuilder) -> Arc { + /// `on_response`, when given, is called after every successful response; + /// the builder must already carry [`response_hook::record`] as observer. + pub(crate) fn new(builder: ClientBuilder, on_response: Option>) -> Arc { Arc::new(Self { builder, closed: AtomicBool::new(false), + on_response, }) } @@ -205,7 +212,8 @@ fn current() -> PyResult> { /// Returns a configuration error when `handle` is closed, or when called from /// inside a runtime worker thread, where `Runtime::block_on` would panic. /// Callers release the GIL (via `Python::detach`) around this so other Python -/// threads make progress. +/// threads make progress. With an `on_response` callback, it runs here after +/// the call, with the GIL re-acquired, and its exception is returned. pub(crate) fn block_on(handle: Arc, operation: F) -> PyResult where F: FnOnce(Client) -> Fut, @@ -221,7 +229,12 @@ where } let inner = current()?; let client = inner.client(&handle)?; - Ok(inner.block_on(operation(client))) + let Some(callback) = &handle.on_response else { + return Ok(inner.block_on(operation(client))); + }; + let (output, seen) = inner.block_on(response_hook::capture(operation(client))); + response_hook::deliver(callback, seen)?; + Ok(output) } /// Mark `handle` closed and drop its pooled client, releasing its idle diff --git a/crates/fmp-py/tests/test_on_response.py b/crates/fmp-py/tests/test_on_response.py new file mode 100644 index 0000000..00e6721 --- /dev/null +++ b/crates/fmp-py/tests/test_on_response.py @@ -0,0 +1,130 @@ +"""Issue #417: ``FmpClient(on_response=...)`` sees allowlisted headers on success.""" + +import threading +from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace +from typing import Any + +import pytest +from conftest import FixtureServer + +QUOTE_SHORT_ROW = {"symbol": "AAPL", "price": 232.8, "change": 2.1, "volume": 44000000} + +VALET_HEADERS = ( + ("X-Proxy-Cache", "HIT"), + ("Set-Cookie", "session=cookie-secret"), + ("Authorization", "Bearer header-secret"), + ("X-Unrelated", "unrelated-value"), + ("X-Proxy-Daily-Remaining", "42"), +) + +Call = tuple[str, int, dict[str, str]] + + +def observed(fixture_server: FixtureServer) -> tuple[Any, list[Call]]: + """A client against ``fixture_server`` whose callback records each call.""" + from fmp import FmpClient + + calls: list[Call] = [] + + def on_response(endpoint_id: str, status: int, headers: dict[str, str]) -> None: + calls.append((endpoint_id, status, headers)) + + client = FmpClient(base_url=fixture_server.base_url, path_prefix="", auth_mode="none", on_response=on_response) + return client, calls + + +def test_on_response_receives_endpoint_status_and_allowlisted_headers(fixture_server: FixtureServer) -> None: + """A valet success reports its cache status and remaining budget, nothing else.""" + fixture_server.route("/quote-short", [QUOTE_SHORT_ROW], headers=VALET_HEADERS) + client, calls = observed(fixture_server) + + rows = client.quote.short("AAPL") + + assert [row.symbol for row in rows] == ["AAPL"] + assert calls == [("quote-short", 200, {"x-proxy-cache": "HIT", "x-proxy-daily-remaining": "42"})] + for secret in ("cookie-secret", "header-secret", "unrelated-value"): + assert secret not in repr(calls) + + +def test_on_response_gets_an_empty_dict_and_first_duplicate(fixture_server: FixtureServer) -> None: + """No allowlisted header gives ``{}``; a repeated name keeps its first value.""" + fixture_server.route("/quote-short", [QUOTE_SHORT_ROW], headers=(("X-Unrelated", "x"),)) + fixture_server.route("/quote", [], headers=(("X-Proxy-Cache", "MISS"), ("X-Proxy-Cache", "HIT"))) + client, calls = observed(fixture_server) + + client.quote.short("AAPL") + client.quote.full("AAPL") + + assert calls == [("quote-short", 200, {}), ("quote", 200, {"x-proxy-cache": "MISS"})] + + +def test_on_response_is_not_called_on_errors(fixture_server: FixtureServer, errors: SimpleNamespace) -> None: + """Status and provider-message errors keep their headers on the exception only.""" + fixture_server.route("/quote-short", b"slow down", status=429, content_type="text/plain", headers=VALET_HEADERS) + fixture_server.route("/quote", b"Invalid name", headers=VALET_HEADERS) + client, calls = observed(fixture_server) + + with pytest.raises(errors.FmpStatusError) as raised: + client.quote.short("AAPL") + assert raised.value.headers == {"x-proxy-cache": "HIT", "x-proxy-daily-remaining": "42"} + with pytest.raises(errors.FmpError): + client.quote.full("AAPL") + + assert calls == [] + + +def test_on_response_exception_propagates_to_the_caller(fixture_server: FixtureServer) -> None: + """An exception raised by the callback replaces the method's return value.""" + from fmp import FmpClient + + fixture_server.route("/quote-short", [QUOTE_SHORT_ROW], headers=VALET_HEADERS) + + class BudgetExhausted(Exception): + pass + + def on_response(_endpoint_id: str, _status: int, headers: dict[str, str]) -> None: + raise BudgetExhausted(headers["x-proxy-daily-remaining"]) + + client = FmpClient(base_url=fixture_server.base_url, path_prefix="", auth_mode="none", on_response=on_response) + with pytest.raises(BudgetExhausted, match="^42$"): + client.quote.short("AAPL") + + +def test_on_response_threads_see_only_their_own_headers(fixture_server: FixtureServer) -> None: + """Concurrent calls each get the headers of their own response, on their own thread.""" + symbols = [f"S{index}" for index in range(16)] + for symbol in symbols: + fixture_server.route( + f"/quote-short?symbol={symbol}", + [dict(QUOTE_SHORT_ROW, symbol=symbol)], + headers=(("X-Proxy-Daily-Remaining", symbol),), + ) + from fmp import FmpClient + + seen: dict[int, list[str]] = {} + + def on_response(_endpoint_id: str, _status: int, headers: dict[str, str]) -> None: + seen.setdefault(threading.get_ident(), []).append(headers["x-proxy-daily-remaining"]) + + client = FmpClient(base_url=fixture_server.base_url, path_prefix="", auth_mode="none", on_response=on_response) + + def call(symbol: str) -> tuple[int, str]: + rows = client.quote.short(symbol) + assert [row.symbol for row in rows] == [symbol] + return threading.get_ident(), seen[threading.get_ident()][-1] + + with ThreadPoolExecutor(max_workers=8) as pool: + results = list(pool.map(call, symbols)) + + assert [remaining for _, remaining in results] == symbols + assert sorted(value for values in seen.values() for value in values) == sorted(symbols) + + +def test_on_response_must_be_callable(fixture_server: FixtureServer, errors: SimpleNamespace) -> None: + """A non-callable ``on_response`` is a configuration error; ``None`` is accepted.""" + from fmp import FmpClient + + with pytest.raises(errors.FmpConfigError, match="on_response must be callable"): + FmpClient(base_url=fixture_server.base_url, path_prefix="", auth_mode="none", on_response=42) + FmpClient(base_url=fixture_server.base_url, path_prefix="", auth_mode="none", on_response=None) diff --git a/crates/libfmp/README.md b/crates/libfmp/README.md index cfe1154..533362e 100644 --- a/crates/libfmp/README.md +++ b/crates/libfmp/README.md @@ -67,6 +67,35 @@ to a non-loopback host is still refused unless the builder opts in with `danger_allow_insecure_authentication(true)`. `Client::builder()` itself never reads either variable. +To read the proxy's cache status and remaining daily budget on a success, run +the endpoint spec through `execute_with_metadata`. It returns a `Response` +with the decoded `data`, the HTTP `status` and the allowlisted, redacted +`headers` (the same allowlist as status errors, below): + +```rust +use libfmp::{Client, endpoints::quote::{QuoteShortQuery, quote_short}, types::Ticker}; + +# async fn example() -> Result<(), Box> { +let client = Client::from_env()?; +let spec = quote_short(QuoteShortQuery::new(Ticker::new("AAPL")?)); +let response = client.execute_with_metadata(&spec).await?; +let cache = response.headers.get("x-proxy-cache"); // Some("HIT") or Some("MISS") +let remaining = response.headers.get("x-proxy-daily-remaining"); +# let _ = (response.data, cache, remaining); +# Ok(()) +# } +``` + +Every endpoint method has a public spec function under `libfmp::endpoints`, +and `execute` returns the same data without the metadata. On failure nothing +changes: the error already carries its headers. + +To watch every call, endpoint methods included, register +`ClientBuilder::on_response(|info| ...)`. After each decoded success it gets a +`ResponseInfo` with the endpoint id, status and the same allowlisted headers. +It never fires on errors, cannot fail the call, and runs inline on the request +task, so keep it cheap and non-blocking. + ## Custom router or proxy Transport choices do not change endpoint code. A caller can select a custom base URL and path prefix, no auth, FMP header or query auth, bearer auth, custom secret header or query auth, and arbitrary default headers. diff --git a/crates/libfmp/src/client.rs b/crates/libfmp/src/client.rs index 2c460ba..0b959e5 100644 --- a/crates/libfmp/src/client.rs +++ b/crates/libfmp/src/client.rs @@ -95,6 +95,7 @@ pub struct ClientBuilder { danger_allow_insecure_authentication: bool, redirect_policy: RedirectPolicy, executor: Option>, + on_response: Option, } impl Default for ClientBuilder { @@ -113,6 +114,7 @@ impl Default for ClientBuilder { danger_allow_insecure_authentication: false, redirect_policy: RedirectPolicy::SameOrigin, executor: None, + on_response: None, } } } @@ -141,6 +143,7 @@ impl fmt::Debug for ClientBuilder { ) .field("redirect_policy", &self.redirect_policy) .field("custom_executor", &self.executor.is_some()) + .field("response_observer", &self.on_response.is_some()) .finish() } } @@ -302,6 +305,39 @@ impl ClientBuilder { self } + /// Registers an observer called after every successful response. + /// + /// It receives a [`ResponseInfo`] with the endpoint id, the HTTP status + /// and the same allowlisted, redacted [`SafeHeaders`] that + /// [`Client::execute_with_metadata`] returns, for every call made through + /// the client, endpoint methods included. It fires once per call, only + /// after the body decoded, never on status, provider-message, decode or + /// transport errors (those carry their headers on the [`Error`]). It + /// returns `()`, so it cannot fail the call. + /// + /// The observer runs inline on the task driving the request, before the + /// call returns: keep it cheap and non-blocking (record a value, send on a + /// channel). A later call replaces an earlier observer. + /// + /// ``` + /// use std::sync::{Arc, Mutex}; + /// + /// let remaining = Arc::new(Mutex::new(None::)); + /// let seen = Arc::clone(&remaining); + /// let builder = libfmp::Client::builder().on_response(move |info| { + /// let value = info.headers.get("x-proxy-daily-remaining").map(str::to_owned); + /// *seen.lock().unwrap() = value; + /// }); + /// # let _ = builder; + /// ``` + pub fn on_response( + mut self, + observer: impl Fn(&ResponseInfo<'_>) + Send + Sync + 'static, + ) -> Self { + self.on_response = Some(Arc::new(observer)); + self + } + /// Validates configuration and builds a cheap-to-clone async client. pub fn build(self) -> Result { if self.authentication_conflict { @@ -376,6 +412,7 @@ impl ClientBuilder { timeout: self.timeout, max_response_body_bytes: self.max_response_body_bytes, redirect_policy: self.redirect_policy, + on_response: self.on_response, }), }) } @@ -393,6 +430,7 @@ struct ClientInner { timeout: Option, max_response_body_bytes: Option, redirect_policy: RedirectPolicy, + on_response: Option, } /// Async Financial Modeling Prep client shared by all endpoint modules. @@ -401,6 +439,39 @@ pub struct Client { inner: Arc, } +/// The observer registered with [`ClientBuilder::on_response`]. +pub type ResponseObserver = Arc) + Send + Sync>; + +/// What a [`ResponseObserver`] learns about one successful response. +/// +/// It never carries the body or any header outside the allowlist. +#[derive(Debug, Clone, Copy, PartialEq)] +#[non_exhaustive] +pub struct ResponseInfo<'a> { + /// The registry id of the endpoint, for example `"quote"`. + pub endpoint: &'static str, + /// The HTTP status of the final (post-redirect) response. + pub status: u16, + /// The allowlisted, redacted headers of the final response. + pub headers: &'a SafeHeaders, +} + +/// A decoded success with the HTTP status and allowlisted response headers. +/// +/// Returned by [`Client::execute_with_metadata`]. `headers` holds only the +/// names accepted by [`crate::error::is_retained_header_name`], redacted and +/// bounded exactly like [`Error::headers`]. +#[derive(Debug, Clone, PartialEq)] +#[non_exhaustive] +pub struct Response { + /// The decoded endpoint data, as [`Client::execute`] returns it. + pub data: T, + /// The HTTP status of the final (post-redirect) response. + pub status: u16, + /// The allowlisted, redacted headers of the final response. + pub headers: SafeHeaders, +} + impl fmt::Debug for Client { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { formatter @@ -440,6 +511,43 @@ impl Client { /// Executes a typed endpoint using only transport-owned default headers. pub async fn execute(&self, endpoint: &EndpointSpec) -> Result + where + Q: QueryParameters, + { + self.execute_with_metadata(endpoint) + .await + .map(|response| response.data) + } + + /// Executes a typed endpoint and keeps the status and allowlisted headers. + /// + /// The request, redirects and decoding are exactly those of + /// [`Self::execute`]; on success the decoded data comes back with the HTTP + /// status and the [`SafeHeaders`] of the final response, filtered by the + /// same allowlist as [`Error::headers`] (`retry-after`, `x-proxy-*`, + /// `x-ratelimit-*`) and redacted. Errors are unchanged and already carry + /// their own headers. + /// + /// Endpoint spec functions are public, so any endpoint method has a + /// metadata form. Reading a proxy's cache status and remaining budget: + /// + /// ```no_run + /// use libfmp::{Client, endpoints::quote::{QuoteQuery, quote}, types::Ticker}; + /// + /// # async fn run() -> libfmp::Result<()> { + /// let client = Client::from_env()?; + /// let spec = quote(QuoteQuery::new(Ticker::new("AAPL")?)); + /// let response = client.execute_with_metadata(&spec).await?; + /// let cache = response.headers.get("x-proxy-cache"); + /// let remaining = response.headers.get("x-proxy-daily-remaining"); + /// println!("{} rows, cache {cache:?}, remaining {remaining:?}", response.data.len()); + /// # Ok(()) + /// # } + /// ``` + pub async fn execute_with_metadata( + &self, + endpoint: &EndpointSpec, + ) -> Result> where Q: QueryParameters, { @@ -454,7 +562,7 @@ impl Client { &self, endpoint: &EndpointSpec, request_headers: &[(&str, &str)], - ) -> Result + ) -> Result> where Q: QueryParameters, { @@ -494,7 +602,7 @@ impl Client { headers: HeaderMap, redactor: &Redactor, timeout: Duration, - ) -> Result { + ) -> Result> { for redirect_count in 0..=MAX_REDIRECTS { apply_query_auth(&mut url, self.inner.auth.query.as_ref()); let max_body_bytes = endpoint @@ -607,6 +715,21 @@ impl Client { response.body_bytes(), ResponseMetadata::new(content_type, content_disposition), ) + .map(|data| { + let success = Response { + data, + status: response.status(), + headers: safe_headers(&response, redactor), + }; + if let Some(observer) = &self.inner.on_response { + observer(&ResponseInfo { + endpoint: endpoint.id(), + status: success.status, + headers: &success.headers, + }); + } + success + }) .map_err(|failure| match failure { DecodeFailure::Shape { path, kind } => Error::decode( Some(endpoint.id()), diff --git a/crates/libfmp/src/lib.rs b/crates/libfmp/src/lib.rs index cd65a1a..84cba45 100644 --- a/crates/libfmp/src/lib.rs +++ b/crates/libfmp/src/lib.rs @@ -16,7 +16,7 @@ pub mod responses; pub mod transport; pub mod types; -pub use client::{Client, ClientBuilder}; +pub use client::{Client, ClientBuilder, Response, ResponseInfo, ResponseObserver}; pub use error::{Error, Result}; pub use bytes; diff --git a/crates/libfmp/tests/success_headers.rs b/crates/libfmp/tests/success_headers.rs new file mode 100644 index 0000000..3a62592 --- /dev/null +++ b/crates/libfmp/tests/success_headers.rs @@ -0,0 +1,174 @@ +#[allow(dead_code)] // Only the fixture executor is used here. +mod support; + +use std::sync::{Arc, Mutex}; + +use libfmp::{ + Client, ClientBuilder, + config::Authentication, + endpoints::quote::{QuoteShortQuery, quote_short}, + error::ErrorCategory, + http::{HeaderMap, HeaderName, HeaderValue}, + transport::TransportResponse, + types::Ticker, +}; + +use support::FixtureExecutor; + +const BODY: &str = r#"[{"symbol":"AAPL","price":232.8,"change":2.1,"volume":44000000}]"#; + +const VALET_HEADERS: &[(&str, &str)] = &[ + ("Content-Type", "application/json"), + ("X-Proxy-Cache", "HIT"), + ("Set-Cookie", "session=cookie-secret"), + ("Authorization", "Bearer header-secret"), + ("X-Unrelated", "unrelated-value"), + ("X-Proxy-Daily-Remaining", "42"), +]; + +fn response(status: u16, pairs: &[(&str, &str)], body: &'static str) -> TransportResponse { + let mut headers = HeaderMap::new(); + for (name, value) in pairs { + headers.append( + HeaderName::from_bytes(name.as_bytes()).unwrap(), + HeaderValue::from_str(value).unwrap(), + ); + } + TransportResponse::new(status, headers, body) +} + +fn builder(response: TransportResponse) -> ClientBuilder { + Client::builder() + .base_url("https://proxy.example/router") + .path_prefix("stable") + .authentication(Authentication::None) + .executor(Arc::new(FixtureExecutor::new([response]))) +} + +fn client(response: TransportResponse) -> Client { + builder(response).build().unwrap() +} + +type Seen = Arc)>>>; + +/// Builds a client whose observer records every call it sees. +fn observed(response: TransportResponse) -> (Client, Seen) { + let seen = Seen::default(); + let sink = Arc::clone(&seen); + let client = builder(response) + .on_response(move |info| { + let headers = info + .headers + .iter() + .map(|(name, value)| (name.to_owned(), value.to_owned())) + .collect(); + sink.lock() + .unwrap() + .push((info.endpoint, info.status, headers)); + }) + .build() + .unwrap(); + (client, seen) +} + +#[tokio::test] +async fn execute_with_metadata_keeps_status_and_proxy_headers_only() { + let client = client(response(200, VALET_HEADERS, BODY)); + let query = QuoteShortQuery::new(Ticker::new("AAPL").unwrap()); + + let response = client + .execute_with_metadata("e_short(query)) + .await + .unwrap(); + + assert_eq!(response.status, 200); + assert_eq!(response.data.len(), 1); + assert_eq!(response.data[0].symbol.as_str(), "AAPL"); + let kept: Vec<_> = response.headers.iter().collect(); + assert_eq!( + kept, + [("x-proxy-cache", "HIT"), ("x-proxy-daily-remaining", "42")] + ); + assert_eq!(response.headers.get("X-Proxy-Cache"), Some("HIT")); + let debug = format!("{response:?}"); + for secret in ["cookie-secret", "header-secret", "unrelated-value"] { + assert!(!debug.contains(secret), "{secret}"); + } +} + +#[tokio::test] +async fn execute_returns_the_same_data_without_metadata() { + let query = QuoteShortQuery::new(Ticker::new("AAPL").unwrap()); + let with_metadata = client(response(200, VALET_HEADERS, BODY)) + .execute_with_metadata("e_short(query.clone())) + .await + .unwrap(); + let rows = client(response(200, VALET_HEADERS, BODY)) + .execute("e_short(query)) + .await + .unwrap(); + + assert_eq!(rows, with_metadata.data); +} + +#[tokio::test] +async fn execute_with_metadata_errors_are_unchanged() { + let failure = response(429, &[("X-Proxy-Daily-Remaining", "0")], "slow down"); + let query = QuoteShortQuery::new(Ticker::new("AAPL").unwrap()); + + let error = client(failure) + .execute_with_metadata("e_short(query)) + .await + .unwrap_err(); + + assert_eq!(error.category(), ErrorCategory::Status); + assert_eq!(error.headers().get("x-proxy-daily-remaining"), Some("0")); +} + +#[tokio::test] +async fn observer_sees_endpoint_method_successes() { + let (client, seen) = observed(response(200, VALET_HEADERS, BODY)); + + let rows = client + .quote_short(Ticker::new("AAPL").unwrap()) + .await + .unwrap(); + + assert_eq!(rows.len(), 1); + let seen = seen.lock().unwrap(); + let pair = |name: &str, value: &str| (name.to_owned(), value.to_owned()); + assert_eq!( + *seen, + [( + "quote-short", + 200, + vec![ + pair("x-proxy-cache", "HIT"), + pair("x-proxy-daily-remaining", "42") + ] + )] + ); +} + +#[tokio::test] +async fn observer_is_silent_on_errors() { + for failure in [ + response(429, VALET_HEADERS, "slow down"), + response( + 200, + VALET_HEADERS, + r#"{"Error Message":"Invalid API KEY."}"#, + ), + response(200, VALET_HEADERS, r#"[{"symbol":7}]"#), + ] { + let (client, seen) = observed(failure); + let query = QuoteShortQuery::new(Ticker::new("AAPL").unwrap()); + + client + .execute_with_metadata("e_short(query)) + .await + .unwrap_err(); + + assert!(seen.lock().unwrap().is_empty()); + } +} diff --git a/sdk/go/README.md b/sdk/go/README.md index 72dd368..60965b3 100644 --- a/sdk/go/README.md +++ b/sdk/go/README.md @@ -99,6 +99,25 @@ the environment key (use `NewClient` to pick another mode). Plaintext `WithDangerAllowInsecureAuthentication`. `NewClient` never reads either variable. +To read the proxy's cache status and remaining daily budget on a success, +attach a `ResponseMetadata` to the call's context. After a decoded success the +client fills it with the endpoint id, the HTTP status and the allowlisted, +redacted headers (the same allowlist as errors: `Retry-After`, `X-Proxy-*`, +`X-RateLimit-*`): + +```go +var meta fmp.ResponseMetadata +quotes, err := client.Quote.Short(fmp.WithResponseMetadata(ctx, &meta), fmp.NewQuoteShortQuery("AAPL")) +if err != nil { + return err // *Error already carries its headers; meta is untouched +} +cache := meta.Headers.Get("X-Proxy-Cache") // "HIT" or "MISS" +remaining := meta.Headers.Get("X-Proxy-Daily-Remaining") +``` + +Use one context and one `ResponseMetadata` per call: concurrent calls sharing +a `meta` race on it. + `client.CloseIdleConnections()` closes the idle keep-alive connections of the transport `NewClient` built; the client stays usable. It is a no-op with `WithHTTPClient`, whose transport stays with the caller. There is no `Close`: diff --git a/sdk/go/binary.go b/sdk/go/binary.go index 69e7c5b..70d4c4a 100644 --- a/sdk/go/binary.go +++ b/sdk/go/binary.go @@ -77,6 +77,7 @@ func (c *Client) getBinary(ctx context.Context, endpointID, relativePath string, if !visibleHeaderText(disposition) { disposition = "" } + c.recordResponseMetadata(ctx, endpointID, resp) return BinaryPayload{Data: resp.body, ContentType: contentType, ContentDisposition: disposition}, nil } diff --git a/sdk/go/client.go b/sdk/go/client.go index 7b41e95..01a211a 100644 --- a/sdk/go/client.go +++ b/sdk/go/client.go @@ -366,6 +366,7 @@ func (c *Client) getJSON(ctx context.Context, endpointID, relativePath string, q decoded.Path, decoded.DecodeKind = c.redactor.Redact(path), kind return decoded } + c.recordResponseMetadata(ctx, endpointID, resp) return nil } diff --git a/sdk/go/csv.go b/sdk/go/csv.go index ccb5da2..3388dde 100644 --- a/sdk/go/csv.go +++ b/sdk/go/csv.go @@ -78,6 +78,7 @@ func getCSV[T any](ctx context.Context, c *Client, endpointID, relativePath stri decoded.Path, decoded.DecodeKind = c.redactor.Redact(path), kind return nil, decoded } + c.recordResponseMetadata(ctx, endpointID, resp) return rows, nil } diff --git a/sdk/go/doc.go b/sdk/go/doc.go index 916107d..777249f 100644 --- a/sdk/go/doc.go +++ b/sdk/go/doc.go @@ -27,6 +27,12 @@ // Without FMP_BASE_URL it targets DefaultBaseURL. Options given to it are // applied after the environment, so WithBaseURL wins. // +// To read such a proxy's X-Proxy-Cache and X-Proxy-Daily-Remaining headers +// on a success, pass WithResponseMetadata(ctx, &meta): after a decoded +// success the client fills meta with the endpoint id, the status and the +// allowlisted headers (see IsRetainedHeaderName). Errors leave meta untouched +// and carry their own headers. Use one ctx/meta pair per call. +// // Header authentication modes (FMPHeader, Bearer, CustomHeader, // CustomHeaderWithPrefix) never place a credential in the URL. Query modes // (FMPQuery, CustomQuery) append the secret as the last query pair of every diff --git a/sdk/go/response_metadata.go b/sdk/go/response_metadata.go new file mode 100644 index 0000000..4b9e16f --- /dev/null +++ b/sdk/go/response_metadata.go @@ -0,0 +1,54 @@ +package fmp + +import ( + "context" + "net/http" +) + +// ResponseMetadata describes one successful response: the registry endpoint +// id, the HTTP status, and the allowlisted response headers. +// +// Headers holds only the names accepted by IsRetainedHeaderName (Retry-After, +// X-Proxy-*, X-RateLimit-*), redacted and bounded exactly like Error.Headers, +// in canonical form; it is nil when none qualify. Read a proxy's cache status +// with meta.Headers.Get("X-Proxy-Cache"). +type ResponseMetadata struct { + EndpointID string + StatusCode int + Headers http.Header +} + +type responseMetadataKey struct{} + +// WithResponseMetadata returns a context that asks the client to fill *meta +// after a successful call made with it. The client writes *meta only once the +// response has been decoded; on any error *meta is left untouched, and the +// *Error already carries its own allowlisted headers. +// +// Use one context and one ResponseMetadata per call: concurrent calls that +// share a meta race on it, and a later call overwrites an earlier one. A nil +// meta returns ctx unchanged. +// +// var meta fmp.ResponseMetadata +// rows, err := client.Quote.Full(fmp.WithResponseMetadata(ctx, &meta), fmp.NewQuoteQuery("AAPL")) +// remaining := meta.Headers.Get("X-Proxy-Daily-Remaining") +func WithResponseMetadata(ctx context.Context, meta *ResponseMetadata) context.Context { + if meta == nil { + return ctx + } + return context.WithValue(ctx, responseMetadataKey{}, meta) +} + +// recordResponseMetadata fills the caller's ResponseMetadata, if any, from a +// successful response. +func (c *Client) recordResponseMetadata(ctx context.Context, endpointID string, resp *bufferedResponse) { + meta, ok := ctx.Value(responseMetadataKey{}).(*ResponseMetadata) + if !ok { + return + } + *meta = ResponseMetadata{ + EndpointID: endpointID, + StatusCode: resp.status, + Headers: retainedHeaders(resp.header, c.redactor), + } +} diff --git a/sdk/go/response_metadata_test.go b/sdk/go/response_metadata_test.go new file mode 100644 index 0000000..d9215a2 --- /dev/null +++ b/sdk/go/response_metadata_test.go @@ -0,0 +1,131 @@ +package fmp + +import ( + "context" + "fmt" + "net/http" + "reflect" + "strings" + "testing" +) + +// valetSuccessServer answers every request with contentType, body, status and the +// headers a valet proxy adds, plus headers that must never be retained. +func valetSuccessServer(t *testing.T, status int, contentType, body string) *Client { + t.Helper() + server, _ := newServer(t, func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", contentType) + w.Header().Set("X-Proxy-Cache", "HIT") + w.Header().Set("X-Proxy-Daily-Remaining", "42") + w.Header().Set("Set-Cookie", "session=cookie-secret") + w.Header().Set("Authorization", "Bearer header-secret") + w.Header().Set("X-Unrelated", "unrelated-value") + w.WriteHeader(status) + _, _ = w.Write([]byte(body)) + }) + return newClient(t, server) +} + +func assertValetMetadata(t *testing.T, meta ResponseMetadata, endpointID string) { + t.Helper() + want := ResponseMetadata{ + EndpointID: endpointID, + StatusCode: http.StatusOK, + Headers: http.Header{"X-Proxy-Cache": {"HIT"}, "X-Proxy-Daily-Remaining": {"42"}}, + } + if !reflect.DeepEqual(meta, want) { + t.Fatalf("meta = %+v, want %+v", meta, want) + } + formatted := fmt.Sprintf("%+v", meta) + for _, leak := range []string{"cookie-secret", "header-secret", "unrelated-value"} { + if strings.Contains(formatted, leak) { + t.Fatalf("meta retained %q: %s", leak, formatted) + } + } +} + +func TestWithResponseMetadataFillsJSONSuccess(t *testing.T) { + t.Parallel() + client := valetSuccessServer(t, http.StatusOK, "application/json", `[{"symbol":"AAPL"}]`) + var meta ResponseMetadata + var rows []probeRow + err := client.getJSON(WithResponseMetadata(context.Background(), &meta), "quote", "quote", nil, &rows) + if err != nil { + t.Fatal(err) + } + if len(rows) != 1 || rows[0].Symbol != "AAPL" { + t.Fatalf("rows = %+v", rows) + } + assertValetMetadata(t, meta, "quote") +} + +func TestWithResponseMetadataFillsCSVSuccess(t *testing.T) { + t.Parallel() + client := valetSuccessServer(t, http.StatusOK, "text/csv", "symbol\nAAPL\n") + var meta ResponseMetadata + rows, err := getCSV[probeRow](WithResponseMetadata(context.Background(), &meta), client, csvEndpoint, "bulk-probe", nil) + if err != nil { + t.Fatal(err) + } + if len(rows) != 1 || rows[0].Symbol != "AAPL" { + t.Fatalf("rows = %+v", rows) + } + assertValetMetadata(t, meta, csvEndpoint) +} + +func TestWithResponseMetadataFillsBinarySuccess(t *testing.T) { + t.Parallel() + client := valetSuccessServer(t, http.StatusOK, binaryMediaType, "payload") + var meta ResponseMetadata + ctx := WithResponseMetadata(context.Background(), &meta) + body, err := client.getBinary(ctx, binaryEndpoint, binaryPath, nil, []string{binaryMediaType}) + if err != nil { + t.Fatal(err) + } + if string(body.Data) != "payload" { + t.Fatalf("payload = %+v", body) + } + assertValetMetadata(t, meta, binaryEndpoint) +} + +func TestWithResponseMetadataLeavesMetaUntouchedOnError(t *testing.T) { + t.Parallel() + sentinel := ResponseMetadata{EndpointID: "sentinel", StatusCode: -1} + cases := []struct { + name string + status int + body string + }{ + {"status error", http.StatusTooManyRequests, `{"error":"slow down"}`}, + {"decode error", http.StatusOK, `{"not":"an array"}`}, + {"provider message", http.StatusOK, `{"Error Message":"Invalid API KEY."}`}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + client := valetSuccessServer(t, tc.status, "application/json", tc.body) + meta := sentinel + var rows []probeRow + err := client.getJSON(WithResponseMetadata(context.Background(), &meta), "quote", "quote", nil, &rows) + if err == nil { + t.Fatal("expected an error") + } + if !reflect.DeepEqual(meta, sentinel) { + t.Fatalf("meta = %+v, want untouched", meta) + } + }) + } +} + +func TestWithResponseMetadataNilAndAbsentAreNoOps(t *testing.T) { + t.Parallel() + ctx := context.Background() + if got := WithResponseMetadata(ctx, nil); got != ctx { + t.Fatal("nil meta must return ctx unchanged") + } + client := valetSuccessServer(t, http.StatusOK, "application/json", `[{"symbol":"AAPL"}]`) + var rows []probeRow + if err := client.getJSON(ctx, "quote", "quote", nil, &rows); err != nil { + t.Fatal(err) + } +}