diff --git a/AGENTS.md b/AGENTS.md index 6c30c82..a1f5c8e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -42,7 +42,7 @@ This crate is a thin async HTTP SDK around a DataHub-style REST API. Entry point - `time_series` (`src/timeseries/`) — `TimeSeries` + datapoint ingestion/retrieval - `units` (`src/unit/`) - `events` (`src/events/`) — event CRUD, filter/search, plus the vocabulary endpoints (`list_types`/`search_types` and the same pair for sub-types, statuses and sources, over `EventDimension`). Those answer "what values does this tenant actually use" for the four categorical fields and back filter dropdowns; they read small server-side dimension tables rather than scanning events, so they are cheap but *eventually consistent* with the events. Note the route asymmetry the SDK hides: `/events/list/{plural}` but `/events/search/{singular}`. `EventUpdate` has **no `event_time`**: an event's time is immutable after creation — the events table is partitioned by it, so ClickHouse refuses the mutation outright, and the api used to validate the field, echo the new value back with a 200 and then fail to apply it. It has been dropped from the update form, so sending it is now a 400. Record a corrected time as a new event. -- `resources` (`src/resources/`) — hierarchical asset-like entities; relationship edges live in `src/relations/` (`EdgeProxy`, `RelForm`, `RelatedNode`) +- `resources` (`src/resources/`) — the generic node service. Its reads span **every** node type and answer with [`Node`](#the-polymorphic-node-type) rather than one flat shape; relationship edges live in `src/relations/` (`EdgeProxy`, `RelForm`, `RelatedNode`) - `edges` (`src/relations/service.rs`) — the `/edges` endpoints: `get`/`by_ids`/`create`/`delete` plus the relationship-type catalogue (`types`/`create_types`). Edges normally come into being through `resources.create(nodes, relations)`; this service is for linking resources that already exist and for reading or deleting an edge on its own. `get` answers an unknown id with 404 and a `problem+json` body; `by_ids`, like every batch lookup, answers 200 with the found subset and silently omits what is missing. (`get` used to be 200-and-nothing despite documenting a 404 — api #275 made single-resource by-id GETs consistently 404 and deliberately left batch lookups alone.) Two further behaviours are worth knowing, and are documented at each call site: - `create_types` fails silently on a duplicate name — the unique-hash collision surfaces at commit, after the handler returned, so the caller gets a 200 with an empty *body*, and in a batch the valid new types are rolled back with it. This one does contradict the OpenAPI: `test_duplicate_relationship_type_conflicts` encodes the intended 409 and is red until the server-side fix lands. @@ -53,6 +53,55 @@ This crate is a thin async HTTP SDK around a DataHub-style REST API. Entry point - `functions` (`src/functions/`) - `labels` (`src/labels/`) — label CRUD (`list`/`get`/`create`/`update`/`delete`). Note the entity type is `labels::Label`, deliberately *not* re-exported at the crate root because `resources::*` already brings a different graph-DTO `Label` there. +### The polymorphic node type (`src/nodes.rs`) + +`/resources` spans six node types, and its reads answer with each row in the shape of its own +kind. `Node` is that: an enum over `Asset`, `TimeSeries`, `Function`, `Resource`, `Dataset` and +`Policy`. `filter`/`search`/`get_by_id` return `DataWrapper`, `by_ids`/`create` and +`EdgesService::by_ids` return `GraphDataWrapper`, and `ResourceNetwork::nodes` is `Vec`. + +**The discriminator is a label, not a field.** There is no `nodeType` key on the wire. A node's +type is the intrinsic type-label the api forces into `labels` on every read — `ASSET`, +`TIMESERIES`, `FUNCTION`, `DATASET`, `POLICY` — and a plain resource carries **none of them**, so +absence is the `RESOURCE` signal. Serde has no mode for a tag inside an array field, so `Node` +hand-writes `Deserialize`: buffer into `serde_json::Value`, canonicalize each label the way the +api's `TextValidator.toSnakeUpperCased` does, dispatch. More than one type-label is an **error**, +mirroring the api's `NodeModelDeserializer`; zero is a `Resource`. Serializing goes the other way, +emitting the variant's own shape and appending its type-label if the caller has not — it never +strips a conflicting one, because a body labelled both `ASSET` and `POLICY` earns the api's 400 +naming both, and quietly picking one for the caller would be worse. + +`Node` is `#[non_exhaustive]`, so a seventh node type is additive. + +Behaviours worth knowing, each pinned by a test in `src/nodes.rs`: + +- **Flat reads never populate `related_resources`.** `get_by_id`, `by_ids`, `filter` and `search` + all answer `[]`; only the graph reads and the create echo fill it. +- **Graph reads are typed but sparse.** Neo4j stores a column subset, so a `TimeSeries` from + `fetch_related` carries **none** of its type-specific fields — the payload is the shared node + keys and nothing else, so `unit`, `value_type`, `table_engine` and `security_categories` are all + `None`. That is why `TimeSeries::value_type` is `Option`: it is always present on a flat + read and never on a graph one, and a required field made any traversal over a timeseries a hard + deserialization error. `metadata` is empty rather than absent. An asset's geometry is + reconstructed as a Point, so a stored Polygon comes back wrong. +- **`update` still echoes flat `Resource`s**, whatever the node's real type — the one read/write + asymmetry left, owned by the api's `NODE_UPDATE_REFACTOR.md`. `ResourceService::update` is + therefore the one method here that does *not* return `Node`. +- **Policies never carry `value`, `template_id` or `data_set_id`** on a read, and their `metadata` + can be outright `null`. +- **`Resource::geolocation` is write-only** server-side: accepted on create, never echoed. Assets + carry it. +- **Every type is creatable through `/resources/create`, timeseries included** — each element of + `nodes` is dispatched by its own labels. `DATASET` and `POLICY` need the all-datasets manage + grant (403 without), and their `data_set_id` is silently dropped. A duplicate `external_id` + surfaces as a constraint violation rather than the clean 409 `/timeseries/create` gives. + +In Python each variant maps to its own pyclass, so `isinstance(node, TimeSeries)` works and an +object from `resources.filter()` behaves exactly like one from `timeseries.by_ids()`. The dispatch +is a hand-written `IntoPyObject` on a non-pyclass `PyNode` wrapper (`datahub_python_bindings/src/nodes.rs`) +— the first such impl in the bindings — which is what lets `Vec` and `Page` stay generic. +Every node class also exposes `node_type` for data-driven dispatch. + ### Blocking client (`src/blocking.rs`) Synchronous mirror of the async API behind the `blocking` cargo feature — the same split as `reqwest` / `reqwest::blocking`. Every wrapper delegates to the async implementation on a dedicated Tokio runtime owned by the client, so there is exactly one implementation of each call. It must not be constructed or called from inside an async context (building its runtime there panics); use the async `ApiService` instead. @@ -162,8 +211,8 @@ list-like, so existing code is unaffected, but carrying `.next_cursor`. It spans **every** node type — assets, timeseries, functions, resources, data sets, policies — narrowed by `nodeType` (`["resource", "timeseries"]`, case-insensitive; omitted = all; a list of only unknown names matches *nothing*). It behaved this way before by omission, with no discriminator -and single-table inheritance doing the rest; the breadth is now stated and narrowable. Every node -carries its type as a label, so a caller can tell what came back. The other three endpoints stay +and single-table inheritance doing the rest; the breadth is now stated and narrowable. What comes +back is typed per row — see [`Node`](#the-polymorphic-node-type). The other three endpoints stay typed. `DatasetFilter` is consequently just the shared criteria — its `writeProtected` and `deactivated` flags were removed server-side as inert. diff --git a/Cargo.toml b/Cargo.toml index 8aa9502..34e6569 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "intellistream-datahub-sdk" -version = "0.2.0" +version = "0.3.0" edition = "2021" rust-version = "1.85" description = "Async Rust SDK for the IntelliStream DataHub REST API: time series and datapoints, events, resources and relationship edges, files, datasets, labels and subscriptions." diff --git a/datahub_python_bindings/Cargo.toml b/datahub_python_bindings/Cargo.toml index c993410..e0efec4 100644 --- a/datahub_python_bindings/Cargo.toml +++ b/datahub_python_bindings/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "datahub_python_bindings" -version = "0.2.0" +version = "0.3.0" edition = "2024" rust-version = "1.85" description = "PyO3 bindings for the IntelliStream DataHub SDK, published to PyPI as intellistream-datahub-sdk." diff --git a/datahub_python_bindings/pyproject.toml b/datahub_python_bindings/pyproject.toml index 1fa3206..8a0f72a 100644 --- a/datahub_python_bindings/pyproject.toml +++ b/datahub_python_bindings/pyproject.toml @@ -11,7 +11,7 @@ exclude = ["**/__pycache__/**"] [project] name = "intellistream-datahub-sdk" -version = "0.2.0" +version = "0.3.0" description = "Python bindings for the IntelliStream DataHub SDK: time series and datapoints, events, resources and relationship edges, files, datasets, labels and subscriptions." readme = "README.md" authors = [{ name = "IntelliStream AS", email = "support@intellistream.ai" }] diff --git a/datahub_python_bindings/python/intellistream_datahub_sdk/__init__.pyi b/datahub_python_bindings/python/intellistream_datahub_sdk/__init__.pyi index 509c2b3..7457942 100644 --- a/datahub_python_bindings/python/intellistream_datahub_sdk/__init__.pyi +++ b/datahub_python_bindings/python/intellistream_datahub_sdk/__init__.pyi @@ -12,6 +12,12 @@ from typing import Any, Iterable, Iterator, Mapping, Optional, Sequence, Union from uuid import UUID +# One node of any type, as the /resources endpoints return them. Which class you get is decided +# by the node's own intrinsic type-label, so `isinstance(n, TimeSeries)` works and every object +# is the same class its own endpoint would hand back. `n.node_type` gives the name as a string +# when dispatching from data rather than by branching. +Node = Union["Asset", "TimeSeries", "Function", "Resource", "Dataset", "Policy"] + # Convenience alias: every entity-like input accepts either the entity itself, # an IdCollection wrapper, a numeric id, or an external_id string. Identifiable = Union["TimeSeries", "Resource", "Unit", "Event", "IdCollection", int, str] @@ -351,6 +357,20 @@ class TimeSeries: source: str | None = None, ) -> None: ... @property + def node_type(self) -> str: + """This node's type as a string ("asset", "timeseries", "function", "resource", + "dataset", "policy"). Present on every node class, for dispatching from data rather + than with an isinstance ladder.""" + @property + def labels(self) -> list[str] | None: + """Always includes the intrinsic "TIMESERIES" type-label.""" + @labels.setter + def labels(self, value: list[str] | None) -> None: ... + @property + def table_engine(self) -> str | None: + """The ClickHouse table engine. On a series reached through `neighbors()` this is the + API's default rather than data — re-read the series by id for the real value.""" + @property def id(self) -> int | None: ... @property def external_id(self) -> str: ... @@ -768,8 +788,8 @@ class Event: @property def last_updated_time(self) -> datetime.datetime | None: ... # --- navigation (only on events returned by the API; raises otherwise) --- - def related_resource_nodes(self) -> list[Resource]: ... - async def related_resource_nodes_async(self) -> list[Resource]: ... + def related_resource_nodes(self) -> list[Node]: ... + async def related_resource_nodes_async(self) -> list[Node]: ... class TimeFilter: @@ -990,6 +1010,22 @@ class Dataset: connected_data_sets: list[int] | None = None, ) -> None: ... @property + def node_type(self) -> str: + """This node's type as a string ("asset", "timeseries", "function", "resource", + "dataset", "policy"). Present on every node class, for dispatching from data rather + than with an isinstance ladder.""" + @property + def labels(self) -> list[str] | None: + """Always includes the intrinsic "DATASET" type-label.""" + @labels.setter + def labels(self, value: list[str] | None) -> None: ... + @property + def source(self) -> str | None: ... + @source.setter + def source(self, value: str | None) -> None: ... + @property + def related_resources(self) -> list[RelatedNode]: ... + @property def external_id(self) -> str: ... @external_id.setter def external_id(self, value: str) -> None: ... @@ -1190,6 +1226,11 @@ class Resource: geolocation: dict[str, Any] | None = None, ) -> None: ... @property + def node_type(self) -> str: + """This node's type as a string ("asset", "timeseries", "function", "resource", + "dataset", "policy"). Present on every node class, for dispatching from data rather + than with an isinstance ladder.""" + @property def name(self) -> str: ... @name.setter def name(self, value: str) -> None: ... @@ -1254,12 +1295,198 @@ class Resource: async def related_events_async(self, limit: int = 100) -> list[Event]: ... +class Asset: + """A resource that carries a geographic location. + + Assets and plain resources share a field set; the API tells them apart by the intrinsic + "ASSET" type-label, and only an asset ever has its `geolocation` echoed back on a read. + """ + + def __init__( + self, + name: str | None = None, + external_id: str | None = None, + id: int | None = None, + metadata: dict[str, str] | None = None, + description: str | None = None, + is_root: bool = False, + data_set_id: int | None = None, + source: str | None = None, + labels: list[str] | None = None, + related_resources: list[RelatedNode] | None = None, + geolocation: dict[str, Any] | None = None, + ) -> None: ... + @property + def node_type(self) -> str: + """Always "asset".""" + @property + def name(self) -> str: ... + @name.setter + def name(self, value: str) -> None: ... + @property + def external_id(self) -> str: ... + @external_id.setter + def external_id(self, value: str) -> None: ... + @property + def id(self) -> int | None: ... + @id.setter + def id(self, value: int | None) -> None: ... + @property + def metadata(self) -> dict[str, str] | None: ... + @metadata.setter + def metadata(self, value: dict[str, str] | None) -> None: ... + @property + def description(self) -> str | None: ... + @description.setter + def description(self, value: str | None) -> None: ... + @property + def is_root(self) -> bool: ... + @is_root.setter + def is_root(self, value: bool) -> None: ... + @property + def data_set_id(self) -> int | None: ... + @data_set_id.setter + def data_set_id(self, value: int | None) -> None: ... + @property + def source(self) -> str | None: ... + @source.setter + def source(self, value: str | None) -> None: ... + @property + def labels(self) -> list[str] | None: ... + @labels.setter + def labels(self, value: list[str] | None) -> None: ... + @property + def related_resources(self) -> list[RelatedNode]: ... + @related_resources.setter + def related_resources(self, value: list[RelatedNode] | None) -> None: ... + @property + def geolocation(self) -> dict[str, Any] | None: + """GeoJSON geometry. On an asset reached through `neighbors()` this is rebuilt from the + graph's native point and is lossy for anything that is not a Point — read the asset by + id when the geometry matters.""" + @geolocation.setter + def geolocation(self, value: dict[str, Any] | None) -> None: ... + @property + def created_time(self) -> datetime.datetime | None: ... + @property + def last_updated_time(self) -> datetime.datetime | None: ... + # --- navigation (only on assets returned by the API; raises otherwise) --- + def neighbors( + self, + depth: int = -1, + relationship_types: list[str] | None = None, + limit: int = 5000, + ) -> ResourceNetwork: ... + async def neighbors_async( + self, + depth: int = -1, + relationship_types: list[str] | None = None, + limit: int = 5000, + ) -> ResourceNetwork: ... + def related_events(self, limit: int = 100) -> list[Event]: ... + async def related_events_async(self, limit: int = 100) -> list[Event]: ... + + +class Policy: + """An access policy, as a node. + + Sparse on every read: the API never sends a policy's `value`, `template_id` or + `data_set_id` back, so those are always None on an object that came from the server. + """ + + def __init__( + self, + name: str | None = None, + external_id: str | None = None, + id: int | None = None, + type: str | None = None, + value: Any | None = None, + deactivated: bool | None = None, + template_id: int | None = None, + metadata: dict[str, str] | None = None, + description: str | None = None, + data_set_id: int | None = None, + source: str | None = None, + labels: list[str] | None = None, + ) -> None: ... + @property + def node_type(self) -> str: + """Always "policy".""" + @property + def name(self) -> str: ... + @name.setter + def name(self, value: str) -> None: ... + @property + def external_id(self) -> str: ... + @external_id.setter + def external_id(self, value: str) -> None: ... + @property + def id(self) -> int | None: ... + @id.setter + def id(self, value: int | None) -> None: ... + @property + def type(self) -> str | None: + """The policy kind, e.g. "IS_WRITE_PROTECTED".""" + @type.setter + def type(self, value: str | None) -> None: ... + @property + def value(self) -> Any | None: + """Never populated on a read — the API does not send it back.""" + @property + def deactivated(self) -> bool | None: ... + @deactivated.setter + def deactivated(self, value: bool | None) -> None: ... + @property + def template_id(self) -> int | None: + """Never populated on a read.""" + @property + def metadata(self) -> dict[str, str] | None: ... + @metadata.setter + def metadata(self, value: dict[str, str] | None) -> None: ... + @property + def description(self) -> str | None: ... + @description.setter + def description(self, value: str | None) -> None: ... + @property + def data_set_id(self) -> int | None: + """Never populated on a read.""" + @property + def source(self) -> str | None: ... + @source.setter + def source(self, value: str | None) -> None: ... + @property + def labels(self) -> list[str] | None: ... + @labels.setter + def labels(self, value: list[str] | None) -> None: ... + @property + def related_resources(self) -> list[RelatedNode]: ... + @property + def created_time(self) -> datetime.datetime | None: ... + @property + def last_updated_time(self) -> datetime.datetime | None: ... + # --- navigation (only on policies returned by the API; raises otherwise) --- + def neighbors( + self, + depth: int = -1, + relationship_types: list[str] | None = None, + limit: int = 5000, + ) -> ResourceNetwork: ... + async def neighbors_async( + self, + depth: int = -1, + relationship_types: list[str] | None = None, + limit: int = 5000, + ) -> ResourceNetwork: ... + def related_events(self, limit: int = 100) -> list[Event]: ... + async def related_events_async(self, limit: int = 100) -> list[Event]: ... + + class ResourceNetwork: """Connected sub-graph returned by `Resource.neighbors` (and the timeseries/dataset/ function equivalents): the reachable `nodes`, the `edges` between them, and their `labels`.""" @property - def nodes(self) -> list[Resource]: ... + def nodes(self) -> list[Node]: ... @property def edges(self) -> list[EdgeProxy]: ... @property @@ -1348,12 +1575,14 @@ class RelForm: class GraphResult: """Nodes and relations returned from a graph operation.""" @property - def nodes(self) -> list[Resource]: ... + def nodes(self) -> list[Node]: ... @property def relations(self) -> list[EdgeProxy]: ... -ResourceIdentifiable = Union[Resource, str, int] +# Any node object, an external id, or a numeric id. Takes every node class, not just Resource, +# because /resources spans them all — a Dataset from filter() can be handed straight to delete(). +ResourceIdentifiable = Union["Node", str, int] class ResourceUpdate: @@ -1406,16 +1635,16 @@ class ResourceFilter: class ResourcesServiceSync: def create( - self, nodes: list[Resource], relations: list[RelForm] | None = None + self, nodes: list[Node], relations: list[RelForm] | None = None ) -> GraphResult: ... - def by_ids(self, input: list[ResourceIdentifiable]) -> list[Resource]: ... + def by_ids(self, input: list[ResourceIdentifiable]) -> list[Node]: ... def delete(self, input: list[ResourceIdentifiable]) -> None: ... def search( self, query: str, filter: ResourceFilter | None = None, limit: int | None = None, - ) -> list[Resource]: + ) -> list[Node]: """Free-text search for ``query``, ranked by relevance. ``filter`` takes the same criteria as ``filter()`` and only ever removes hits from the @@ -1424,7 +1653,7 @@ class ResourcesServiceSync: which is easy to conflate. """ def update(self, input: list[ResourceUpdate]) -> GraphResult: ... - def get_by_id(self, id: int) -> Resource | None: ... + def get_by_id(self, id: int) -> Node | None: ... def filter( self, filter: ResourceFilter | None = None, @@ -1460,18 +1689,18 @@ class ResourcesServiceSync: class ResourcesServiceAsync: async def create( - self, nodes: list[Resource], relations: list[RelForm] | None = None + self, nodes: list[Node], relations: list[RelForm] | None = None ) -> GraphResult: ... - async def by_ids(self, input: list[ResourceIdentifiable]) -> list[Resource]: ... + async def by_ids(self, input: list[ResourceIdentifiable]) -> list[Node]: ... async def delete(self, input: list[ResourceIdentifiable]) -> None: ... async def search( self, query: str, filter: ResourceFilter | None = None, limit: int | None = None, - ) -> list[Resource]: ... + ) -> list[Node]: ... async def update(self, input: list[ResourceUpdate]) -> GraphResult: ... - async def get_by_id(self, id: int) -> Resource | None: ... + async def get_by_id(self, id: int) -> Node | None: ... async def filter( self, filter: ResourceFilter | None = None, @@ -1692,8 +1921,8 @@ class INode: def security_categories(self) -> list[int] | None: ... # --- navigation (only on inodes returned by the API; raises otherwise) --- # `related_resources` (above) returns the raw ids; these resolve them to Resource objects. - def related_resource_nodes(self) -> list[Resource]: ... - async def related_resource_nodes_async(self) -> list[Resource]: ... + def related_resource_nodes(self) -> list[Node]: ... + async def related_resource_nodes_async(self) -> list[Node]: ... class FileUpload: @@ -1985,10 +2214,27 @@ class Function: @property def name(self) -> str | None: ... @property + def node_type(self) -> str: + """This node's type as a string ("asset", "timeseries", "function", "resource", + "dataset", "policy"). Present on every node class, for dispatching from data rather + than with an isinstance ladder.""" + @property def labels(self) -> list[str]: ... @property def metadata(self) -> dict[str, str]: ... @property + def description(self) -> str | None: ... + @description.setter + def description(self, value: str | None) -> None: ... + @property + def source(self) -> str | None: ... + @source.setter + def source(self, value: str | None) -> None: ... + @property + def data_set_id(self) -> int | None: ... + @data_set_id.setter + def data_set_id(self, value: int | None) -> None: ... + @property def created_time(self) -> datetime.datetime | None: ... @property def last_updated_time(self) -> datetime.datetime | None: ... diff --git a/datahub_python_bindings/src/datasets/mod.rs b/datahub_python_bindings/src/datasets/mod.rs index 2590565..290b753 100644 --- a/datahub_python_bindings/src/datasets/mod.rs +++ b/datahub_python_bindings/src/datasets/mod.rs @@ -109,6 +109,10 @@ impl PyDataset { policies, metadata: metadata.unwrap_or_default(), connected_data_sets: connected_data_sets.unwrap_or_default(), + labels: None, + source: None, + data_set_id: None, + related_resources: vec![], created_time: None, last_updated_time: None, }, @@ -171,6 +175,39 @@ impl PyDataset { pub fn set_connected_data_sets(&mut self, value: Vec) { self.inner.connected_data_sets = value; } + /// The labels on this node, always including the intrinsic `DATASET` type-label. It is what + /// identifies a data set in a heterogeneous `resources.filter()` result. + #[getter] + pub fn labels(&self) -> Option<&Vec> { + self.inner.labels.as_ref() + } + #[setter] + pub fn set_labels(&mut self, value: Option>) { + self.inner.labels = value; + } + #[getter] + pub fn source(&self) -> Option<&str> { + self.inner.source.as_deref() + } + #[setter] + pub fn set_source(&mut self, value: Option) { + self.inner.source = value; + } + #[getter] + pub fn related_resources(&self) -> Vec { + self.inner + .related_resources + .iter() + .cloned() + .map(crate::relations::PyRelatedNode::from) + .collect() + } + /// Always `"dataset"`. Present on every node class so data-driven code can dispatch without + /// an `isinstance` ladder. + #[getter] + pub fn node_type(&self) -> &'static str { + crate::nodes::node_type_name(intellistream_datahub_sdk::nodes::NodeType::Dataset) + } } #[derive(FromPyObject)] diff --git a/datahub_python_bindings/src/events/general.rs b/datahub_python_bindings/src/events/general.rs index 1b13c0c..a3f816f 100644 --- a/datahub_python_bindings/src/events/general.rs +++ b/datahub_python_bindings/src/events/general.rs @@ -168,7 +168,7 @@ impl PyEvent { impl PyEvent { /// Fetch the resources this event references (its `related_resources`), resolved via the /// resources service. Blocking; see [`related_resource_nodes_async`] for the awaitable variant. - fn related_resource_nodes(&self, py: Python<'_>) -> PyResult> { + fn related_resource_nodes(&self, py: Python<'_>) -> PyResult> { let service = self.client.clone().ok_or_else(crate::missing_client_err)?; let ids = self.related_id_collections(); if ids.is_empty() { @@ -182,7 +182,7 @@ impl PyEvent { .nodes() .unwrap_or_default() .into_iter() - .map(|r| PyResource::with_client(r, service.clone())) + .map(|r| crate::nodes::PyNode::with_client(r, service.clone())) .collect()) }) } @@ -193,7 +193,7 @@ impl PyEvent { let ids = self.related_id_collections(); future_into_py(py, async move { if ids.is_empty() { - return Ok(Vec::::new()); + return Ok(Vec::::new()); } let result = service .resources @@ -204,7 +204,7 @@ impl PyEvent { .nodes() .unwrap_or_default() .into_iter() - .map(|r| PyResource::with_client(r, service.clone())) + .map(|r| crate::nodes::PyNode::with_client(r, service.clone())) .collect()) }) } diff --git a/datahub_python_bindings/src/files/mod.rs b/datahub_python_bindings/src/files/mod.rs index 3e8f16b..df7f9f4 100644 --- a/datahub_python_bindings/src/files/mod.rs +++ b/datahub_python_bindings/src/files/mod.rs @@ -221,7 +221,7 @@ impl PyINode { /// Fetch the resources this file references (its `related_resources` ids), resolved to /// `Resource` objects via the resources service. (The `related_resources` *property* returns /// the raw ids; this resolves them.) Blocking; see [`related_resource_nodes_async`]. - fn related_resource_nodes(&self, py: Python<'_>) -> PyResult> { + fn related_resource_nodes(&self, py: Python<'_>) -> PyResult> { let service = self.client.clone().ok_or_else(crate::missing_client_err)?; let ids = self.related_id_collections(); if ids.is_empty() { @@ -235,7 +235,7 @@ impl PyINode { .nodes() .unwrap_or_default() .into_iter() - .map(|r| PyResource::with_client(r, service.clone())) + .map(|r| crate::nodes::PyNode::with_client(r, service.clone())) .collect()) }) } @@ -246,7 +246,7 @@ impl PyINode { let ids = self.related_id_collections(); future_into_py(py, async move { if ids.is_empty() { - return Ok(Vec::::new()); + return Ok(Vec::::new()); } let result = service .resources @@ -257,7 +257,7 @@ impl PyINode { .nodes() .unwrap_or_default() .into_iter() - .map(|r| PyResource::with_client(r, service.clone())) + .map(|r| crate::nodes::PyNode::with_client(r, service.clone())) .collect()) }) } diff --git a/datahub_python_bindings/src/functions/mod.rs b/datahub_python_bindings/src/functions/mod.rs index 5d95be7..80991fd 100644 --- a/datahub_python_bindings/src/functions/mod.rs +++ b/datahub_python_bindings/src/functions/mod.rs @@ -18,7 +18,7 @@ use std::sync::Arc; pub mod async_service; pub mod sync_service; -#[pyclass(module = "intellistream_datahub_sdk", name = "Function")] +#[pyclass(module = "intellistream_datahub_sdk", name = "Function", from_py_object)] #[derive(Clone)] pub struct PyFunction { pub inner: Function, @@ -101,6 +101,40 @@ impl PyFunction { self.inner.labels.clone() } + /// Always `"function"`. Present on every node class so data-driven code can dispatch without + /// an `isinstance` ladder. + #[getter] + fn node_type(&self) -> &'static str { + crate::nodes::node_type_name(intellistream_datahub_sdk::nodes::NodeType::Function) + } + + #[getter] + fn description(&self) -> Option<&str> { + self.inner.description.as_deref() + } + #[setter] + fn set_description(&mut self, value: Option) { + self.inner.description = value; + } + + #[getter] + fn source(&self) -> Option<&str> { + self.inner.source.as_deref() + } + #[setter] + fn set_source(&mut self, value: Option) { + self.inner.source = value; + } + + #[getter] + fn data_set_id(&self) -> Option { + self.inner.data_set_id + } + #[setter] + fn set_data_set_id(&mut self, value: Option) { + self.inner.data_set_id = value; + } + #[getter] fn metadata(&self) -> std::collections::HashMap { self.inner.metadata.clone() diff --git a/datahub_python_bindings/src/lib.rs b/datahub_python_bindings/src/lib.rs index 571eb85..a1daad5 100644 --- a/datahub_python_bindings/src/lib.rs +++ b/datahub_python_bindings/src/lib.rs @@ -3,6 +3,7 @@ mod datetime; mod events; mod files; mod labels; +mod nodes; mod relations; mod resources; mod subscriptions; @@ -1260,5 +1261,6 @@ fn _core(m: &Bound<'_, PyModule>) -> PyResult<()> { subscriptions::register(m)?; functions::register(m)?; relations::register(m)?; + nodes::register(m)?; Ok(()) } diff --git a/datahub_python_bindings/src/nodes.rs b/datahub_python_bindings/src/nodes.rs new file mode 100644 index 0000000..9cfc9bf --- /dev/null +++ b/datahub_python_bindings/src/nodes.rs @@ -0,0 +1,729 @@ +//! The polymorphic node surface: `Asset`, `Policy`, and the dispatch that turns a Rust +//! [`Node`] into whichever Python class matches its type. +//! +//! `/resources` spans every node type, so its reads hand back a mixed list. Each element is the +//! *same* class a caller would get from the type's own endpoint — a timeseries from +//! `resources.filter()` is the `TimeSeries` class, not a lookalike — so `isinstance` works and +//! object-level navigation behaves identically wherever the object came from. Where dispatch is +//! data-driven rather than branching, every node class also exposes `node_type`. + +use std::collections::HashMap; +use std::sync::Arc; + +use chrono::{DateTime, Utc}; +use intellistream_datahub_sdk::nodes::{Asset, Node, NodeType, Policy}; +use intellistream_datahub_sdk::relations::RelatedNode; +use intellistream_datahub_sdk::resources::RelatedResourcesForm; +use intellistream_datahub_sdk::ApiService; +use pyo3::exceptions::PyValueError; +use pyo3::types::{PyAny, PyModuleMethods}; +use pyo3::{Bound, IntoPyObject, PyErr, PyResult, Python, pyclass, pymethods}; +use pyo3_async_runtimes::tokio::future_into_py; +use pythonize::{depythonize, pythonize}; + +use crate::datasets::PyDataset; +use crate::events::PyEvent; +use intellistream_datahub_sdk::filters::{EventFilter, EventFilterForm}; +use crate::functions::PyFunction; +use crate::relations::PyRelatedNode; +use crate::resources::{PyResource, PyResourceNetwork}; +use crate::timeseries::PyTimeSeries; + +fn geometry_from_py(obj: Bound<'_, PyAny>) -> PyResult { + depythonize(&obj).map_err(|e| PyValueError::new_err(format!("invalid geolocation: {e}"))) +} + +/// The name a node of this type answers to in Python's `node_type` and in the `node_type=` +/// filter argument. +pub fn node_type_name(kind: NodeType) -> &'static str { + kind.filter_name() +} + +/// One node of any type on its way to Python. +/// +/// Deliberately **not** a `#[pyclass]`: Python never sees a `Node` object, it sees an `Asset`, a +/// `TimeSeries`, a `Dataset` and so on. This type exists only so `Vec` and +/// `Page::new(py, nodes, ..)` compose — both take anything that is `IntoPyObject`, which is what +/// the impl below provides. It is the first hand-written `IntoPyObject` in these bindings. +#[derive(Clone)] +pub struct PyNode { + pub inner: Node, + pub client: Arc, +} + +impl PyNode { + pub fn with_client(inner: Node, client: Arc) -> Self { + Self { inner, client } + } + + /// Wrap a whole list, stamping the client so navigation works off every element. + pub fn many(nodes: Vec, client: Arc) -> Vec { + nodes + .into_iter() + .map(|n| PyNode::with_client(n, client.clone())) + .collect() + } +} + +impl<'py> IntoPyObject<'py> for PyNode { + type Target = PyAny; + type Output = Bound<'py, PyAny>; + type Error = PyErr; + + fn into_pyobject(self, py: Python<'py>) -> Result { + let client = self.client; + Ok(match self.inner { + Node::Asset(n) => PyAsset::with_client(n, client).into_pyobject(py)?.into_any(), + Node::TimeSeries(n) => PyTimeSeries::with_client(n, client) + .into_pyobject(py)? + .into_any(), + Node::Function(n) => PyFunction::with_client(n, client) + .into_pyobject(py)? + .into_any(), + Node::Resource(n) => PyResource::with_client(n, client) + .into_pyobject(py)? + .into_any(), + Node::Dataset(n) => PyDataset::with_client(n, client) + .into_pyobject(py)? + .into_any(), + Node::Policy(n) => PyPolicy::with_client(n, client) + .into_pyobject(py)? + .into_any(), + // `Node` is `#[non_exhaustive]`: a node type added to the api arrives here before + // this crate knows about it. Say so plainly rather than mapping it onto the wrong + // class. + other => { + return Err(PyValueError::new_err(format!( + "this SDK does not yet know the node type {:?}; upgrade the SDK", + other.kind() + ))); + } + }) + } +} + +/// What `resources.create(...)` accepts as a node: any of the six node classes. +/// +/// Each is dispatched server-side by its own type-label, so one heterogeneous list can create a +/// data set, an asset and a timeseries in a single call. +#[derive(Clone, pyo3::FromPyObject)] +pub enum NodeInput { + Asset(PyAsset), + TimeSeries(PyTimeSeries), + Function(PyFunction), + Dataset(PyDataset), + Policy(PyPolicy), + Resource(PyResource), +} + +impl From for Node { + fn from(v: NodeInput) -> Self { + match v { + NodeInput::Asset(n) => Node::Asset(n.inner), + NodeInput::TimeSeries(n) => Node::TimeSeries(n.inner), + NodeInput::Function(n) => Node::Function(n.inner), + NodeInput::Dataset(n) => Node::Dataset(n.inner), + NodeInput::Policy(n) => Node::Policy(n.inner), + NodeInput::Resource(n) => Node::Resource(n.inner), + } + } +} + +/// Generate the object-level navigation every node class carries: `neighbors` walks outward +/// from this node through the graph, and `related_events` goes the other way, finding the events +/// that reference it. +/// +/// The four older node classes each spell these out by hand. The two defined here share one +/// macro instead, because writing them out a fifth and sixth time is how `Asset` came to have +/// `neighbors` but not `related_events` — a node that had both as a `Resource` silently lost one +/// by being typed. +macro_rules! node_navigation { + ($ty:ty) => { + impl $ty { + /// The events filter selecting events that reference this node, by id when the + /// server has assigned one and by external id otherwise. + fn related_events_filter(&self, limit: u64) -> EventFilterForm { + let mut basic = EventFilter::default(); + match self.inner.id { + Some(id) => { + basic.set_related_resource_ids(&[id]); + } + None => { + basic.set_related_resource_external_ids(&[self.inner.external_id.as_str()]); + } + } + let mut filter = EventFilterForm::default(); + filter.set_filter(basic); + filter.set_limit(limit); + filter + } + } + + #[pymethods] + impl $ty { + /// Fetch the events whose `related_resources` include this node (matched by graph-node + /// id when present, else external id). `limit` caps the results. Blocking; see + /// `related_events_async`. + #[pyo3(signature = (limit=100))] + fn related_events(&self, py: Python<'_>, limit: u64) -> PyResult> { + let service = self.client.clone().ok_or_else(crate::missing_client_err)?; + let filter = self.related_events_filter(limit); + py.detach(|| { + let result = crate::nav_runtime() + .block_on(service.events.filter(&filter)) + .map_err(crate::datahub_err)?; + Ok(result + .get_items() + .iter() + .cloned() + .map(|e| PyEvent::with_client(e, service.clone())) + .collect()) + }) + } + + /// Awaitable variant of `related_events`. + #[pyo3(signature = (limit=100))] + fn related_events_async<'py>( + &self, + py: Python<'py>, + limit: u64, + ) -> PyResult> { + let service = self.client.clone().ok_or_else(crate::missing_client_err)?; + let filter = self.related_events_filter(limit); + future_into_py(py, async move { + let result = service + .events + .filter(&filter) + .await + .map_err(crate::datahub_err)?; + Ok(result + .get_items() + .iter() + .cloned() + .map(|e| PyEvent::with_client(e, service.clone())) + .collect::>()) + }) + } + } + + #[pymethods] + impl $ty { + /// Walk the graph from this node and return the connected sub-graph (its `nodes`, the + /// `edges` between them, and their `labels`). `depth` bounds the traversal in hops + /// (`-1`, the default, = the whole connected component); `relationship_types` filters + /// which edge types to follow (`None` = all); `limit` caps the node count. + /// + /// Nodes reached this way are typed but sparse — the graph stores only a subset of + /// each node's columns. Re-read one by id for its full field set. + #[pyo3(signature = (depth=-1, relationship_types=None, limit=5000))] + fn neighbors( + &self, + py: Python<'_>, + depth: i32, + relationship_types: Option>, + limit: i32, + ) -> PyResult { + let service = self.client.clone().ok_or_else(crate::missing_client_err)?; + let form = self.related_form(depth, relationship_types, limit); + py.detach(|| { + let result = crate::nav_runtime() + .block_on(service.resources.fetch_related(&form)) + .map_err(crate::datahub_err)?; + Ok(PyResourceNetwork::from_network(result, service.clone())) + }) + } + + /// Awaitable variant of `neighbors`. + #[pyo3(signature = (depth=-1, relationship_types=None, limit=5000))] + fn neighbors_async<'py>( + &self, + py: Python<'py>, + depth: i32, + relationship_types: Option>, + limit: i32, + ) -> PyResult> { + let service = self.client.clone().ok_or_else(crate::missing_client_err)?; + let form = self.related_form(depth, relationship_types, limit); + future_into_py(py, async move { + let result = service + .resources + .fetch_related(&form) + .await + .map_err(crate::datahub_err)?; + Ok(PyResourceNetwork::from_network(result, service.clone())) + }) + } + } + }; +} + +/// An asset — a resource that carries a geographic location. +#[pyclass(module = "intellistream_datahub_sdk", name = "Asset", from_py_object)] +#[derive(Clone)] +pub struct PyAsset { + pub inner: Asset, + /// `None` on locally-constructed assets — navigation then raises. + pub client: Option>, +} + +impl From for PyAsset { + fn from(inner: Asset) -> Self { + Self { + inner, + client: None, + } + } +} +impl From for Asset { + fn from(v: PyAsset) -> Self { + v.inner + } +} + +impl PyAsset { + pub fn with_client(inner: Asset, client: Arc) -> Self { + Self { + inner, + client: Some(client), + } + } + + fn related_form( + &self, + depth: i32, + relationship_types: Option>, + limit: i32, + ) -> RelatedResourcesForm { + RelatedResourcesForm { + id: self.inner.id, + external_id: Some(self.inner.external_id.clone()), + depth, + relationship_types, + limit, + excluded_labels: vec![], + } + } +} + +#[pymethods] +impl PyAsset { + #[new] + #[pyo3(signature=( + name=None, + external_id=None, + id=None, + metadata=None, + description=None, + is_root=false, + data_set_id=None, + source=None, + labels=None, + related_resources=None, + geolocation=None))] + pub fn new( + name: Option, + external_id: Option, + id: Option, + metadata: Option>, + description: Option, + is_root: bool, + data_set_id: Option, + source: Option, + labels: Option>, + related_resources: Option>, + geolocation: Option>, + ) -> PyResult { + let geolocation = geolocation.map(geometry_from_py).transpose()?; + let (final_name, final_ext_id) = crate::resources::name_and_external_id(name, external_id)?; + Ok(Self { + inner: Asset { + id, + external_id: final_ext_id, + name: final_name, + metadata, + description, + is_root, + data_set_id, + source, + labels, + related_resources: related_resources + .map(|v| v.into_iter().map(RelatedNode::from).collect()) + .unwrap_or_default(), + geolocation, + created_time: None, + last_updated_time: None, + }, + client: None, + }) + } + + /// Always `"asset"`. Present on every node class so data-driven code can dispatch without + /// an `isinstance` ladder. + #[getter] + pub fn node_type(&self) -> &'static str { + node_type_name(NodeType::Asset) + } + #[getter] + pub fn name(&self) -> &str { + &self.inner.name + } + #[setter] + pub fn set_name(&mut self, value: String) { + self.inner.name = value; + } + #[getter] + pub fn external_id(&self) -> &str { + &self.inner.external_id + } + #[setter] + pub fn set_external_id(&mut self, value: String) { + self.inner.external_id = value; + } + #[getter] + pub fn id(&self) -> Option { + self.inner.id + } + #[setter] + pub fn set_id(&mut self, value: Option) { + self.inner.id = value; + } + #[getter] + pub fn metadata(&self) -> Option<&HashMap> { + self.inner.metadata.as_ref() + } + #[setter] + pub fn set_metadata(&mut self, value: Option>) { + self.inner.metadata = value; + } + #[getter] + pub fn description(&self) -> Option<&str> { + self.inner.description.as_deref() + } + #[setter] + pub fn set_description(&mut self, value: Option) { + self.inner.description = value; + } + #[getter] + pub fn is_root(&self) -> bool { + self.inner.is_root + } + #[setter] + pub fn set_is_root(&mut self, value: bool) { + self.inner.is_root = value; + } + #[getter] + pub fn data_set_id(&self) -> Option { + self.inner.data_set_id + } + #[setter] + pub fn set_data_set_id(&mut self, value: Option) { + self.inner.data_set_id = value; + } + #[getter] + pub fn source(&self) -> Option<&str> { + self.inner.source.as_deref() + } + #[setter] + pub fn set_source(&mut self, value: Option) { + self.inner.source = value; + } + #[getter] + pub fn labels(&self) -> Option<&Vec> { + self.inner.labels.as_ref() + } + #[setter] + pub fn set_labels(&mut self, value: Option>) { + self.inner.labels = value; + } + #[getter] + pub fn related_resources(&self) -> Vec { + self.inner + .related_resources + .iter() + .cloned() + .map(PyRelatedNode::from) + .collect() + } + #[setter] + pub fn set_related_resources(&mut self, value: Option>) { + self.inner.related_resources = value + .map(|v| v.into_iter().map(RelatedNode::from).collect()) + .unwrap_or_default(); + } + /// The GeoJSON geometry as a Python `dict`, or `None`. + /// + /// On an asset reached through `neighbors()` this is reconstructed from the graph's native + /// point, which is lossy for anything that is not a `Point` — read the asset by id when the + /// geometry matters. + #[getter] + pub fn geolocation<'py>(&self, py: Python<'py>) -> PyResult>> { + match &self.inner.geolocation { + Some(geom) => Ok(Some(pythonize(py, geom).map_err(|e| { + PyValueError::new_err(format!("could not serialize geolocation: {e}")) + })?)), + None => Ok(None), + } + } + #[setter] + pub fn set_geolocation(&mut self, value: Option>) -> PyResult<()> { + self.inner.geolocation = value.map(geometry_from_py).transpose()?; + Ok(()) + } + #[getter] + pub fn created_time(&self) -> Option> { + self.inner.created_time + } + #[getter] + pub fn last_updated_time(&self) -> Option> { + self.inner.last_updated_time + } + fn __repr__(&self) -> String { + format!( + "Asset(external_id={:?}, name={:?})", + self.inner.external_id, self.inner.name + ) + } +} + +node_navigation!(PyAsset); + +/// An access policy, as a node. +/// +/// Sparse on every read: the api never sends a policy's `value`, `template_id` or `data_set_id` +/// back, so those are `None` regardless of what is stored. +#[pyclass(module = "intellistream_datahub_sdk", name = "Policy", from_py_object)] +#[derive(Clone)] +pub struct PyPolicy { + pub inner: Policy, + pub client: Option>, +} + +impl From for PyPolicy { + fn from(inner: Policy) -> Self { + Self { + inner, + client: None, + } + } +} +impl From for Policy { + fn from(v: PyPolicy) -> Self { + v.inner + } +} + +impl PyPolicy { + pub fn with_client(inner: Policy, client: Arc) -> Self { + Self { + inner, + client: Some(client), + } + } + + fn related_form( + &self, + depth: i32, + relationship_types: Option>, + limit: i32, + ) -> RelatedResourcesForm { + RelatedResourcesForm { + id: self.inner.id, + external_id: Some(self.inner.external_id.clone()), + depth, + relationship_types, + limit, + excluded_labels: vec![], + } + } +} + +#[pymethods] +impl PyPolicy { + #[new] + #[pyo3(signature=( + name=None, + external_id=None, + id=None, + r#type=None, + value=None, + deactivated=None, + template_id=None, + metadata=None, + description=None, + data_set_id=None, + source=None, + labels=None))] + #[allow(clippy::too_many_arguments)] + pub fn new( + name: Option, + external_id: Option, + id: Option, + r#type: Option, + value: Option>, + deactivated: Option, + template_id: Option, + metadata: Option>, + description: Option, + data_set_id: Option, + source: Option, + labels: Option>, + ) -> PyResult { + let value = value + .map(|v| { + depythonize::(&v) + .map_err(|e| PyValueError::new_err(format!("invalid policy value: {e}"))) + }) + .transpose()?; + let (final_name, final_ext_id) = crate::resources::name_and_external_id(name, external_id)?; + Ok(Self { + inner: Policy { + id, + external_id: final_ext_id, + name: final_name, + description, + policy_type: r#type, + value, + deactivated, + template_id, + data_set_id, + source, + metadata, + labels, + related_resources: vec![], + created_time: None, + last_updated_time: None, + }, + client: None, + }) + } + + /// Always `"policy"`. + #[getter] + pub fn node_type(&self) -> &'static str { + node_type_name(NodeType::Policy) + } + #[getter] + pub fn name(&self) -> &str { + &self.inner.name + } + #[setter] + pub fn set_name(&mut self, value: String) { + self.inner.name = value; + } + #[getter] + pub fn external_id(&self) -> &str { + &self.inner.external_id + } + #[setter] + pub fn set_external_id(&mut self, value: String) { + self.inner.external_id = value; + } + #[getter] + pub fn id(&self) -> Option { + self.inner.id + } + #[setter] + pub fn set_id(&mut self, value: Option) { + self.inner.id = value; + } + /// The policy kind, e.g. `"IS_WRITE_PROTECTED"`. + #[getter(r#type)] + pub fn policy_type(&self) -> Option<&str> { + self.inner.policy_type.as_deref() + } + #[setter(r#type)] + pub fn set_policy_type(&mut self, value: Option) { + self.inner.policy_type = value; + } + /// The policy's value. Never populated on a read — the api does not send it back. + #[getter] + pub fn value<'py>(&self, py: Python<'py>) -> PyResult>> { + match &self.inner.value { + Some(v) => Ok(Some(pythonize(py, v).map_err(|e| { + PyValueError::new_err(format!("could not serialize policy value: {e}")) + })?)), + None => Ok(None), + } + } + #[getter] + pub fn deactivated(&self) -> Option { + self.inner.deactivated + } + #[setter] + pub fn set_deactivated(&mut self, value: Option) { + self.inner.deactivated = value; + } + /// Never populated on a read. + #[getter] + pub fn template_id(&self) -> Option { + self.inner.template_id + } + #[getter] + pub fn metadata(&self) -> Option<&HashMap> { + self.inner.metadata.as_ref() + } + #[setter] + pub fn set_metadata(&mut self, value: Option>) { + self.inner.metadata = value; + } + #[getter] + pub fn description(&self) -> Option<&str> { + self.inner.description.as_deref() + } + #[setter] + pub fn set_description(&mut self, value: Option) { + self.inner.description = value; + } + /// Never populated on a read. + #[getter] + pub fn data_set_id(&self) -> Option { + self.inner.data_set_id + } + #[getter] + pub fn source(&self) -> Option<&str> { + self.inner.source.as_deref() + } + #[setter] + pub fn set_source(&mut self, value: Option) { + self.inner.source = value; + } + #[getter] + pub fn labels(&self) -> Option<&Vec> { + self.inner.labels.as_ref() + } + #[setter] + pub fn set_labels(&mut self, value: Option>) { + self.inner.labels = value; + } + #[getter] + pub fn related_resources(&self) -> Vec { + self.inner + .related_resources + .iter() + .cloned() + .map(PyRelatedNode::from) + .collect() + } + #[getter] + pub fn created_time(&self) -> Option> { + self.inner.created_time + } + #[getter] + pub fn last_updated_time(&self) -> Option> { + self.inner.last_updated_time + } + fn __repr__(&self) -> String { + format!( + "Policy(external_id={:?}, name={:?})", + self.inner.external_id, self.inner.name + ) + } +} + +node_navigation!(PyPolicy); + +pub fn register(m: &Bound<'_, pyo3::types::PyModule>) -> PyResult<()> { + m.add_class::()?; + m.add_class::()?; + Ok(()) +} diff --git a/datahub_python_bindings/src/relations/mod.rs b/datahub_python_bindings/src/relations/mod.rs index d6c3a40..f152591 100644 --- a/datahub_python_bindings/src/relations/mod.rs +++ b/datahub_python_bindings/src/relations/mod.rs @@ -1,12 +1,12 @@ pub(crate) mod async_service; pub(crate) mod sync_service; -use crate::resources::PyResource; use intellistream_datahub_sdk::graph_data_wrapper::GraphDataWrapper; use intellistream_datahub_sdk::generic::IdAndExtId; use intellistream_datahub_sdk::relations::{ EdgeProxy, RelForm, RelTypeForm, RelatedNode, RelationDirection, RelationshipType, }; +use intellistream_datahub_sdk::nodes::Node; use intellistream_datahub_sdk::{ApiService, Resource}; use pyo3::prelude::*; use pyo3::{Bound, PyResult, pyclass, pymethods}; @@ -348,37 +348,58 @@ impl PyRelForm { } } -/// Python view of `GraphDataWrapper`: the nodes and relations returned -/// from a graph operation. `.nodes` is `list[Resource]`, `.relations` is `list[EdgeProxy]`. +/// Python view of a graph response: the nodes and relations returned from a graph operation. +/// `.nodes` is a list of node objects (`Asset`, `TimeSeries`, `Dataset`, …), `.relations` is +/// `list[EdgeProxy]`. #[pyclass(module = "intellistream_datahub_sdk", name = "GraphResult")] #[derive(Clone)] pub struct PyGraphResult { - pub nodes: Vec, + pub nodes: Vec, pub relations: Vec, } impl PyGraphResult { - /// Build the Python view of a create/graph response, stamping `client` onto every node so - /// callers can chain navigation off the returned resources. - pub fn from_wrapper(wrapper: GraphDataWrapper, client: Arc) -> Self { + /// Build the Python view of a typed graph response (create, `by_ids`), stamping `client` onto + /// every node so callers can chain navigation off the results. + pub fn from_wrapper(wrapper: GraphDataWrapper, client: Arc) -> Self { + let nodes = crate::nodes::PyNode::many(wrapper.nodes().unwrap_or_default(), client.clone()); + Self { + nodes, + relations: Self::relations_of(&wrapper.relations()), + } + } + + /// Build the view of a response the api echoes as flat resources whatever the node's real + /// type — today that is `/resources/update` alone. The nodes really are resource-shaped here, + /// so they are presented as such rather than being guessed back into their types out of data + /// the echo does not carry. + pub fn from_resource_wrapper( + wrapper: GraphDataWrapper, + client: Arc, + ) -> Self { let nodes = wrapper .nodes() .unwrap_or_default() .into_iter() - .map(|r| PyResource::with_client(r, client.clone())) + .map(|r| crate::nodes::PyNode::with_client(Node::Resource(r), client.clone())) .collect(); - let relations = wrapper - .relations() + Self { + nodes, + relations: Self::relations_of(&wrapper.relations()), + } + } + + fn relations_of(relations: &Option<&Vec>) -> Vec { + relations .map(|v| v.iter().cloned().map(PyEdgeProxy::from).collect()) - .unwrap_or_default(); - Self { nodes, relations } + .unwrap_or_default() } } #[pymethods] impl PyGraphResult { #[getter] - fn nodes(&self) -> Vec { + fn nodes(&self) -> Vec { self.nodes.clone() } diff --git a/datahub_python_bindings/src/resources/async_service.rs b/datahub_python_bindings/src/resources/async_service.rs index cc8ed9e..6a66d21 100644 --- a/datahub_python_bindings/src/resources/async_service.rs +++ b/datahub_python_bindings/src/resources/async_service.rs @@ -1,11 +1,12 @@ +use intellistream_datahub_sdk::nodes::Node; use crate::relations::{PyGraphResult, PyRelForm}; use crate::resources::{PyResourceFilter, PyResourceNetwork, PyResourceUpdate, ResourceIdentifiable}; use intellistream_datahub_sdk::resources::ResourceUpdate; -use crate::{DataSetRef, PyResource, StringOrList}; +use crate::{DataSetRef, StringOrList}; use intellistream_datahub_sdk::generic::IdAndExtId; use intellistream_datahub_sdk::relations::RelForm; use intellistream_datahub_sdk::resources::RelatedResourcesForm; -use intellistream_datahub_sdk::{ApiService, Resource}; +use intellistream_datahub_sdk::ApiService; use pyo3::{Bound, PyAny, PyResult, Python, pyclass, pymethods}; use pyo3_async_runtimes::tokio::future_into_py; use std::collections::HashMap; @@ -22,10 +23,10 @@ impl PyResourcesServiceAsync { fn create<'py>( &self, py: Python<'py>, - nodes: Vec, + nodes: Vec, relations: Option>, ) -> PyResult> { - let resources: Vec = nodes.into_iter().map(Resource::from).collect(); + let nodes: Vec = nodes.into_iter().map(Node::from).collect(); let rel_forms: Vec = relations .unwrap_or_default() .into_iter() @@ -35,7 +36,7 @@ impl PyResourcesServiceAsync { future_into_py(py, async move { let result = service .resources - .create(resources, rel_forms) + .create(nodes, rel_forms) .await .map_err(|e| crate::datahub_err(e))?; Ok(PyGraphResult::from_wrapper(result, service.clone())) @@ -60,11 +61,11 @@ impl PyResourcesServiceAsync { .await .map_err(|e| crate::datahub_err(e))?; - let py_units: Vec = result + let py_units: Vec = result .nodes() .unwrap() .iter() - .map(|u| PyResource::with_client(u.clone(), service.clone())) + .map(|u| crate::nodes::PyNode::with_client(u.clone(), service.clone())) .collect(); Ok(py_units) }) @@ -87,10 +88,12 @@ impl PyResourcesServiceAsync { .await .map_err(|e| crate::datahub_err(e))?; - let py_ts: Vec = result + // `delete` answers 204 with no body, so this list is always empty; the api types + // the echo as flat resources, so that is what it is wrapped as. + let py_ts: Vec = result .nodes().unwrap_or_default() .into_iter() - .map(|res| PyResource::with_client(res.clone(), service.clone())) + .map(|res| crate::nodes::PyNode::with_client(Node::Resource(res), service.clone())) .collect(); Ok(py_ts) }) @@ -113,18 +116,23 @@ impl PyResourcesServiceAsync { .await .map_err(|e| crate::datahub_err(e))?; - let py_res: Vec = result + let py_res: Vec = result .get_items() .iter() - .map(|r| PyResource::with_client(r.clone(), service.clone())) + .map(|r| crate::nodes::PyNode::with_client(r.clone(), service.clone())) .collect(); Ok(py_res) }) } - /// Update resources in place. Each [`ResourceUpdate`] targets one resource and carries only - /// the fields to change. Returns the updated graph, whose node labels reflect what the server - /// stored (the intrinsic type-label is always kept). + /// Update nodes in place. Each [`ResourceUpdate`] targets one node and carries only the + /// fields to change; every field it can set is shared by all node types, so one update form + /// covers them all. + /// + /// **The echo is flat.** Unlike every read on this service, the api answers here with each + /// node shaped as a plain `Resource` whatever its real type, so `.nodes` holds `Resource` + /// objects even for a timeseries. Re-read the node if you need its typed form. The `labels` + /// do reflect what the server stored, intrinsic type-label included. fn update<'py>( &self, py: Python<'py>, @@ -138,7 +146,7 @@ impl PyResourcesServiceAsync { .update(&updates) .await .map_err(|e| crate::datahub_err(e))?; - Ok(PyGraphResult::from_wrapper(result, service.clone())) + Ok(PyGraphResult::from_resource_wrapper(result, service.clone())) }) } @@ -155,7 +163,7 @@ impl PyResourcesServiceAsync { Ok(result .get_items() .first() - .map(|r| PyResource::with_client(r.clone(), service.clone()))) + .map(|r| crate::nodes::PyNode::with_client(r.clone(), service.clone()))) }) } @@ -199,10 +207,10 @@ impl PyResourcesServiceAsync { .await .map_err(|e| crate::datahub_err(e))?; let next_cursor = result.next_cursor().map(str::to_string); - let items: Vec = result + let items: Vec = result .get_items() .iter() - .map(|r| PyResource::with_client(r.clone(), service.clone())) + .map(|r| crate::nodes::PyNode::with_client(r.clone(), service.clone())) .collect(); Python::attach(|py| crate::PyPage::new(py, items, next_cursor)) }) diff --git a/datahub_python_bindings/src/resources/mod.rs b/datahub_python_bindings/src/resources/mod.rs index 5c33dc6..33cc3ff 100644 --- a/datahub_python_bindings/src/resources/mod.rs +++ b/datahub_python_bindings/src/resources/mod.rs @@ -22,6 +22,26 @@ fn geometry_from_py(obj: Bound<'_, PyAny>) -> PyResult { depythonize(&obj).map_err(|e| PyValueError::new_err(format!("invalid geolocation: {e}"))) } +/// Resolve the `name`/`external_id` pair every node constructor accepts: either may be omitted +/// and is derived from the other, but not both. Shared by `Resource`, `Asset` and `Policy` so +/// the three cannot drift. +pub(crate) fn name_and_external_id( + name: Option, + external_id: Option, +) -> PyResult<(String, String)> { + match (name, external_id) { + (Some(name), Some(external_id)) => Ok((name, external_id)), + (None, Some(external_id)) => Ok((external_id.clone(), external_id)), + (Some(name), None) => { + let ext = to_snake_lower_cased_allow_start_with_digits(&name); + Ok((name, ext)) + } + (None, None) => Err(PyValueError::new_err( + "name or external_id must be provided", + )), + } +} + pub mod async_service; pub mod sync_service; @@ -103,23 +123,37 @@ impl PyResourceUpdate { } } -/// Things accepted as a resource identifier when fetching by_ids or deleting. -/// Mirrors the `FunctionIdentifyable` pattern so callers can pass a `Resource`, +/// Things accepted as a node identifier when fetching by_ids or deleting. +/// Mirrors the `FunctionIdentifyable` pattern so callers can pass any node object, /// an external id string, or a numeric id directly. +/// +/// It takes every node class, not just `Resource`, because `/resources` spans them all — a +/// `Dataset` that came back from `filter()` can be handed straight to `delete()`. #[derive(Clone, FromPyObject)] pub enum ResourceIdentifiable { Resource(PyResource), + Asset(crate::nodes::PyAsset), + TimeSeries(crate::timeseries::PyTimeSeries), + Function(crate::functions::PyFunction), + Dataset(crate::datasets::PyDataset), + Policy(crate::nodes::PyPolicy), ExternalId(String), Id(u64), } impl From for IdAndExtId { fn from(value: ResourceIdentifiable) -> Self { + let pair = |id: Option, ext: &str| Self { + id, + external_id: Some(ext.to_string()), + }; match value { - ResourceIdentifiable::Resource(r) => Self { - id: r.inner.id, - external_id: Some(r.inner.external_id.clone()), - }, + ResourceIdentifiable::Resource(r) => pair(r.inner.id, &r.inner.external_id), + ResourceIdentifiable::Asset(r) => pair(r.inner.id, &r.inner.external_id), + ResourceIdentifiable::TimeSeries(r) => pair(r.inner.id, &r.inner.external_id), + ResourceIdentifiable::Function(r) => pair(r.inner.id, &r.inner.external_id), + ResourceIdentifiable::Dataset(r) => pair(r.inner.id, &r.inner.external_id), + ResourceIdentifiable::Policy(r) => pair(r.inner.id, &r.inner.external_id), ResourceIdentifiable::ExternalId(ext) => Self { id: None, external_id: Some(ext), @@ -213,19 +247,7 @@ impl PyResource { geolocation: Option>, ) -> PyResult { let geolocation = geolocation.map(geometry_from_py).transpose()?; - let (final_name, final_ext_id) = match (name, external_id) { - (Some(name), Some(external_id)) => (name, external_id), - (None, Some(external_id)) => (external_id.clone(), external_id), - (Some(name), None) => ( - name.clone(), - to_snake_lower_cased_allow_start_with_digits(&name), - ), - (None, None) => { - return Err(PyValueError::new_err( - "name or external_id must be provided", - )); - } - }; + let (final_name, final_ext_id) = name_and_external_id(name, external_id)?; Ok(Self { inner: Resource { name: final_name, @@ -247,6 +269,12 @@ impl PyResource { client: None, }) } + /// Always `"resource"`. Present on every node class so data-driven code can dispatch without + /// an `isinstance` ladder. + #[getter] + pub fn node_type(&self) -> &'static str { + crate::nodes::node_type_name(intellistream_datahub_sdk::nodes::NodeType::Resource) + } #[getter] pub fn name(&self) -> &str { self.inner.name.as_str() @@ -369,7 +397,7 @@ use crate::labels::PyLabel; #[pyclass(module = "intellistream_datahub_sdk", name = "ResourceNetwork")] #[derive(Clone)] pub struct PyResourceNetwork { - pub nodes: Vec, + pub nodes: Vec, pub edges: Vec, pub labels: Vec, } @@ -382,11 +410,7 @@ impl PyResourceNetwork { client: Arc, ) -> Self { Self { - nodes: network - .nodes - .into_iter() - .map(|r| PyResource::with_client(r, client.clone())) - .collect(), + nodes: crate::nodes::PyNode::many(network.nodes, client.clone()), edges: network.edges.into_iter().map(PyEdgeProxy::from).collect(), labels: network.labels.into_iter().map(PyLabel::from).collect(), } @@ -395,8 +419,10 @@ impl PyResourceNetwork { #[pymethods] impl PyResourceNetwork { + /// The nodes in the traversed sub-graph, each as its own class (`Asset`, `TimeSeries`, + /// `Dataset`, …). Typed but sparse — the graph carries only a subset of each node's columns. #[getter] - fn nodes(&self) -> Vec { + fn nodes(&self) -> Vec { self.nodes.clone() } #[getter] diff --git a/datahub_python_bindings/src/resources/sync_service.rs b/datahub_python_bindings/src/resources/sync_service.rs index b3d7418..4adc966 100644 --- a/datahub_python_bindings/src/resources/sync_service.rs +++ b/datahub_python_bindings/src/resources/sync_service.rs @@ -1,6 +1,7 @@ +use intellistream_datahub_sdk::nodes::Node; use crate::relations::{PyGraphResult, PyRelForm}; use crate::resources::{PyResourceFilter, ResourceIdentifiable}; -use crate::resources::{PyResource, PyResourceNetwork, PyResourceUpdate}; +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}; @@ -10,7 +11,7 @@ use intellistream_datahub_sdk::relations::RelForm; use intellistream_datahub_sdk::resources::{ FetchNearestResourcesForm, RelatedResourcesForm, ResourceFilter, ResourceFilterForm, }; -use intellistream_datahub_sdk::{ApiService, Resource}; +use intellistream_datahub_sdk::ApiService; use pyo3::{PyResult, Python, pyclass, pymethods}; use std::collections::HashMap; use std::sync::Arc; @@ -27,10 +28,10 @@ impl PyResourcesServiceSync { fn create<'py>( &self, py: Python<'py>, - nodes: Vec, + nodes: Vec, relations: Option>, ) -> PyResult { - let resources: Vec = nodes.into_iter().map(Resource::from).collect(); + let nodes: Vec = nodes.into_iter().map(Node::from).collect(); let rel_forms: Vec = relations .unwrap_or_default() .into_iter() @@ -39,7 +40,7 @@ impl PyResourcesServiceSync { let service = self.api_service.clone(); let result = py.detach(|| { self.runtime - .block_on(service.resources.create(resources, rel_forms)) + .block_on(service.resources.create(nodes, rel_forms)) }); let result = result.map_err(|e| crate::datahub_err(e))?; @@ -50,7 +51,7 @@ impl PyResourcesServiceSync { &self, py: Python<'py>, input: Vec, - ) -> PyResult> { + ) -> PyResult> { let service = self.api_service.clone(); let input_ids = input .into_iter() @@ -61,12 +62,12 @@ impl PyResourcesServiceSync { let result = result.map_err(|e| crate::datahub_err(e))?; - let py_res: Vec = result + let py_res: Vec = result .nodes() .as_ref() .unwrap() .iter() - .map(|ts| PyResource::with_client(ts.clone(), service.clone())) + .map(|ts| crate::nodes::PyNode::with_client(ts.clone(), service.clone())) .collect(); Ok(py_res) } @@ -92,7 +93,7 @@ impl PyResourcesServiceSync { query: String, filter: Option, limit: Option, - ) -> PyResult> { + ) -> PyResult> { let form = crate::search_form(query, filter.map(|f| f.inner), limit); let service = self.api_service.clone(); @@ -102,18 +103,23 @@ impl PyResourcesServiceSync { .block_on(service.resources.search(&form)) .map_err(|e| crate::datahub_err(e))?; - let py_res: Vec = result + let py_res: Vec = result .get_items() .iter() - .map(|r| PyResource::with_client(r.clone(), service.clone())) + .map(|r| crate::nodes::PyNode::with_client(r.clone(), service.clone())) .collect(); Ok(py_res) }) } - /// Update resources in place. Each [`ResourceUpdate`] targets one resource and carries only - /// the fields to change. Returns the updated graph, whose node labels reflect what the server - /// stored (the intrinsic type-label is always kept). + /// Update nodes in place. Each [`ResourceUpdate`] targets one node and carries only the + /// fields to change; every field it can set is shared by all node types, so one update form + /// covers them all. + /// + /// **The echo is flat.** Unlike every read on this service, the api answers here with each + /// node shaped as a plain `Resource` whatever its real type, so `.nodes` holds `Resource` + /// objects even for a timeseries. Re-read the node if you need its typed form. The `labels` + /// do reflect what the server stored, intrinsic type-label included. fn update<'py>( &self, py: Python<'py>, @@ -123,20 +129,20 @@ impl PyResourcesServiceSync { let service = self.api_service.clone(); let result = py.detach(|| self.runtime.block_on(service.resources.update(&updates))); let result = result.map_err(|e| crate::datahub_err(e))?; - Ok(PyGraphResult::from_wrapper(result, service.clone())) + Ok(PyGraphResult::from_resource_wrapper(result, service.clone())) } /// `GET /resources/{id}` — one resource by numeric id. Raises when it does not exist, /// unlike `by_ids`, which silently omits what it cannot find. - fn get_by_id<'py>(&self, py: Python<'py>, id: u64) -> PyResult> { + fn get_by_id<'py>(&self, py: Python<'py>, id: u64) -> PyResult> { let service = self.api_service.clone(); let result = py.detach(|| self.runtime.block_on(service.resources.get_by_id(id))); let result = result.map_err(|e| crate::datahub_err(e))?; Ok(result .get_items() .first() - .map(|r| PyResource::with_client(r.clone(), service.clone()))) + .map(|r| crate::nodes::PyNode::with_client(r.clone(), service.clone()))) } /// `POST /resources/filter` — structured lookup; every criterion is combined with AND. @@ -189,10 +195,10 @@ impl PyResourcesServiceSync { .block_on(service.resources.filter(&form)) .map_err(|e| crate::datahub_err(e))?; let next_cursor = result.next_cursor().map(str::to_string); - let items: Vec = result + let items: Vec = result .get_items() .iter() - .map(|r| PyResource::with_client(r.clone(), service.clone())) + .map(|r| crate::nodes::PyNode::with_client(r.clone(), service.clone())) .collect(); Ok::<_, pyo3::PyErr>((items, next_cursor)) })?; diff --git a/datahub_python_bindings/src/timeseries/construction.rs b/datahub_python_bindings/src/timeseries/construction.rs index c7cc269..c6a77f2 100644 --- a/datahub_python_bindings/src/timeseries/construction.rs +++ b/datahub_python_bindings/src/timeseries/construction.rs @@ -77,13 +77,15 @@ impl PyTimeSeries { unit_external_id, security_categories, data_set_id, - value_type: value_type.to_string(), + value_type: Some(value_type.to_string()), source, created_time: None, last_updated_time: None, related_resources: related_resources .map(|r| r.into_iter().map(RelatedNode::from).collect()) .unwrap_or_default(), + labels: None, + table_engine: None, }; Ok(PyTimeSeries { inner, diff --git a/datahub_python_bindings/src/timeseries/general.rs b/datahub_python_bindings/src/timeseries/general.rs index 7c3c743..84c521b 100644 --- a/datahub_python_bindings/src/timeseries/general.rs +++ b/datahub_python_bindings/src/timeseries/general.rs @@ -78,12 +78,14 @@ impl PyTimeSeries { self.inner.data_set_id = value; } #[getter] - pub fn value_type(&self) -> &str { - &self.inner.value_type.as_str() + /// `None` on a series reached through `neighbors()` — the graph does not carry the column. + /// Re-read the series by id when the value type matters. + pub fn value_type(&self) -> Option<&str> { + self.inner.value_type.as_deref() } #[setter] pub fn set_value_type(&mut self, value: ValueType) { - self.inner.value_type = value.to_string(); + self.inner.value_type = Some(value.to_string()); } #[getter] pub fn source(&self) -> Option<&str> { @@ -93,6 +95,30 @@ impl PyTimeSeries { pub fn set_source(&mut self, value: Option) { self.inner.source = value; } + /// The labels on this node, always including the intrinsic `TIMESERIES` type-label. It is + /// what identifies a timeseries in a heterogeneous `resources.filter()` result. + #[getter] + pub fn labels(&self) -> Option<&Vec> { + self.inner.labels.as_ref() + } + #[setter] + pub fn set_labels(&mut self, value: Option>) { + self.inner.labels = value; + } + /// The ClickHouse table engine backing this series. Server-assigned. + /// + /// On a series reached through `neighbors()` this is the api's DTO default rather than data — + /// the graph does not store the column. Re-read the series by id for the real value. + #[getter] + pub fn table_engine(&self) -> Option<&str> { + self.inner.table_engine.as_deref() + } + /// Always `"timeseries"`. Present on every node class so data-driven code can dispatch + /// without an `isinstance` ladder. + #[getter] + pub fn node_type(&self) -> &'static str { + crate::nodes::node_type_name(intellistream_datahub_sdk::nodes::NodeType::TimeSeries) + } #[getter] pub fn created_time(&self) -> Option> { self.inner.created_time diff --git a/datahub_python_bindings/src/timeseries/mod.rs b/datahub_python_bindings/src/timeseries/mod.rs index 3b863b0..dd91a75 100644 --- a/datahub_python_bindings/src/timeseries/mod.rs +++ b/datahub_python_bindings/src/timeseries/mod.rs @@ -84,7 +84,7 @@ pub mod sync_service; /// the connected node (a Timeseries, Dataset, Asset or Policy) /// /// -#[pyclass(module = "intellistream_datahub_sdk", name = "TimeSeries")] +#[pyclass(module = "intellistream_datahub_sdk", name = "TimeSeries", from_py_object)] #[derive(Clone)] pub struct PyTimeSeries { pub inner: TimeSeries, diff --git a/python_tests/conftest.py b/python_tests/conftest.py index 1a7444a..fd8b2d1 100644 --- a/python_tests/conftest.py +++ b/python_tests/conftest.py @@ -41,11 +41,14 @@ from fixtures import ENV_FILE, TEST_PREFIX, _safe_delete_each -# `/resources/filter` labels every node with its type; these are the ones with a typed +# `/resources/filter` returns every node as its own type; these are the ones with a typed # delete endpoint of their own. Anything else is deleted as a plain resource. -_TIMESERIES = "TIMESERIES" -_DATASET = "DATASET" -_FUNCTION = "FUNCTION" +# +# Dispatch is on `node_type` rather than on the labels the type is derived from: it is present +# on every node class and says the same thing without the caller re-deriving it. +_TIMESERIES = "timeseries" +_DATASET = "dataset" +_FUNCTION = "function" def _is_test(value) -> bool: @@ -79,8 +82,8 @@ def _delete_nodes(client, nodes) -> None: while remaining: by_type = {_TIMESERIES: [], _DATASET: [], _FUNCTION: [], "other": []} for node in remaining: - labels = {str(label).upper() for label in (node.labels or [])} - key = next((k for k in (_TIMESERIES, _DATASET, _FUNCTION) if k in labels), "other") + node_type = getattr(node, "node_type", "resource") + key = node_type if node_type in by_type else "other" by_type[key].append(node.external_id) # Data sets last: everything else may belong to one. diff --git a/python_tests/test_polymorphic_nodes.py b/python_tests/test_polymorphic_nodes.py new file mode 100644 index 0000000..d09dcd4 --- /dev/null +++ b/python_tests/test_polymorphic_nodes.py @@ -0,0 +1,261 @@ +"""``/resources`` returns nodes typed by their intrinsic label, not one flat shape. + +``/resources`` is the generic node query — it spans assets, timeseries, functions, resources, +data sets and policies — but it used to answer with every row coerced into a ``Resource``. A +timeseries came back with no ``unit``, a data set with no ``policies``, and both carried an +``is_root`` flag meaningless for their type. Now each row arrives as its own class. + +Two things this suite pins that nothing else can: + +* the **dispatch** — which class you get for which node, and that a node with no type-label is a + plain ``Resource`` rather than an error; +* the **sparseness boundary** — a node read flatly carries its type's full field set, while the + same node reached through the graph does not, because the graph stores only a column subset. + Getting this backwards is how a caller ends up trusting a ``value_type`` that is really a + server-side default. +""" +import pytest + +import intellistream_datahub_sdk +from intellistream_datahub_sdk import ( + Asset, + Dataset, + Function, + Resource, + TimeSeries, +) + +from fixtures import TEST_LABEL, async_client, sync_client, unique_id # noqa: F401 (fixtures) +from polling import poll_until + + +@pytest.fixture +def corpus(sync_client): + """One node of each creatable type, all sharing an external-id stem. + + Created through ``/resources/create`` in a single call, which is itself the thing being + covered: each element of ``nodes`` is dispatched server-side by its own type-label, so one + heterogeneous list creates four different kinds of node. + """ + stem = unique_id("poly") + ds = Dataset(external_id=f"{stem}_ds", name=f"Poly DS {stem}") + asset = Asset( + external_id=f"{stem}_asset", + name=f"Poly Asset {stem}", + labels=[TEST_LABEL], + is_root=True, + geolocation={"type": "Point", "coordinates": [10.75, 59.91]}, + ) + func = Function(external_id=f"{stem}_fn", name=f"Poly Fn {stem}") + # Every node needs at least one label; a plain resource is the one that carries no *type* + # label, not one that carries none at all. + plain = Resource( + external_id=f"{stem}_plain", name=f"Poly Plain {stem}", labels=[TEST_LABEL] + ) + + created = sync_client.resources.create([ds, asset, func, plain]) + try: + yield {"stem": stem, "nodes": created.nodes} + finally: + for ext in (f"{stem}_plain", f"{stem}_fn", f"{stem}_asset", f"{stem}_ds"): + try: + sync_client.resources.delete([ext]) + except Exception: + pass + + +def by_ext(nodes): + return {n.external_id: n for n in nodes} + + +# --------------------------------------------------------------------------- # +# dispatch +# --------------------------------------------------------------------------- # + +def test_one_create_call_builds_four_different_node_types(corpus): + """The write side dispatches per element, not per request.""" + found = by_ext(corpus["nodes"]) + stem = corpus["stem"] + assert isinstance(found[f"{stem}_ds"], Dataset) + assert isinstance(found[f"{stem}_asset"], Asset) + assert isinstance(found[f"{stem}_fn"], Function) + assert isinstance(found[f"{stem}_plain"], Resource) + + +def test_filter_returns_each_node_as_its_own_class(sync_client, corpus): + stem = corpus["stem"] + found = by_ext(sync_client.resources.filter(external_id=f"{stem}*", limit=100)) + assert isinstance(found[f"{stem}_ds"], Dataset) + assert isinstance(found[f"{stem}_asset"], Asset) + assert isinstance(found[f"{stem}_fn"], Function) + assert isinstance(found[f"{stem}_plain"], Resource) + + +def test_node_type_names_the_type_without_an_isinstance_ladder(sync_client, corpus): + stem = corpus["stem"] + found = by_ext(sync_client.resources.filter(external_id=f"{stem}*", limit=100)) + assert {ext.rsplit("_", 1)[1]: n.node_type for ext, n in found.items()} == { + "ds": "dataset", + "asset": "asset", + "fn": "function", + "plain": "resource", + } + + +def test_a_node_with_no_type_label_is_a_plain_resource(sync_client, corpus): + """Absence of a type-label *is* the resource signal — there is no ``RESOURCE`` label.""" + plain = sync_client.resources.filter(external_id=f"{corpus['stem']}_plain")[0] + assert isinstance(plain, Resource) + assert not {l.upper() for l in (plain.labels or [])} & { + "ASSET", "TIMESERIES", "FUNCTION", "DATASET", "POLICY" + } + + +def test_a_domain_label_does_not_change_the_type(sync_client, corpus): + """``TEST`` is an ordinary label; only the five privileged ones name a type.""" + asset = sync_client.resources.filter(external_id=f"{corpus['stem']}_asset")[0] + assert isinstance(asset, Asset) + assert TEST_LABEL in (asset.labels or []) + + +def test_every_node_still_carries_its_type_as_a_label(sync_client, corpus): + """The label the dispatch reads is not consumed by it — callers can still see it.""" + found = by_ext(sync_client.resources.filter(external_id=f"{corpus['stem']}*", limit=100)) + stem = corpus["stem"] + assert "DATASET" in (found[f"{stem}_ds"].labels or []) + assert "ASSET" in (found[f"{stem}_asset"].labels or []) + assert "FUNCTION" in found[f"{stem}_fn"].labels + + +# --------------------------------------------------------------------------- # +# per-type fields, which the flat shape used to drop on the floor +# --------------------------------------------------------------------------- # + +def test_a_timeseries_read_through_resources_carries_its_timeseries_fields( + sync_client, corpus +): + """This is the whole point: `unit` and `value_type` used to be unreachable here.""" + stem = corpus["stem"] + ext = f"{stem}_ts" + sync_client.timeseries.create([ + TimeSeries(external_id=ext, name=f"Poly TS {stem}", unit="bar", value_type="float") + ]) + try: + node = sync_client.resources.filter(external_id=ext)[0] + assert isinstance(node, TimeSeries) + assert node.unit == "bar" + assert node.value_type == "float" + assert node.table_engine # server-assigned, present on a flat read + finally: + try: + sync_client.timeseries.delete([ext]) + except Exception: + pass + + +def test_only_assets_and_resources_carry_is_root(sync_client, corpus): + """A data set has no such column; the flat shape used to give it one anyway.""" + stem = corpus["stem"] + found = by_ext(sync_client.resources.filter(external_id=f"{stem}*", limit=100)) + assert found[f"{stem}_asset"].is_root is True + assert found[f"{stem}_plain"].is_root is False + assert not hasattr(found[f"{stem}_ds"], "is_root") + assert not hasattr(found[f"{stem}_fn"], "is_root") + + +def test_an_asset_echoes_its_geometry_where_a_plain_resource_does_not(sync_client, corpus): + """`geoLocation` is write-only on a resource and read-write on an asset.""" + stem = corpus["stem"] + found = by_ext(sync_client.resources.filter(external_id=f"{stem}*", limit=100)) + assert found[f"{stem}_asset"].geolocation["type"] == "Point" + assert found[f"{stem}_plain"].geolocation is None + + +# --------------------------------------------------------------------------- # +# the sparseness boundary +# --------------------------------------------------------------------------- # + +def test_a_node_reached_through_the_graph_is_typed_but_sparse(sync_client, corpus): + """The graph stores a column subset, so the same timeseries is thinner here. + + Typed either way — that part is the fix. But a caller who reads `unit` off a graph-sourced + node gets nothing, and must re-read the node by id. + """ + stem = corpus["stem"] + ts_ext = f"{stem}_ts_graph" + sync_client.timeseries.create([ + TimeSeries(external_id=ts_ext, name=f"Graph TS {stem}", unit="bar", value_type="float") + ]) + try: + sync_client.edges.create([ + intellistream_datahub_sdk.RelForm( + relationship_type="MEASURES", + from_external_id=f"{stem}_asset", + to_external_id=ts_ext, + ) + ]) + asset = sync_client.resources.filter(external_id=f"{stem}_asset")[0] + + # The graph projection lags the write, so poll rather than skip: skipping would let this + # assertion quietly never run, which is the whole point of the test. + def reached(): + return next( + (n for n in asset.neighbors(depth=1).nodes if n.external_id == ts_ext), None + ) + + graph_ts = poll_until(reached, lambda n: n is not None) + assert graph_ts is not None, "the timeseries never appeared in the graph projection" + assert isinstance(graph_ts, TimeSeries) + # The graph carries the shared node fields and nothing else, so every type-specific + # field is absent rather than defaulted. `value_type` is the one that bit: as a required + # field it made any traversal over a timeseries a hard deserialization error. + assert graph_ts.unit is None, "the graph does not carry the unit column" + assert graph_ts.value_type is None + assert graph_ts.table_engine is None + assert graph_ts.security_categories is None + finally: + try: + sync_client.timeseries.delete([ts_ext]) + except Exception: + pass + + +# --------------------------------------------------------------------------- # +# the write side +# --------------------------------------------------------------------------- # + +def test_a_labelled_resource_still_creates_a_typed_node(sync_client): + """The pre-existing idiom: put the type-label on a bare `Resource`. + + Kept working on purpose — serialization never strips or rewrites a caller's labels — so code + written before the typed classes existed reads back as the right type. + """ + ext = unique_id("poly_legacy") + sync_client.resources.create([Resource(external_id=ext, name="Legacy Asset", + labels=["ASSET"])]) + try: + assert isinstance(sync_client.resources.filter(external_id=ext)[0], Asset) + finally: + try: + sync_client.resources.delete([ext]) + except Exception: + pass + + +def test_two_type_labels_are_refused(sync_client): + """A node's type is intrinsic; the api will not guess which of two the caller meant.""" + ext = unique_id("poly_ambiguous") + with pytest.raises(intellistream_datahub_sdk.DataHubException): + sync_client.resources.create([ + Resource(external_id=ext, name="Ambiguous", labels=["ASSET", "DATASET"]) + ]) + + +def test_a_node_from_filter_can_be_deleted_directly(sync_client): + """`delete` takes any node object, not just a `Resource`.""" + ext = unique_id("poly_delete") + sync_client.datasets.create([Dataset(external_id=ext, name="Poly Delete DS")]) + node = sync_client.resources.filter(external_id=ext)[0] + assert isinstance(node, Dataset) + sync_client.datasets.delete([node]) + assert sync_client.resources.filter(external_id=ext) == [] diff --git a/src/blocking.rs b/src/blocking.rs index 538df59..4ee9794 100644 --- a/src/blocking.rs +++ b/src/blocking.rs @@ -40,6 +40,7 @@ use crate::generic::{ use crate::graph_data_wrapper::GraphDataWrapper; use crate::http::ResponseError; use crate::labels::Label; +use crate::nodes::Node; use crate::relations::{EdgeProxy, RelForm, RelTypeForm, RelationshipType}; use crate::resources::{ RelatedResourcesForm, Resource, ResourceFilter, ResourceNetwork, ResourceUpdate, @@ -187,14 +188,21 @@ pub struct ResourceService { impl ResourceService { delegate! { resources => - fn create(nodes: Vec, relations: Vec) -> Result, ResponseError>; - fn search(payload: &SearchAndFilterForm) -> Result, ResponseError>; + fn search(payload: &SearchAndFilterForm) -> Result, ResponseError>; fn fetch_related(form: &RelatedResourcesForm) -> Result; } - // These two return GraphDataWrapper, not DataWrapper; delegated by hand. + // Generic or GraphDataWrapper-returning; delegated by hand. - pub fn by_ids(&self, input: &I) -> Result, ResponseError> + pub fn create>( + &self, + nodes: Vec, + relations: Vec, + ) -> Result, ResponseError> { + self.rt.block_on(self.api.resources.create(nodes, relations)) + } + + pub fn by_ids(&self, input: &I) -> Result, ResponseError> where for<'a> &'a I: Into>, { @@ -208,6 +216,7 @@ impl ResourceService { self.rt.block_on(self.api.resources.delete(input)) } + /// Mirrors the async `update`: the echo is a flat [`Resource`] whatever the node's type. pub fn update(&self, input: &I) -> Result, ResponseError> where for<'a> &'a I: Into>, @@ -372,7 +381,7 @@ impl EdgesService { } delegate_into! { edges => - fn by_ids(input: Into>) -> Result, ResponseError>; + fn by_ids(input: Into>) -> Result, ResponseError>; fn create(data: Into>) -> Result, ResponseError>; fn delete(json: Into>) -> Result, ResponseError>; fn create_types(data: Into>) -> Result, ResponseError>; diff --git a/src/datasets/mod.rs b/src/datasets/mod.rs index b6f9484..85c1684 100644 --- a/src/datasets/mod.rs +++ b/src/datasets/mod.rs @@ -9,7 +9,7 @@ use crate::graph_data_wrapper::{GraphDataWrapper, GraphNode}; use crate::http::ResponseError; use crate::resources::Resource; use crate::ApiService; -use chrono::{DateTime, FixedOffset, Utc}; +use chrono::{DateTime, Utc}; use maplit::hashmap; use serde::{Deserialize, Serialize}; use std::collections::HashMap; @@ -177,10 +177,38 @@ pub struct Dataset { pub name: String, pub description: Option, pub policies: Option>, + #[serde(default)] pub metadata: HashMap, + /// Input-only in practice: the api declares it but never populates it on a read, so this is + /// empty on everything that comes back from the server. + /// + /// The wire carries these ids as JSON *strings* (`["5"]`), not numbers. + #[serde(default, with = "crate::serde_helper::string_id_vec")] pub connected_data_sets: Vec, - pub created_time: Option>, - pub last_updated_time: Option>, + /// The labels carried by this node, always including the intrinsic `DATASET` type-label the + /// api forces back on every read. It is what identifies a data set in a heterogeneous + /// `/resources` result — see [`crate::nodes::Node`]. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub labels: Option>, + /// The name of the system this data set's primary information comes from — the `source` + /// column shared by every node type. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub source: Option, + /// Declared by the shared node base, but the api refuses to set it on a data set (a data set + /// belonging to a data set would orphan its own ACL grant), so this is always `None` on a + /// read and silently dropped on a create. + #[serde( + default, + skip_serializing_if = "Option::is_none", + with = "crate::serde_helper::opt_string_id" + )] + pub data_set_id: Option, + /// The nodes this data set is connected to. Populated only on graph reads and the + /// `/resources/create` echo; the flat reads answer with an empty list. + #[serde(default, skip_serializing)] + pub related_resources: Vec, + pub created_time: Option>, + pub last_updated_time: Option>, } impl DataHubEntity for Dataset { fn ext_id(&self) -> &String { @@ -200,6 +228,10 @@ impl Dataset { name, policies: None, connected_data_sets: vec![], + labels: None, + source: None, + data_set_id: None, + related_resources: vec![], created_time: None, last_updated_time: None, @@ -249,10 +281,10 @@ impl Dataset { self.description = Some(description); self } - pub fn created_time(&self) -> Option<&DateTime> { + pub fn created_time(&self) -> Option<&DateTime> { self.created_time.as_ref() } - pub fn last_updated_time(&self) -> Option<&DateTime> { + pub fn last_updated_time(&self) -> Option<&DateTime> { self.last_updated_time.as_ref() } pub fn build(&self) -> Self { diff --git a/src/functions/mod.rs b/src/functions/mod.rs index 7979927..8e14856 100644 --- a/src/functions/mod.rs +++ b/src/functions/mod.rs @@ -109,7 +109,7 @@ impl FunctionsService { /// API representation of a function. Mirrors `ai.intellistream.datahub.function.Function`, which /// extends the shared node base and adds nothing of its own — so this is exactly the node fields. -#[derive(Debug, Serialize, Deserialize, Clone)] +#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)] #[serde(rename_all = "camelCase")] pub struct Function { #[serde(skip_serializing_if = "Option::is_none")] @@ -123,6 +123,18 @@ pub struct Function { pub labels: Vec, #[serde(default, skip_serializing_if = "HashMap::is_empty")] pub metadata: HashMap, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub description: Option, + /// The name of the system this function's primary information comes from — the `source` + /// column shared by every node type. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub source: Option, + #[serde( + default, + skip_serializing_if = "Option::is_none", + with = "crate::serde_helper::opt_string_id" + )] + pub data_set_id: Option, /// The nodes this function is connected to, with relationship type and direction. /// Populated server-side by `FunctionService.list()`. #[serde(default, skip_serializing)] @@ -141,6 +153,9 @@ impl Function { name: None, labels: vec![], metadata: HashMap::new(), + description: None, + source: None, + data_set_id: None, related_resources: vec![], created_time: None, last_updated_time: None, diff --git a/src/lib.rs b/src/lib.rs index 94bce7b..0410a7e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -38,6 +38,7 @@ pub mod labels; mod mcp_integration; #[cfg(test)] mod multi_tenant_integration; +pub mod nodes; pub mod relations; pub mod resources; pub mod serde_helper; @@ -49,6 +50,7 @@ pub mod unit; pub mod functions; pub use resources::*; +pub use nodes::{Asset, Node, NodeType, Policy}; /// GeoJSON geometry type used by [`Resource::geolocation`]; re-exported so callers /// don't need a direct dependency on the `geojson` crate. pub use geojson::Geometry; @@ -56,6 +58,7 @@ pub use events::*; pub use timeseries::*; pub use relations::{EdgeProxy, RelForm, RelatedNode, RelationDirection}; use crate::datasets::*; +pub use crate::datasets::Dataset; pub use subscriptions::{ DataCollectionString, DataSort, DataWrapperMessage, EventAction, EventObject, ListenError, diff --git a/src/mcp_integration.rs b/src/mcp_integration.rs index 23ae171..14eb12e 100644 --- a/src/mcp_integration.rs +++ b/src/mcp_integration.rs @@ -1713,7 +1713,7 @@ async fn sweep_resources_and_edges( .map(|wrapper| wrapper.nodes.clone().unwrap_or_default()) .unwrap_or_default() }, - |found: &Vec| !found.is_empty(), + |found: &Vec| !found.is_empty(), ) .await; assert!( diff --git a/src/multi_tenant_integration.rs b/src/multi_tenant_integration.rs index f666222..90ee926 100644 --- a/src/multi_tenant_integration.rs +++ b/src/multi_tenant_integration.rs @@ -207,6 +207,7 @@ use crate::generic::{DataWrapper, IdAndExtId, SearchAndFilterForm}; use crate::graph_data_wrapper::GraphDataWrapper; use crate::http::ResponseError; use crate::resources::{RelatedResourcesForm, Resource}; +use crate::nodes::Node; use crate::tests::cleanup::{cleanup_datasets_as, cleanup_resources_as, cleanup_timeseries_as}; use crate::{ApiService, TimeSeries}; use chrono::Utc; @@ -363,7 +364,7 @@ where /// narrowed by the tenant's database (or by the dataset ACL) answers 200 with an empty node list, /// while a single-item read of something in another tenant is a 404. Both mean the caller cannot /// see it, which is what these tests are about. -fn is_absent(result: &Result, ResponseError>) -> bool { +fn is_absent(result: &Result, ResponseError>) -> bool { match result { Ok(wrapper) => wrapper.nodes().map_or(true, |n| n.is_empty()), Err(e) => e.get_status().as_u16() == 404, @@ -644,12 +645,12 @@ async fn multi_tenant_same_external_id_in_two_orgs_are_independent() -> Result<( .by_ids(&by_external_id(&external_id)) .await?; assert_eq!( - read_a.nodes().and_then(|n| n.first().and_then(|r| r.description.clone())), + read_a.nodes().and_then(|n| n.first().and_then(|r| r.description().map(str::to_string))), Some("belongs to org A".to_string()), "org A must read back its own entity" ); assert_eq!( - read_b.nodes().and_then(|n| n.first().and_then(|r| r.description.clone())), + read_b.nodes().and_then(|n| n.first().and_then(|r| r.description().map(str::to_string))), Some("belongs to org B".to_string()), "org B must read back its own entity, not org A's" ); @@ -723,7 +724,7 @@ async fn multi_tenant_entity_created_in_one_org_is_invisible_from_the_other( seen_by_a .get_items() .iter() - .any(|r| r.external_id == external_id), + .any(|r| r.external_id() == external_id), "org A should find its own entity by marker '{marker}'" ); @@ -732,7 +733,7 @@ async fn multi_tenant_entity_created_in_one_org_is_invisible_from_the_other( !hits .get_items() .iter() - .any(|r| r.external_id == external_id), + .any(|r| r.external_id() == external_id), "org B's search must not surface org A's entity" ); @@ -950,7 +951,7 @@ async fn acl_list_and_search_omit_rows_rather_than_denying() -> Result<(), Respo seen_by_admin .get_items() .iter() - .any(|r| r.external_id == seeded), + .any(|r| r.external_id() == seeded), "the seeding admin should find its own resource" ); @@ -959,7 +960,7 @@ async fn acl_list_and_search_omit_rows_rather_than_denying() -> Result<(), Respo !seen_by_outsider .get_items() .iter() - .any(|r| r.external_id == seeded), + .any(|r| r.external_id() == seeded), "an ungranted caller's search must omit the row" ); diff --git a/src/nodes.rs b/src/nodes.rs new file mode 100644 index 0000000..89518a0 --- /dev/null +++ b/src/nodes.rs @@ -0,0 +1,1038 @@ +//! The polymorphic node type returned by the `/resources` endpoints. +//! +//! `/resources` is the *generic* node query: assets, timeseries, functions, resources, data sets +//! and policies share one table and one set of criteria, and a single result page can hold any +//! mixture of them. [`Node`] is that mixture in the type system — one variant per node type, each +//! carrying the concrete struct for its type. +//! +//! # The discriminator is a label, not a field +//! +//! There is no `nodeType` key on the wire. A node's type is the *intrinsic type-label* the api +//! forces into its `labels` array on every read — `ASSET`, `TIMESERIES`, `FUNCTION`, `DATASET`, +//! `POLICY` — and a plain resource carries **none of them**, so absence is the `RESOURCE` signal. +//! [`Node`]'s `Deserialize` follows the api's own `NodeModelDeserializer`: the same label +//! canonicalization ([`to_snake_upper_cased`]), the same fallback to `RESOURCE`, and the same +//! refusal to guess when a node carries more than one type-label. +//! +//! # What a node actually carries depends on where you read it +//! +//! - **Flat reads** (`get_by_id`, `by_ids`, `filter`, `search`) are fully populated *except* +//! `related_resources`, which is always empty — the api does not join the edges in. +//! - **Graph reads** ([`ResourceService::fetch_related`](crate::resources::ResourceService::fetch_related), +//! `fetch_nearest`) are **typed but sparse**: Neo4j stores only a subset of the columns, so a +//! [`TimeSeries`] from there carries **none** of its type-specific fields — no `unit`, +//! `unit_external_id`, `value_type`, `table_engine` or `security_categories`. They are absent +//! from the payload, not defaulted, which is why every one of them is `Option`. `metadata` is +//! silently empty rather than absent, and `related_resources` *is* populated there. +//! - **Policies** never carry `value`, `template_id` or `data_set_id` on any read — the api's +//! transformer does not set them. + +use std::collections::HashMap; + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; + +use crate::datasets::Dataset; +use crate::functions::Function; +use crate::generic::Identifiable; +use crate::graph_data_wrapper::GraphNode; +use crate::relations::RelatedNode; +use crate::resources::Resource; +use crate::timeseries::TimeSeries; + +/// The six node types `/resources` spans. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum NodeType { + Asset, + TimeSeries, + Function, + Resource, + Dataset, + Policy, +} + +impl NodeType { + /// The intrinsic type-label a node of this type carries. + /// + /// `None` for [`NodeType::Resource`], and that is the whole design: a plain resource carries + /// no type-label at all, so an empty match is what identifies one. Mirrors the api's + /// `TypeLabels.forEntity` returning an empty `Optional`. + pub const fn type_label(self) -> Option<&'static str> { + match self { + NodeType::Asset => Some("ASSET"), + NodeType::TimeSeries => Some("TIMESERIES"), + NodeType::Function => Some("FUNCTION"), + NodeType::Dataset => Some("DATASET"), + NodeType::Policy => Some("POLICY"), + NodeType::Resource => None, + } + } + + /// The name this type answers to in [`ResourceFilter::node_type`](crate::resources::ResourceFilter). + pub const fn filter_name(self) -> &'static str { + match self { + NodeType::Asset => "asset", + NodeType::TimeSeries => "timeseries", + NodeType::Function => "function", + NodeType::Resource => "resource", + NodeType::Dataset => "dataset", + NodeType::Policy => "policy", + } + } + + /// The type this label names, or `None` when it is an ordinary domain label. + /// The label is canonicalized first, so `"dataset"`, `"DataSet"` and `"data-set"` all resolve. + pub fn from_type_label(label: &str) -> Option { + match to_snake_upper_cased(label).as_str() { + "ASSET" => Some(NodeType::Asset), + "TIMESERIES" => Some(NodeType::TimeSeries), + "FUNCTION" => Some(NodeType::Function), + "DATASET" => Some(NodeType::Dataset), + "POLICY" => Some(NodeType::Policy), + _ => None, + } + } + + /// Which type a label set names. Empty ⇒ [`NodeType::Resource`]; more than one distinct + /// type-label is an error, as it is on the api. + pub fn from_labels>(labels: &[S]) -> Result { + let mut found: Vec = Vec::new(); + for label in labels { + if let Some(t) = NodeType::from_type_label(label.as_ref()) { + if !found.contains(&t) { + found.push(t); + } + } + } + match found.len() { + 0 => Ok(NodeType::Resource), + 1 => Ok(found[0]), + _ => Err(AmbiguousNodeType(found)), + } + } +} + +/// A node carrying more than one type-label, which names no single type. +/// +/// The api rejects this on every `NodeModel` request body and at create time, so it cannot arise +/// from a healthy tenant — but note that the api's *graph* mapper silently takes the first +/// type-label instead, so the two server paths disagree about a row that has somehow acquired two. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct AmbiguousNodeType(pub Vec); + +impl std::fmt::Display for AmbiguousNodeType { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let names: Vec<&str> = self + .0 + .iter() + .filter_map(|t| t.type_label()) + .collect(); + write!( + f, + "a node may have at most one type-label; got {}", + names.join(", ") + ) + } +} + +impl std::error::Error for AmbiguousNodeType {} + +/// Port of the api's `TextValidator.toSnakeUpperCased`, which is what canonicalizes every label +/// name server-side before it is stored or compared. +/// +/// Leading digits are dropped, the rest is upper-cased, and each whitespace character — or each +/// run of other non-word characters — becomes a single `_`. Blank input passes through unchanged. +pub fn to_snake_upper_cased(s: &str) -> String { + if s.trim().is_empty() { + return s.to_string(); + } + let without_leading_digits: String = { + let mut chars = s.chars().peekable(); + while chars.peek().is_some_and(|c| c.is_numeric()) { + chars.next(); + } + chars.collect() + }; + + let is_word = |c: char| c.is_alphanumeric() || c == '_'; + let upper = without_leading_digits.to_uppercase(); + let mut out = String::with_capacity(upper.len()); + let mut chars = upper.chars().peekable(); + while let Some(c) = chars.next() { + if c.is_whitespace() { + // The api's pattern tries a single whitespace before a run of non-word characters, + // so "a b" keeps both separators where "a--b" collapses to one. + out.push('_'); + } else if !is_word(c) { + out.push('_'); + while chars.peek().is_some_and(|n| !is_word(*n)) { + chars.next(); + } + } else { + out.push(c); + } + } + out +} + +/// An asset — a resource that can carry a geographic location. +/// +/// Field-identical to [`Resource`] plus a meaningful `geolocation`: the api models both on the +/// same node base and distinguishes them only by the `ASSET` type-label. `geolocation` is the one +/// practical difference, since a plain resource accepts it on write but never echoes it back. +#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct Asset { + #[serde(default, with = "crate::serde_helper::opt_string_id")] + pub id: Option, + pub external_id: String, + pub name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub metadata: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub description: Option, + #[serde(default)] + pub is_root: bool, + #[serde( + default, + skip_serializing_if = "Option::is_none", + with = "crate::serde_helper::opt_string_id" + )] + pub data_set_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub source: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub labels: Option>, + /// Populated on graph reads and the create echo; always empty on the flat reads. + #[serde(default)] + pub related_resources: Vec, + /// GeoJSON geometry, as a nested object under `geoLocation`. + /// + /// A node reached through the graph has this reconstructed from Neo4j's native WGS-84 point, + /// which is lossy for anything that is not a `Point` — read the asset flatly if the geometry + /// matters. + #[serde( + rename = "geoLocation", + default, + skip_serializing_if = "Option::is_none" + )] + pub geolocation: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub created_time: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_updated_time: Option>, +} + +impl Asset { + pub fn new(external_id: &str, name: &str) -> Self { + Self { + id: None, + external_id: external_id.to_string(), + name: name.to_string(), + metadata: None, + description: None, + is_root: false, + data_set_id: None, + source: None, + labels: None, + related_resources: vec![], + geolocation: None, + created_time: None, + last_updated_time: None, + } + } +} + +impl GraphNode for Asset {} + +/// An access policy, as a node. +/// +/// Reads are sparse by construction: the api's policy transformer never sets `value`, +/// `template_id` or `data_set_id`, so those are `None` on anything that came back from the server +/// regardless of what is stored. +#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct Policy { + #[serde(default, with = "crate::serde_helper::opt_string_id")] + pub id: Option, + pub external_id: String, + pub name: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub description: Option, + /// The policy kind, e.g. `IS_WRITE_PROTECTED`. Named `type` on the wire. + #[serde(rename = "type", default, skip_serializing_if = "Option::is_none")] + pub policy_type: Option, + /// The policy's value. The api declares it `Object`, so it is any JSON scalar. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub value: Option, + /// Wire key is `deactivated`, not `isDeactivated` — the api's getter naming decides this. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub deactivated: Option, + #[serde( + default, + skip_serializing_if = "Option::is_none", + with = "crate::serde_helper::opt_string_id" + )] + pub template_id: Option, + #[serde( + default, + skip_serializing_if = "Option::is_none", + with = "crate::serde_helper::opt_string_id" + )] + pub data_set_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub source: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub metadata: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub labels: Option>, + #[serde(default)] + pub related_resources: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub created_time: Option>, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_updated_time: Option>, +} + +impl Policy { + pub fn new(external_id: &str, name: &str) -> Self { + Self { + id: None, + external_id: external_id.to_string(), + name: name.to_string(), + description: None, + policy_type: None, + value: None, + deactivated: None, + template_id: None, + data_set_id: None, + source: None, + metadata: None, + labels: None, + related_resources: vec![], + created_time: None, + last_updated_time: None, + } + } +} + +impl GraphNode for Policy {} + +/// One node of any type, as `/resources` returns them. +/// +/// `#[non_exhaustive]`: a seventh node type on the api is an additive change, so `match` on this +/// needs a `_` arm. Where you only want one field, prefer the accessors ([`Node::external_id`], +/// [`Node::labels`], …) over matching. +#[non_exhaustive] +#[derive(Debug, Clone, PartialEq)] +pub enum Node { + Asset(Asset), + TimeSeries(TimeSeries), + Function(Function), + Resource(Resource), + Dataset(Dataset), + Policy(Policy), +} + +macro_rules! on_node { + ($self:expr, $inner:ident => $body:expr) => { + match $self { + Node::Asset($inner) => $body, + Node::TimeSeries($inner) => $body, + Node::Function($inner) => $body, + Node::Resource($inner) => $body, + Node::Dataset($inner) => $body, + Node::Policy($inner) => $body, + } + }; +} + +impl Node { + /// Which of the six types this is. + pub fn kind(&self) -> NodeType { + match self { + Node::Asset(_) => NodeType::Asset, + Node::TimeSeries(_) => NodeType::TimeSeries, + Node::Function(_) => NodeType::Function, + Node::Resource(_) => NodeType::Resource, + Node::Dataset(_) => NodeType::Dataset, + Node::Policy(_) => NodeType::Policy, + } + } + + /// The server-assigned numeric id, absent on a node you built locally. + pub fn id(&self) -> Option { + on_node!(self, n => n.id) + } + + pub fn external_id(&self) -> &str { + on_node!(self, n => &n.external_id) + } + + /// The node's name. `None` only on a [`Function`], the one type whose name is optional. + pub fn name(&self) -> Option<&str> { + match self { + Node::Asset(n) => Some(&n.name), + Node::TimeSeries(n) => Some(&n.name), + Node::Function(n) => n.name.as_deref(), + Node::Resource(n) => Some(&n.name), + Node::Dataset(n) => Some(&n.name), + Node::Policy(n) => Some(&n.name), + } + } + + /// Every label on the node, including the intrinsic type-label. + pub fn labels(&self) -> &[String] { + match self { + Node::Asset(n) => n.labels.as_deref().unwrap_or(&[]), + Node::TimeSeries(n) => n.labels.as_deref().unwrap_or(&[]), + Node::Function(n) => &n.labels, + Node::Resource(n) => n.labels.as_deref().unwrap_or(&[]), + Node::Dataset(n) => n.labels.as_deref().unwrap_or(&[]), + Node::Policy(n) => n.labels.as_deref().unwrap_or(&[]), + } + } + + pub fn description(&self) -> Option<&str> { + on_node!(self, n => n.description.as_deref()) + } + + /// The name of the system this node's primary information comes from. + pub fn source(&self) -> Option<&str> { + on_node!(self, n => n.source.as_deref()) + } + + /// The node's metadata map. Empty rather than absent on a node read through the graph — that + /// path does not carry the column at all. + pub fn metadata(&self) -> Option<&HashMap> { + match self { + Node::Asset(n) => n.metadata.as_ref(), + Node::TimeSeries(n) => n.metadata.as_ref(), + Node::Function(n) => Some(&n.metadata), + Node::Resource(n) => n.metadata.as_ref(), + Node::Dataset(n) => Some(&n.metadata), + Node::Policy(n) => n.metadata.as_ref(), + } + } + + /// Whether this node is a root of the graph. `None` on the types that have no such column — + /// only assets and plain resources carry it. + pub fn is_root(&self) -> Option { + match self { + Node::Asset(n) => Some(n.is_root), + Node::Resource(n) => Some(n.is_root), + _ => None, + } + } + + pub fn data_set_id(&self) -> Option { + on_node!(self, n => n.data_set_id) + } + + /// The nodes this one is connected to. **Empty on the flat reads** — only graph reads and the + /// `/resources/create` echo populate it. + pub fn related_resources(&self) -> &[RelatedNode] { + on_node!(self, n => &n.related_resources) + } + + pub fn created_time(&self) -> Option> { + on_node!(self, n => n.created_time) + } + + pub fn last_updated_time(&self) -> Option> { + on_node!(self, n => n.last_updated_time) + } + + pub fn as_asset(&self) -> Option<&Asset> { + match self { + Node::Asset(n) => Some(n), + _ => None, + } + } + + pub fn as_time_series(&self) -> Option<&TimeSeries> { + match self { + Node::TimeSeries(n) => Some(n), + _ => None, + } + } + + pub fn as_function(&self) -> Option<&Function> { + match self { + Node::Function(n) => Some(n), + _ => None, + } + } + + pub fn as_resource(&self) -> Option<&Resource> { + match self { + Node::Resource(n) => Some(n), + _ => None, + } + } + + pub fn as_dataset(&self) -> Option<&Dataset> { + match self { + Node::Dataset(n) => Some(n), + _ => None, + } + } + + pub fn as_policy(&self) -> Option<&Policy> { + match self { + Node::Policy(n) => Some(n), + _ => None, + } + } + + pub fn into_asset(self) -> Option { + match self { + Node::Asset(n) => Some(n), + _ => None, + } + } + + pub fn into_time_series(self) -> Option { + match self { + Node::TimeSeries(n) => Some(n), + _ => None, + } + } + + pub fn into_function(self) -> Option { + match self { + Node::Function(n) => Some(n), + _ => None, + } + } + + pub fn into_resource(self) -> Option { + match self { + Node::Resource(n) => Some(n), + _ => None, + } + } + + pub fn into_dataset(self) -> Option { + match self { + Node::Dataset(n) => Some(n), + _ => None, + } + } + + pub fn into_policy(self) -> Option { + match self { + Node::Policy(n) => Some(n), + _ => None, + } + } +} + +impl From for Node { + fn from(v: Asset) -> Self { + Node::Asset(v) + } +} +impl From for Node { + fn from(v: TimeSeries) -> Self { + Node::TimeSeries(v) + } +} +impl From for Node { + fn from(v: Function) -> Self { + Node::Function(v) + } +} +impl From for Node { + fn from(v: Resource) -> Self { + Node::Resource(v) + } +} +impl From for Node { + fn from(v: Dataset) -> Self { + Node::Dataset(v) + } +} +impl From for Node { + fn from(v: Policy) -> Self { + Node::Policy(v) + } +} + +fn ensure_label_opt(labels: &mut Option>, label: &str) { + let entries = labels.get_or_insert_with(Vec::new); + if !entries.iter().any(|l| l.eq_ignore_ascii_case(label)) { + entries.push(label.to_string()); + } +} + +fn ensure_label(labels: &mut Vec, label: &str) { + if !labels.iter().any(|l| l.eq_ignore_ascii_case(label)) { + labels.push(label.to_string()); + } +} + +/// Serializes as the variant's own shape, with the type-label added to `labels` when the caller +/// has not already put it there. +/// +/// `/resources/create` dispatches each element of `nodes` by its own labels, so this is what makes +/// `create(vec![Dataset::new(…)])` build a data set. Two deliberate non-behaviours: +/// +/// - a conflicting type-label is **not** stripped. A body labelled both `ASSET` and `POLICY` earns +/// the api's 400 naming both, which is the accurate answer to an ambiguous intent; quietly +/// picking one for the caller is not. +/// - [`Node::Resource`] is passed through verbatim, so the long-standing idiom of creating a typed +/// node by putting its label on a bare [`Resource`] keeps working unchanged. +impl Serialize for Node { + fn serialize(&self, serializer: S) -> Result { + match self { + Node::Resource(r) => r.serialize(serializer), + Node::Asset(a) => { + let mut a = a.clone(); + ensure_label_opt(&mut a.labels, "ASSET"); + a.serialize(serializer) + } + Node::TimeSeries(t) => { + let mut t = t.clone(); + ensure_label_opt(&mut t.labels, "TIMESERIES"); + t.serialize(serializer) + } + Node::Function(f) => { + let mut f = f.clone(); + ensure_label(&mut f.labels, "FUNCTION"); + f.serialize(serializer) + } + Node::Dataset(d) => { + let mut d = d.clone(); + ensure_label_opt(&mut d.labels, "DATASET"); + d.serialize(serializer) + } + Node::Policy(p) => { + let mut p = p.clone(); + ensure_label_opt(&mut p.labels, "POLICY"); + p.serialize(serializer) + } + } + } +} + +/// Dispatches on the intrinsic type-label in `labels`, mirroring the api's `NodeModelDeserializer`. +/// +/// The buffer-then-dispatch shape is deliberate: serde has no mode for a tag that lives *inside an +/// array field*, so the tag has to be read before the body can be bound. `serde_json::Value` is +/// the right buffer here rather than serde's private `Content` (an unstable internal API) because +/// this SDK only ever speaks JSON — the whole deserialization path starts from a `&str`. +impl<'de> Deserialize<'de> for Node { + fn deserialize>(deserializer: D) -> Result { + use serde::de::Error; + + let value = serde_json::Value::deserialize(deserializer)?; + let labels: Vec<&str> = value + .get("labels") + .and_then(|l| l.as_array()) + .map(|entries| entries.iter().filter_map(|e| e.as_str()).collect()) + .unwrap_or_default(); + + let kind = NodeType::from_labels(&labels).map_err(D::Error::custom)?; + + macro_rules! bind { + ($variant:path) => { + serde_json::from_value(value) + .map($variant) + .map_err(D::Error::custom) + }; + } + + match kind { + NodeType::Asset => bind!(Node::Asset), + NodeType::TimeSeries => bind!(Node::TimeSeries), + NodeType::Function => bind!(Node::Function), + NodeType::Resource => bind!(Node::Resource), + NodeType::Dataset => bind!(Node::Dataset), + NodeType::Policy => bind!(Node::Policy), + } + } +} + +impl GraphNode for Node {} + +impl Identifiable for Node { + fn id(&self) -> u64 { + // The inherent `Node::id` returns `Option` and shadows this one at call sites; name + // it explicitly so this is not an infinite recursion. + Node::id(self).unwrap_or(0) + } + + fn external_id(&self) -> &str { + Node::external_id(self) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + /// The full response shape for a node of each type, transcribed from the api's wire-contract + /// tests (`AssetWireContractTest`, `ResourceWireContractTest`, `TimeseriesWireContractTest`, + /// `DataSetModelWireContractTest`, `PolicyWireContractTest`). Ids arrive as JSON *strings*; + /// `securityCategories` as raw numbers. + fn asset_json() -> serde_json::Value { + json!({ + "id": "34", "externalId": "pump_a", "name": "Pump A", "isRoot": true, + "geoLocation": { "type": "Point", "coordinates": [10.75, 59.91] }, + "relatedResources": [], "metadata": { "k": "v" }, "description": "d", + "dataSetId": "12", "source": "sap", "labels": ["ASSET", "PLANT"], + "createdTime": "2024-06-17T12:34:56Z", "lastUpdatedTime": "2024-06-17T12:34:56Z" + }) + } + + fn timeseries_json() -> serde_json::Value { + json!({ + "id": "7", "externalId": "Engine.Temp", "name": "Engine temp", + "metadata": {}, "unit": "deg C", "unitExternalId": "deg_c", + "relatedResources": [], "description": null, "securityCategories": [1, 2], + "dataSetId": "21", "source": null, "labels": ["TIMESERIES"], + "tableEngine": "MERGETREE", "valueType": "float", + "createdTime": "2024-06-17T12:34:56Z", "lastUpdatedTime": "2024-06-17T12:34:56Z" + }) + } + + fn dataset_json() -> serde_json::Value { + json!({ + "id": "3", "externalId": "raw_data", "name": "Raw data", "description": null, + "policies": ["policy_a"], "metadata": {}, "connectedDataSets": ["5"], + "labels": ["DATASET"], "relatedResources": [], "source": null, + "createdTime": "2024-06-17T12:34:56Z", "lastUpdatedTime": "2024-06-17T12:34:56Z" + }) + } + + fn policy_json() -> serde_json::Value { + json!({ + "id": "9", "name": "Write protect", "description": null, + "type": "IS_WRITE_PROTECTED", "value": "TRUE", "deactivated": false, + "externalId": "write_protect", "metadata": null, "templateId": "3", + "dataSetId": null, "source": null, "relatedResources": [], "labels": ["POLICY"], + "createdTime": "2024-06-17T12:34:56Z", "lastUpdatedTime": "2024-06-17T12:34:56Z" + }) + } + + fn function_json() -> serde_json::Value { + json!({ + "id": "11", "externalId": "nifi_ingest", "name": "Ingest", + "labels": ["FUNCTION"], "metadata": {}, "description": null, "source": null, + "relatedResources": [], + "createdTime": "2024-06-17T12:34:56Z", "lastUpdatedTime": "2024-06-17T12:34:56Z" + }) + } + + fn resource_json() -> serde_json::Value { + json!({ + "id": "1", "externalId": "plain", "name": "Plain", "isRoot": false, + "relatedResources": [], "metadata": {}, "description": "d", "dataSetId": "12", + "source": "sap", "labels": ["PIPE"], + "createdTime": "2024-06-17T12:34:56Z", "lastUpdatedTime": "2024-06-17T12:34:56Z" + }) + } + + #[test] + fn dispatches_on_the_intrinsic_type_label() { + let nodes: Vec = serde_json::from_value(json!([ + asset_json(), timeseries_json(), function_json(), + resource_json(), dataset_json(), policy_json() + ])) + .unwrap(); + + let kinds: Vec = nodes.iter().map(Node::kind).collect(); + assert_eq!( + kinds, + vec![ + NodeType::Asset, + NodeType::TimeSeries, + NodeType::Function, + NodeType::Resource, + NodeType::Dataset, + NodeType::Policy + ] + ); + } + + #[test] + fn a_node_with_no_type_label_is_a_resource() { + // `PIPE` is an ordinary domain label, so it names no type. + let node: Node = serde_json::from_value(resource_json()).unwrap(); + assert_eq!(node.kind(), NodeType::Resource); + assert_eq!(node.labels(), ["PIPE"]); + } + + #[test] + fn absent_labels_is_a_resource() { + let node: Node = + serde_json::from_value(json!({ "externalId": "x", "name": "X", "isRoot": false })) + .unwrap(); + assert_eq!(node.kind(), NodeType::Resource); + assert!(node.labels().is_empty()); + } + + #[test] + fn the_type_label_is_matched_case_insensitively() { + let mut v = dataset_json(); + v["labels"] = json!(["dataset"]); + let node: Node = serde_json::from_value(v).unwrap(); + assert_eq!(node.kind(), NodeType::Dataset); + } + + #[test] + fn two_type_labels_are_refused_rather_than_guessed() { + let mut v = asset_json(); + v["labels"] = json!(["ASSET", "POLICY"]); + let err = serde_json::from_value::(v).unwrap_err().to_string(); + assert!( + err.contains("at most one type-label"), + "unexpected error: {err}" + ); + } + + #[test] + fn an_asset_carries_its_geometry_and_root_flag() { + let asset = serde_json::from_value::(asset_json()) + .unwrap() + .into_asset() + .expect("asset"); + assert_eq!(asset.id, Some(34)); + assert_eq!(asset.data_set_id, Some(12)); + assert!(asset.is_root); + assert!(asset.geolocation.is_some()); + } + + #[test] + fn a_timeseries_read_through_resources_carries_its_own_fields() { + let ts = serde_json::from_value::(timeseries_json()) + .unwrap() + .into_time_series() + .expect("timeseries"); + assert_eq!(ts.unit.as_deref(), Some("deg C")); + assert_eq!(ts.unit_external_id.as_deref(), Some("deg_c")); + assert_eq!(ts.value_type.as_deref(), Some("float")); + assert_eq!(ts.table_engine.as_deref(), Some("MERGETREE")); + // Raw numbers on the wire, unlike every id in the family. + assert_eq!(ts.security_categories, Some(vec![1, 2])); + assert_eq!(ts.data_set_id, Some(21)); + assert_eq!(ts.labels.as_deref(), Some(&["TIMESERIES".to_string()][..])); + } + + #[test] + fn connected_data_sets_arrive_as_strings() { + let ds = serde_json::from_value::(dataset_json()) + .unwrap() + .into_dataset() + .expect("dataset"); + assert_eq!(ds.connected_data_sets, vec![5]); + assert_eq!(ds.policies, Some(vec!["policy_a".to_string()])); + } + + #[test] + fn a_policy_binds_the_deactivated_key_not_is_deactivated() { + let p = serde_json::from_value::(policy_json()) + .unwrap() + .into_policy() + .expect("policy"); + assert_eq!(p.deactivated, Some(false)); + assert_eq!(p.policy_type.as_deref(), Some("IS_WRITE_PROTECTED")); + assert_eq!(p.value, Some(json!("TRUE"))); + assert_eq!(p.template_id, Some(3)); + assert_eq!(p.data_set_id, None); + } + + #[test] + fn a_graph_sourced_timeseries_has_none_of_its_type_specific_fields() { + // What `/resources/fetch-related` actually sends for a TIMESERIES: the shared node + // fields and nothing else. Neo4j does not store the rest, and the api omits them rather + // than emitting defaults — so every one of these must be optional, `valueType` included. + // A required `valueType` made any graph traversal over a timeseries a hard error. + let node: Node = serde_json::from_value(json!({ + "id": "793", "externalId": "well_qgl", "name": "Well QGL", + "createdTime": "2024-06-17T12:34:56Z", "lastUpdatedTime": "2024-06-17T12:34:56Z", + "dataSetId": "731", "labels": ["TIMESERIES"], "metadata": {}, "relatedResources": [] + })) + .unwrap(); + let ts = node.into_time_series().expect("timeseries"); + assert_eq!(ts.value_type, None, "not told, rather than a wrong default"); + assert_eq!(ts.unit, None); + assert_eq!(ts.table_engine, None); + assert_eq!(ts.security_categories, None); + } + + #[test] + fn an_unknown_key_is_tolerated_on_a_read() { + // `/resources/update` echoes every node as a flat resource, so `isRoot` turns up on types + // that have no such field. A read must not fail on it. + let mut v = timeseries_json(); + v["isRoot"] = json!(false); + let node: Node = serde_json::from_value(v).unwrap(); + assert_eq!(node.kind(), NodeType::TimeSeries); + } + + #[test] + fn serializing_injects_the_type_label() { + for (node, expected) in [ + (Node::Asset(Asset::new("a", "A")), "ASSET"), + (Node::TimeSeries(TimeSeries::new("t", "T")), "TIMESERIES"), + (Node::Function(Function::new("f".into())), "FUNCTION"), + (Node::Dataset(Dataset::new("D".into())), "DATASET"), + (Node::Policy(Policy::new("p", "P")), "POLICY"), + ] { + let v = serde_json::to_value(&node).unwrap(); + let labels = v["labels"].as_array().expect("labels"); + assert!( + labels.iter().any(|l| l == expected), + "{expected} missing from {v}" + ); + } + } + + #[test] + fn serializing_a_resource_leaves_its_labels_alone() { + // Creating a typed node by putting its label on a bare `Resource` is a long-standing + // idiom; a plain resource has no type-label of its own to add. + let mut r = Resource::new(); + r.external_id = "a".into(); + r.labels = Some(vec!["ASSET".into()]); + let v = serde_json::to_value(Node::Resource(r)).unwrap(); + assert_eq!(v["labels"], json!(["ASSET"])); + } + + #[test] + fn serializing_never_emits_another_types_fields() { + // The api's request-body mapper rejects a body naming a field the target type does not + // have, so a node must serialize as its own shape and nothing wider. + let foreign = [ + "unit", + "unitExternalId", + "securityCategories", + "tableEngine", + "valueType", + "policies", + "connectedDataSets", + "type", + "value", + "deactivated", + "templateId", + "geoLocation", + "isRoot", + ]; + let own: HashMap<&str, &[&str]> = HashMap::from([ + ("ASSET", &["geoLocation", "isRoot"][..]), + ( + "TIMESERIES", + &["unit", "unitExternalId", "securityCategories", "tableEngine", "valueType"][..], + ), + ("FUNCTION", &[][..]), + ("DATASET", &["policies", "connectedDataSets"][..]), + ("POLICY", &["type", "value", "deactivated", "templateId"][..]), + ]); + + for node in [ + Node::Asset(Asset::new("a", "A")), + Node::TimeSeries(TimeSeries::new("t", "T")), + Node::Function(Function::new("f".into())), + Node::Dataset(Dataset::new("D".into())), + Node::Policy(Policy::new("p", "P")), + ] { + let label = node.kind().type_label().unwrap(); + let v = serde_json::to_value(&node).unwrap(); + let obj = v.as_object().unwrap(); + for key in foreign { + if own[label].contains(&key) { + continue; + } + assert!( + !obj.contains_key(key), + "a {label} body must not name `{key}`: {v}" + ); + } + } + } + + #[test] + fn every_typed_variant_round_trips_through_its_own_serialization() { + for node in [ + Node::Asset(Asset::new("a", "A")), + Node::TimeSeries(TimeSeries::new("t", "T")), + Node::Function(Function::new("f".into())), + Node::Dataset(Dataset::new("D".into())), + Node::Policy(Policy::new("p", "P")), + ] { + let kind = node.kind(); + let round_tripped: Node = + serde_json::from_value(serde_json::to_value(&node).unwrap()).unwrap(); + assert_eq!(round_tripped.kind(), kind); + } + } + + #[test] + fn a_data_wrapper_carries_a_mixed_page() { + use crate::generic::{DataWrapper, DataWrapperDeserialization}; + + let body = json!({ + "items": [asset_json(), timeseries_json(), dataset_json()], + "nextCursor": "abc" + }) + .to_string(); + let wrapper = DataWrapper::::deserialize_and_set_status(&body, 200).unwrap(); + assert_eq!(wrapper.length(), 3); + assert_eq!(wrapper.next_cursor(), Some("abc")); + assert_eq!(wrapper.get_items()[1].kind(), NodeType::TimeSeries); + } + + #[test] + fn a_graph_wrapper_carries_a_mixed_node_list_under_either_key() { + use crate::generic::DataWrapperDeserialization; + use crate::graph_data_wrapper::GraphDataWrapper; + + for key in ["nodes", "items"] { + let body = json!({ key: [policy_json(), function_json()] }).to_string(); + let wrapper = + GraphDataWrapper::::deserialize_and_set_status(&body, 200).unwrap(); + let nodes = wrapper.nodes.expect("nodes"); + assert_eq!(nodes[0].kind(), NodeType::Policy); + assert_eq!(nodes[1].kind(), NodeType::Function); + } + } + + #[test] + fn accessors_read_the_shared_node_fields_off_any_variant() { + let node: Node = serde_json::from_value(timeseries_json()).unwrap(); + assert_eq!(node.id(), Some(7)); + assert_eq!(node.external_id(), "Engine.Temp"); + assert_eq!(node.name(), Some("Engine temp")); + assert_eq!(node.data_set_id(), Some(21)); + assert!(node.created_time().is_some()); + assert!(node.related_resources().is_empty()); + } + + #[test] + fn the_identifiable_impl_does_not_shadow_itself_into_recursion() { + // `Identifiable::id` and the inherent `Node::id` share a name and differ in return type; + // the impl has to name the inherent one explicitly or it calls itself forever. + use crate::generic::Identifiable; + let node: Node = serde_json::from_value(asset_json()).unwrap(); + assert_eq!(Identifiable::id(&node), 34); + assert_eq!(Identifiable::external_id(&node), "pump_a"); + + let local = Node::Asset(Asset::new("x", "X")); + assert_eq!(Identifiable::id(&local), 0, "an unsaved node has no id"); + } + + #[test] + fn labels_canonicalize_the_way_the_api_does() { + assert_eq!(to_snake_upper_cased("dataset"), "DATASET"); + assert_eq!(to_snake_upper_cased("data-set"), "DATA_SET"); + assert_eq!(to_snake_upper_cased("data set"), "DATA_SET"); + assert_eq!(to_snake_upper_cased("12timeseries"), "TIMESERIES"); + assert_eq!(to_snake_upper_cased(" "), " "); + // "data set" is not a type-label, and must not be mistaken for one. + assert_eq!(NodeType::from_type_label("data set"), None); + assert_eq!(NodeType::from_type_label("policy"), Some(NodeType::Policy)); + } +} diff --git a/src/relations/service.rs b/src/relations/service.rs index 7182cc1..534cb16 100644 --- a/src/relations/service.rs +++ b/src/relations/service.rs @@ -9,7 +9,7 @@ use crate::generic::{ApiServiceProvider, DataWrapper, IdAndExtId}; use crate::graph_data_wrapper::GraphDataWrapper; use crate::http::ResponseError; use crate::relations::{EdgeProxy, RelForm, RelTypeForm, RelationshipType}; -use crate::resources::Resource; +use crate::nodes::Node; use crate::ApiService; use std::sync::Weak; @@ -43,13 +43,14 @@ impl EdgesService { /// `POST /edges/byids` — several relationships plus the resources they connect. /// - /// The response is a graph, not a list: `nodes()` holds the resources at both ends and + /// The response is a graph, not a list: `nodes()` holds the nodes at both ends and /// `relations()` the edges themselves, so no follow-up call is needed to resolve endpoints. + /// An edge can join any two node types, so the endpoints come back as [`Node`]s. /// /// Unlike [`get`](Self::get), this does **not** 404 on unmatched ids: a batch lookup answers /// 200 with empty `nodes` and `relations`, the same as every other `/byids` endpoint. "Absent /// from the result" is the only coherent answer when some of a batch exist and some do not. - pub async fn by_ids(&self, input: &I) -> Result, ResponseError> + pub async fn by_ids(&self, input: &I) -> Result, ResponseError> where for<'a> &'a I: Into>, { diff --git a/src/relations/tests.rs b/src/relations/tests.rs index 24a0765..86ac6b1 100644 --- a/src/relations/tests.rs +++ b/src/relations/tests.rs @@ -384,8 +384,8 @@ mod live { let id_of = |ext: &str| { nodes .iter() - .find(|n| n.external_id == ext) - .and_then(|n| n.id) + .find(|n| n.external_id() == ext) + .and_then(|n| n.id()) }; let edge_between = |from: &str, to: &str| { let (f, t) = (id_of(from), id_of(to)); @@ -412,8 +412,8 @@ mod live { assert_eq!(graph.relations().map(|r| r.len()), Some(1)); let resolved = graph.nodes().unwrap_or_default(); assert_eq!(resolved.len(), 2, "byids should resolve both endpoints"); - assert!(resolved.iter().any(|n| n.external_id == a)); - assert!(resolved.iter().any(|n| n.external_id == c)); + assert!(resolved.iter().any(|n| n.external_id() == a)); + assert!(resolved.iter().any(|n| n.external_id() == c)); await_graph(&api, a, "all three links visible", |n| n.edges().len() >= 3).await; diff --git a/src/resources/mod.rs b/src/resources/mod.rs index ddcde7c..3ec7d69 100644 --- a/src/resources/mod.rs +++ b/src/resources/mod.rs @@ -9,6 +9,7 @@ use crate::generic::{ SearchAndFilterForm, }; use crate::graph_data_wrapper::{GraphDataWrapper, GraphNode}; +use crate::nodes::Node; use crate::http::{process_response, ResponseError}; use crate::relations::{EdgeProxy, RelForm, RelatedNode}; use crate::ApiService; @@ -36,29 +37,52 @@ impl ResourceService { } } - /// Create resources, optionally with relations between them. Mirrors Java's - /// `POST /resources/create` body shape `GraphDataWrapper`; - /// the response is the graph in its post-create form, with each relation - /// returned as an `EdgeProxy` carrying the server-assigned id. Pass an - /// empty `Vec` for `relations` to create nodes only. - pub async fn create( + /// Create nodes of any type, optionally with relations between them + /// (`POST /resources/create`). The response is the graph in its post-create form: each node + /// typed, with `related_resources` populated from the relations created in the same call, and + /// each relation returned as an `EdgeProxy` carrying the server-assigned id. Pass an empty + /// `Vec` for `relations` to create nodes only. + /// + /// Each node is dispatched server-side by its own type-label, so + /// `create(vec![Dataset::new(..)], vec![])` builds a data set and + /// `create(vec![Asset::new(..)], vec![])` an asset. A bare [`Resource`] with the label set by + /// hand works the same way. Every node type is creatable here, timeseries included — datapoint + /// ingestion stays on `/timeseries`. + /// + /// Two server-side behaviours worth knowing: + /// + /// - **`DATASET` and `POLICY` nodes need the all-datasets manage grant**, and are a 403 + /// without it. + /// - **`data_set_id` is silently dropped on those two types.** A data set or policy that + /// belonged to a data set would orphan its own ACL grant, so the api refuses to set it + /// rather than erroring. + /// + /// Note also that a duplicate `external_id` surfaces here as a constraint violation rather + /// than the clean 409 `/timeseries/create` answers with — a known asymmetry in the api. + pub async fn create>( &self, - nodes: Vec, + nodes: Vec, relations: Vec, - ) -> Result, ResponseError> { - let payload: GraphDataWrapper = + ) -> Result, ResponseError> { + let nodes: Vec = nodes.into_iter().map(Into::into).collect(); + let payload: GraphDataWrapper = GraphDataWrapper::with_relations(nodes, relations); let url = &format!("{}/create", self.base_url); - self.execute_post_request::, _>(&url, &payload) + self.execute_post_request::, _>(&url, &payload) .await } - pub async fn by_ids(&self, input: &I) -> Result, ResponseError> + /// Look nodes up by id or external id (`POST /resources/byids`). Like every batch lookup, + /// this answers 200 with the found subset and silently omits what is missing. + /// + /// Each node comes back as its own type. `related_resources` is empty on this path — the api + /// does not join the edges in for a flat read. + pub async fn by_ids(&self, input: &I) -> Result, ResponseError> where for<'a> &'a I: Into>, { let payload = input.into(); let url = &format!("{}/byids", self.base_url); - self.execute_post_request::, _>(&url, &payload) + self.execute_post_request::, _>(&url, &payload) .await } @@ -75,9 +99,9 @@ impl ResourceService { pub async fn search( &self, payload: &SearchAndFilterForm, - ) -> Result, ResponseError> { + ) -> Result, ResponseError> { let url = &format!("{}/search", self.base_url); - self.execute_post_request::, _>(&url, &payload) + self.execute_post_request::, _>(&url, &payload) .await } @@ -97,10 +121,17 @@ impl ResourceService { self.execute_post_request::(&url, &form) .await } - /// Update resources in place (`POST /resources/update`). Each [`ResourceUpdate`] targets one - /// resource by id or external id and carries only the fields to change (PATCH semantics). The - /// server returns the resources after the update, so the returned `labels` reflect what the - /// backend actually stored — including the intrinsic type-label it always forces back. + /// Update nodes in place (`POST /resources/update`). Each [`ResourceUpdate`] targets one node + /// by id or external id and carries only the fields to change (PATCH semantics). Every field + /// it can set is a shared node field, so one update form covers all six node types — except + /// `geolocation`, which only an asset stores. + /// + /// **The echo is flat.** Unlike every read on this service, the api answers here with each + /// node serialized as a plain [`Resource`], whatever its type — so a timeseries updated + /// through this endpoint comes back without its `unit`, though the same node reads back as a + /// [`Node::TimeSeries`] from [`get_by_id`](Self::get_by_id). Re-read the node if you need its + /// typed form. The returned `labels` do reflect what the backend stored, including the + /// intrinsic type-label it always forces back. pub async fn update(&self, input: &I) -> Result, ResponseError> where for<'a> &'a I: Into>, @@ -119,9 +150,9 @@ impl ResourceService { /// /// Unlike [`by_ids`](Self::by_ids), which omits what it cannot find, this is a 404 when the /// resource does not exist. - pub async fn get_by_id(&self, id: u64) -> Result, ResponseError> { + pub async fn get_by_id(&self, id: u64) -> Result, ResponseError> { let url = &format!("{}/{}", self.base_url, id); - self.execute_get_request::, ()>(url, None) + self.execute_get_request::, ()>(url, None) .await } @@ -133,9 +164,9 @@ impl ResourceService { pub async fn filter( &self, form: &ResourceFilterForm, - ) -> Result, ResponseError> { + ) -> Result, ResponseError> { let url = &format!("{}/filter", self.base_url); - self.execute_post_request::, _>(url, form) + self.execute_post_request::, _>(url, form) .await } @@ -182,6 +213,10 @@ pub struct Resource { /// `{"type":"Point","coordinates":[10.75,59.91]}`. Build one with /// [`geojson::Geometry::new_point`] and friends (re-exported as /// [`crate::Geometry`]). + /// + /// **Write-only on a plain resource.** The api accepts it on create/update but never echoes + /// it back on a `Resource`, so this is always `None` on a read. A node created with the + /// `ASSET` type-label comes back as [`crate::nodes::Asset`], which does carry it. #[serde(rename = "geoLocation", skip_serializing_if = "Option::is_none")] pub geolocation: Option, #[serde(skip_serializing_if = "Option::is_none")] @@ -405,7 +440,7 @@ pub struct Label { #[serde(rename_all = "camelCase")] pub struct ResourceNetwork { #[serde(default)] - pub nodes: Vec, + pub nodes: Vec, #[serde(default)] pub edges: Vec, #[serde(default)] @@ -413,7 +448,7 @@ pub struct ResourceNetwork { } impl ResourceNetwork { - pub fn nodes(&self) -> &Vec { + pub fn nodes(&self) -> &Vec { &self.nodes } pub fn edges(&self) -> &Vec { diff --git a/src/resources/tests.rs b/src/resources/tests.rs index 6b0e6f6..9be7278 100644 --- a/src/resources/tests.rs +++ b/src/resources/tests.rs @@ -10,6 +10,7 @@ use maplit::hashmap; use uuid::Uuid; use crate::tests::ids::unique_id; use crate::tests::polling::poll_until; +use crate::nodes::NodeType; fn create_test_resources() -> Vec { // helper function to create test resources will @@ -83,7 +84,7 @@ async fn test_create_and_delete_resources() -> Result<(), ResponseError> { .nodes() .unwrap() .iter() - .map(|r| to_snake_lower_cased_allow_start_with_digits(&r.external_id)) + .map(|r| to_snake_lower_cased_allow_start_with_digits(r.external_id())) .collect::>(); let input_ids = test_resources .iter() @@ -134,9 +135,9 @@ async fn test_search_resources() -> Result<(), ResponseError> { assert!(search_result.get_items().iter().all(|r| { let haystack = format!( "{} {} {}", - r.name, - r.external_id, - r.description.as_deref().unwrap_or("") + r.name().unwrap_or(""), + r.external_id(), + r.description().unwrap_or("") ) .to_lowercase(); haystack.contains("test") @@ -145,7 +146,7 @@ async fn test_search_resources() -> Result<(), ResponseError> { .nodes() .unwrap() .iter() - .map(|r| IdAndExtId::from_external_id(&r.external_id)) + .map(|r| IdAndExtId::from_external_id(r.external_id())) .collect::>(); api_service.resources.delete(&resulting_ids).await?; resource_cleanup.disarm(); // explicit delete succeeded; skip the drop teardown @@ -304,7 +305,7 @@ async fn neo4j_persists_expected_fields_per_node_type() -> Result<(), Box::new(), vec![RelForm::by_external_ids(&asset_ext, &func_ext, "uses")], ) .await?; @@ -314,22 +315,32 @@ async fn neo4j_persists_expected_fields_per_node_type() -> Result<(), Box Resource { + let find = |ext: &str| -> Node { net.nodes() .iter() - .find(|n| n.external_id == ext) + .find(|n| n.external_id() == ext) .unwrap_or_else(|| panic!("node {ext} not found in network after propagation")) .clone() }; - // --- asset / resource node --- - let a = find(&asset_ext); + // Each node comes back as its own type, discriminated by its intrinsic type-label. + assert_eq!( + ( + find(&asset_ext).kind(), + find(&ts_ext).kind(), + find(&func_ext).kind() + ), + (NodeType::Asset, NodeType::TimeSeries, NodeType::Function) + ); + + // --- asset node --- + let a = find(&asset_ext).into_asset().expect("asset variant"); assert_eq!(a.name, "Neo Fields Asset"); assert_eq!(a.description.as_deref(), Some("asset description")); assert!(a.is_root, "asset isRoot should persist as true"); @@ -365,14 +376,30 @@ async fn neo4j_persists_expected_fields_per_node_type() -> Result<(), Box Result<(), Box Result<(), ResponseError> { let api = create_api_service(); @@ -555,12 +581,14 @@ async fn test_resource_geolocation_round_trips() -> Result<(), ResponseError> { .map(|dw| dw.nodes().unwrap_or_default()) .unwrap_or_default() .into_iter() - .find(|r| r.external_id == ext) + .find(|r| r.external_id() == ext) }, - |found: &Option| found.is_some(), + |found: &Option| found.is_some(), ) .await - .expect("resource should be readable via by_ids after create"); + .expect("resource should be readable via by_ids after create") + .into_asset() + .expect("an ASSET-labelled node reads back as the asset variant"); let geom = fetched .geolocation diff --git a/src/serde_helper.rs b/src/serde_helper.rs index af63ab8..94712de 100644 --- a/src/serde_helper.rs +++ b/src/serde_helper.rs @@ -107,6 +107,41 @@ pub mod opt_string_id_vec { } } +/// `Vec` of ids as JSON strings (accepts strings or numbers on input). +/// +/// The non-optional form of [`opt_string_id_vec`], for fields the backend always sends as an +/// array — `DataSetModel.connectedDataSets` is declared `List` and serialized with +/// `ToStringSerializer`, so the wire carries `["5"]` where the Rust type says `u64`. +pub mod string_id_vec { + use super::*; + + pub fn serialize(value: &Vec, s: S) -> Result { + use serde::ser::SerializeSeq; + let mut seq = s.serialize_seq(Some(value.len()))?; + for id in value { + seq.serialize_element(&id.to_string())?; + } + seq.end() + } + + pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result, D::Error> { + #[derive(Deserialize)] + #[serde(untagged)] + enum StringOrNumber { + Str(String), + Num(u64), + } + Option::>::deserialize(d)? + .unwrap_or_default() + .into_iter() + .map(|i| match i { + StringOrNumber::Str(s) => s.parse().map_err(serde::de::Error::custom), + StringOrNumber::Num(n) => Ok(n), + }) + .collect() + } +} + #[cfg(test)] mod tests { use serde::{Deserialize, Serialize}; diff --git a/src/timeseries/mod.rs b/src/timeseries/mod.rs index 278b13f..652c7b6 100644 --- a/src/timeseries/mod.rs +++ b/src/timeseries/mod.rs @@ -568,7 +568,7 @@ impl TimeSeriesFilterForm { } } -#[derive(Debug, Serialize, Deserialize, Clone)] +#[derive(Debug, Serialize, Deserialize, Clone, PartialEq)] pub struct TimeSeries { #[serde(default, with = "crate::serde_helper::opt_string_id")] pub id: Option, @@ -585,8 +585,18 @@ pub struct TimeSeries { #[serde(rename = "dataSetId")] #[serde(default, with = "crate::serde_helper::opt_string_id")] pub data_set_id: Option, - #[serde(rename = "valueType")] - pub value_type: String, + /// 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. + #[serde( + rename = "valueType", + default, + skip_serializing_if = "Option::is_none" + )] + pub value_type: Option, /// The name of the system this series' primary information comes from. Shared by every node /// type — it is the `source` column of the one `node` table — and answered on every /// timeseries response. @@ -601,6 +611,18 @@ pub struct TimeSeries { /// turned into an edge server-side. #[serde(rename = "relatedResources", default)] pub related_resources: Vec, + /// The labels carried by this node, always including the intrinsic `TIMESERIES` type-label + /// the api forces back on every read. It is what makes a timeseries recognisable in a + /// heterogeneous `/resources` result — see [`crate::nodes::Node`]. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub labels: Option>, + /// ClickHouse table engine backing this series (`MERGETREE` by default). Server-assigned; + /// sent back on every timeseries read. + /// + /// Beware on a node reached through the graph (`fetch_related`/`fetch_nearest`): Neo4j does + /// not store this column, so the api fills it from its DTO default rather than from data. + #[serde(rename = "tableEngine", default, skip_serializing_if = "Option::is_none")] + pub table_engine: Option, } impl TimeSeries { @@ -615,11 +637,13 @@ impl TimeSeries { unit_external_id: None, security_categories: None, data_set_id: None, - value_type: "float".to_string(), + value_type: Some("float".to_string()), source: None, created_time: None, last_updated_time: None, related_resources: vec![], + labels: None, + table_engine: None, } } pub fn from_dict(dict: HashMap) -> Self { @@ -637,11 +661,13 @@ impl TimeSeries { .get("securityCategories") .map(|v| serde_json::from_str(v).unwrap()), data_set_id: dict.get("dataSetId").map(|v| v.parse::().unwrap()), - value_type: dict.get("valueType").unwrap().to_string(), + value_type: dict.get("valueType").map(|v| v.to_string()), source: dict.get("source").map(|v| v.to_string()), created_time: None, last_updated_time: None, related_resources: vec![], + labels: None, + table_engine: None, } } @@ -690,7 +716,7 @@ impl TimeSeries { } pub fn set_value_type(&mut self, value_type: &str) -> &mut TimeSeries { - self.value_type = value_type.to_string(); + self.value_type = Some(value_type.to_string()); self }