“

系列：S12 开发者教程 · 第 3 篇 | 难度：入门 | 阅读时间：15 分钟

第一条数据：写入和查询实体

#引言

在上一篇教程中，我们创建了 ObjectType 和 RelationType 来定义业务模型。但模型只是骨架，数据才是血肉。在本教程中，你将学习如何在 coomia-dip 中写入实体数据、创建关系实例、执行查询，以及进行批量数据操作。

coomia-dip 的数据层（Data Layer）基于 Apache Iceberg 和 Nessie 构建，提供了 ACID 事务、时间旅行查询和分支管理等企业级特性。所有数据操作通过 gRPC 协议与 Data Layer 通信，Python SDK 对此进行了友好的封装。

#前置准备

确保你已经完成了前两篇教程：

本地部署 coomia-dip（S12-01）
创建了 Employee、Department、Project 三个 ObjectType（S12-02）

Python

from ontology_sdk import OntoPlatform

platform = OntoPlatform(
    control_plane_url="localhost:50051",
    data_plane_url="localhost:50052"
)

#创建实体（Create）

#创建单个实体

Python

# 创建一个员工实体
employee = platform.objects.create(
    object_type="Employee",
    properties={
        "employee_id": "EMP-000001",
        "name": "张三",
        "email": "zhangsan@example.com",
        "title": "高级工程师",
        "level": 8,
        "hire_date": "2020-03-15",
        "salary": 35000.00,
        "is_active": True,
        "skills": ["python", "java", "kubernetes"]
    }
)

print(f"创建成功: {employee.rid}")
print(f"  名称: {employee.properties['name']}")
print(f"  RID: {employee.rid}")

每个创建的实体都会获得一个全局唯一的 RID（Resource Identifier），格式为 ri.ontology.object.<type>.<uuid>。

#批量创建实体

Python

# 批量创建多个员工
employees_data = [
    {
        "employee_id": "EMP-000002",
        "name": "李四",
        "email": "lisi@example.com",
        "title": "产品经理",
        "level": 7,
        "hire_date": "2021-06-01",
        "salary": 30000.00,
        "skills": ["product-design", "data-analysis"]
    },
    {
        "employee_id": "EMP-000003",
        "name": "王五",
        "email": "wangwu@example.com",
        "title": "数据工程师",
        "level": 6,
        "hire_date": "2022-01-10",
        "salary": 28000.00,
        "skills": ["spark", "flink", "sql"]
    },
    {
        "employee_id": "EMP-000004",
        "name": "赵六",
        "email": "zhaoliu@example.com",
        "title": "前端工程师",
        "level": 5,
        "hire_date": "2023-03-20",
        "salary": 22000.00,
        "skills": ["react", "typescript", "css"]
    }
]

results = platform.objects.batch_create(
    object_type="Employee",
    items=employees_data
)

print(f"批量创建: {len(results)} 条记录")
for r in results:
    print(f"  {r.properties['name']} -> {r.rid}")

#创建部门和项目数据

Python

# 创建部门
engineering = platform.objects.create(
    object_type="Department",
    properties={
        "dept_id": "DEPT-001",
        "name": "工程部",
        "code": "ENG",
        "budget": 5000000.00
    }
)

product = platform.objects.create(
    object_type="Department",
    properties={
        "dept_id": "DEPT-002",
        "name": "产品部",
        "code": "PROD",
        "budget": 2000000.00
    }
)

# 创建项目
project_alpha = platform.objects.create(
    object_type="Project",
    properties={
        "project_id": "PRJ-001",
        "name": "Alpha 平台",
        "status": "in_progress",
        "priority": "high",
        "budget": 1000000.00,
        "start_date": "2024-01-01",
        "end_date": "2024-12-31",
        "tags": ["platform", "core"]
    }
)

print(f"部门: {engineering.properties['name']}, {product.properties['name']}")
print(f"项目: {project_alpha.properties['name']}")

#创建关系实例

有了实体数据后，我们可以创建关系来连接它们：

Python

