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/Cargo.toml b/Cargo.toml index 8bc63be..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"] } 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/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/datahub.rs b/src/datahub.rs index eadfaec..5091cae 100644 --- a/src/datahub.rs +++ b/src/datahub.rs @@ -1,3 +1,21 @@ +//! 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, Utc}; use dotenv::from_path; @@ -24,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"; @@ -32,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")] @@ -134,6 +157,12 @@ 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, @@ -525,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 diff --git a/src/datasets/mod.rs b/src/datasets/mod.rs index 84a3418..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; @@ -15,6 +32,7 @@ use serde::{Deserialize, Serialize}; use std::collections::HashMap; 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 { 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 da8f11a..87a39cf 100644 --- a/src/events/mod.rs +++ b/src/events/mod.rs @@ -1,3 +1,32 @@ +//! 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; @@ -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 { 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 cc8c93c..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>, diff --git a/src/generic.rs b/src/generic.rs index a48259c..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 { @@ -552,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, @@ -677,6 +721,7 @@ impl DataWrapper { } } +#[doc(hidden)] pub trait ApiServiceProvider { fn api_service(&self) -> &Weak; @@ -954,6 +999,7 @@ impl ApiServiceProvider for crate::labels::LabelsService { } // A marker trait +#[doc(hidden)] pub trait DataWrapperDeserialization where Self: Sized, @@ -1016,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")] 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 eedb1a6..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; @@ -43,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)] @@ -72,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, } @@ -109,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/relations/mod.rs b/src/relations/mod.rs index 7e068ec..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"`. 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/resources/mod.rs b/src/resources/mod.rs index 00fb248..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)] @@ -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 { 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 90c3ec6..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 { @@ -134,6 +163,8 @@ impl Subscription { 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 { @@ -141,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/timeseries/mod.rs b/src/timeseries/mod.rs index 0c276e1..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; @@ -32,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, @@ -70,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, @@ -79,6 +111,7 @@ impl TimeSeriesService { .await } + /// [`create`](Self::create) for a single series. pub async fn create_one( &self, ts: &TimeSeries, @@ -88,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, @@ -114,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, @@ -123,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, @@ -169,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, @@ -423,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, @@ -449,6 +494,7 @@ impl TimeSeriesService { .await } + /// The most recent datapoint of each named series. pub async fn retrieve_latest_datapoint( &self, json: &DataWrapper, @@ -581,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")] @@ -598,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, @@ -764,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")] @@ -773,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, 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 {