Introducing Industrial Model - Type-safe, Pythonic ORM

Hi everyone,

I’d like to share industrial-model — a Python ORM built on top of the Cognite Data Modeling Service API.
It lets you define DMS views using Pydantic models, query them with a fluent, expressive API (filter, search, aggregate), and get fully typed results with IDE autocomplete out of the box.

The SDK also supports upserts, deletes, and async workflows, making it a natural fit for modern Python stacks. In addition, it includes configuration for injecting instance spaces directly into queries, which can significantly improve query performance when working with large or complex data models.

We’ve been using industrial-model for the past couple of months, and it has significantly reduced boilerplate when querying complex graphs, while also speeding up onboarding for new developers.

Documentation:
https://github.com/lucasrosaalves/industrial-model

 Feedback and feature requests are very welcome!

Sample code:

from pathlib import Path

from industrial_model import (
    Engine,
    ViewInstance,
    ViewInstanceConfig,
    AggregatedViewInstance,
    InstanceId,
    select,
    search,
    aggregate,
    col,
    and_,
    or_,
)

class Asset(ViewInstance):
    view_config = ViewInstanceConfig(
        view_external_id="CogniteAsset",
        instance_spaces_prefix="Industrial-",
    )
    name: str
    description: str | None = None
    aliases: list[str] = []
    parent: "Asset" | None = None


class AssetCountByParent(AggregatedViewInstance):
    view_config = ViewInstanceConfig(view_external_id="CogniteAsset")
    parent: InstanceId | None = None


engine = Engine.from_config_file(Path("cognite-sdk-config.yaml"))

# Query with filters
results = engine.query(
    select(Asset)
    .where(
        and_(
            col(Asset.aliases).contains_any_(["pump", "equipment"]),
            or_(
                col(Asset.parent).nested_(col(Asset.name) == "Root Asset"),
                col(Asset.name).prefix("Pump-"),
            ),
        )
    )
    .asc(Asset.name)
    .limit(100)
)

# Search
search_results = engine.search(
    search(Asset)
    .where(col(Asset.aliases).contains_any_(["pump"]))
    .query_by("industrial equipment", [Asset.name, Asset.description])
)

# Aggregate
aggregate_stmt = (
    aggregate(AssetCountByParent, "count")
    .group_by(col(AssetCountByParent.parent))
)