From f320776e0e3fae292ab336a7a98839868af4801a Mon Sep 17 00:00:00 2001 From: benbenbang Date: Tue, 6 Oct 2026 18:34:31 +0800 Subject: [PATCH 1/4] feat(client): add execute_with_metadata for status and headers This change introduces a metadata-aware execution path so callers can read the HTTP status and allowlisted response headers on success, not only on failure. **New capability:** * Added `Client::execute_with_metadata`, which runs the same request, redirects and decoding as `execute` but returns a `Response` carrying the decoded `data`, the final `status`, and the redacted `SafeHeaders`. * Added the public `Response` struct and re-exported it from the crate root alongside `Client` and `ClientBuilder`. * `execute` now delegates to `execute_with_metadata` and returns just the data, so existing behaviour is unchanged; the internal header filtering uses the same allowlist as `Error::headers` (`retry-after`, `x-proxy-*`, `x-ratelimit-*`). **Why:** * This lets callers inspect proxy cache status and remaining daily budget (`x-proxy-cache`, `x-proxy-daily-remaining`) on successful responses without dropping down to the transport layer. **Docs and tests:** * Documented the metadata form in the README and on the method, showing how to read cache and budget headers. * Added `tests/success_headers.rs` covering the headers surfaced on success. --- crates/libfmp/README.md | 23 ++++++ crates/libfmp/src/client.rs | 62 ++++++++++++++- crates/libfmp/src/lib.rs | 2 +- crates/libfmp/tests/success_headers.rs | 102 +++++++++++++++++++++++++ 4 files changed, 186 insertions(+), 3 deletions(-) create mode 100644 crates/libfmp/tests/success_headers.rs diff --git a/crates/libfmp/README.md b/crates/libfmp/README.md index cfe11545..926bc9fa 100644 --- a/crates/libfmp/README.md +++ b/crates/libfmp/README.md @@ -67,6 +67,29 @@ 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. + ## 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 2c460bae..773b365a 100644 --- a/crates/libfmp/src/client.rs +++ b/crates/libfmp/src/client.rs @@ -401,6 +401,22 @@ pub struct Client { inner: Arc, } +/// 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 +456,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 +507,7 @@ impl Client { &self, endpoint: &EndpointSpec, request_headers: &[(&str, &str)], - ) -> Result + ) -> Result> where Q: QueryParameters, { @@ -494,7 +547,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 +660,11 @@ impl Client { response.body_bytes(), ResponseMetadata::new(content_type, content_disposition), ) + .map(|data| Response { + data, + status: response.status(), + headers: safe_headers(&response, redactor), + }) .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 cd65a1a8..721d5190 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}; 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 00000000..a7caf714 --- /dev/null +++ b/crates/libfmp/tests/success_headers.rs @@ -0,0 +1,102 @@ +#[allow(dead_code)] // Only the fixture executor is used here. +mod support; + +use std::sync::Arc; + +use libfmp::{ + Client, + 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 client(response: TransportResponse) -> Client { + Client::builder() + .base_url("https://proxy.example/router") + .path_prefix("stable") + .authentication(Authentication::None) + .executor(Arc::new(FixtureExecutor::new([response]))) + .build() + .unwrap() +} + +#[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")); +} From eb601d4cdbad1985effa4041230d7162c67a1d78 Mon Sep 17 00:00:00 2001 From: benbenbang Date: Tue, 6 Oct 2026 18:39:00 +0800 Subject: [PATCH 2/4] feat(sdk/go): expose response metadata on successful calls This commit adds a way to read the proxy's cache status and remaining daily budget after a successful call. Callers attach a ResponseMetadata to the call's context with WithResponseMetadata, and the client fills it once the response decodes successfully. **New capability:** * Added response_metadata.go introducing ResponseMetadata and WithResponseMetadata, which carry a *ResponseMetadata through the call's context. * On a decoded success the client records the endpoint id, the HTTP status and the allowlisted, redacted headers (the same allowlist as errors: Retry-After, X-Proxy-*, X-RateLimit-*) via a new recordResponseMetadata call in binary.go, client.go and csv.go. * Errors leave the metadata untouched since *Error already carries its own headers. Each call needs its own ctx/meta pair; sharing one meta across concurrent calls races on it. **Documentation and tests:** * Documented the feature in README.md and doc.go, including the X-Proxy-Cache and X-Proxy-Daily-Remaining use case. * Added response_metadata_test.go covering the new behaviour. --- sdk/go/README.md | 19 +++++ sdk/go/binary.go | 1 + sdk/go/client.go | 1 + sdk/go/csv.go | 1 + sdk/go/doc.go | 6 ++ sdk/go/response_metadata.go | 54 +++++++++++++ sdk/go/response_metadata_test.go | 131 +++++++++++++++++++++++++++++++ 7 files changed, 213 insertions(+) create mode 100644 sdk/go/response_metadata.go create mode 100644 sdk/go/response_metadata_test.go diff --git a/sdk/go/README.md b/sdk/go/README.md index 72dd3686..60965b3d 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 69e7c5b4..70d4c4ae 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 7b41e95a..01a211a3 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 ccb5da23..3388dde3 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 916107db..777249f7 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 00000000..4b9e16fc --- /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 00000000..d9215a2e --- /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) + } +} From a4848288c442b8e071ba25bc48e438e3751a23fc Mon Sep 17 00:00:00 2001 From: benbenbang Date: Tue, 6 Oct 2026 18:40:30 +0800 Subject: [PATCH 3/4] feat(client): add on_response observer for successful calls This pull request adds an optional response observer to the client so callers can watch every successful response made through it, including those from endpoint methods. The main changes introduce the registration API, the data passed to observers, and the firing logic. **Observer registration and types:** * Added `ClientBuilder::on_response`, which stores a `ResponseObserver` closure and replaces any previously registered observer, carried through the builder into the client configuration. * Added the `ResponseObserver` type alias and the `ResponseInfo` struct, exposing the endpoint id, the final HTTP status, and the allowlisted, redacted headers, and re-exported both from the crate root. **Firing behaviour:** * The observer runs inline once per call, only after the body decodes successfully, and never fires on status, provider-message, decode, or transport errors, since those already carry their headers on the error. * Included the observer state in the builder and client `Debug` output as a presence flag. **Documentation and tests:** * Documented the observer in the README and on the builder method, noting it cannot fail the call and should stay cheap and non-blocking. * Added tests covering that the observer sees endpoint-method successes with the expected endpoint, status, and headers, and stays silent across status, provider-message, and decode errors. --- crates/libfmp/README.md | 6 ++ crates/libfmp/src/client.rs | 73 +++++++++++++++++++++-- crates/libfmp/src/lib.rs | 2 +- crates/libfmp/tests/success_headers.rs | 80 ++++++++++++++++++++++++-- 4 files changed, 152 insertions(+), 9 deletions(-) diff --git a/crates/libfmp/README.md b/crates/libfmp/README.md index 926bc9fa..533362e5 100644 --- a/crates/libfmp/README.md +++ b/crates/libfmp/README.md @@ -90,6 +90,12 @@ 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 773b365a..0b959e5f 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,23 @@ 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 @@ -660,10 +715,20 @@ impl Client { response.body_bytes(), ResponseMetadata::new(content_type, content_disposition), ) - .map(|data| Response { - data, - status: response.status(), - headers: safe_headers(&response, redactor), + .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( diff --git a/crates/libfmp/src/lib.rs b/crates/libfmp/src/lib.rs index 721d5190..84cba457 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, Response}; +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 index a7caf714..3a62592c 100644 --- a/crates/libfmp/tests/success_headers.rs +++ b/crates/libfmp/tests/success_headers.rs @@ -1,10 +1,10 @@ #[allow(dead_code)] // Only the fixture executor is used here. mod support; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use libfmp::{ - Client, + Client, ClientBuilder, config::Authentication, endpoints::quote::{QuoteShortQuery, quote_short}, error::ErrorCategory, @@ -37,14 +37,38 @@ fn response(status: u16, pairs: &[(&str, &str)], body: &'static str) -> Transpor TransportResponse::new(status, headers, body) } -fn client(response: TransportResponse) -> Client { +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() + .unwrap(); + (client, seen) } #[tokio::test] @@ -100,3 +124,51 @@ async fn execute_with_metadata_errors_are_unchanged() { 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()); + } +} From 29c829a20c082a9d611558b9983d9949d16dc19a Mon Sep 17 00:00:00 2001 From: benbenbang Date: Tue, 6 Oct 2026 18:58:25 +0800 Subject: [PATCH 4/4] feat(fmp-py): add FmpClient on_response success callback This pull request adds an optional `on_response` callback to `FmpClient` so callers can read the proxy's cache status and remaining daily budget after every successful request, across all namespace methods. **New response callback:** * `FmpClient(on_response=...)` accepts a callable invoked as `on_response(endpoint_id, status, headers)` after each successful response, before the method returns. * `headers` carries only the allowlisted response headers (`retry-after`, `x-proxy-*`, `x-ratelimit-*`), lowercase and redacted, keeping the first value for a repeated name, and is `{}` rather than `None` when none were sent, mirroring an error's `headers`. * The callback runs on the calling thread with the GIL held, making it safe to use from several threads at once; it is never invoked for errors, which already carry their own `headers`, and if it raises, the method raises that exception and discards the decoded data. **Implementation:** * Added `response_hook` module to capture and deliver allowlisted headers, wired through `ClientHandle` and `block_on`. * Validated that `on_response` is callable or `None` at construction time. **Docs and tests:** * Documented the callback in the README and the `__init__.pyi` and Rust docstrings. * Added `tests/test_on_response.py` covering the new behaviour. close #417 --- crates/fmp-py/README.md | 24 ++++ crates/fmp-py/python/fmp/_native/__init__.pyi | 18 +++ crates/fmp-py/src/client.rs | 29 +++- crates/fmp-py/src/lib.rs | 1 + crates/fmp-py/src/response_hook.rs | 77 +++++++++++ crates/fmp-py/src/runtime.rs | 23 +++- crates/fmp-py/tests/test_on_response.py | 130 ++++++++++++++++++ 7 files changed, 295 insertions(+), 7 deletions(-) create mode 100644 crates/fmp-py/src/response_hook.rs create mode 100644 crates/fmp-py/tests/test_on_response.py diff --git a/crates/fmp-py/README.md b/crates/fmp-py/README.md index f1313569..90aa5037 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 b9e6ebda..26e33dc0 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 c72bfe1c..095e13d8 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 acae2230..2338a1b0 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 00000000..970d86a5 --- /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 9408b498..f5e87174 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 00000000..00e67217 --- /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)