| Name | industrial-model JSON |
| Version |
1.0.1
JSON |
| download |
| home_page | None |
| Summary | Industrial Model ORM |
| upload_time | 2025-07-17 21:15:12 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.11 |
| license | None |
| keywords |
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# 📦 industrial-model
`industrial-model` is a Python ORM-style abstraction for querying views in Cognite Data Fusion (CDF). It provides a declarative and type-safe way to model CDF views using `pydantic`, build queries, and interact with the CDF API in a Pythonic fashion.
---
## ✨ Features
- Define CDF views using Pydantic-style classes.
- Build complex queries using fluent and composable filters.
- Easily fetch data using standard or paginated query execution.
- Automatic alias and field transformation support.
---
## 📦 Installation
```bash
pip install industrial-model
```
---
# Usage Example
This section shows how to interact with **Cognite Data Fusion (CDF)** using the `industrial_model` package.
We use the simplified version of the `CogniteAsset`view in the `CogniteCore` data model (version `v1`) as a sample for all the examples below.
---
```graphql
type CogniteAsset {
name: String
description: String
tags: [String]
parent: CogniteAsset
root: CogniteAsset
}
```
---
## 🚀 Getting Started
### 1. Define Your Model (You only need to add the properties that you want to retrieve)
```python
from industrial_model import ViewInstance
class CogniteAsset(ViewInstance):
name: str
description: str
aliases: list[str]
```
### 2. Create the Engine
#### Option A: From Configuration File
Create a `cognite-sdk-config.yaml` file with your credentials and model configuration:
```yaml
cognite:
project: "${CDF_PROJECT}"
client_name: "${CDF_CLIENT_NAME}"
base_url: "https://${CDF_CLUSTER}.cognitedata.com"
credentials:
client_credentials:
token_url: "${CDF_TOKEN_URL}"
client_id: "${CDF_CLIENT_ID}"
client_secret: "${CDF_CLIENT_SECRET}"
scopes: ["https://${CDF_CLUSTER}.cognitedata.com/.default"]
data_model:
external_id: "CogniteCore"
space: "cdf_cdm"
version: "v1"
```
```python
from industrial_model import Engine
from pathlib import Path
engine = Engine.from_config_file(Path("cognite-sdk-config.yaml"))
```
#### Option B: Manually
```python
from cognite.client import CogniteClient
from industrial_model import Engine, DataModelId
engine = Engine(
cognite_client=CogniteClient(), # you need to create a valid cognite client
data_model_id=DataModelId(external_id="CogniteCore", space="cdf_cdm", version="v1")
)
```
---
## 🔎 Querying Assets by Alias
```python
from industrial_model import select, col
statement = (
select(CogniteAsset)
.where(col(CogniteAsset.aliases).contains_any_(["my_alias"]))
.limit(1000)
)
results = engine.query_all_pages(statement)
```
---
## 🔗 Filtering by Parent Name
```python
class CogniteAsset(ViewInstance):
name: str
description: str
aliases: list[str]
parent: CogniteAsset | None = None
```
```python
statement = (
select(CogniteAsset)
.where(
col(CogniteAsset.aliases).contains_any_(["my_alias"]) &
col(CogniteAsset.parent).nested_(col(CogniteAsset.name) == "Parent Asset Name")
)
)
results = engine.query(statement)
```
---
## 🔗 Filtering by Parent Name with bool operators
```python
from industrial_model import select, col, or_, and_
statement = select(CogniteAsset).where(
and_(
col(CogniteAsset.aliases).contains_any_(["my_alias"]),
or_(
col(CogniteAsset.parent).nested_(
col(CogniteAsset.name) == "Parent Asset Name 1"
),
col(CogniteAsset.parent).nested_(
col(CogniteAsset.name) == "Parent Asset Name 2"
),
),
)
)
results = engine.query(statement)
```
---
## 🔗 Paginating with cursor and sort by name
```python
class CogniteAsset(ViewInstance):
name: str
description: str
aliases: list[str]
parent: CogniteAsset | None = None
```
```python
statement = select(CogniteAsset).asc(CogniteAsset.name).cursor("NEXT_CURSOR")
results = engine.query(statement)
```
---
## 🔗 Proving an alias for a property
```python
from pydantic import Field
class CogniteAsset(ViewInstance):
another_name: str = Field(alias="name")
```
---
## 🎯 Optimize Query with View Config - The spaces will be appended in every query
```python
from industrial_model import ViewInstanceConfig
class CogniteAsset(ViewInstance):
view_config = ViewInstanceConfig(
view_external_id="CogniteAsset", # Maps this class to the 'CogniteAsset' view
instance_spaces_prefix="Industr-", # Filters queries to spaces with this prefix
instance_spaces=[
"Industrial-Data"
], # Alternatively, explicitly filter by these spaces
)
name: str
description: str
aliases: list[str]
parent: CogniteAsset | None = None
```
---
## 🔍 Search by Fuzzy Name
```python
from industrial_model import search
search_statement = (
search(CogniteAsset)
.where(col(CogniteAsset.aliases).contains_any_(["my_alias"]))
.query_by("my fuzzy name", [CogniteAsset.name])
)
search_result = engine.search(search_statement)
```
---
## 📊 Aggregating Data
```python
from industrial_model import aggregate, AggregatedViewInstance
class CogniteAssetByName(AggregatedViewInstance):
view_config = ViewInstanceConfig(view_external_id="CogniteAsset")
name: str
aggregate_statement = aggregate(CogniteAssetByName, "count").group_by(
col(CogniteAssetByName.name)
)
aggregate_result = engine.aggregate(aggregate_statement)
```
---
## 🗑️ Deleting Instances
```python
instances_to_delete = engine.search(
search(CogniteAsset)
.where(col(CogniteAsset.aliases).contains_any_(["my_alias"]))
.query_by("my fuzzy name", [CogniteAsset.name])
)
engine.delete(instances_to_delete)
```
---
## ✏️ Upserting Instances
```python
from industrial_model import WritableViewInstance, InstanceId
class CogniteAsset(WritableViewInstance):
view_config = ViewInstanceConfig(view_external_id="CogniteAsset")
name: str
aliases: list[str]
def edge_id_factory(self, target_node: InstanceId, edge_type: InstanceId) -> InstanceId:
return InstanceId(
external_id=f"{self.external_id}-{target_node.external_id}-{edge_type.external_id}",
space=self.space,
)
```
```python
instances = engine.query_all_pages(
select(CogniteAsset).where(col(CogniteAsset.aliases).contains_any_(["my_alias"]))
)
for instance in instances:
instance.aliases.append("new_alias")
engine.upsert(instances, replace=False)
```
---
## ✏️ Async version
All methods have a async equivalent version
```python
await engine.query_async(...)
await engine.query_all_pages_async(...)
await engine.search_async(...)
await engine.aggregate_async(...)
await engine.delete_async(...)
await engine.upsert_async(...)
```
---
Raw data
{
"_id": null,
"home_page": null,
"name": "industrial-model",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": null,
"author": null,
"author_email": "Lucas Alves <lucasrosaalves@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/9f/f0/b88375970956f82ae578da0d07959e48a430c57fc3fe9cd549320c193f94/industrial_model-1.0.1.tar.gz",
"platform": null,
"description": "# \ud83d\udce6 industrial-model\n\n`industrial-model` is a Python ORM-style abstraction for querying views in Cognite Data Fusion (CDF). It provides a declarative and type-safe way to model CDF views using `pydantic`, build queries, and interact with the CDF API in a Pythonic fashion.\n\n---\n\n## \u2728 Features\n\n- Define CDF views using Pydantic-style classes.\n- Build complex queries using fluent and composable filters.\n- Easily fetch data using standard or paginated query execution.\n- Automatic alias and field transformation support.\n\n---\n\n## \ud83d\udce6 Installation\n\n```bash\npip install industrial-model\n```\n\n---\n\n# Usage Example\n\nThis section shows how to interact with **Cognite Data Fusion (CDF)** using the `industrial_model` package. \nWe use the simplified version of the `CogniteAsset`view in the `CogniteCore` data model (version `v1`) as a sample for all the examples below.\n\n---\n\n```graphql\ntype CogniteAsset {\n name: String\n description: String\n tags: [String]\n parent: CogniteAsset\n root: CogniteAsset\n}\n```\n\n---\n\n## \ud83d\ude80 Getting Started\n\n### 1. Define Your Model (You only need to add the properties that you want to retrieve)\n\n```python\nfrom industrial_model import ViewInstance\n\nclass CogniteAsset(ViewInstance):\n name: str\n description: str\n aliases: list[str]\n```\n\n### 2. Create the Engine\n\n#### Option A: From Configuration File\n\nCreate a `cognite-sdk-config.yaml` file with your credentials and model configuration:\n\n```yaml\ncognite:\n project: \"${CDF_PROJECT}\"\n client_name: \"${CDF_CLIENT_NAME}\"\n base_url: \"https://${CDF_CLUSTER}.cognitedata.com\"\n credentials:\n client_credentials:\n token_url: \"${CDF_TOKEN_URL}\"\n client_id: \"${CDF_CLIENT_ID}\"\n client_secret: \"${CDF_CLIENT_SECRET}\"\n scopes: [\"https://${CDF_CLUSTER}.cognitedata.com/.default\"]\n\ndata_model:\n external_id: \"CogniteCore\"\n space: \"cdf_cdm\"\n version: \"v1\"\n```\n\n```python\nfrom industrial_model import Engine\nfrom pathlib import Path\n\nengine = Engine.from_config_file(Path(\"cognite-sdk-config.yaml\"))\n```\n\n#### Option B: Manually\n\n```python\nfrom cognite.client import CogniteClient\nfrom industrial_model import Engine, DataModelId\n\nengine = Engine(\n cognite_client=CogniteClient(), # you need to create a valid cognite client\n data_model_id=DataModelId(external_id=\"CogniteCore\", space=\"cdf_cdm\", version=\"v1\")\n)\n```\n\n---\n\n## \ud83d\udd0e Querying Assets by Alias\n\n```python\nfrom industrial_model import select, col\n\nstatement = (\n select(CogniteAsset)\n .where(col(CogniteAsset.aliases).contains_any_([\"my_alias\"]))\n .limit(1000)\n)\n\nresults = engine.query_all_pages(statement)\n```\n\n---\n\n## \ud83d\udd17 Filtering by Parent Name\n\n```python\nclass CogniteAsset(ViewInstance):\n name: str\n description: str\n aliases: list[str]\n parent: CogniteAsset | None = None\n```\n\n```python\nstatement = (\n select(CogniteAsset)\n .where(\n col(CogniteAsset.aliases).contains_any_([\"my_alias\"]) &\n col(CogniteAsset.parent).nested_(col(CogniteAsset.name) == \"Parent Asset Name\")\n )\n)\n\nresults = engine.query(statement)\n```\n\n---\n\n## \ud83d\udd17 Filtering by Parent Name with bool operators\n\n```python\nfrom industrial_model import select, col, or_, and_\n\nstatement = select(CogniteAsset).where(\n and_(\n col(CogniteAsset.aliases).contains_any_([\"my_alias\"]),\n or_(\n col(CogniteAsset.parent).nested_(\n col(CogniteAsset.name) == \"Parent Asset Name 1\"\n ),\n col(CogniteAsset.parent).nested_(\n col(CogniteAsset.name) == \"Parent Asset Name 2\"\n ),\n ),\n )\n)\n\nresults = engine.query(statement)\n```\n\n---\n\n## \ud83d\udd17 Paginating with cursor and sort by name\n\n```python\nclass CogniteAsset(ViewInstance):\n name: str\n description: str\n aliases: list[str]\n parent: CogniteAsset | None = None\n```\n\n```python\nstatement = select(CogniteAsset).asc(CogniteAsset.name).cursor(\"NEXT_CURSOR\")\n\nresults = engine.query(statement)\n```\n\n---\n\n## \ud83d\udd17 Proving an alias for a property\n\n```python\nfrom pydantic import Field\n\nclass CogniteAsset(ViewInstance):\n another_name: str = Field(alias=\"name\")\n```\n\n---\n\n## \ud83c\udfaf Optimize Query with View Config - The spaces will be appended in every query\n\n```python\nfrom industrial_model import ViewInstanceConfig\n\nclass CogniteAsset(ViewInstance):\n view_config = ViewInstanceConfig(\n view_external_id=\"CogniteAsset\", # Maps this class to the 'CogniteAsset' view\n instance_spaces_prefix=\"Industr-\", # Filters queries to spaces with this prefix\n instance_spaces=[\n \"Industrial-Data\"\n ], # Alternatively, explicitly filter by these spaces\n )\n name: str\n description: str\n aliases: list[str]\n parent: CogniteAsset | None = None\n```\n\n---\n\n## \ud83d\udd0d Search by Fuzzy Name\n\n```python\nfrom industrial_model import search\n\nsearch_statement = (\n search(CogniteAsset)\n .where(col(CogniteAsset.aliases).contains_any_([\"my_alias\"]))\n .query_by(\"my fuzzy name\", [CogniteAsset.name])\n)\n\nsearch_result = engine.search(search_statement)\n```\n\n---\n\n## \ud83d\udcca Aggregating Data\n\n```python\nfrom industrial_model import aggregate, AggregatedViewInstance\n\nclass CogniteAssetByName(AggregatedViewInstance):\n view_config = ViewInstanceConfig(view_external_id=\"CogniteAsset\")\n name: str\n\naggregate_statement = aggregate(CogniteAssetByName, \"count\").group_by(\n col(CogniteAssetByName.name)\n)\n\naggregate_result = engine.aggregate(aggregate_statement)\n```\n\n---\n\n## \ud83d\uddd1\ufe0f Deleting Instances\n\n```python\ninstances_to_delete = engine.search(\n search(CogniteAsset)\n .where(col(CogniteAsset.aliases).contains_any_([\"my_alias\"]))\n .query_by(\"my fuzzy name\", [CogniteAsset.name])\n)\n\nengine.delete(instances_to_delete)\n```\n\n---\n\n## \u270f\ufe0f Upserting Instances\n\n```python\nfrom industrial_model import WritableViewInstance, InstanceId\n\nclass CogniteAsset(WritableViewInstance):\n view_config = ViewInstanceConfig(view_external_id=\"CogniteAsset\")\n name: str\n aliases: list[str]\n\n def edge_id_factory(self, target_node: InstanceId, edge_type: InstanceId) -> InstanceId:\n return InstanceId(\n external_id=f\"{self.external_id}-{target_node.external_id}-{edge_type.external_id}\",\n space=self.space,\n )\n```\n\n```python\ninstances = engine.query_all_pages(\n select(CogniteAsset).where(col(CogniteAsset.aliases).contains_any_([\"my_alias\"]))\n)\n\nfor instance in instances:\n instance.aliases.append(\"new_alias\")\n\nengine.upsert(instances, replace=False)\n```\n\n---\n\n## \u270f\ufe0f Async version\n\nAll methods have a async equivalent version\n\n```python\nawait engine.query_async(...)\nawait engine.query_all_pages_async(...)\nawait engine.search_async(...)\nawait engine.aggregate_async(...)\nawait engine.delete_async(...)\nawait engine.upsert_async(...)\n```\n\n---\n",
"bugtrack_url": null,
"license": null,
"summary": "Industrial Model ORM",
"version": "1.0.1",
"project_urls": {
"Homepage": "https://github.com/lucasrosaalves/industrial-model",
"Source": "https://github.com/lucasrosaalves/industrial-model"
},
"split_keywords": [],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "1b61a6ff59ba052ef707872559866c107b648d5ccc4944264223836201ca2550",
"md5": "0bb0020b7a3a21e7a05dacb4dac7c077",
"sha256": "dc4f34683b89345702c3881336178fb1140de1d917fdf5d633460edb2abbfbcf"
},
"downloads": -1,
"filename": "industrial_model-1.0.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "0bb0020b7a3a21e7a05dacb4dac7c077",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 35227,
"upload_time": "2025-07-17T21:15:10",
"upload_time_iso_8601": "2025-07-17T21:15:10.954014Z",
"url": "https://files.pythonhosted.org/packages/1b/61/a6ff59ba052ef707872559866c107b648d5ccc4944264223836201ca2550/industrial_model-1.0.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "9ff0b88375970956f82ae578da0d07959e48a430c57fc3fe9cd549320c193f94",
"md5": "734990c36539e61da22738547119b68f",
"sha256": "4f82883cd619a11ceb9fa91e882aec1c985e7db76feaece0435ee58c356d4dff"
},
"downloads": -1,
"filename": "industrial_model-1.0.1.tar.gz",
"has_sig": false,
"md5_digest": "734990c36539e61da22738547119b68f",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 23907,
"upload_time": "2025-07-17T21:15:12",
"upload_time_iso_8601": "2025-07-17T21:15:12.230193Z",
"url": "https://files.pythonhosted.org/packages/9f/f0/b88375970956f82ae578da0d07959e48a430c57fc3fe9cd549320c193f94/industrial_model-1.0.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-17 21:15:12",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "lucasrosaalves",
"github_project": "industrial-model",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "industrial-model"
}
