How-To: Simple Streamlit App for CRUD Operations on CDF Data Model view [Cognite Official]

Objective

By the end of this guide, you'll learn how to build and run a simple Streamlit web application that lets you create, read, update, and delete (CRUD) data model instances in Cognite Data Fusion (CDF) using the Python SDK. This app is ideal for demos, quick data manipulation, or onboarding users to structured data modeling concepts within CDF.

Introduction

In Cognite Data Fusion, data modeling is used to define structured digital representations of business entities such as assets, equipment, or time series. Often, you need a fast, friendly interface to explore and manipulate these modeled entities — especially for prototyping or internal data reviews.

This guide walks you through a fully functional Streamlit app that connects to CDF, reads data model instances from a defined view, and lets you interact with those instances in real time. The app is lightweight, self-contained, and uses the cognite-sdk-python for all data operations.

Explanation (Steps)

We’ll break this demo into 4 interactive features:

1. Read – List instances from a specific view in a selected space, with filtering and formatted display.
2. Create – Add a new instance by specifying an external ID and a name.
3. Update – Modify an existing instance by changing its attributes (e.g., name).
4. Delete – Remove an instance using its external ID.

The app uses a single data model view (CogniteSourceSystem) and a specific instance space (springfield_instances). You can adjust these for your own environment.

Conclusion

This CRUD demo app is a practical entry point to working with Cognite’s data modeling API. It highlights:

• The simplicity of apply() and delete() operations
• How to build an intuitive frontend using Streamlit
• The importance of structuring data in views for efficient interaction

You can extend this example to support multiple views, richer property types, or even edge relationships.

Technical Details

• Framework: Streamlit
• SDK: cognite-sdk-python
• Backend: Cognite Data Fusion (CDF) API v2023-01-01
• Key Methods: apply(nodes=...), instances.list(), instances.delete()
• Dependencies: pandas, streamlit, cognite-sdk-python

Code

Note: Navigate to Data Management → Build solutions → Streamlit apps → Create app and copy/paste the code below to get started.

import streamlit as st

import pandas as pd

from datetime import datetime

from cognite.client import CogniteClient

from cognite.client.data_classes import data_modeling as dm

from cognite.client.data_classes.filters import Exists, Prefix, And

client = CogniteClient()

SPACE = "springfield_instances"   # ADD YOUR INSTANCE SPACE HERE

VIEW_NAME = "CogniteSourceSystem"

VIEW_ID = dm.ViewId("cdf_cdm", VIEW_NAME, "v1")

def load_data():

    filter = And(

        Exists(property=["cdf_cdm", f"{VIEW_NAME}/v1", "name"]),

        Prefix(property=["cdf_cdm", f"{VIEW_NAME}/v1", "name"], value="P")

    )

    instances = client.data_modeling.instances.list(

        # space=SPACE,            # Update space name or comment out to list all

        sources=[VIEW_ID], 

        # filter=filter,          # Update filter example or comment out to list all

        limit=-1

    )

    rows = []

    for i in instances:

        row = {"externalId": i.external_id}

        for k, v in i.properties.items():

            if isinstance(v, dm.DirectRelationReference):

                row[k] = v.external_id

            elif isinstance(v, dict):

                for sub_k, sub_v in v.items():

                    row[f"{k}.{sub_k}"] = sub_v

            elif isinstance(v, list):

                row[k] = ", ".join(map(str, v))

            else:

                row[k] = v

        rows.append(row)

    df = pd.DataFrame(rows)

    # Clean column headers

    df.rename(columns={c: c.split(".", 1)[-1] if "." in c else c for c in df.columns}, inplace=True)

    return df

def apply_instance(external_id, name):

    node = dm.NodeApply(

        space=SPACE,

        external_id=external_id,

        sources=[dm.NodeOrEdgeData(source=VIEW_ID, properties={"name": name})],

    )

    client.data_modeling.instances.apply(nodes=[node])

    st.success(f"Upserted: {external_id}")

def delete_instance(external_id):

    client.data_modeling.instances.delete(nodes=[dm.NodeId(space=SPACE, external_id=external_id)])

    st.warning(f"Deleted: {external_id}")

# --- UI ---

st.title(f"Core Data Model - CRUD example")

tab1, tab2, tab3, tab4 = st.tabs(["Read", "Create", "Update", "Delete"])

with tab1:

    st.subheader(f"Existing instances in {VIEW_NAME}")

    if st.button("Refresh view"):

        st.session_state["refresh_trigger"] = datetime.utcnow().isoformat()

    if "refresh_trigger" not in st.session_state:

        st.session_state["refresh_trigger"] = datetime.utcnow().isoformat()

    df = load_data()

    if not df.empty:

        st.dataframe(df, use_container_width=True, hide_index=True)

        st.write(f"Total instances: {len(df)}")

    else:

        st.info("No instances found.")

with tab2:

    st.subheader(f"Create New Instance in {VIEW_NAME}")

    new_ext_id = st.text_input("External ID (unique)", key="create_ext_id")

    new_name = st.text_input("Name", key="create_name")

    if st.button("Create"):

        apply_instance(new_ext_id, new_name)

with tab3:

    st.subheader(f"Update Existing Instance in {VIEW_NAME}")

    upd_ext_id = st.text_input("External ID to Update", key="update_ext_id")

    upd_name = st.text_input("New Name", key="update_name")

    if st.button("Update"):

        apply_instance(upd_ext_id, upd_name)

with tab4:

    st.subheader(f"Delete Instance from {VIEW_NAME}")

    del_ext_id = st.text_input("External ID to Delete", key="delete_ext_id")

    if st.button("Delete"):

        delete_instance(del_ext_id)