diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 05d3a5e..d7746f2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -36,6 +36,8 @@ jobs: # without failing a release. `blocking` did exactly that. - run: cargo build --all-features - run: cargo test --no-run --all-features + # Only --doc: every other test needs a live backend. + - run: cargo test --doc --all-features - run: cargo publish --dry-run wheels: diff --git a/AGENTS.md b/AGENTS.md index da76ca8..9e32b81 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -42,7 +42,7 @@ rather than run unoptimised, and the same applies to the PyO3 module, which need Most tests are integration tests that call a live backend via `create_api_service()`. They read configuration from a local `.env` file (gitignored). Required: - `BASE_URL` — backend root, e.g. `http://localhost:8081` -- Either `TOKEN` (bearer token used as-is, no expiry) **or** the OAuth2 client-credentials set: `CLIENT_ID`, `CLIENT_SECRET`, `TOKEN_URI` (optional: `PROJECT_NAME`) +- Either `TOKEN` (bearer token used as-is, no expiry) **or** the OAuth2 client-credentials set: `CLIENT_ID`, `CLIENT_SECRET`, `TOKEN_URI` Tests that mutate backend state (create/delete) often `sleep` a few seconds between operations and are sensitive to race conditions — prefer running them serially or isolating by unique external IDs. @@ -185,7 +185,7 @@ The API wraps collections in `{ "items": [...] }`. `DataWrapper` mirrors that ### Entity → request-body conversion -`DataHubEntity` is a marker trait (`ext_id()` + `Clone + Serialize`) that unlocks generic `From` impls so a `T`, `&T`, `Vec`, or `&Vec` can be passed to service methods and auto-wrapped into `DataWrapper`. Service methods accept `&I where for<'a> &'a I: Into>` (see `EventsService::create`) — implement `DataHubEntity` on new entity types to get this ergonomics for free. `IdAndExtId` has its own parallel set of `From` impls for delete/byids endpoints. +`DataHubEntity` is a marker trait (`Clone + Serialize`, no methods) that unlocks generic `From` impls so a `T`, `&T`, `Vec`, or `&Vec` can be passed to service methods and auto-wrapped into `DataWrapper`. Service methods accept `&I where for<'a> &'a I: Into>` (see `EventsService::create`) — implement `DataHubEntity` on new entity types to get this ergonomics for free. `IdAndExtId` has its own parallel set of `From` impls for delete/byids endpoints. ### Auth (`src/datahub.rs`) @@ -245,10 +245,10 @@ refusal, so a by-id 404 carries `type: …/errors/not-found` and **`slug()` matc stays as the regression guard. 401s carry a problem document too, with `type: …/errors/unauthorized` and a `detail` naming the -check that failed — a missing, empty, malformed or ambiguous `organization` claim. That is the -reason `src/auth_diagnostics.rs` reconstructs from the token it just sent, so the SDK now appends -a near-duplicate of what the server already said. Note every entry-point 401 shares the -`unauthorized` slug, so the cause is in the prose and cannot be branched on. +check that failed — a missing, empty, malformed or ambiguous `organization` claim. That retired +the SDK's own `auth_diagnostics` module, which existed only to reconstruct the reason from the +token it had just sent. Note every entry-point 401 shares the `unauthorized` slug, so the cause is +in the prose and cannot be branched on. ### Filters (`src/filters.rs`) diff --git a/Cargo.toml b/Cargo.toml index 2d07d25..1c2b3d6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -20,6 +20,11 @@ exclude = [ # Synchronous client (intellistream_datahub_sdk::blocking), same split as reqwest::blocking. blocking = [] +# docs.rs builds with default features only, which would leave `blocking` undocumented even +# though the README points callers at it. +[package.metadata.docs.rs] +all-features = true + [dependencies] maplit = "1" tokio = { version = "1", features = ["full"] } @@ -39,9 +44,6 @@ infer = { version = "0.19", default-features = false, features = ["std"] } thiserror = "2.0.17" zstd = "0.13" geojson = "1" -# Reading (never verifying) the payload of a JWT the SDK already holds, to explain an -# otherwise-unexplained 401 — see `auth_diagnostics`. -base64 = "0.22" arrow-array = "59.3" arrow-schema = "59.3" arrow-ipc = "59.3" diff --git a/README.md b/README.md index 0653c0b..fdf472b 100644 --- a/README.md +++ b/README.md @@ -68,7 +68,6 @@ environment: - `BASE_URL` — backend root, e.g. `http://localhost:8081` (required) - Either `TOKEN` (bearer token used as-is, never considered expired), **or** the OAuth2 client-credentials set: `CLIENT_ID`, `CLIENT_SECRET`, `TOKEN_URI` -- `PROJECT_NAME` — optional - `SCOPE` — against DataHub, needed when the realm uses Keycloak Organizations: that claim comes from a dynamic client scope, so the request must name it (`organization:*`, or `organization:` to pin one tenant). Without it the token carries no tenant and every call diff --git a/datahub_python_bindings/src/datasets/async_service.rs b/datahub_python_bindings/src/datasets/async_service.rs index 63efd6f..1fe3384 100644 --- a/datahub_python_bindings/src/datasets/async_service.rs +++ b/datahub_python_bindings/src/datasets/async_service.rs @@ -2,10 +2,9 @@ use crate::datasets::{ DatasetIdentifiable, PyDataset, PyDatasetUpdate, }; use crate::resources::PyResource; -use crate::{DatahubIdentity, Identifiable, PyIdCollection}; use intellistream_datahub_sdk::ApiService; use intellistream_datahub_sdk::datasets::{Dataset, DatasetUpdate}; -use intellistream_datahub_sdk::generic::{DataWrapper, IdAndExtId}; +use intellistream_datahub_sdk::generic::IdAndExtId; use pyo3::{Bound, PyAny, PyResult, Python, pyclass, pymethods}; use pyo3_async_runtimes::tokio::future_into_py; use std::sync::Arc; diff --git a/datahub_python_bindings/src/datasets/mod.rs b/datahub_python_bindings/src/datasets/mod.rs index 290b753..3861057 100644 --- a/datahub_python_bindings/src/datasets/mod.rs +++ b/datahub_python_bindings/src/datasets/mod.rs @@ -5,9 +5,8 @@ pub(crate) mod sync_service; use crate::PyIdCollection; use crate::events::{PyEvent, PyTimeFilter}; use crate::resources::PyResourceNetwork; -use crate::{PyFieldBool, PyFieldStr, PyListFieldStr, PyMapField}; +use crate::{PyFieldStr, PyListFieldStr, PyMapField}; use intellistream_datahub_sdk::filters::{EventFilter, EventFilterForm}; -use intellistream_datahub_sdk::datahub::to_snake_lower_cased_allow_start_with_digits; use intellistream_datahub_sdk::datasets::{ DatasetFilter, Dataset, DatasetFilterForm, DatasetUpdate, DatasetUpdateFields, }; @@ -19,7 +18,6 @@ use pyo3::{Bound, PyResult, pyclass, pymethods}; use pyo3_async_runtimes::tokio::future_into_py; use std::collections::HashMap; use std::sync::Arc; -use uuid::Uuid; #[pyclass(module = "intellistream_datahub_sdk", name = "Dataset", from_py_object)] #[derive(Clone)] diff --git a/datahub_python_bindings/src/datasets/sync_service.rs b/datahub_python_bindings/src/datasets/sync_service.rs index ac65c2d..899f9bb 100644 --- a/datahub_python_bindings/src/datasets/sync_service.rs +++ b/datahub_python_bindings/src/datasets/sync_service.rs @@ -2,11 +2,10 @@ use crate::datasets::{ DatasetIdentifiable, PyDataset, PyDatasetUpdate, }; use crate::resources::PyResource; -use crate::{PyIdCollection}; use intellistream_datahub_sdk::datasets::{Dataset, DatasetUpdate}; use intellistream_datahub_sdk::generic::IdAndExtId; -use intellistream_datahub_sdk::{ApiService, Resource}; -use pyo3::{Bound, PyAny, PyResult, Python, pyclass, pymethods}; +use intellistream_datahub_sdk::ApiService; +use pyo3::{PyResult, Python, pyclass, pymethods}; use std::sync::Arc; #[pyclass(module = "intellistream_datahub_sdk", name = "DatasetsServiceSync")] diff --git a/datahub_python_bindings/src/events/async_service.rs b/datahub_python_bindings/src/events/async_service.rs index 5d7328d..8cd7df3 100644 --- a/datahub_python_bindings/src/events/async_service.rs +++ b/datahub_python_bindings/src/events/async_service.rs @@ -1,13 +1,9 @@ use crate::events::{ EventIdentifyable, PyEventFilter, PyEvent, PyEventDimension, PyEventUpdate, }; -use crate::timeseries::async_service::PyTimeSeriesServiceAsync; -use crate::timeseries::{PyTimeSeries, PyTimeSeriesUpdate}; -use crate::{PyIdCollection}; use intellistream_datahub_sdk::events::{EventDimension, EventIdCollection, EventUpdate}; -use intellistream_datahub_sdk::generic::DataWrapper; use intellistream_datahub_sdk::{ - ApiService, Event, TimeSeries, TimeSeriesUpdate, TimeSeriesUpdateCollection, + ApiService, Event, }; use pyo3::{Bound, PyAny, PyResult, Python, pyclass, pymethods}; use pyo3_async_runtimes::tokio::future_into_py; diff --git a/datahub_python_bindings/src/events/general.rs b/datahub_python_bindings/src/events/general.rs index a3f816f..92893ac 100644 --- a/datahub_python_bindings/src/events/general.rs +++ b/datahub_python_bindings/src/events/general.rs @@ -1,7 +1,6 @@ use crate::PyEvent; use crate::PyIdCollection; use crate::datetime::py_datetime_to_utc; -use crate::resources::PyResource; use chrono::{DateTime, Utc}; use intellistream_datahub_sdk::generic::IdAndExtId; use pyo3::prelude::*; diff --git a/datahub_python_bindings/src/events/mod.rs b/datahub_python_bindings/src/events/mod.rs index 7911c1b..2b79711 100644 --- a/datahub_python_bindings/src/events/mod.rs +++ b/datahub_python_bindings/src/events/mod.rs @@ -1,17 +1,12 @@ use crate::{DataSetRef, PyIdCollection, StringOrList, opt_data_set_refs, opt_patterns}; use crate::datetime::opt_py_datetime_to_utc; -use crate::timeseries::datapoints::{ - PyDatapointString, PyDatapointsCollectionDatapoints, PyDatapointsCollectionString, - PyRetrieveFilter, -}; -use crate::timeseries::{PyDeleteFilter, PyTimeSeries, PyTimeSeriesUpdate}; use crate::{PyFieldStr, PyFieldU64, PyListFieldIdCollection, PyMapField}; use intellistream_datahub_sdk::filters::{EventFilter, DataSort, EventFilterForm, TimeFilter}; use intellistream_datahub_sdk::events::{ EventDimension, EventIdCollection, EventUpdate, EventUpdateFields, }; use intellistream_datahub_sdk::generic::IdAndExtId; -use intellistream_datahub_sdk::{ApiService, Event, TimeSeries}; +use intellistream_datahub_sdk::{ApiService, Event}; use std::sync::Arc; use pyo3::exceptions::PyValueError; use pyo3::prelude::*; diff --git a/datahub_python_bindings/src/events/sync_service.rs b/datahub_python_bindings/src/events/sync_service.rs index b851051..8ed7fea 100644 --- a/datahub_python_bindings/src/events/sync_service.rs +++ b/datahub_python_bindings/src/events/sync_service.rs @@ -1,17 +1,12 @@ use crate::events::{ EventIdentifyable, PyEventFilter, PyEvent, PyEventDimension, PyEventUpdate, }; -use crate::{PyIdCollection}; use intellistream_datahub_sdk::events::{EventDimension, EventIdCollection, EventUpdate}; -use intellistream_datahub_sdk::filters::EventFilterForm; -use intellistream_datahub_sdk::generic::DataWrapper; use intellistream_datahub_sdk::{ - ApiService, Event, TimeSeries, TimeSeriesUpdate, TimeSeriesUpdateCollection, + ApiService, Event, }; -use pyo3::{Bound, PyAny, PyResult, Python, pyclass, pymethods}; -use pyo3_async_runtimes::tokio::future_into_py; +use pyo3::{PyResult, Python, pyclass, pymethods}; use std::sync::Arc; -use tokio::runtime; use uuid::Uuid; #[pyclass(module = "intellistream_datahub_sdk", name = "EventsServiceSync")] @@ -96,8 +91,7 @@ impl PyEventsServiceSync { input.iter().map(|u| EventIdCollection::from(u.clone())).collect(); py.detach(|| { - let result = self - .runtime + self.runtime .block_on(service.events.delete(&input_ids)) .map_err(|e| crate::datahub_err(e))?; diff --git a/datahub_python_bindings/src/files/mod.rs b/datahub_python_bindings/src/files/mod.rs index df7f9f4..50ef03e 100644 --- a/datahub_python_bindings/src/files/mod.rs +++ b/datahub_python_bindings/src/files/mod.rs @@ -2,7 +2,6 @@ pub mod async_service; pub mod sync_service; use crate::datetime::opt_py_datetime_to_utc; -use crate::resources::PyResource; use chrono::{DateTime, Utc}; use intellistream_datahub_sdk::files::{FileDownload, FileUpdate}; use intellistream_datahub_sdk::generic::{INode, IdAndExtId}; @@ -10,7 +9,6 @@ use intellistream_datahub_sdk::{ApiService, FileUpload}; use pyo3::prelude::*; use pyo3::types::{PyBytes, PyType}; use pyo3_async_runtimes::tokio::future_into_py; -use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::sync::Arc; diff --git a/datahub_python_bindings/src/files/sync_service.rs b/datahub_python_bindings/src/files/sync_service.rs index d4bdaf3..9085889 100644 --- a/datahub_python_bindings/src/files/sync_service.rs +++ b/datahub_python_bindings/src/files/sync_service.rs @@ -72,8 +72,7 @@ impl PyFilesServiceSync { let input_ids: Vec = input.into_iter().map(|u| IdAndExtId::from(u)).collect(); py.detach(|| { - let result = self - .runtime + self.runtime .block_on(service.files.delete(&DataWrapper::from_vec(input_ids))) .map_err(|e| crate::datahub_err(e))?; diff --git a/datahub_python_bindings/src/lib.rs b/datahub_python_bindings/src/lib.rs index 5732966..3a55d17 100644 --- a/datahub_python_bindings/src/lib.rs +++ b/datahub_python_bindings/src/lib.rs @@ -12,7 +12,6 @@ pub mod timeseries; pub mod units; mod functions; -use crate::datasets::PyDataset; use crate::datasets::async_service::PyDatasetsServiceAsync; use crate::datasets::sync_service::PyDatasetsServiceSync; use crate::files::async_service::PyFilesServiceAsync; @@ -33,7 +32,7 @@ use crate::subscriptions::sync_service::PySubscriptionsServiceSync; use crate::timeseries::async_service::PyTimeSeriesServiceAsync; use crate::timeseries::datapoints::PyRetrieveFilter; use crate::timeseries::sync_service::PyTimeSeriesServiceSync; -use crate::timeseries::{PyDeleteFilter, PyTimeSeries}; +use crate::timeseries::PyTimeSeries; use crate::units::PyUnit; use crate::assets::async_service::PyAssetsServiceAsync; use crate::assets::sync_service::PyAssetsServiceSync; @@ -52,11 +51,9 @@ use pyo3::create_exception; use pyo3::exceptions::{PyException, PyValueError}; use pyo3::prelude::*; use pyo3::types::PyType; -use pyo3_async_runtimes::tokio::future_into_py; use std::collections::HashMap; use std::sync::Arc; use std::sync::OnceLock; -use units::*; create_exception!( intellistream_datahub_sdk, @@ -153,7 +150,6 @@ fn build_buffered_config( token_url: Option, client_id: Option, client_secret: Option, - project_name: Option, enable_buffering: bool, buffer_retention_secs: Option, buffer_max_bytes: Option, @@ -174,7 +170,6 @@ fn build_buffered_config( token_url, client_id, client_secret, - project_name, ); if let Some(secs) = buffer_retention_secs { config.set_buffer_retention_secs(secs); @@ -229,7 +224,6 @@ impl PySyncClient { token_url=None, client_id=None, client_secret=None, - project_name=None, enable_buffering=false, buffer_retention_secs=None, buffer_max_bytes=None, @@ -250,7 +244,6 @@ impl PySyncClient { token_url: Option, client_id: Option, client_secret: Option, - project_name: Option, enable_buffering: bool, buffer_retention_secs: Option, buffer_max_bytes: Option, @@ -272,7 +265,6 @@ impl PySyncClient { token_url, client_id, client_secret, - project_name, enable_buffering, buffer_retention_secs, buffer_max_bytes, @@ -291,14 +283,14 @@ impl PySyncClient { } } #[classmethod] - fn from_env(py: Py) -> PyResult { + fn from_env(_py: Py) -> PyResult { Ok(Self { inner: ApiService::new(DataHubConfig::from_env().unwrap()), runtime: Arc::new(tokio::runtime::Runtime::new().unwrap()), }) } #[classmethod] - fn from_envfile(py: Py, path: Option<&str>) -> PyResult { + fn from_envfile(_py: Py, path: Option<&str>) -> PyResult { Ok(Self { inner: ApiService::new(DataHubConfig::from_envfile(path).unwrap()), runtime: Arc::new(tokio::runtime::Runtime::new().unwrap()), @@ -407,7 +399,6 @@ impl PyAsyncClient { token_url=None, client_id=None, client_secret=None, - project_name=None, enable_buffering=false, buffer_retention_secs=None, buffer_max_bytes=None, @@ -428,7 +419,6 @@ impl PyAsyncClient { token_url: Option, client_id: Option, client_secret: Option, - project_name: Option, enable_buffering: bool, buffer_retention_secs: Option, buffer_max_bytes: Option, @@ -450,7 +440,6 @@ impl PyAsyncClient { token_url, client_id, client_secret, - project_name, enable_buffering, buffer_retention_secs, buffer_max_bytes, @@ -468,13 +457,13 @@ impl PyAsyncClient { } } #[classmethod] - fn from_env(py: Py) -> PyResult { + fn from_env(_py: Py) -> PyResult { Ok(Self { inner: ApiService::new(DataHubConfig::from_env().unwrap()), }) } #[classmethod] - fn from_envfile(py: Py, path: Option<&str>) -> PyResult { + fn from_envfile(_py: Py, path: Option<&str>) -> PyResult { Ok(Self { inner: ApiService::new(DataHubConfig::from_envfile(path).unwrap()), }) diff --git a/datahub_python_bindings/src/resources/sync_service.rs b/datahub_python_bindings/src/resources/sync_service.rs index b4ad2e3..5be2f2d 100644 --- a/datahub_python_bindings/src/resources/sync_service.rs +++ b/datahub_python_bindings/src/resources/sync_service.rs @@ -3,7 +3,6 @@ use crate::relations::{PyGraphResult, PyRelForm}; use crate::resources::{PyResourceFilter, ResourceIdentifiable}; use crate::resources::{PyResourceNetwork, PyResourceUpdate}; use intellistream_datahub_sdk::resources::ResourceUpdate; -use crate::resources::async_service::PyResourcesServiceAsync; use crate::{DataSetRef, StringOrList, opt_data_set_refs, opt_patterns}; use intellistream_datahub_sdk::filters::NodeFilter; use intellistream_datahub_sdk::generic::IdAndExtId; diff --git a/datahub_python_bindings/src/timeseries/async_service.rs b/datahub_python_bindings/src/timeseries/async_service.rs index d5d0f6e..24999ea 100644 --- a/datahub_python_bindings/src/timeseries/async_service.rs +++ b/datahub_python_bindings/src/timeseries/async_service.rs @@ -1,18 +1,17 @@ use crate::timeseries::datapoints::{ - PyDatapoint, PyDatapointsCollectionDatapoints, PyDatapointsCollectionString, + PyDatapointsCollectionDatapoints, PyDatapointsCollectionString, }; use crate::timeseries::{ PyDeleteFilter, PyTimeSeries, PyTimeSeriesUpdate, PyTimeseriesIdentifiable, }; use crate::{ - DatahubIdentity, Identifiable, PyIdCollection, PyRetrieveFilter, + DatahubIdentity, Identifiable, PyRetrieveFilter, }; use crate::datetime::py_datetime_to_utc; use intellistream_datahub_sdk::generic::{ DataWrapper, DatapointString, DatapointsCollection, DeleteFilter, IdAndExtId, RetrieveFilter, }; use intellistream_datahub_sdk::{ApiService, TimeSeries, TimeSeriesUpdate, TimeSeriesUpdateCollection}; -use pyo3::prelude::*; use pyo3::{Bound, PyAny, PyResult, Python, pyclass, pymethods}; use pyo3_async_runtimes::tokio::future_into_py; use std::sync::Arc; @@ -236,7 +235,6 @@ impl PyTimeSeriesServiceAsync { .insert_datapoints(&mut wrapper) .await .map_err(|e| crate::datahub_err(e))?; - let val = result.get_items().clone(); Ok(result.get_items().clone()) }) } @@ -274,7 +272,6 @@ impl PyTimeSeriesServiceAsync { .insert_datapoints(&mut wrapper) .await .map_err(|e| crate::datahub_err(e))?; - let val = result.get_items().clone(); Ok(result.get_items().clone()) }) } @@ -309,7 +306,7 @@ impl PyTimeSeriesServiceAsync { let wrapper = DataWrapper::::from_vec(input.into_iter().map(|f| f.into()).collect()); future_into_py(py, async move { - let result = service + service .time_series .delete_datapoints(&wrapper) .await diff --git a/datahub_python_bindings/src/timeseries/datapoints.rs b/datahub_python_bindings/src/timeseries/datapoints.rs index 1b942cb..c9f834e 100644 --- a/datahub_python_bindings/src/timeseries/datapoints.rs +++ b/datahub_python_bindings/src/timeseries/datapoints.rs @@ -2,11 +2,11 @@ use crate::datetime::{opt_py_datetime_to_utc, py_datetime_to_utc}; use crate::timeseries::PyTimeseriesIdentifiable; use chrono::{DateTime, Utc}; use intellistream_datahub_sdk::generic::{ - Datapoint, DatapointString, DatapointsCollection, IdAndExtId, Identifiable, RetrieveFilter, + Datapoint, DatapointString, DatapointsCollection, IdAndExtId, RetrieveFilter, }; use pyo3::prelude::*; use pyo3::types::{PyDict, PyType}; -use pyo3::{Bound, Py, Python, pyclass, pymethods}; +use pyo3::{Bound, Python, pyclass, pymethods}; #[pyclass(module = "intellistream_datahub_sdk", name = "DatapointsCollectionString")] #[derive(Clone, Debug)] diff --git a/datahub_python_bindings/src/timeseries/mod.rs b/datahub_python_bindings/src/timeseries/mod.rs index d4de4f0..caf5049 100644 --- a/datahub_python_bindings/src/timeseries/mod.rs +++ b/datahub_python_bindings/src/timeseries/mod.rs @@ -1,33 +1,26 @@ use crate::datetime::opt_py_datetime_to_utc; -use crate::events::EventIdentifyable; -use crate::timeseries::async_service::PyTimeSeriesServiceAsync; use crate::timeseries::datapoints::{ PyDatapointString, PyDatapointsCollectionDatapoints, PyDatapointsCollectionString, PyRetrieveFilter, }; -use crate::timeseries::sync_service::PyTimeSeriesServiceSync; use crate::{ DatahubIdentity, Identifiable, PyFieldStr, PyFieldU64, PyIdCollection, PyMapField, }; use chrono::{DateTime, Utc}; -use intellistream_datahub_sdk::fields::{Field, ListField, MapField}; use intellistream_datahub_sdk::generic::{ - Datapoint, DatapointString, DatapointsCollection, DeleteFilter, IdAndExtId, + DatapointString, DatapointsCollection, DeleteFilter, IdAndExtId, RetrieveFilter, }; use crate::events::PyEvent; -use crate::resources::{PyResource, PyResourceNetwork}; +use crate::resources::PyResourceNetwork; use intellistream_datahub_sdk::filters::{EventFilter, EventFilterForm}; use intellistream_datahub_sdk::resources::RelatedResourcesForm; use intellistream_datahub_sdk::{ApiService, TimeSeries, TimeSeriesUpdate, TimeSeriesUpdateFields}; use pyo3::exceptions::PyValueError; use pyo3::prelude::*; use pyo3::pyclass; -use pyo3::types::{PyDict, PyList, PyTuple}; use pyo3_async_runtimes::tokio::future_into_py; -use serde::de::Unexpected::Map; -use std::collections::HashMap; use std::str::FromStr; use std::sync::Arc; use strum::{Display, EnumString}; diff --git a/datahub_python_bindings/src/timeseries/sync_service.rs b/datahub_python_bindings/src/timeseries/sync_service.rs index dad14d9..470f6a0 100644 --- a/datahub_python_bindings/src/timeseries/sync_service.rs +++ b/datahub_python_bindings/src/timeseries/sync_service.rs @@ -4,10 +4,9 @@ use crate::timeseries::datapoints::{ PyDatapointsCollectionDatapoints, PyDatapointsCollectionString, }; use crate::{DatahubIdentity, Identifiable}; -use crate::{PyIdCollection, PyRetrieveFilter}; +use crate::PyRetrieveFilter; use intellistream_datahub_sdk::generic::{DataWrapper, IdAndExtId}; use intellistream_datahub_sdk::{ApiService, TimeSeriesUpdateCollection}; -use pyo3_async_runtimes::tokio::future_into_py; use std::sync::Arc; #[pyclass(module = "intellistream_datahub_sdk", name = "TimeSeriesServiceSync")] @@ -96,8 +95,7 @@ impl PyTimeSeriesServiceSync { let wrapper = DataWrapper::from_vec(input_ids); py.detach(|| { - let result = self - .runtime + self.runtime .block_on(service.time_series.delete(&wrapper)) .map_err(|e| crate::datahub_err(e))?; @@ -331,7 +329,7 @@ impl PyTimeSeriesServiceSync { .block_on(service.time_series.delete_datapoints(&wrapper)) }); - let result = result.map_err(|e| crate::datahub_err(e))?; + result.map_err(|e| crate::datahub_err(e))?; Ok(()) } diff --git a/datahub_python_bindings/src/units/mod.rs b/datahub_python_bindings/src/units/mod.rs index d3e69cb..eb3d370 100644 --- a/datahub_python_bindings/src/units/mod.rs +++ b/datahub_python_bindings/src/units/mod.rs @@ -1,4 +1,4 @@ -use intellistream_datahub_sdk::{TimeSeries, Unit}; +use intellistream_datahub_sdk::Unit; use pyo3::{pyclass, pymethods}; use std::collections::HashMap; diff --git a/python_tests/README.md b/python_tests/README.md index 25f2ca7..20f250b 100644 --- a/python_tests/README.md +++ b/python_tests/README.md @@ -108,7 +108,7 @@ The fixtures build a client from an env file containing at least: - `BASE_URL` — backend root, e.g. `http://localhost:8081` - **either** `TOKEN` (a bearer token used as-is) **or** the OAuth2 client-credentials set - `CLIENT_ID` / `CLIENT_SECRET` / `TOKEN_URI` (optional `PROJECT_NAME`) + `CLIENT_ID` / `CLIENT_SECRET` / `TOKEN_URI` A gitignored `.env` already exists at the repo root. Make sure it points at a backend you can reach and that the credentials are valid. diff --git a/python_tests/test_auth_constructor.py b/python_tests/test_auth_constructor.py index 1070e82..c22f8af 100644 --- a/python_tests/test_auth_constructor.py +++ b/python_tests/test_auth_constructor.py @@ -57,7 +57,6 @@ def _client_credentials_kwargs(env): client_secret=client_secret, scope=env.get("SCOPE"), audience=env.get("AUDIENCE"), - project_name=env.get("PROJECT_NAME"), ) @@ -75,7 +74,6 @@ def _federated_kwargs(env): token_url=env["TOKEN_URI"], scope=env.get("SCOPE"), audience=env.get("AUDIENCE"), - project_name=env.get("PROJECT_NAME"), assertion=env.get("ASSERTION"), assertion_token_url=env.get("ASSERTION_TOKEN_URI"), assertion_client_id=env.get("ASSERTION_CLIENT_ID"), diff --git a/src/assets/mod.rs b/src/assets/mod.rs index 1d06b92..3c9fa54 100644 --- a/src/assets/mod.rs +++ b/src/assets/mod.rs @@ -1,3 +1,18 @@ +//! Assets — the typed `/assets` family, and the [`Asset`] shape its reads +//! answer with. +//! +//! [`AssetsService`] is reached as `api.assets`. Every call is the generic `/resources` pipeline +//! with the `ASSET` discriminator pinned server-side, so the two families cannot drift on ACLs or +//! status codes; what differs is that reads answer [`Asset`] rather than the +//! polymorphic [`Node`], putting `geolocation` and `is_root` in reach without a +//! match. Filtering and searching reuse the resource bodies, and a `node_type` set on either is +//! *replaced* with `["asset"]` rather than merged. +//! +//! Two things to expect: [`list`](AssetsService::list) does not page, so narrow with +//! [`filter`](AssetsService::filter) instead of raising `limit`; and +//! [`update`](AssetsService::update) echoes [`Node`]s rather than `Asset`s, +//! because an update may touch relations whose other end is a different node type. + #[cfg(test)] mod tests; diff --git a/src/auth_diagnostics.rs b/src/auth_diagnostics.rs deleted file mode 100644 index 4e7dbda..0000000 --- a/src/auth_diagnostics.rs +++ /dev/null @@ -1,221 +0,0 @@ -//! Explaining a 401 that the API declines to explain. -//! -//! DataHub resolves the caller's tenant from the access token's `organization` claim, and rejects -//! the token outright when that claim is missing, malformed, or names more than one organization. -//! The reason never reaches the caller: the API installs a custom authentication entry point that -//! answers with a bare `WWW-Authenticate: Bearer realm="Restricted Content"` and a generic body, -//! logging the real reason server-side only. What arrives here is -//! `ResponseError { status: 401, message: "" }` — indistinguishable from a rotated secret, which -//! is why the usual first response is to go and rotate the secret. -//! -//! The token itself carries the answer, so the SDK reconstructs it locally: decode the payload, -//! look at `organization`, and report which case the server would have hit. -//! -//! # This discloses nothing -//! -//! Only the caller's *own* token is inspected — one the SDK already holds in memory — and a JWT -//! payload is base64, not encrypted, so its holder can read it at any time (`cut -d. -f2 | -//! base64 -d`). Nothing crosses the network, no signature is verified, and no server-side state is -//! consulted. This is diagnosis, never an access decision: the API remains the only thing that -//! decides whether a token is acceptable, and a token this module considers well-formed can still -//! be rejected for reasons it cannot see (expiry, revocation, audience, signature). - -use base64::engine::general_purpose::URL_SAFE_NO_PAD; -use base64::Engine; - -/// What the `organization` claim of a token looks like, in the terms the API's -/// `OrganizationValidator` reasons about. -#[derive(Debug, Clone, PartialEq, Eq)] -pub(crate) enum OrganizationClaim { - /// Not a readable JWT — an opaque token supplied through `TOKEN`, say. Nothing to say about it. - NotAJwt, - /// A JWT, but with no `organization` claim: the token request asked for no organization scope, - /// or the realm emits the claim only under a selector. - Absent, - /// Present but empty (`{}`), which the API treats the same as absent. - Empty, - /// A JSON array rather than an object. Keycloak merges same-named multivalued claims, so this - /// means two mappers are writing `organization`, or the membership mapper has - /// `addOrganizationId` switched off and is emitting bare aliases. - Array, - /// Exactly one organization — the shape the API accepts. - One, - /// Several organizations. The API refuses to guess which tenant is meant. - Many(Vec), -} - -/// Inspect the `organization` claim of `token`. Best-effort and infallible: anything that does not -/// parse is [`OrganizationClaim::NotAJwt`], because the alternative — guessing — would blame the -/// wrong thing for an opaque token. -pub(crate) fn inspect_organization_claim(token: &str) -> OrganizationClaim { - // A JWS is `header.payload.signature`; we want the middle segment and never the signature. - let Some(payload) = token.split('.').nth(1).filter(|_| token.split('.').count() == 3) else { - return OrganizationClaim::NotAJwt; - }; - let Ok(bytes) = URL_SAFE_NO_PAD.decode(payload) else { - return OrganizationClaim::NotAJwt; - }; - let Ok(claims) = serde_json::from_slice::(&bytes) else { - return OrganizationClaim::NotAJwt; - }; - if !claims.is_object() { - return OrganizationClaim::NotAJwt; - } - - match claims.get("organization") { - None | Some(serde_json::Value::Null) => OrganizationClaim::Absent, - Some(serde_json::Value::Array(_)) => OrganizationClaim::Array, - Some(serde_json::Value::Object(orgs)) if orgs.is_empty() => OrganizationClaim::Empty, - Some(serde_json::Value::Object(orgs)) if orgs.len() == 1 => OrganizationClaim::One, - Some(serde_json::Value::Object(orgs)) => { - // Sorted: a JSON object's key order is not meaningful (and depends on whether - // serde_json is built with `preserve_order`), while an error message that reshuffles - // between runs is hard to grep for and hard to assert on. - let mut aliases: Vec = orgs.keys().cloned().collect(); - aliases.sort(); - OrganizationClaim::Many(aliases) - } - // Any other JSON type is malformed in a way the API also rejects, and "array" is the - // closest actionable advice we have. - Some(_) => OrganizationClaim::Array, - } -} - -/// A sentence explaining why this token's tenant could not be resolved, or `None` when the claim is -/// well-formed and the 401 must have another cause. -pub(crate) fn organization_hint(token: &str) -> Option { - let advice = match inspect_organization_claim(token) { - OrganizationClaim::NotAJwt | OrganizationClaim::One => return None, - OrganizationClaim::Absent => "the access token carries no `organization` claim, so the \ - server cannot tell which tenant you mean. DataHub resolves the tenant from that \ - claim; set the OAuth2 scope (`SCOPE=organization:*`, or `organization:` to \ - pin one) if your realm emits it under a selector." - .to_string(), - OrganizationClaim::Empty => "the access token's `organization` claim is empty, so the \ - server cannot tell which tenant you mean. Check that this principal is a member of \ - an organization, and that the token request asks for the organization scope." - .to_string(), - OrganizationClaim::Array => "the access token's `organization` claim is a JSON array, \ - which the server rejects. Two protocol mappers are writing that claim, or the \ - organization membership mapper has `addOrganizationId` switched off." - .to_string(), - OrganizationClaim::Many(aliases) => { - format!( - "the access token names {} organizations ({}), and the server will not choose one \ - for you. Pin a single tenant with `SCOPE=organization:`.", - aliases.len(), - aliases.join(", ") - ) - } - }; - Some(format!( - "{advice} (Diagnosed by the SDK from your own token; the server does not report this.)" - )) -} - -#[cfg(test)] -mod tests { - use super::*; - use serde_json::json; - - /// A token whose payload is `claims`. The header and signature are junk on purpose — nothing - /// here verifies them, and a test that supplied a real signature would imply otherwise. - fn jwt(claims: serde_json::Value) -> String { - format!( - "aGVhZGVy.{}.c2ln", - URL_SAFE_NO_PAD.encode(claims.to_string()) - ) - } - - #[test] - fn a_single_organization_is_well_formed_and_gets_no_hint() { - let token = jwt(json!({"organization": {"acme": {"id": "abc"}}})); - assert_eq!(inspect_organization_claim(&token), OrganizationClaim::One); - // The token is fine, so whatever caused the 401 is something this module can't see — - // inventing an explanation would send the reader down the wrong path. - assert_eq!(organization_hint(&token), None); - } - - #[test] - fn several_organizations_are_named_in_the_message() { - let token = jwt(json!({"organization": { - "beta": {"id": "2"}, - "acme": {"id": "1"}, - }})); - // Sorted, not in the order they appeared in the claim. - assert_eq!( - inspect_organization_claim(&token), - OrganizationClaim::Many(vec!["acme".into(), "beta".into()]) - ); - - let hint = organization_hint(&token).expect("a two-organization token should be explained"); - assert!(hint.contains("names 2 organizations"), "{hint}"); - // Sorted, so the message is stable across runs rather than following JSON key order. - assert!(hint.contains("(acme, beta)"), "{hint}"); - // The actionable part: what to actually change. - assert!(hint.contains("SCOPE=organization:"), "{hint}"); - } - - #[test] - fn a_missing_claim_points_at_the_scope() { - let token = jwt(json!({"sub": "service-account-x"})); - assert_eq!(inspect_organization_claim(&token), OrganizationClaim::Absent); - let hint = organization_hint(&token).unwrap(); - assert!(hint.contains("no `organization` claim"), "{hint}"); - assert!(hint.contains("SCOPE=organization:*"), "{hint}"); - } - - #[test] - fn a_null_claim_is_treated_as_absent() { - // Keycloak emits `"organization": null` for a principal with no resolvable membership. - let token = jwt(json!({"organization": null})); - assert_eq!(inspect_organization_claim(&token), OrganizationClaim::Absent); - assert!(organization_hint(&token).is_some()); - } - - #[test] - fn an_empty_claim_is_reported_separately_from_a_missing_one() { - let token = jwt(json!({"organization": {}})); - assert_eq!(inspect_organization_claim(&token), OrganizationClaim::Empty); - let hint = organization_hint(&token).unwrap(); - assert!(hint.contains("is empty"), "{hint}"); - } - - #[test] - fn an_array_claim_points_at_the_mapper_configuration() { - // What two mappers writing the same claim produces — the shape that cost this project a - // day: one mapper emitting bare aliases, another the nested object, merged by Keycloak. - let token = jwt(json!({"organization": ["acme", {"acme": {"id": "1"}}]})); - assert_eq!(inspect_organization_claim(&token), OrganizationClaim::Array); - let hint = organization_hint(&token).unwrap(); - assert!(hint.contains("JSON array"), "{hint}"); - assert!(hint.contains("addOrganizationId"), "{hint}"); - } - - #[test] - fn a_flat_alias_array_is_also_reported_as_an_array() { - // `addOrganizationId` off, single mapper: `["acme"]`. Same rejection, same advice. - let token = jwt(json!({"organization": ["acme"]})); - assert_eq!(inspect_organization_claim(&token), OrganizationClaim::Array); - } - - #[test] - fn an_opaque_token_is_never_speculated_about() { - // A user-supplied `TOKEN=` need not be a JWT. Reporting "you have no organization claim" - // for one would blame the wrong thing entirely. - for opaque in ["", "not-a-jwt", "two.parts", "a.b.c.d", "aGVhZGVy.!!!.c2ln"] { - assert_eq!( - inspect_organization_claim(opaque), - OrganizationClaim::NotAJwt, - "{opaque:?} should not be treated as a JWT" - ); - assert_eq!(organization_hint(opaque), None, "{opaque:?}"); - } - } - - #[test] - fn a_payload_that_is_not_a_json_object_is_not_a_jwt() { - let token = format!("aGVhZGVy.{}.c2ln", URL_SAFE_NO_PAD.encode("[1,2,3]")); - assert_eq!(inspect_organization_claim(&token), OrganizationClaim::NotAJwt); - } -} diff --git a/src/blocking.rs b/src/blocking.rs index aec476c..8d922d4 100644 --- a/src/blocking.rs +++ b/src/blocking.rs @@ -1,7 +1,14 @@ //! Blocking (synchronous) client, mirroring the async API — the same split as //! `reqwest` / `reqwest::blocking`. //! -//! Enable it with the `blocking` cargo feature. Every wrapper delegates to the async +//! **Requires the `blocking` cargo feature**, which is off by default: +//! +//! ```toml +//! [dependencies] +//! intellistream-datahub-sdk = { version = "0.4", features = ["blocking"] } +//! ``` +//! +//! Every wrapper delegates to the async //! implementation on a dedicated Tokio runtime owned by the client, so behavior //! (durable buffering, retries, OAuth token refresh) is identical to the async API — //! there is exactly one implementation of each call. diff --git a/src/buffer_integration.rs b/src/buffer_integration.rs index f6fd6b4..fe56bf5 100644 --- a/src/buffer_integration.rs +++ b/src/buffer_integration.rs @@ -55,7 +55,6 @@ fn unreachable_buffered_service(dir: &PathBuf) -> Arc { None, None, None, - None, ); config .set_buffer_dir(dir.clone()) diff --git a/src/datahub.rs b/src/datahub.rs index 9d2f94c..5091cae 100644 --- a/src/datahub.rs +++ b/src/datahub.rs @@ -1,19 +1,36 @@ +//! Configuration and authentication. [`DataHubConfig`] holds the API base URL and the credentials +//! every call is made with, and mints, caches and refreshes the bearer token behind them. +//! +//! Build one from the process environment and a local `.env` with [`DataHubConfig::from_env`], or +//! from a named file with [`from_envfile`](DataHubConfig::from_envfile), then pass it to +//! [`ApiService::new`](crate::ApiService::new). `BASE_URL` is always required; for credentials, +//! either set `TOKEN` to a token you manage yourself — that one is never treated as expired and +//! never refreshed — or give `CLIENT_ID`, `CLIENT_SECRET` and `TOKEN_URI` for the +//! client-credentials flow, or the `ASSERTION*` variables for the RFC 7523 `jwt-bearer` exchange. +//! The setters adjust the same values in code, and also switch on the durable ingest buffer +//! ([`enable_buffering`](DataHubConfig::enable_buffering)). +//! +//! `openid` is **always** requested and `SCOPE` *adds* to it rather than replacing it, because the +//! API resolves dataset grants by calling the identity provider's UserInfo endpoint with your own +//! token — on a Keycloak Organizations realm that means `SCOPE=organization:`. Everything +//! here fails with [`DataHubError`], never with +//! [`ResponseError`](crate::http::ResponseError). + use crate::errors::DataHubError; -use chrono::{DateTime, Duration, Utc}; +use chrono::{DateTime, Utc}; use dotenv::from_path; -use maplit::hashmap; use oauth2::basic::{BasicClient, BasicTokenResponse, BasicTokenType}; use oauth2::{ reqwest, AccessToken, ClientId, ClientSecret, EmptyExtraTokenFields, Scope, TokenResponse, TokenUrl, }; use regex::Regex; -use serde::{Deserialize, Serialize}; +use serde::Deserialize; use std::borrow::Cow; use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::sync::Arc; -use std::{default, env}; +use std::env; use tokio::sync::RwLock; /// Default durable-buffer time window applied when buffering is enabled without an explicit value. @@ -25,7 +42,8 @@ pub const DEFAULT_BUFFER_DIR: &str = ".datahub-spool"; /// Default number of datapoint insert requests in flight at once. Each carries up to 100 000 points, /// which the api holds in memory while it parses them, so this bounds what one insert costs its heap. pub const DEFAULT_DATAPOINT_INSERT_PARALLELISM: usize = 4; -/// Scope sent with a token request when none is configured. See [`OAuthConfig::effective_scope`]. +/// Scope sent with a token request when none is configured. `SCOPE` adds to this rather than +/// replacing it, so `SCOPE=organization:acme` asks for `openid organization:acme`. pub const DEFAULT_SCOPE: &str = "openid"; /// RFC 7523 grant type: exchange an externally-issued JWT assertion for a token. const JWT_BEARER_GRANT: &str = "urn:ietf:params:oauth:grant-type:jwt-bearer"; @@ -33,6 +51,10 @@ const JWT_BEARER_GRANT: &str = "urn:ietf:params:oauth:grant-type:jwt-bearer"; /// (Keycloak "Signed JWT - Federated") instead of a client secret. const JWT_BEARER_CLIENT_ASSERTION: &str = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"; +/// The credential half of [`DataHubConfig`]: which OAuth2 flow to use and what to send with it. +/// +/// Populated from the environment by [`DataHubConfig::from_env`]; the `ASSERTION*` fields select +/// the RFC 7523 `jwt-bearer` exchange in place of plain client credentials. #[derive(Default, Deserialize, Debug, Clone)] pub struct OAuthConfig { #[serde(alias = "CLIENT_ID")] @@ -95,8 +117,6 @@ pub struct OAuthConfig { #[serde(alias = "ASSERTION_GRANT")] pub(crate) assertion_grant: Option, - #[serde(alias = "PROJECT_NAME")] - pub(crate) project_name: Option, } impl OAuthConfig { @@ -133,10 +153,16 @@ impl OAuthConfig { } } #[derive(Default, Debug, Clone)] -struct AuthState { +pub(crate) struct AuthState { pub token: Option, pub expire_time: Option>, } +/// Base URL, credentials and the cached bearer token — everything an [`ApiService`](crate::ApiService) +/// needs to talk to one backend as one tenant. +/// +/// Build it with [`from_env`](Self::from_env) or [`from_envfile`](Self::from_envfile), adjust it +/// with the setters, then hand it to [`ApiService::new`](crate::ApiService::new). Cloning is cheap +/// and shares the token cache. #[derive(Debug, Clone)] pub struct DataHubConfig { pub(crate) config: Arc, @@ -202,7 +228,6 @@ impl DataHubConfig { token_uri: Option, client_id: Option, client_secret: Option, - project_name: Option, ) -> DataHubConfig { let oauthconfig = OAuthConfig { client_id, @@ -217,7 +242,6 @@ impl DataHubConfig { assertion_scope: None, assertion_audience: None, assertion_grant: None, - project_name, }; // Oauth client will only be configured if all required fields are present @@ -443,10 +467,10 @@ impl DataHubConfig { .unwrap_or_else(|| PathBuf::from(DEFAULT_BUFFER_DIR)) } + /// Uses the refresh token when present, otherwise makes a new client-credentials request. async fn refresh_token( &self, ) -> Result, DataHubError> { - /// will use refresh token if present otherwise it will make a new client credentials request let refresh_token = { let authstate = self.auth_state.read().await; authstate @@ -530,6 +554,10 @@ impl DataHubConfig { auth_state.expire_time = None; } + /// The current bearer token, minting or refreshing it if the cached one is missing or expired. + /// + /// A `TOKEN` supplied through the environment is returned as-is and never refreshed — it is + /// assumed to be managed outside the SDK. pub async fn get_api_token(&self) -> Result { { // lock scope. read and if expired refresh token @@ -852,7 +880,6 @@ mod jwt_bearer_tests { Some(token_uri), Some("datahub-jwt-grant".to_string()), Some("kc-secret".to_string()), - None, ) } @@ -916,7 +943,6 @@ mod jwt_bearer_tests { Some(url), Some("datahub-exchange".to_string()), None, - None, ); api.set_assertion("header.payload.signature"); let token = api.get_api_token().await.unwrap(); @@ -953,7 +979,6 @@ mod jwt_bearer_tests { Some(url), None, None, - None, ); api.set_assertion("header.payload.signature"); api.set_assertion_grant("jwt-bearer"); @@ -981,7 +1006,6 @@ mod jwt_bearer_tests { Some("http://127.0.0.1:1".to_string()), None, None, - None, ); api.set_assertion("header.payload.signature"); api.set_assertion_grant("password"); diff --git a/src/datasets/mod.rs b/src/datasets/mod.rs index d844314..6869ca8 100644 --- a/src/datasets/mod.rs +++ b/src/datasets/mod.rs @@ -1,3 +1,20 @@ +//! Data sets — the container the rest of the platform hangs off, and the unit access is granted on. +//! +//! [`DatasetsService`] is reached as `api.datasets` and covers create, list, by-ids, filter, +//! search, update and delete, all over [`Dataset`]. [`list`](DatasetsService::list) is a capped, +//! unpaged sample of the tenant; criteria, ordering and cursor paging live on +//! [`filter`](DatasetsService::filter). +//! +//! Because a data set is what read and write grants attach to, editing one is an operator action: +//! [`update`](DatasetsService::update) requires the all-datasets write grant and answers **403** +//! without it, even for a caller who may write the data set's *contents*. Two fields never come +//! back populated — `connected_data_sets` is input-only, and `data_set_id` is always `None`, since +//! a data set cannot belong to another one. +//! +//! [`policies`](DatasetsService::policies) is wired to the documented endpoint but has been +//! observed answering with an empty body even where policies exist, so treat its result as +//! unreliable rather than authoritative. + #[cfg(test)] mod tests; @@ -5,7 +22,7 @@ use crate::datahub::to_snake_lower_cased_allow_start_with_digits; use crate::fields::{Field, ListField, MapField}; use crate::filters::{MetadataFilter, NodeFilter, TimeFilter}; use crate::generic::{ApiServiceProvider, DataHubEntity, DataWrapper, IdAndExtId, SearchAndFilterForm}; -use crate::graph_data_wrapper::{GraphDataWrapper, GraphNode}; +use crate::graph_data_wrapper::GraphNode; use crate::http::ResponseError; use crate::resources::Resource; use crate::ApiService; @@ -13,8 +30,9 @@ use chrono::{DateTime, Utc}; use maplit::hashmap; use serde::{Deserialize, Serialize}; use std::collections::HashMap; -use std::sync::{Arc, Weak}; +use std::sync::Weak; +/// Data set CRUD, filtering and search. Reached as `api.datasets`; see the [module docs](self). pub struct DatasetsService { pub(crate) api_service: Weak, base_url: String, @@ -34,6 +52,9 @@ impl DatasetsService { } } + /// `POST /datasets/create` — create one or more data sets. + /// + /// Accepts a [`Dataset`], a `Vec` of them, or a reference to either. pub async fn create(&self, data: &I) -> Result, ResponseError> where for<'a> &'a I: Into>, @@ -44,6 +65,10 @@ impl DatasetsService { .await } + /// `POST /datasets/delete` — delete data sets by id or external id. + /// + /// A data set stands above everything that belongs to it, so this is refused while it still has + /// members. pub async fn delete(&self, json: &I) -> Result, ResponseError> where for<'a> &'a I: Into>, @@ -84,6 +109,9 @@ impl DatasetsService { self.execute_post_request(path, filter).await } + /// `POST /datasets/byids` — fetch data sets by id or external id. + /// + /// Like every batch lookup, this answers 200 with the subset it found and silently omits the rest. pub async fn by_ids(&self, id_collection: &I) -> Result, ResponseError> where for<'a> &'a I: Into>, @@ -166,6 +194,10 @@ impl DatasetsService { } } +/// A data set: the container entities belong to, and the unit read and write grants attach to. +/// +/// `connected_data_sets` is input-only and `data_set_id` is always `None` on a read — a data set +/// cannot belong to another one. #[derive(Debug, Serialize, Deserialize, Clone, PartialEq)] #[serde(rename_all = "camelCase")] pub struct Dataset { @@ -212,11 +244,7 @@ pub struct Dataset { pub created_time: Option>, pub last_updated_time: Option>, } -impl DataHubEntity for Dataset { - fn ext_id(&self) -> &String { - &self.external_id - } -} +impl DataHubEntity for Dataset {} impl GraphNode for Dataset {} impl Dataset { diff --git a/src/datasets/tests.rs b/src/datasets/tests.rs index 5cb7bfa..693d51e 100644 --- a/src/datasets/tests.rs +++ b/src/datasets/tests.rs @@ -4,7 +4,6 @@ use crate::fields::{Field, MapField}; use crate::generic::IdAndExtId; use crate::http::ResponseError; use crate::tests::cleanup::cleanup_datasets; -use maplit::hashmap; use crate::tests::ids::unique_id; fn create_test_dataset() -> Vec { @@ -52,7 +51,7 @@ async fn test_dataset_crud() -> Result<(), ResponseError> { true )); - let create_res = api_service.datasets.create(&test_dataset).await?; + api_service.datasets.create(&test_dataset).await?; let mut dataset_cleanup = cleanup_datasets( test_dataset .iter() diff --git a/src/errors.rs b/src/errors.rs index 27697a0..5a74b72 100644 --- a/src/errors.rs +++ b/src/errors.rs @@ -1,3 +1,11 @@ +//! [`DataHubError`], the error type for configuration and authentication — what can go wrong +//! *before* a request is made. +//! +//! [`DataHubConfig::from_env`](crate::datahub::DataHubConfig::from_env) and its siblings return it +//! for a missing or unparseable setting, and the token exchange returns it for an OAuth2 or URL +//! failure. Service methods never surface it: the SDK maps a failed token acquisition to a 401 +//! [`ResponseError`](crate::http::ResponseError) so every API call has one error type. + use thiserror::Error; #[derive(Error, Debug)] diff --git a/src/events/mod.rs b/src/events/mod.rs index c347e29..87a39cf 100644 --- a/src/events/mod.rs +++ b/src/events/mod.rs @@ -1,8 +1,37 @@ +//! Events — time-stamped records of something that happened, plus the vocabulary endpoints behind +//! their filter dropdowns. +//! +//! [`EventsService`] is reached as `api.events`. An [`Event`] is identified by a UUID rather than a +//! numeric id, and [`create`](EventsService::create) stamps a UUID v7 client-side on any event that +//! arrives without one — so a retry carries the same id and the server collapses the duplicate +//! instead of storing it twice. +//! +//! - **CRUD and queries** — [`create`](EventsService::create), [`get`](EventsService::get), +//! [`by_ids`](EventsService::by_ids), [`update`](EventsService::update), +//! [`delete`](EventsService::delete), plus [`filter`](EventsService::filter), +//! [`search`](EventsService::search) and [`count`](EventsService::count). +//! - **Vocabulary** — [`list_types`](EventsService::list_types) / +//! [`search_types`](EventsService::search_types) and the same pair for sub-types, statuses and +//! sources (see [`EventDimension`]). They answer "what values does this tenant actually use" for +//! the four categorical fields, reading small server-side dimension tables instead of scanning +//! events — cheap, but only *eventually consistent* with the events themselves. +//! +//! Two surprises worth knowing. [`list`](EventsService::list) returns the **oldest** events, not +//! the newest: it is `filter` with an empty body, whose default sort is `eventTime` ascending, so +//! ask "what just happened" through `filter` with a descending sort. And [`EventUpdate`] cannot +//! change `event_time` or `external_id` — both identify an event rather than describe it, and +//! sending either is a 400; record a correction by creating a new event and deleting the old one. +//! +//! With durable buffering enabled on the client, [`create`](EventsService::create) can answer +//! **202 with no items** because the batch was spooled to disk rather than sent. Check +//! [`get_http_status_code`](crate::generic::DataWrapper::get_http_status_code) or +//! [`buffered_count`](EventsService::buffered_count) when you need to tell the two apart. + #[cfg(test)] mod tests; use crate::buffer::DurableSpool; -use crate::datahub::{to_snake_lower_cased_allow_start_with_digits, DataHubConfig}; +use crate::datahub::DataHubConfig; use crate::fields::{Field, ListField, MapField}; use crate::filters::{EventFilter, EventFilterForm}; use crate::generic::{ @@ -17,6 +46,8 @@ use std::collections::HashMap; use std::sync::{Mutex, Weak}; use uuid::Uuid; +/// Event CRUD, filtering, search and the vocabulary endpoints. Reached as `api.events`; see the +/// [module docs](self). pub struct EventsService { pub(crate) api_service: Weak, base_url: String, @@ -34,6 +65,13 @@ impl EventsService { } } + /// `POST /events/create` — create one or more events. + /// + /// Any event without an `id` is given a time-ordered UUID v7 before sending, so a retry carries + /// the same id and the server collapses the duplicate rather than storing it twice. + /// + /// With durable buffering enabled, a batch that cannot reach the server is spooled to disk and + /// this answers **202 with no items** instead of failing. pub async fn create(&self, data: &I) -> Result, ResponseError> where for<'a> &'a I: Into>, @@ -160,6 +198,7 @@ impl EventsService { } } + /// `POST /events/delete` — delete events by UUID. pub async fn delete(&self, json: &I) -> Result, ResponseError> where for<'a> &'a I: Into>, @@ -191,11 +230,15 @@ impl EventsService { .await } + /// `POST /events/filter` — events matching [`EventFilterForm`], with sorting and keyset paging. + /// + /// Defaults to `eventTime` ascending. Echo the response's `next_cursor` back to walk the rest. pub async fn filter(&self, filter: &EventFilterForm) -> Result, ResponseError> { let path = &format!("{}/filter", self.base_url); self.execute_post_request(path, &filter).await } + /// `POST /events/byids` — fetch events by UUID, answering the subset it found. pub async fn by_ids(&self, id_collection: &I) -> Result, ResponseError> where for<'a> &'a I: Into>, @@ -435,6 +478,10 @@ fn buffered_wrapper() -> DataWrapper { // Not PartialEq: it carries `Vec`, which is intentionally non-comparable (see // `IdAndExtId`) — the same resource can be named as {id}, {externalId} or both. Same reasoning as // `EventFilter`. +/// Something that happened, at a time. +/// +/// Identified by a UUID rather than a numeric id. `external_id` is deliberately *not* unique: the +/// events sharing one are the lifecycle of a single logical event. #[derive(Debug, Serialize, Deserialize, Clone)] #[serde(rename_all = "camelCase")] pub struct Event { @@ -465,11 +512,7 @@ pub struct Event { pub source: Option, pub event_time: DateTime, } -impl DataHubEntity for Event { - fn ext_id(&self) -> &String { - &self.external_id - } -} +impl DataHubEntity for Event {} impl Event { /// The three fields the server will not accept an event without, and nothing else — everything diff --git a/src/events/tests.rs b/src/events/tests.rs index 61c8268..144fdff 100644 --- a/src/events/tests.rs +++ b/src/events/tests.rs @@ -9,24 +9,6 @@ use chrono::{DateTime, Duration, TimeZone, Utc}; use maplit::hashmap; use std::collections::HashMap; -async fn delete_events( - api_service: &ApiService, - events: Vec, -) -> Result<(), Box> { - let delete_result = api_service.events.delete(&events).await; - match delete_result { - Ok(events) => { - assert_eq!(events.length(), 0); - } - Err(e) => { - eprintln!("{:?}", e.get_message()); - assert_eq!(e.status.as_u16(), 200); - } - } - tokio::time::sleep(std::time::Duration::from_secs(5)).await; - Ok(()) -} - fn create_test_events(dataset_id: u64) -> Vec { let unique_id: u64 = 7110; let total_events = 89; diff --git a/src/fields.rs b/src/fields.rs index abeb379..e9f693f 100644 --- a/src/fields.rs +++ b/src/fields.rs @@ -1,6 +1,22 @@ +//! The update primitives the `POST //update` request bodies are built from: +//! [`Field`] for a scalar, [`ListField`] for a list, [`MapField`] for `metadata`. +//! +//! An update names its target and then describes only the fields it changes, so each of these has +//! to distinguish "leave alone" (absent) from "set to this" and from "clear". Set or clear a +//! scalar with [`Field::value`] and [`Field::null`]. Lists and maps go further: [`ListField::set`] +//! and [`MapField::set`] replace the whole collection, while `add` / `remove` apply a delta and +//! leave the rest in place — the two are mutually exclusive by construction, so a body carrying +//! both cannot be built. +//! +//! You meet these as the fields of each entity's update form, e.g. +//! [`ResourceUpdateFields`](crate::resources::ResourceUpdateFields). + use serde::{Deserialize, Serialize}; use std::collections::HashMap; +/// One scalar field of an update: absent (leave alone), set to a value, or explicitly cleared. +/// +/// Build with [`Field::value`] or [`Field::null`]; the default is absent. #[derive(Debug, Deserialize, Serialize, Clone, Default)] #[serde(rename_all = "camelCase")] pub struct Field { diff --git a/src/files/mod.rs b/src/files/mod.rs index 2cbf2cc..55c6e9b 100644 --- a/src/files/mod.rs +++ b/src/files/mod.rs @@ -1,3 +1,20 @@ +//! Files — upload, download, and the directory tree they live in. +//! +//! [`FileService`] is reached as `api.files`. Files and folders are both +//! [`INode`]s: walk the tree with +//! [`list_root_directory`](FileService::list_root_directory) and +//! [`list_directory_by_path`](FileService::list_directory_by_path), look one up with +//! [`get_by_id`](FileService::get_by_id), [`get_by_external_id`](FileService::get_by_external_id) +//! or [`search`](FileService::search), rename, move or re-assign a data set with +//! [`update`](FileService::update), and soft-delete, [`list_trash`](FileService::list_trash) and +//! [`restore`](FileService::restore). +//! +//! Upload is a raw `PUT`: the file content is the request body and all metadata rides in +//! `X-Datahub-*` headers, which [`FileUpload`] builds for you — but note that [`FileUpload::new`] +//! inspects the path eagerly and **panics** if it is not a readable regular file, so validate it +//! first. Download comes in two shapes: [`download`](FileService::download) buffers the whole file +//! into memory, while [`download_to_path`](FileService::download_to_path) streams it to disk. + mod test; use crate::datahub::to_snake_lower_cased_allow_start_with_digits; @@ -15,6 +32,7 @@ use tokio::fs::File; use tokio::io::AsyncWriteExt; use tokio_util::codec::{BytesCodec, FramedRead}; +/// File upload, download and the directory tree. Reached as `api.files`; see the [module docs](self). pub struct FileService { pub(crate) api_service: Weak, base_url: String, @@ -29,6 +47,10 @@ impl FileService { } } + /// `PUT /files` — upload a file's contents, with its metadata in `X-Datahub-*` headers. + /// + /// The body is the raw bytes rather than a multipart form. Build the [`FileUpload`] first; it + /// reads the path eagerly and panics if it is not a readable regular file. pub async fn upload_file( &self, file_upload: FileUpload, @@ -41,6 +63,7 @@ impl FileService { .await } + /// The top level of the file tree, as [`INode`]s — files and folders alike. pub async fn list_root_directory(&self) -> Result, ResponseError> { // Create and send an HTTP GET request let full_path = format!("{}/list", self.base_url.as_str()); @@ -48,6 +71,7 @@ impl FileService { .await } + /// The contents of one folder, named by its path. pub async fn list_directory_by_path( &self, path: &str, @@ -57,6 +81,10 @@ impl FileService { .await } + /// `POST /files/delete` — move files to the trash by id or external id. + /// + /// Soft: see [`list_trash`](Self::list_trash) for what is in there and + /// [`restore`](Self::restore) to bring one back. pub async fn delete( &self, id_collection: &DataWrapper, @@ -341,6 +369,10 @@ impl FileUpdate { } } +/// A file staged for upload: the local path plus the metadata that rides in `X-Datahub-*` headers. +/// +/// [`new`](Self::new) inspects the path immediately and **panics** if it is not a readable regular +/// file, so check it first if the path came from outside your program. #[derive(Debug, Serialize, Deserialize, Clone)] pub struct FileUpload { #[serde(rename = "externalId")] diff --git a/src/filters.rs b/src/filters.rs index 9c57963..4de2ab9 100644 --- a/src/filters.rs +++ b/src/filters.rs @@ -1,3 +1,25 @@ +//! Filter criteria, and the sorting and paging types the filter endpoints share. +//! +//! **Two types per entity, named the same way every time.** `XFilter` is the *criteria* — the +//! fields a row is matched on. `XFilterForm` is the *request body*, wrapping those criteria with +//! `limit`, `sort` and `cursor`. So [`ResourceFilter`](crate::resources::ResourceFilter) goes +//! inside [`ResourceFilterForm`](crate::resources::ResourceFilterForm), and likewise for +//! timeseries, datasets and subscriptions. The same `XFilter` is what +//! [`SearchAndFilterForm`](crate::generic::SearchAndFilterForm) narrows a search by, which is why +//! paging belongs to the form and not to the criteria. +//! +//! Each entity's pair lives in its own module; what is *here* is the shared machinery. +//! [`NodeFilter`] is the criteria every node type can be filtered by, flattened into all three +//! node filters, so on the wire its fields sit alongside the type-specific ones. [`TimeFilter`] is +//! every timestamp window (inclusive at **both** ends), [`MetadataFilter`] the metadata criterion, +//! and [`DataSort`] + [`PageRequest`] the ordering and keyset paging. Events are the one entity +//! whose pair lives here in full — [`EventFilter`] and [`EventFilterForm`] — because events are +//! not nodes and the filter deliberately does not extend [`NodeFilter`]. +//! +//! Across all of them: fields **AND** together, entries within a list **OR** (except `labels` and +//! `metadata`, where every entry must be present), and an empty list, a blank entry or `None` +//! places no restriction. + use crate::generic::IdAndExtId; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; @@ -315,7 +337,7 @@ impl DataSort { /// /// Flattened into each of them, so `sort` and `cursor` sit beside `filter` and `limit` on the wire. /// Events carry the same two fields but declare them directly on -/// [`EventFilterForm`](crate::filters::EventFilterForm), which also has `advancedFilter`. +/// [`EventFilterForm`], which also has `advancedFilter`. #[derive(Debug, Serialize, Deserialize, Clone, Default, PartialEq)] #[serde(rename_all = "camelCase")] pub struct PageRequest { @@ -358,6 +380,8 @@ impl PageRequest { } // Not PartialEq: holds `Option`, which is non-comparable (see `IdAndExtId`). +/// The request body of `POST /events/filter`: [`EventFilter`] criteria plus `limit`, `sort`, +/// `cursor` and the `advancedFilter` expression. #[derive(Debug, Serialize, Deserialize, Clone)] #[serde(rename_all = "camelCase")] pub struct EventFilterForm { @@ -455,7 +479,9 @@ impl EventFilterForm { self.limit = limit; self } - /// Set the boolean filter expression. See [`EventFilterForm::advanced_filter`]. + /// Set the boolean filter expression, in the api's PostgreSQL-flavoured filter language — + /// for example `type NOT LIKE 'pump' AND (subType = 'water' OR subType = 'gas')`. The api + /// parses and validates it, so an invalid expression comes back as a 400 carrying an offset. pub fn set_advanced_filter(&mut self, expression: impl Into) -> &mut Self { self.advanced_filter = Some(expression.into()); self diff --git a/src/functions/mod.rs b/src/functions/mod.rs index fb51a85..05dd017 100644 --- a/src/functions/mod.rs +++ b/src/functions/mod.rs @@ -1,3 +1,21 @@ +//! Function nodes — a plain graph node, distinguished only by its intrinsic `FUNCTION` type-label. +//! +//! [`FunctionsService`] is reached as `api.functions` and covers +//! [`create`](FunctionsService::create), [`list`](FunctionsService::list), +//! [`get_by_id`](FunctionsService::get_by_id), [`update`](FunctionsService::update) and +//! [`delete`](FunctionsService::delete). A [`Function`] adds no fields of its own to the shared +//! node base, and its `name` is `Option` here but required by the server, so a create that omits +//! it is a 400. +//! +//! Two caller-visible limits. `Function::related_resources` is **always empty** on everything this +//! service returns, the create echo included — read a function's edges through +//! [`ResourceService::fetch_related`](crate::resources::ResourceService::fetch_related) or +//! [`EdgesService`](crate::relations::EdgesService) instead. And +//! [`by_ids`](FunctionsService::by_ids) / [`by_external_id`](FunctionsService::by_external_id) are +//! implemented client-side over a single 10000-row listing, so a tenant holding more functions +//! than that silently misses its oldest; prefer [`get_by_id`](FunctionsService::get_by_id) +//! whenever you have the numeric id. + #[cfg(test)] mod test; @@ -36,6 +54,9 @@ impl FunctionsService { } } + /// `POST /functions/create` — create one or more function nodes. + /// + /// `name` is required by the server even though the field is `Option` here. pub async fn create(&self, data: &I) -> Result, ResponseError> where for<'a> &'a I: Into>, @@ -248,8 +269,4 @@ impl Function { } } -impl DataHubEntity for Function { - fn ext_id(&self) -> &String { - &self.external_id - } -} +impl DataHubEntity for Function {} diff --git a/src/generic.rs b/src/generic.rs index feddc72..4c3b9cd 100644 --- a/src/generic.rs +++ b/src/generic.rs @@ -1,3 +1,20 @@ +//! The response envelope every collection endpoint answers with, plus the id selectors and +//! datapoint types the services share. +//! +//! [`DataWrapper`] models the API's `{ "items": [...] }`: call +//! [`get_items`](DataWrapper::get_items) for the rows and +//! [`next_cursor`](DataWrapper::next_cursor) to continue a paged read. It doubles as a *request* +//! body — [`IdAndExtId`] names one entity by id or external id, and the `byids`/`delete` methods +//! accept anything that converts into a `DataWrapper`, so a bare [`IdAndExtId`], a +//! `Vec` of them, or a reference to either can be passed straight in. The entity types have the +//! same conversions, so `&event` or `&events` is accepted wherever a create body is wanted. +//! +//! Datapoints live here because timeseries reads and writes share them: +//! [`DatapointsCollection`] is one series plus its points, carrying [`DatapointString`] on the +//! way in and [`Datapoint`] on the way back. [`RetrieveFilter`] and [`DeleteFilter`] are the +//! per-series windows for reading and deleting those points, and [`SearchAndFilterForm`] is the +//! body every `/search` endpoint takes. + use crate::events::EventsService; use crate::files::FileService; use crate::http::{process_response, ResponseError}; @@ -20,6 +37,10 @@ use crate::subscriptions::SubscriptionsService; // object can be named three ways — {id, externalId}, {id, None}, {None, externalId}. A derived // (structural) equality would call those unequal, and a correct semantic equality is impossible // here without resolving against the backend. So the type simply isn't comparable. +/// Names one entity by numeric `id`, by `external_id`, or by both — the selector the `byids` and +/// `delete` endpoints take. +/// +/// The backend resolves whichever side is missing and returns both. #[derive(Debug, Deserialize, Serialize, Clone)] pub struct IdAndExtId { // todo Implement this as an enum, would allow for better validation @@ -56,6 +77,10 @@ impl From<&Vec> for DataWrapper { } } +/// One datapoint on the way **in**: timestamp and value both as strings. +/// +/// Strings because a datapoint's value type is per-series — a text series and a decimal series +/// share this shape — and because a decimal must not go through an `f64` on the way to the wire. #[derive(Debug, Deserialize, Serialize, Clone)] pub struct DatapointString { pub timestamp: String, @@ -78,6 +103,8 @@ impl DatapointString { } } +/// One datapoint on the way **out**: a parsed timestamp, the value, and the aggregate columns +/// when the read asked for aggregates. #[derive(Debug, Deserialize, Serialize, Clone)] pub struct Datapoint { // Read from "isoTime" when deserializing, but emit "timestamp" on serialization @@ -163,6 +190,9 @@ impl Datapoint { } } +/// One series, named by `id` or `external_id`, plus its datapoints. +/// +/// Carries [`DatapointString`] when writing and [`Datapoint`] when reading. #[derive(Debug, Deserialize, Serialize, Clone)] pub struct DatapointEpoch { pub(crate) timestamp: i64, @@ -410,6 +440,10 @@ impl DeleteFilter { } } +/// The per-series window of a datapoint read: time range, `limit`, and optionally `aggregates` +/// with a `granularity`. +/// +/// Its setters are crate-internal; build one with `..Default::default()` and set the public fields. #[derive(Debug, Deserialize, Serialize, Clone, Default)] #[serde(rename_all = "camelCase")] pub struct RetrieveFilter { @@ -458,27 +492,11 @@ impl RetrieveFilter { self } - pub(crate) fn add_aggregate(&mut self, aggregate: &str) -> &mut RetrieveFilter { - if self.aggregates.is_none() { - self.aggregates = Some(vec![]); - } - self.aggregates - .as_mut() - .unwrap() - .push(aggregate.to_string()); - self - } - pub(crate) fn set_granularity(&mut self, granularity: &str) -> &mut RetrieveFilter { self.granularity = Some(granularity.to_string()); self } - pub(crate) fn set_id(&mut self, id: u64) -> &mut RetrieveFilter { - self.id = Some(id); - self - } - pub(crate) fn set_external_id(&mut self, external_id: &str) -> &mut RetrieveFilter { self.external_id = Some(external_id.to_string()); self @@ -531,9 +549,9 @@ impl From> for DataWrapper { } } } -pub(crate) trait DataHubEntity: Clone + Serialize { - fn ext_id(&self) -> &String; -} +/// Marker trait: implementing it unlocks the generic `From` impls that wrap a `T`, `&T`, +/// `Vec` or `&Vec` into a `DataWrapper` for a request body. +pub(crate) trait DataHubEntity: Clone + Serialize {} impl From for DataWrapper { fn from(value: T) -> Self { DataWrapper { @@ -568,11 +586,21 @@ impl From<&T> for DataWrapper { } } +/// Implemented by entities that carry both a numeric `id` and an `external_id`, so a +/// [`DataWrapper`] of them can be searched by either. pub trait Identifiable { fn id(&self) -> u64; fn external_id(&self) -> &str; } +/// The `{ "items": [...] }` envelope every collection endpoint answers with. +/// +/// Read the rows with [`get_items`](Self::get_items), and continue a paged read by echoing +/// [`next_cursor`](Self::next_cursor) back as the next request's cursor. It doubles as a request +/// body, which is why `&entity` and `&vec_of_entities` can be passed straight to `create`. +/// +/// A non-2xx response is kept rather than thrown away: the body lands in `error_body` and the +/// status in [`get_http_status_code`](Self::get_http_status_code). #[derive(Debug, Deserialize, Serialize, Clone)] pub struct DataWrapper { items: Vec, @@ -693,6 +721,7 @@ impl DataWrapper { } } +#[doc(hidden)] pub trait ApiServiceProvider { fn api_service(&self) -> &Weak; @@ -707,11 +736,11 @@ pub trait ApiServiceProvider { /// for its whole lifetime. Clearing it here means the next call mints a fresh one, so a /// client that started a few seconds too early recovers on its next attempt instead of /// re-sending the same rejected credential until it expires. - async fn on_request_error(&self, error: ResponseError, token: &str) -> ResponseError { + async fn on_request_error(&self, error: ResponseError, _token: &str) -> ResponseError { if error.get_status() == http::StatusCode::UNAUTHORIZED { self.get_api_service().config.invalidate_token().await; } - explain_auth_failure(error, token) + error } async fn get_token(&self) -> Result { @@ -918,50 +947,14 @@ pub trait ApiServiceProvider { .get(reqwest::header::CONTENT_TYPE) .and_then(|value| value.to_str().ok()) .map(str::to_string); - Err(explain_auth_failure( - ResponseError { - status, - message: response - .text() - .await - .unwrap_or_else(|_| "Failed to read response body".to_string()), - content_type, - }, - &token, - )) - } -} - -/// Add a reason to a 401 that arrived without one. -/// -/// The API rejects a token whose `organization` claim is missing, malformed or ambiguous, but its -/// authentication entry point sends no `error_description` and an empty body — so the caller gets -/// `401` and nothing else, which reads as a bad credential. The token the SDK just sent carries -/// enough to say which it was; see [`crate::auth_diagnostics`] for why reading it discloses -/// nothing. -/// -/// Anything already explained is left alone: a non-401, or a 401 that did come with a body, keeps -/// its own message, and a well-formed claim adds nothing (the 401 then has a cause this cannot -/// see — expiry, revocation, audience, signature). -fn explain_auth_failure(error: ResponseError, token: &str) -> ResponseError { - if error.get_status() != http::StatusCode::UNAUTHORIZED { - return error; - } - let Some(hint) = crate::auth_diagnostics::organization_hint(token) else { - return error; - }; - let existing = error.get_message(); - let message = if existing.trim().is_empty() { - hint - } else { - format!("{existing} — {hint}") - }; - ResponseError { - status: error.get_status(), - message, - // Preserved: appending the organization hint does not change what the server sent, and - // dropping it here would make an explained 401 look like one that never reached the wire. - content_type: error.content_type().map(str::to_string), + Err(ResponseError { + status, + message: response + .text() + .await + .unwrap_or_else(|_| "Failed to read response body".to_string()), + content_type, + }) } } @@ -1006,6 +999,7 @@ impl ApiServiceProvider for crate::labels::LabelsService { } // A marker trait +#[doc(hidden)] pub trait DataWrapperDeserialization where Self: Sized, @@ -1026,13 +1020,8 @@ where wrapper.set_http_status_code(status_code); return Ok(wrapper); } - // For 2xx responses, we expect the body to be a valid DataWrapper - // If body is empty, it's fine for `from_str` to fail and return an error - // Or, if you specifically want an empty wrapper for 2xx with empty body: - // let mut wrapper = DataWrapper::new(); - // wrapper.set_http_status_code(status_code); - // return Ok(wrapper); - // However, typically a successful response with a body should be parsed. + // A 2xx with an empty body fails to parse here and surfaces as an error; the + // 204/empty cases are handled above, before this point. serde_json::from_str(body).map(|mut wrapper: DataWrapper| { wrapper.set_http_status_code(status_code); wrapper @@ -1073,6 +1062,7 @@ impl DataWrapperDeserialization for String { } } +/// A node in the file tree — a file or a folder — as the `/files` endpoints return it. #[derive(Debug, Deserialize, Serialize, Clone)] pub struct INode { #[serde(default, with = "crate::serde_helper::opt_string_id")] @@ -1169,79 +1159,6 @@ mod search_body_tests { } } -#[cfg(test)] -mod auth_failure_tests { - use super::explain_auth_failure; - use crate::http::ResponseError; - use base64::engine::general_purpose::URL_SAFE_NO_PAD; - use base64::Engine; - use oauth2::http::StatusCode; - - fn jwt(payload: &str) -> String { - format!("aGVhZGVy.{}.c2ln", URL_SAFE_NO_PAD.encode(payload)) - } - - fn error(code: u16, message: &str) -> ResponseError { - ResponseError { - status: StatusCode::from_u16(code).unwrap(), - message: message.to_string(), - content_type: None, - } - } - - #[test] - fn an_unexplained_401_gains_the_reason_the_server_withheld() { - // What a caller in two organizations actually gets back: 401, empty body, no - // `WWW-Authenticate` detail. Without this the only signal is "401", which reads as a bad - // secret and sends people off to rotate credentials. - let token = jwt(r#"{"organization":{"beta":{"id":"2"},"acme":{"id":"1"}}}"#); - let explained = explain_auth_failure(error(401, ""), &token); - - assert_eq!(explained.get_status(), StatusCode::UNAUTHORIZED, "status is untouched"); - let message = explained.get_message(); - assert!(message.contains("names 2 organizations"), "{message}"); - assert!(message.contains("acme, beta"), "{message}"); - assert!(message.contains("SCOPE=organization:"), "{message}"); - } - - #[test] - fn a_401_that_came_with_a_body_keeps_it() { - // Should the API ever start explaining itself, its words win and ours are appended — - // never replace a real server message with a guess. - let token = jwt(r#"{"organization":{"acme":{"id":"1"},"beta":{"id":"2"}}}"#); - let explained = explain_auth_failure(error(401, "Bearer token expired"), &token); - let message = explained.get_message(); - assert!(message.starts_with("Bearer token expired"), "{message}"); - assert!(message.contains("names 2 organizations"), "{message}"); - } - - #[test] - fn a_well_formed_token_is_left_alone() { - // Exactly one organization, so the 401 has some other cause (expiry, revocation, audience, - // signature). Volunteering an organization explanation here would misdirect the reader. - let token = jwt(r#"{"organization":{"acme":{"id":"1"}}}"#); - let explained = explain_auth_failure(error(401, ""), &token); - assert_eq!(explained.get_message(), ""); - } - - #[test] - fn an_opaque_token_is_left_alone() { - // A user-supplied `TOKEN=` need not be a JWT at all. - let explained = explain_auth_failure(error(401, ""), "an-opaque-api-key"); - assert_eq!(explained.get_message(), ""); - } - - #[test] - fn only_401s_are_touched() { - // A dataset-ACL 403 already carries a problem+json body explaining itself; appending - // organization advice to it would be noise, and wrong. - let token = jwt(r#"{"organization":{"acme":{"id":"1"},"beta":{"id":"2"}}}"#); - for code in [400u16, 403, 404, 500] { - let explained = explain_auth_failure(error(code, "original"), &token); - assert_eq!(explained.get_message(), "original", "{code} should pass through"); - } - } -} #[cfg(test)] mod delete_filter_tests { diff --git a/src/graph_data_wrapper.rs b/src/graph_data_wrapper.rs index ee5b283..455e29a 100644 --- a/src/graph_data_wrapper.rs +++ b/src/graph_data_wrapper.rs @@ -1,9 +1,31 @@ +//! [`GraphDataWrapper`], the response shape of the graph endpoints: nodes *and* the edges +//! between them, where [`DataWrapper`](crate::generic::DataWrapper) carries rows alone. +//! +//! [`ResourceService::create`](crate::resources::ResourceService::create), +//! [`by_ids`](crate::resources::ResourceService::by_ids) and +//! [`update`](crate::resources::ResourceService::update) answer one, as do +//! [`AssetsService::update`](crate::assets::AssetsService::update), +//! [`FunctionsService::update`](crate::functions::FunctionsService::update) and +//! [`EdgesService::by_ids`](crate::relations::EdgesService::by_ids). Read it with +//! [`nodes`](GraphDataWrapper::nodes) and [`relations`](GraphDataWrapper::relations) — both are +//! `Option`, absent rather than empty when the response carried none. +//! +//! `R` defaults to [`EdgeProxy`], the response form of an edge, so +//! `GraphDataWrapper` is the shape you receive; the explicit `GraphDataWrapper` +//! is the request form. The same struct is the update request body, which is why a bare +//! `&ResourceUpdate` or `&Vec` can be handed directly to `update` — [`GraphNode`] +//! is the marker that unlocks those conversions. + use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; use crate::generic::DataWrapperDeserialization; use crate::relations::EdgeProxy; +/// Marker for the node types a [`GraphDataWrapper`] can carry. +/// +/// Implementing it is what lets `&update` and `&vec_of_updates` convert into a request body, so +/// they can be handed straight to `resources.update` and its siblings. pub trait GraphNode: Clone + Serialize { fn into_wrapper(self) -> GraphDataWrapper { GraphDataWrapper::from(&self) diff --git a/src/http.rs b/src/http.rs index 0c1f625..ae83a63 100644 --- a/src/http.rs +++ b/src/http.rs @@ -1,3 +1,19 @@ +//! [`ResponseError`], the error every service method returns. +//! +//! It carries the HTTP status ([`get_status`](ResponseError::get_status)) and the raw response +//! body ([`get_message`](ResponseError::get_message)) and, when the API refused in the documented +//! way, the RFC 9457 problem document behind it: reach that with +//! [`problem`](ResponseError::problem) and branch on its +//! [`slug`](crate::problem::ProblemDetail::slug) rather than on the prose in `title` or `detail`. +//! Failures the SDK raised itself — a connect timeout, a client-side rejection, an unobtainable +//! token — arrive as the same type with no problem document. +//! +//! [`is_transient`](ResponseError::is_transient), +//! [`is_auth_failure`](ResponseError::is_auth_failure) and +//! [`is_bufferable`](ResponseError::is_bufferable) classify a failure for retry; the durable +//! ingest buffer spools on the last of them, so a 401/403 costs a rotated credential rather than +//! the batch. + use crate::generic::DataWrapperDeserialization; use oauth2::http::StatusCode; use reqwest::{Error, Response}; @@ -5,6 +21,11 @@ use serde::de::DeserializeOwned; use std::fmt; use thiserror::Error; +/// A failed API call: the status, the raw body, and the problem document behind it when there +/// was one. +/// +/// Also covers failures that never reached the server — a connect timeout, an unobtainable token — +/// which arrive with no problem document. #[derive(Debug, Error, Clone)] pub struct ResponseError { pub(crate) status: StatusCode, @@ -136,6 +157,7 @@ impl fmt::Display for ResponseError { } } +#[doc(hidden)] pub async fn process_response(response: Response, path: &str) -> Result where T: DeserializeOwned + DataWrapperDeserialization, diff --git a/src/labels/mod.rs b/src/labels/mod.rs index d1a4c22..9576b5f 100644 --- a/src/labels/mod.rs +++ b/src/labels/mod.rs @@ -1,3 +1,16 @@ +//! Labels — the tenant's catalogue of the categories resources and timeseries are filed under. +//! +//! Every resource carries at least one label (`PIPE`, `SENSOR`, `DOCUMENT`, …), and creating a +//! resource with a name that does not exist yet auto-creates the label — so [`LabelsService`] +//! (`api.labels`) is mostly for admin flows: pre-seeding names, colours and i18n codes, listing +//! what exists, and cleaning up. Names are tenant-scoped and canonicalised to `SNAKE_UPPER_CASE` +//! server-side, so `Heat Exchanger` and `heat_exchanger` are the same label and a duplicate is a +//! **409**. +//! +//! The entity is `labels::Label`, deliberately *not* re-exported at the crate root because +//! `resources::*` already brings a different graph-DTO `Label` there — spell it `labels::Label`. +//! [`delete`](LabelsService::delete) is refused while any resource still references the label. + #[cfg(test)] mod test; @@ -33,7 +46,8 @@ impl LabelsService { self.execute_get_request(&self.base_url, None::<&str>).await } - /// Look up a single label by its numeric `id`. Returns an empty `items` if it doesn't exist. + /// Look up a single label by its numeric `id`. A miss is a **404** carrying a `not-found` + /// problem, not an empty `items`. pub async fn get(&self, id: u64) -> Result, ResponseError> { let path = &format!("{}/{}", self.base_url, id); self.execute_get_request(path, None::<&str>).await diff --git a/src/lib.rs b/src/lib.rs index 23fd824..a9e2578 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,3 +1,98 @@ +//! An async Rust client for the **IntelliStream DataHub** REST API — time series and +//! datapoints, events, the resource graph, files, datasets and subscriptions. +//! +//! Everything starts at [`create_api_service()`], which reads its configuration from the +//! environment and returns an [`ApiService`]. Each API family is a field on it: +//! +//! ```no_run +//! use intellistream_datahub_sdk::create_api_service; +//! +//! # async fn run() { +//! let api = create_api_service(); +//! let hits = api.time_series.search_by_query("engine").await.unwrap(); +//! for ts in hits.get_items() { +//! println!("{}", ts.external_id); +//! } +//! # } +//! ``` +//! +//! # The services +//! +//! | Field | Module | Covers | +//! |---|---|---| +//! | `time_series` | [`timeseries`] | Time series CRUD, plus datapoint ingestion and retrieval | +//! | `events` | [`events`] | Event CRUD, filter and search, and the vocabulary endpoints | +//! | `resources` | [`resources`] | The generic node service — reads span every node type | +//! | `assets` | [`assets`] | The typed `/assets` family | +//! | `datasets` | [`datasets`] | Data sets, the unit access is granted on | +//! | `files` | [`files`] | Upload, download, directory listing, trash and restore | +//! | `functions` | [`functions`] | Function nodes | +//! | `edges` | [`relations`] | Relationship edges between resources, and the type catalogue | +//! | `labels` | [`labels`] | The tenant's label catalogue | +//! | `units` | [`mod@unit`] | The unit catalogue | +//! | `subscriptions` | [`subscriptions`] | Subscription CRUD and WebSocket listening | +//! +//! # Reading a response +//! +//! Collection endpoints answer `{ "items": [...] }`, which this crate models as +//! [`DataWrapper`](generic::DataWrapper) — call +//! [`get_items()`](generic::DataWrapper::get_items) for the rows. Graph endpoints answer +//! [`GraphDataWrapper`](graph_data_wrapper::GraphDataWrapper), which carries edges +//! alongside the nodes. +//! +//! `/resources` spans six node types and answers each row in the shape of its own kind, so +//! its reads return the [`Node`] enum rather than one flat struct. Match on it, or narrow +//! with [`as_time_series()`](nodes::Node::as_time_series), +//! [`into_asset()`](nodes::Node::into_asset) and their siblings. +//! +//! # Identifying things +//! +//! Most entities have both a server-assigned numeric `id` and a caller-supplied +//! `externalId`, and endpoints accept either. [`IdAndExtId`](generic::IdAndExtId) models +//! that choice for delete and by-id calls. +//! +//! # Errors +//! +//! Service methods return [`ResponseError`](http::ResponseError), which carries the status +//! and the raw body. When the API refused in the documented way, the body is an RFC 9457 +//! problem document: reach it with +//! [`ResponseError::problem()`](http::ResponseError::problem) and branch on +//! [`ProblemDetail::slug()`](problem::ProblemDetail::slug) — never on the prose in `title` +//! or `detail`, which may be reworded at any time. +//! +//! Configuration and authentication failures surface separately, as +//! [`DataHubError`](errors::DataHubError). +//! +//! # Filtering, searching and listing +//! +//! Each collection offers three reads, and they are not interchangeable: +//! +//! - **`list(limit)`** — the criteria-free first page. No paging, ever. +//! - **`filter(form)`** — structured criteria, with `sort` and a keyset `cursor`. This is +//! the one that pages: echo `next_cursor` back until it stops coming. +//! - **`search(form)`** — a free-text phrase, optionally narrowed by the same criteria. The +//! phrase selects and the filter only removes; it can never widen a search. +//! +//! The [`filters`] module explains the `XFilter` (criteria) versus `XFilterForm` (request +//! body) split that all three share. +//! +//! # Configuration +//! +//! [`create_api_service()`] loads a local `.env` file and the process environment. It needs +//! `BASE_URL`, plus either `TOKEN` or the OAuth2 client-credentials set `CLIENT_ID` / +//! `CLIENT_SECRET` / `TOKEN_URI`. See [`datahub::DataHubConfig`] for the full set, including +//! the `SCOPE` a Keycloak Organizations realm requires and the RFC 7523 assertion flow. +//! +//! # Cargo features +//! +//! - **`blocking`** — a synchronous mirror of the whole API in [`blocking`], the same split +//! as `reqwest` / `reqwest::blocking`. Every call delegates to the async implementation on +//! a runtime the client owns, so there is one implementation of each call. It must not be +//! constructed or used from inside an async context. +//! +//! If you only need to make a handful of calls from a synchronous `main`, [`block_on`] runs +//! a future on a self-contained runtime without taking a Tokio dependency of your own. + use dotenv::dotenv; use reqwest::Client; use reqwest::ClientBuilder; @@ -17,8 +112,6 @@ pub use crate::labels::LabelsService; pub use crate::relations::EdgesService; pub use crate::subscriptions::SubscriptionsService; -/// Explaining an unexplained 401 from the token the SDK already holds. -pub(crate) mod auth_diagnostics; #[cfg(feature = "blocking")] pub mod blocking; pub mod assets; @@ -45,6 +138,7 @@ pub mod nodes; pub mod problem; pub mod relations; pub mod resources; +#[doc(hidden)] // id-as-string serde adapters; see the module doc for why they stay `pub` pub mod serde_helper; pub mod subscriptions; #[cfg(test)] @@ -74,18 +168,39 @@ pub use subscriptions::{ use crate::functions::FunctionsService; //pub use filters::Filter; +/// The client: one configured connection to a DataHub backend, with every API family on it. +/// +/// Build one with [`create_api_service()`] and call through its fields — `api.events.create(…)`, +/// `api.time_series.insert_datapoints(…)`. It is held in an [`Arc`] and every service borrows +/// the same HTTP client and token cache, so clone the `Arc` freely rather than building a +/// second one; a second client means a second token cache. +/// +/// One `ApiService` is one tenant. Tenant identity rides in the access token's `organization` +/// claim and there is no per-call override, so talking to two tenants means two clients, each +/// with its own [`datahub::DataHubConfig`] and scope. pub struct ApiService { config: Box, + /// Time series and their datapoints — see [`timeseries`]. pub time_series: TimeSeriesService, + /// The unit catalogue — see [`mod@unit`]. pub units: UnitsService, + /// Events — see [`events`]. pub events: EventsService, + /// The generic node service, spanning every node type — see [`resources`]. pub resources: ResourceService, + /// The typed `/assets` family — see [`assets`]. pub assets: AssetsService, + /// Data sets — see [`datasets`]. pub datasets: DatasetsService, + /// Files — see [`files`]. pub files: FileService, + /// Subscriptions, including WebSocket listening — see [`subscriptions`]. pub subscriptions: SubscriptionsService, + /// Function nodes — see [`functions`]. pub functions: FunctionsService, + /// The label catalogue — see [`labels`]. pub labels: LabelsService, + /// Relationship edges and their type catalogue — see [`relations`]. pub edges: EdgesService, pub(crate) http_client: Client, } @@ -111,6 +226,28 @@ pub fn block_on(future: F) -> F::Output { .block_on(future) } +/// Build an [`ApiService`] from the environment. +/// +/// Loads a local `.env` file if there is one, then reads the process environment. It needs +/// `BASE_URL`, plus either `TOKEN` or the OAuth2 client-credentials trio `CLIENT_ID` / +/// `CLIENT_SECRET` / `TOKEN_URI`. [`datahub::DataHubConfig`] documents the rest, including the +/// `SCOPE` a Keycloak Organizations realm needs and the RFC 7523 assertion flow. +/// +/// ```no_run +/// use intellistream_datahub_sdk::create_api_service; +/// +/// # async fn run() { +/// let api = create_api_service(); +/// let units = api.units.list().await.unwrap(); +/// # } +/// ``` +/// +/// # Panics +/// +/// Configuration is read eagerly, so a missing `BASE_URL` or an unusable credential set panics +/// here rather than failing on the first call. Build the [`datahub::DataHubConfig`] +/// yourself with [`DataHubConfig::from_env`](datahub::DataHubConfig::from_env), which returns a +/// `Result`, and pass it to [`ApiService::new`]. pub fn create_api_service() -> Arc { dotenv().ok(); // Reads the .env file let dataplatform_api: DataHubConfig /* Type */ = DataHubConfig::create_default(); diff --git a/src/multi_tenant_integration.rs b/src/multi_tenant_integration.rs index 69baa2b..e5206a7 100644 --- a/src/multi_tenant_integration.rs +++ b/src/multi_tenant_integration.rs @@ -66,11 +66,9 @@ //! server's own words it reads. `WWW-Authenticate` still carries no `error_description`; the //! reason is in the body. //! -//! [`crate::auth_diagnostics`] reconstructs the same reason from the token the SDK just sent and -//! appends it, so the message a caller sees can carry both. Every 401 here shares the -//! `unauthorized` slug, so the cause is only in the prose: a caller cannot branch on `type` to -//! tell "no token" from "ambiguous organization". Cases that differ only by fixture are asserted -//! on **status alone**. +//! Every 401 here shares the `unauthorized` slug, so the cause is only in the prose: a caller +//! cannot branch on `type` to tell "no token" from "ambiguous organization". Cases that differ +//! only by fixture are asserted on **status alone**. //! //! (403s are different: those carry a real RFC 9457 `problem+json` body from the server, with //! `dataSetId` and `permission`, which the ACL tests assert on directly.) @@ -283,7 +281,6 @@ fn principal(test: &str, prefix: &str, scope: Option<&str>) -> Option Some(token_uri), Some(client_id.clone()), Some(client_secret), - None, ); if let Some(scope) = scope { config.set_scope(scope); @@ -463,10 +460,9 @@ async fn multi_tenant_multi_org_principal_with_wildcard_scope_is_rejected( "a token naming two organizations", ); - // The reason reaches the caller two ways now: the server writes it into the problem's - // `detail`, and `crate::auth_diagnostics` appends its own from the token just used. Asserting - // it against a real two-organization token from a real realm is what proves the explanation - // survives the whole path, entry point to `ResponseError`. + // The reason is the server's own: the organization validator writes it into the problem's + // `detail`. Asserting it against a real two-organization token from a real realm is what + // proves the explanation survives the whole path, entry point to `ResponseError`. let message = error.get_message(); assert!( message.contains("names 2 organizations"), @@ -484,7 +480,7 @@ async fn multi_tenant_multi_org_principal_with_wildcard_scope_is_rejected( } } assert!( - message.contains("SCOPE=organization:"), + message.contains("scope=organization:"), "the message should say what to change — got {message:?}" ); Ok(()) diff --git a/src/nodes.rs b/src/nodes.rs index 3281d9c..ad48841 100644 --- a/src/nodes.rs +++ b/src/nodes.rs @@ -254,11 +254,7 @@ impl Asset { impl GraphNode for Asset {} -impl crate::generic::DataHubEntity for Asset { - fn ext_id(&self) -> &String { - &self.external_id - } -} +impl crate::generic::DataHubEntity for Asset {} /// An access policy, as a node. /// diff --git a/src/problem_integration.rs b/src/problem_integration.rs index 35a5b81..c1fc888 100644 --- a/src/problem_integration.rs +++ b/src/problem_integration.rs @@ -550,9 +550,9 @@ mod unified { } /// An unauthenticated call answers 401 with a full problem document — `type`, `detail`, - /// `requestId` and `retry` — rather than the empty body it used to send. That empty body is - /// why [`crate::auth_diagnostics`] exists, reconstructing the reason from the token just - /// sent; the server now says it directly. + /// `requestId` and `retry` — rather than the empty body it used to send. The SDK carried an + /// `auth_diagnostics` module to reconstruct that reason from the token it had just sent; the + /// server saying it directly is what retired it. #[tokio::test] async fn an_unauthenticated_call_says_why() { let refusal = Probe::new().get_unauthenticated("/timeseries?limit=1").await; diff --git a/src/relations/mod.rs b/src/relations/mod.rs index 1b5aeb8..51404aa 100644 --- a/src/relations/mod.rs +++ b/src/relations/mod.rs @@ -1,3 +1,24 @@ +//! Relationships between resources, as first-class objects — the `/edges` endpoints and the shapes +//! they exchange. +//! +//! Edges normally come into being as a side effect of +//! [`ResourceService::create`](crate::resources::ResourceService::create), which makes the nodes +//! and their links in one call. [`EdgesService`] (`api.edges`) is for the rest: linking resources +//! that already exist, reading an edge back, deleting one without touching its endpoints, and +//! managing the relationship-type catalogue ([`types`](EdgesService::types) / +//! [`create_types`](EdgesService::create_types), over [`RelationshipType`] and [`RelTypeForm`]). +//! +//! The module also holds the edge shapes the rest of the SDK exchanges: [`RelForm`] describes an +//! edge you want created, [`EdgeProxy`] is a stored edge as the server returns it, and +//! [`RelatedNode`] with [`RelationDirection`] is the node-centric view — "this node relates to that +//! one, this way round" — that hangs off a node's `related_resources`. +//! +//! Deleting is the asymmetric part. [`EdgesService::delete`] refuses with **409** `would-strand` +//! rather than cut a node's only route to the graph root, so an edge is separately deletable only +//! when both endpoints stay reachable without it; otherwise it goes away with the resources it +//! connects. That check reads a graph projection which lags the write, so a delete issued +//! immediately after the edge was created can be wrongly *allowed*. + #[cfg(test)] mod tests; @@ -10,8 +31,8 @@ use serde::{Deserialize, Serialize}; use std::collections::HashMap; /// Server-assigned edge between two resources. Returned from `/resources/create`, -/// `/resources/update`, `/resources/fetch-related`, `GET /functions`, etc. — always -/// the response shape for graph operations. +/// `/resources/update` and `/resources/fetch-related` — the response shape for graph +/// operations. /// /// `relationship_type` is renamed from `type` because `type` is a Rust keyword. /// The wire field stays `"type"`. @@ -234,20 +255,6 @@ impl RelTypeForm { } } -impl crate::generic::DataHubEntity for RelTypeForm { - fn ext_id(&self) -> &String { - &self.name - } -} +impl crate::generic::DataHubEntity for RelTypeForm {} -impl crate::generic::DataHubEntity for RelForm { - fn ext_id(&self) -> &String { - // Edges are identified by their endpoints, not by an external id of their own. The - // `From for DataWrapper` impls only need *some* borrow, and the from-side external - // id is the closest thing an edge form has to a name. - static EMPTY: std::sync::OnceLock = std::sync::OnceLock::new(); - self.from_external_id - .as_ref() - .unwrap_or_else(|| EMPTY.get_or_init(String::new)) - } -} +impl crate::generic::DataHubEntity for RelForm {} diff --git a/src/relations/service.rs b/src/relations/service.rs index 591d205..6fb0859 100644 --- a/src/relations/service.rs +++ b/src/relations/service.rs @@ -13,6 +13,8 @@ use crate::nodes::Node; use crate::ApiService; use std::sync::Weak; +/// The `/edges` endpoints and the relationship-type catalogue. Reached as `api.edges`; see the +/// [`relations`](crate::relations) module docs. pub struct EdgesService { pub(crate) api_service: Weak, base_url: String, diff --git a/src/relations/tests.rs b/src/relations/tests.rs index 549a61f..0a8789d 100644 --- a/src/relations/tests.rs +++ b/src/relations/tests.rs @@ -623,9 +623,10 @@ mod live { Some(201), "creating relationships answers 201, not 200" ); - let edge_id = created.get_items()[0] - .id - .expect("the new edge should have an id"); + assert!( + created.get_items()[0].id.is_some(), + "the new edge should have an id" + ); // The (start, end, type) triple is unique, so the same link again is a conflict. let dup = api.edges.create(&vec![form]).await; diff --git a/src/resources/mod.rs b/src/resources/mod.rs index f9df949..005dba3 100644 --- a/src/resources/mod.rs +++ b/src/resources/mod.rs @@ -1,3 +1,21 @@ +//! [`ResourceService`], the generic node service over `/resources`, and the types its calls take +//! and answer. +//! +//! It is the one service that spans **every** node type — assets, timeseries, functions, data +//! sets, policies and plain resources — so its reads answer the [`Node`] enum +//! rather than one flat struct, each row in the shape of its own kind. Create nodes, and +//! optionally the edges between them, with [`create`](ResourceService::create); read with +//! [`get_by_id`](ResourceService::get_by_id), [`by_ids`](ResourceService::by_ids), +//! [`list`](ResourceService::list), [`filter`](ResourceService::filter) and +//! [`search`](ResourceService::search); change fields with [`update`](ResourceService::update), +//! whose body is a [`ResourceUpdate`] naming one node plus the [`ResourceUpdateFields`] to apply. +//! +//! [`fetch_related`](ResourceService::fetch_related) is the graph read: bounded by a +//! [`RelatedResourcesForm`], it walks outward from a starting node and answers a +//! [`ResourceNetwork`] of nodes, edges and labels. Note that the [`Label`] in this module is that +//! traversal's graph DTO, not the label entity — the entity is +//! [`labels::Label`](crate::labels::Label). + #[cfg(test)] mod tests; #[cfg(test)] @@ -10,7 +28,7 @@ use crate::generic::{ }; use crate::graph_data_wrapper::{GraphDataWrapper, GraphNode}; use crate::nodes::Node; -use crate::http::{process_response, ResponseError}; +use crate::http::ResponseError; use crate::relations::{EdgeProxy, RelForm, RelatedNode}; use crate::ApiService; use chrono::{DateTime, Utc}; @@ -18,6 +36,8 @@ use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::sync::Weak; +/// The generic node service, spanning every node type. Reached as `api.resources`; see the +/// [module docs](self). pub struct ResourceService { api_service: Weak, base_url: String, @@ -88,6 +108,10 @@ impl ResourceService { .await } + /// `POST /resources/delete` — delete nodes by id or external id. + /// + /// Refused with **409** `would-strand` if it would cut a surviving node's only route to the graph + /// root; the problem's `blockedBy` names what is in the way. pub async fn delete(&self, input: &I) -> Result, ResponseError> where for<'a> &'a I: Into>, @@ -98,6 +122,10 @@ impl ResourceService { self.execute_post_request::, _>(&url, &payload) .await } + /// `POST /resources/search` — full-text search across every node type. + /// + /// The phrase selects and the filter only narrows; results are ranked by `ts_rank` and tie-broken + /// on id. Answers each row as its own [`Node`] variant. pub async fn search( &self, payload: &SearchAndFilterForm, @@ -167,7 +195,7 @@ impl ResourceService { /// /// The cheap "what have I got" read: no body, no criteria. Like every read under /// `/resources` it spans **every** node type and answers each row in the shape of its own - /// kind — see [`Node`](crate::nodes::Node) — and it is narrowed to the data sets you may read + /// kind — see [`crate::nodes::Node`] — and it is narrowed to the data sets you may read /// exactly as [`filter`](Self::filter) is. /// /// Resources are the bulk of a tenant, so treat this as a sample rather than an inventory: it @@ -214,6 +242,10 @@ impl ResourceService { .await } } +/// A plain graph node — one carrying none of the intrinsic type-labels. +/// +/// The flat shape `/resources` answers with for an unlabelled node; a labelled one comes back as +/// the matching [`Node`] variant instead. #[derive(Debug, Serialize, Deserialize, Clone, PartialEq)] #[serde(rename_all = "camelCase")] pub struct Resource { @@ -275,8 +307,6 @@ impl Resource { } impl GraphNode for Resource {} impl Identifiable for Resource { - //todo!() - fn id(&self) -> u64 { self.id.unwrap_or(0) } diff --git a/src/serde_helper.rs b/src/serde_helper.rs index 94712de..a4242db 100644 --- a/src/serde_helper.rs +++ b/src/serde_helper.rs @@ -1,3 +1,14 @@ +//! Serde adapters for DataHub's id encoding: entity ids are 64-bit but travel as JSON **strings**, +//! because a value past 2^53 loses precision in a JavaScript client. +//! +//! Each submodule is a `serialize`/`deserialize` pair for one shape, used through +//! `#[serde(with = "...")]` rather than called directly. The SDK's own types already carry the +//! right one, so an id field is a numeric `u64`/`i64` in Rust and a string on the wire. +//! Deserialization also accepts a bare JSON number, so a payload from an older server still parses. +//! +//! Hidden from the rendered documentation because callers do not reach for these directly, but +//! left `pub` so a downstream type can reuse them in its own `#[serde(with = …)]`. + use serde::{Deserialize, Deserializer, Serializer}; pub fn is_zero(x: &u64) -> bool { @@ -67,7 +78,7 @@ pub mod string_id { /// `Option>` of ids as JSON strings (accepts strings or numbers on input). /// -/// The list form of [`opt_string_id`], for filter fields like `DataSetFilter.ids` that the backend +/// The list form of `opt_string_id`, for filter fields like `DataSetFilter.ids` that the backend /// declares as `List` but serializes with `ToStringSerializer`. pub mod opt_string_id_vec { use super::*; diff --git a/src/subscriptions/listen.rs b/src/subscriptions/listen.rs index 285e2e9..c3f9211 100644 --- a/src/subscriptions/listen.rs +++ b/src/subscriptions/listen.rs @@ -91,6 +91,8 @@ pub enum EventObject { ResourceAndRelation, } +/// One series' worth of datapoints as they arrive over the subscription WebSocket, with values +/// left as strings. #[derive(Debug, Serialize, Deserialize, Clone)] #[serde(rename_all = "camelCase")] pub struct DataCollectionString { diff --git a/src/subscriptions/mod.rs b/src/subscriptions/mod.rs index d261c3d..64289b6 100644 --- a/src/subscriptions/mod.rs +++ b/src/subscriptions/mod.rs @@ -1,3 +1,23 @@ +//! Subscriptions — a named set of timeseries to be notified about, and the WebSocket stream that +//! delivers the notifications. +//! +//! [`SubscriptionsService`] is reached as `api.subscriptions`. +//! +//! - **Managing subscriptions** — [`create`](SubscriptionsService::create), +//! [`list`](SubscriptionsService::list), [`filter`](SubscriptionsService::filter) and +//! [`delete`](SubscriptionsService::delete), over [`Subscription`]. There is no update or +//! single-get. `list` is a capped, unpaged sample; criteria live on `filter` and +//! [`SubscriptionFilterForm`], which carries only the `timeseries` criterion of the several the +//! endpoint accepts. +//! - **Listening** — [`listen`](SubscriptionsService::listen) opens a WebSocket and returns a +//! [`SubscriptionListener`] multiplexing the named subscriptions' streams. +//! +//! The listener has to be driven: call [`next`](SubscriptionListener::next) in a loop and +//! [`ack`](SubscriptionListener::ack) what you have processed. `next` is also what answers the +//! server's pings and transparently re-establishes a dropped connection, so a listener that is not +//! being polled is closed as idle after roughly 45 seconds — run heavy per-message work on another +//! task. Anything left unacked is redelivered to the next listener on the same subscription. + pub mod listen; mod test; @@ -14,6 +34,8 @@ use crate::ApiService; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; +/// Subscription management and WebSocket listening. Reached as `api.subscriptions`; see the +/// [module docs](self). pub struct SubscriptionsService { pub(crate) api_service: Weak, base_url: String, @@ -31,6 +53,7 @@ impl SubscriptionsService { } } + /// `POST /subscriptions/create` — create one or more subscriptions. pub async fn create(&self, data: &I) -> Result, ResponseError> where for<'a> &'a I: Into>, @@ -76,6 +99,10 @@ impl SubscriptionsService { .await } + /// `POST /subscriptions/delete` — delete subscriptions by id or external id. + /// + /// Close any [`SubscriptionListener`] on the subscription first; deleting one with a live + /// listener attached is refused. pub async fn delete(&self, json: &I) -> Result, ResponseError> where for<'a> &'a I: Into>, @@ -104,6 +131,8 @@ impl SubscriptionsService { } } +/// A named standing interest in a set of timeseries, which +/// [`listen`](SubscriptionsService::listen) then streams datapoints for. #[derive(Debug, Serialize, Deserialize, Clone)] #[serde(rename_all = "camelCase")] pub struct Subscription { @@ -132,12 +161,10 @@ impl Subscription { } } -impl DataHubEntity for Subscription { - fn ext_id(&self) -> &String { - &self.external_id - } -} +impl DataHubEntity for Subscription {} +/// Criteria for [`SubscriptionsService::filter`]. Currently only `timeseries`, a subset of what +/// the endpoint accepts. #[derive(Debug, Serialize, Deserialize, Clone, Default)] #[serde(rename_all = "camelCase")] pub struct SubscriptionFilter { @@ -145,6 +172,7 @@ pub struct SubscriptionFilter { pub timeseries: Vec, } +/// The request body of `POST /subscriptions/filter`: [`SubscriptionFilter`] criteria plus paging. #[derive(Debug, Serialize, Deserialize, Clone)] #[serde(rename_all = "camelCase")] pub struct SubscriptionFilterForm { diff --git a/src/subscriptions/test.rs b/src/subscriptions/test.rs index b1499aa..dd01d2c 100644 --- a/src/subscriptions/test.rs +++ b/src/subscriptions/test.rs @@ -538,7 +538,6 @@ mod tests { result } - #[allow(dead_code)] fn _require_send(_: &T) {} // Compile-time check: the returned listener is Send so it can be moved to other tasks. diff --git a/src/timeseries/mod.rs b/src/timeseries/mod.rs index ee9752d..bddcfc2 100644 --- a/src/timeseries/mod.rs +++ b/src/timeseries/mod.rs @@ -1,3 +1,30 @@ +//! Time series and their datapoints. +//! +//! [`TimeSeriesService`] is reached as `api.time_series` and does two distinct jobs: +//! +//! - **The series themselves** — [`create`](TimeSeriesService::create), +//! [`list`](TimeSeriesService::list), [`by_ids`](TimeSeriesService::by_ids), +//! [`filter`](TimeSeriesService::filter), [`search`](TimeSeriesService::search), +//! [`update`](TimeSeriesService::update) and [`delete`](TimeSeriesService::delete), over +//! [`TimeSeries`]. `list` is a capped, unpaged sample; criteria and paging live on `filter`, +//! whose `data_set_id` expands down the data-set hierarchy server-side. +//! - **Datapoints** — [`insert_datapoint`](TimeSeriesService::insert_datapoint) and +//! [`insert_datapoints`](TimeSeriesService::insert_datapoints) on the JSON ingest path, +//! [`insert_datapoints_binary`](TimeSeriesService::insert_datapoints_binary) on the Arrow/zstd +//! one (see [`binary`]), then +//! [`retrieve_datapoints`](TimeSeriesService::retrieve_datapoints), +//! [`retrieve_latest_datapoint`](TimeSeriesService::retrieve_latest_datapoint) and +//! [`delete_datapoints`](TimeSeriesService::delete_datapoints) to read and clear them. +//! +//! Ingest is retry-safe — datapoints dedup on `(series, timestamp)` server-side — and with durable +//! buffering enabled on the client, `insert_datapoints` spools to disk when the server is +//! unreachable and answers **202 with no items** instead of failing, flushing the backlog on a +//! later call. +//! +//! Mind which delete you want: [`delete`](TimeSeriesService::delete) removes the series *and* its +//! datapoints, while [`delete_datapoints`](TimeSeriesService::delete_datapoints) clears a window +//! and keeps the definition. + pub mod binary; mod test; @@ -12,8 +39,7 @@ use crate::generic::{ }; use crate::filters::NodeFilter; use crate::relations::RelatedNode; -use crate::http::{process_response, ResponseError}; -use crate::serde_helper::is_zero; +use crate::http::ResponseError; use crate::ApiService; use chrono::{DateTime, Utc}; use futures::stream::{self, StreamExt}; @@ -33,6 +59,7 @@ struct SpoolDatapoint { value: String, } +/// Time series and datapoint calls. Reached as `api.time_series`; see the [module docs](self). pub struct TimeSeriesService { pub(crate) api_service: Weak, base_url: String, @@ -71,6 +98,10 @@ impl TimeSeriesService { .await } + /// `POST /timeseries/create` — create one or more series. + /// + /// A duplicate `external_id` answers a clean 409. See [`create_one`](Self::create_one) and + /// [`create_from_list`](Self::create_from_list) for the common shapes. pub async fn create( &self, json: &DataWrapper, @@ -80,6 +111,7 @@ impl TimeSeriesService { .await } + /// [`create`](Self::create) for a single series. pub async fn create_one( &self, ts: &TimeSeries, @@ -89,6 +121,7 @@ impl TimeSeriesService { self.create(&dw).await } + /// [`create`](Self::create) for a slice of series. pub async fn create_from_list( &self, ts_list: &Vec, @@ -115,6 +148,10 @@ impl TimeSeriesService { self.execute_post_request(path, json).await } + /// `POST /timeseries/update` — partial update of one or more series. + /// + /// Each [`TimeSeriesUpdate`] names its target and carries only the fields it changes. `value_type` + /// is not among them: it is fixed at creation. pub async fn update( &self, json: &TimeSeriesUpdateCollection, @@ -124,6 +161,7 @@ impl TimeSeriesService { .await } + /// `POST /timeseries/byids` — fetch series by id or external id, answering the subset it found. pub async fn by_ids( &self, json: &DataWrapper, @@ -170,6 +208,8 @@ impl TimeSeriesService { self.search(&SearchAndFilterForm::new(query)).await } + /// Insert a single datapoint into one series — [`insert_datapoints`](Self::insert_datapoints) + /// for a batch, which is what you want for anything but a one-off. pub async fn insert_datapoint( &self, id: Option, @@ -424,6 +464,10 @@ impl TimeSeriesService { .await } + /// `POST /timeseries/data/list` — read datapoints for one or more series. + /// + /// Each [`RetrieveFilter`] names a series and a window, and may ask for `aggregates` at a + /// `granularity` instead of raw points. The window is half-open: `[start, end)`. pub async fn retrieve_datapoints( &self, json: &DataWrapper, @@ -450,6 +494,7 @@ impl TimeSeriesService { .await } + /// The most recent datapoint of each named series. pub async fn retrieve_latest_datapoint( &self, json: &DataWrapper, @@ -582,6 +627,9 @@ impl TimeSeriesFilterForm { } } +/// A time series: the definition datapoints hang off, not the points themselves. +/// +/// `value_type` fixes what its datapoints may hold and cannot be changed after creation. #[derive(Debug, Serialize, Deserialize, Clone, PartialEq)] pub struct TimeSeries { #[serde(default, with = "crate::serde_helper::opt_string_id")] @@ -599,10 +647,10 @@ pub struct TimeSeries { pub data_set_id: Option, /// The series' value type (`float`, `bigint`, `text`, …). /// - /// `None` means the endpoint did not say, not that the series has no type. A flat read always - /// carries it; a node reached through the graph (`fetch_related`/`fetch_nearest`) does not, - /// because Neo4j stores only a subset of the columns — re-read the series by id when the - /// value type matters. + /// `None` means the endpoint did not say, not that the series has no type. Both the flat and + /// the graph reads (`fetch_related`/`fetch_nearest`) carry it now; a series last written + /// before the graph projection included the field still reports it absent, so treat `None` as + /// "not stated" rather than as a default. #[serde( rename = "valueType", default, @@ -765,6 +813,7 @@ impl TimeSeriesUpdateFields { } } +/// A partial update of one series: how to find it, plus only the fields being changed. #[derive(Debug, Serialize, Deserialize, Clone)] pub struct TimeSeriesUpdate { #[serde(default, with = "crate::serde_helper::opt_string_id")] @@ -774,6 +823,7 @@ pub struct TimeSeriesUpdate { pub update: TimeSeriesUpdateFields, } +/// A batch of [`TimeSeriesUpdate`]s for one `POST /timeseries/update` call. #[derive(Debug, Deserialize, Serialize, Clone)] pub struct TimeSeriesUpdateCollection { items: Vec, @@ -883,7 +933,6 @@ mod unbuffered_insert_tests { None, None, None, - None, ); if let Some(parallelism) = parallelism { config.set_datapoint_insert_parallelism(parallelism); diff --git a/src/timeseries/test.rs b/src/timeseries/test.rs index d895c51..574ca19 100644 --- a/src/timeseries/test.rs +++ b/src/timeseries/test.rs @@ -3,7 +3,6 @@ mod tests { use std::fs::File; use std::io::Read; - use std::rc::Rc; use std::sync::Arc; use chrono::{DateTime, Duration, TimeZone, Utc}; use maplit::hashmap; @@ -463,20 +462,7 @@ mod tests { }) .set_value_type("float").clone(); - let new_ts_name = format!("Rust SDK Test {id} TimeSeries", id = unique_id + 1); - let ts2 = TimeSeries::builder() - .set_external_id(new_ts_ext_id2.as_str()) - .set_name(new_ts_name.as_str()) - .set_description("This is test timeseries generated by rust sdk test code.") - .set_unit("bar") - .set_metadata(hashmap! { - "foodda".to_string() => "bardda".to_string(), - "bardda".to_string() => "bazdda".to_string() - }) - .set_value_type("float").clone(); - ts_collection.add_item(ts1); - // ts_collection.add_item(ts2); let result = api_service.time_series.create(&ts_collection).await; @@ -518,17 +504,6 @@ mod tests { } } - //println!("Prepare datapoints for second time series..."); - //let mut data_request: DataWrapper> = DataWrapper::new(); - //let mut dp_collection = DatapointsCollection::from_external_id(new_ts_ext_id.as_str()); - - //let datetime = Utc.with_ymd_and_hms(2025, 2, 4, 9, 0, 0).unwrap(); - //dp_collection.datapoints = create_daily_datapoints(datetime); - //for dp in &mut dp_collection.datapoints { - // dp.value = dp.value.clone(); - //} - - //data_request.get_items_mut().push(dp_collection); println!("Start datapoint insert for second time series!"); //let result = api_service.time_series.insert_datapoints(&mut data_request).await; @@ -774,7 +749,6 @@ mod tests { .expect("could not create the benchmark series"); let mut cleanup = cleanup_timeseries(external_ids.clone()); - let per_series = chunk / series_count; let mut latencies: Vec = Vec::new(); let mut sent = 0usize; let mut offset = 0i64; diff --git a/src/unit/mod.rs b/src/unit/mod.rs index 35cca88..0775a12 100644 --- a/src/unit/mod.rs +++ b/src/unit/mod.rs @@ -1,3 +1,14 @@ +//! The unit catalogue — the engineering units a +//! [`TimeSeries`](crate::timeseries::TimeSeries) can be expressed in. +//! +//! [`UnitsService`] is reached as `api.units` and is read-only: the SDK exposes no create, update +//! or delete here, because the catalogue is maintained server-side. Fetch all of it with +//! [`list`](UnitsService::list) — it is a small, slow-changing set — or resolve individual entries +//! with [`by_external_id`](UnitsService::by_external_id) and [`by_ids`](UnitsService::by_ids). +//! +//! A [`Unit`] carries its display `symbol`, the physical `quantity` it measures, its `alias_names` +//! and a `conversion` table. Attach one to a series through `TimeSeries::unit_external_id`. + mod test; use crate::generic::{ApiServiceProvider, DataWrapper, IdAndExtId}; @@ -8,6 +19,7 @@ use std::clone::Clone; use std::collections::HashMap; use std::sync::Weak; +/// Read-only access to the unit catalogue. Reached as `api.units`; see the [module docs](self). pub struct UnitsService { pub(crate) api_service: Weak, base_url: String, @@ -22,15 +34,18 @@ impl UnitsService { } } + /// Every unit in the catalogue. It is a small, slow-changing set, so fetching it whole is cheap. pub async fn list(&self) -> Result, ResponseError> { self.execute_get_request(&self.base_url, None::<&str>).await } + /// One unit by its external id, e.g. `temperature_deg_c`. pub async fn by_external_id(&self, value: &str) -> Result, ResponseError> { let path = &format!("{}/{value}", self.base_url, value = value); self.execute_get_request(path, None::<&str>).await } + /// Units by id or external id, answering the subset it found. pub async fn by_ids( &self, json: &DataWrapper, @@ -40,6 +55,8 @@ impl UnitsService { } } +/// An engineering unit from the platform's catalogue, referenced by a series through +/// `unit_external_id`. #[derive(Debug, Serialize, Deserialize, Clone)] #[serde(rename_all = "camelCase")] pub struct Unit {