# 张三属于工程部
platform.relations.create(
    relation_type="belongs_to",
    source_rid=employee.rid,  # 张三
    target_rid=engineering.rid,  # 工程部
    properties={
        "joined_at": "2020-03-15",
        "role_in_dept": "技术负责人"
    }
)

# 张三参与 Alpha 项目
platform.relations.create(
    relation_type="participates_in",
    source_rid=employee.rid,
    target_rid=project_alpha.rid,
    properties={
        "role": "owner",
        "allocation_pct": 60
    }
)

# 工程部负责 Alpha 项目
platform.relations.create(
    relation_type="owns_project",
    source_rid=engineering.rid,
    target_rid=project_alpha.rid
)

print("关系创建完成")

#查询实体（Read）

#按主键查询

Python

# 通过主键查找
emp = platform.objects.get(
    object_type="Employee",
    primary_key="EMP-000001"
)
print(f"找到: {emp.properties['name']} ({emp.properties['title']})")

#按 RID 查询

Python

# 通过 RID 查找（跨类型通用）
obj = platform.objects.get_by_rid(rid="ri.ontology.object.employee.xxx")
print(f"找到: {obj.object_type} - {obj.properties['name']}")

#条件查询

Python

# 查找所有在职的高级工程师
results = platform.objects.search(
    object_type="Employee",
    filter={
        "is_active": {"eq": True},
        "level": {"gte": 7},
        "title": {"contains": "工程师"}
    },
    sort=[{"field": "level", "order": "desc"}],
    limit=10
)

print(f"查询结果: {len(results)} 条")
for emp in results:
    print(f"  {emp.properties['name']} - L{emp.properties['level']} {emp.properties['title']}")

#过滤操作符

coomia-dip 支持丰富的过滤操作符：

操作符	说明	示例
`eq`	等于	`{"status": {"eq": "active"}}`
`neq`	不等于	`{"status": {"neq": "cancelled"}}`
`gt`	大于	`{"level": {"gt": 5}}`
`gte`	大于等于	`{"salary": {"gte": 30000}}`
`lt`	小于	`{"level": {"lt": 10}}`
`lte`	小于等于	`{"budget": {"lte": 100000}}`
`in`	在列表中	`{"status": {"in": ["active", "planning"]}}`
`not_in`	不在列表中	`{"priority": {"not_in": ["low"]}}`
`contains`	包含子串	`{"name": {"contains": "工程"}}`
`starts_with`	前缀匹配	`{"email": {"starts_with": "zhang"}}`
`is_null`	为空	`{"phone": {"is_null": True}}`
`is_not_null`	不为空	`{"salary": {"is_not_null": True}}`
`between`	范围	`{"hire_date": {"between": ["2020-01-01", "2023-12-31"]}}`
`array_contains`	数组包含	`{"skills": {"array_contains": "python"}}`

#关系查询

Python

# 查找张三所属的部门
depts = platform.relations.get_targets(
    relation_type="belongs_to",
    source_rid=employee.rid
)
for dept in depts:
    print(f"所属部门: {dept.properties['name']}")

# 查找工程部的所有员工
members = platform.relations.get_sources(
    relation_type="belongs_to",
    target_rid=engineering.rid
)
print(f"工程部成员: {len(members)} 人")

# 查找张三参与的所有项目
projects = platform.relations.get_targets(
    relation_type="participates_in",
    source_rid=employee.rid
)
for proj in projects:
    print(f"参与项目: {proj.properties['name']}")

#多跳关系查询

Python

# 查找某个部门下所有员工参与的项目（两跳查询）
dept_projects = platform.objects.search(
    object_type="Project",
    filter={
        "__relation": {
            "participates_in": {
                "source": {
                    "__relation": {
                        "belongs_to": {
                            "target": {"dept_id": {"eq": "DEPT-001"}}
                        }
                    }
                }
            }
        }
    }
)

#更新实体（Update）

#部分更新

Python

# 更新张三的薪资和职级
platform.objects.update(
    object_type="Employee",
    primary_key="EMP-000001",
    properties={
        "salary": 40000.00,
        "level": 9
    }
)
print("更新完成")

#原子操作

Python

# 原子递增操作
platform.objects.atomic_update(
    object_type="Employee",
    primary_key="EMP-000001",
    operations=[
        {"field": "level", "op": "increment", "value": 1}
    ]
)

#删除实体（Delete）

Python

# 删除单个实体
platform.objects.delete(
    object_type="Employee",
    primary_key="EMP-000004"
)

# 批量删除
platform.objects.batch_delete(
    object_type="Employee",
    primary_keys=["EMP-000003", "EMP-000004"]
)

# 条件删除
platform.objects.delete_where(
    object_type="Employee",
    filter={"is_active": {"eq": False}}
)

#事务操作

coomia-dip 支持跨实体的 ACID 事务：

Python

with platform.transaction() as tx:
    # 创建员工
    new_emp = tx.objects.create(
        object_type="Employee",
        properties={
            "employee_id": "EMP-000005",
            "name": "孙七",
            "email": "sunqi@example.com",
            "hire_date": "2024-03-01"
        }
    )

    # 分配到部门
    tx.relations.create(
        relation_type="belongs_to",
        source_rid=new_emp.rid,
        target_rid=engineering.rid,
        properties={"joined_at": "2024-03-01"}
    )

    # 如果任一操作失败，整个事务回滚
    tx.commit()
    print("事务提交成功")

#时间旅行查询

基于 Iceberg 的时间旅行能力，你可以查询历史数据：

Python

# 查询某个时间点的数据
historical = platform.objects.search(
    object_type="Employee",
    filter={"employee_id": {"eq": "EMP-000001"}},
    as_of="2024-01-01T00:00:00Z"
)

# 查看某个实体的变更历史
changelog = platform.objects.get_changelog(
    object_type="Employee",
    primary_key="EMP-000001"
)
for entry in changelog:
    print(f"  {entry.timestamp}: {entry.change_type}")
    if entry.changed_properties:
        for prop, change in entry.changed_properties.items():
            print(f"    {prop}: {change['old']} -> {change['new']}")

#数据导入导出

#从 CSV 导入

Python

# 从 CSV 文件批量导入
result = platform.data.import_csv(
    object_type="Employee",
    file_path="employees.csv",
    column_mapping={
        "工号": "employee_id",
        "姓名": "name",
        "邮箱": "email",
        "入职日期": "hire_date"
    },
    batch_size=1000
)
print(f"导入完成: {result.success_count} 成功, {result.error_count} 失败")

#导出为 JSON

Python

# 导出所有员工数据
platform.data.export_json(
    object_type="Employee",
    output_path="employees_export.json",
    filter={"is_active": {"eq": True}}
)

#性能优化建议

批量操作：优先使用 batch_create 和 batch_delete，减少 gRPC 往返次数
分页查询：大数据量查询使用 limit 和 offset，或使用游标分页
选择性加载：使用 select 参数只返回需要的属性
索引优化：对频繁查询的属性添加索引
事务合并：将相关操作放在同一个事务中

Python

# 选择性加载示例
results = platform.objects.search(
    object_type="Employee",
    filter={"is_active": {"eq": True}},
    select=["employee_id", "name", "email"],  # 只返回指定属性
    limit=100,
    offset=0
)

#总结

在本教程中，我们学习了 coomia-dip 数据操作的完整生命周期：

创建：单个创建和批量创建实体与关系
查询：主键查询、条件过滤、关系遍历、多跳查询
更新：部分更新和原子操作
删除：单个删除、批量删除、条件删除
事务：跨实体 ACID 事务
时间旅行：历史数据查询和变更追踪

这些操作构成了 coomia-dip 数据管理的基础。结合 Ontology 的语义能力，你可以构建出比传统 CRUD 更加智能和灵活的数据应用。

本文是 coomia-dip 开发者教程系列的第 3 篇。 项目地址：https://github.com/coomia-dip/coomia-dip↗ | 许可证：Apache License 2.0