OpenMetadata Python SDK 实战类型化实体门面、配置体系与搜索血缘 API 完整指南【免费下载链接】OpenMetadataThe Open Context Layer for Data and AI , OpenMetadata is the open platform for building trusted data context and business semantics for humans, AI assistants, and agents.项目地址: https://gitcode.com/GitHub_Trending/op/OpenMetadataOpenMetadata 的 Python SDK 位于openmetadata-ingestion包内以“复数门面类 生成的 Pydantic 实体模型 底层 ingestion 客户端”三层结构封装了表、数据库、用户等常见元数据操作。本文基于仓库中的 SDK 文档 与ingestion/src/metadata/sdk/下的源码实现完整讲解 SDK 的安装配置、CRUD 与分页、部分更新、表级辅助操作、治理标签、CSV 导入导出、搜索与血缘 API 的使用方式与底层原理帮助你在自动化脚本、数据治理工具或 AI Agent 中直接复用这套类型化 API。SDK 架构门面、生成模型与底层客户端理解 SDK 的第一步是分清三个层次这也是源码中类职责的划分方式门面类Facades位于 entities 包如Tables、Databases、Users全部采用复数命名提供create/retrieve/list/update等 SDK 方法。生成的 Pydantic 实体模型位于metadata/generated/schema/entity/...如单数的Table、Database它们是纯数据模型不暴露任何 SDK 方法只用于承载与传输数据。底层 ingestion 客户端OMeta client即metadata.ingestion.ometa中的OpenMetadata客户端负责实际的 HTTP 请求。SDK 门面只是对它的类型化包装。这一分层在 client.py 中体现得很清楚OpenMetadata.initialize(config)会将 SDK 配置翻译成OpenMetadataConnection并构造出内部ometa客户端见 client.py#L24-L56而 BaseEntity 则通过_get_client()统一获取该默认客户端所有门面类的create、retrieve、update、delete最终都委托给它。from metadata.sdk import Tables from metadata.generated.schema.entity.data.table import Table table: Table Tables.retrieve_by_name(service.database.schema.table)文档特别强调请使用复数门面类做 SDK 操作单数生成类如metadata.generated.schema.entity.data.table.Table只作为数据类型注解使用不能调用create()/update()。安装与配置安装SDK 随主包发布无需单独安装pip install openmetadata-ingestion若需要运行数据质量相关示例再按工作负载安装对应 extrapip install openmetadata-ingestion[pandas] pip install openmetadata-ingestion[mysql] pip install openmetadata-ingestion[postgres]使用 configure() 建立全局客户端应用代码推荐入口是 configure()它会初始化所有实体门面共用的全局客户端from metadata.sdk import configure configure(hosthttp://localhost:8585/api, jwt_tokenyour-jwt-token)也可以完全不传参从环境变量读取from metadata.sdk import configure # Reads OPENMETADATA_HOST or OPENMETADATA_SERVER_URL. # Reads OPENMETADATA_JWT_TOKEN or OPENMETADATA_API_KEY. configure()configure()支持三种传参方式OpenMetadataConfig实例、Mapping、或关键字参数host/server_url、jwt_token、其余透传项混合使用 config 对象与关键字参数会抛出TypeError见init.py#L117-L145。支持的环境变量由 config.py 的 from_env() 实现取值规则与默认值如下环境变量含义默认值OPENMETADATA_HOST或OPENMETADATA_SERVER_URL服务端 URL必填二者其一无缺失时抛ValueErrorOPENMETADATA_JWT_TOKEN或OPENMETADATA_API_KEYJWT 令牌 / API Key无OPENMETADATA_VERIFY_SSL是否校验 SSL布尔字符串falseOPENMETADATA_CA_BUNDLECA 证书包路径无OPENMETADATA_CLIENT_TIMEOUT客户端超时秒数30从源码看server_url会在构造时自动去除末尾斜杠server_url.rstrip(/)而jwt_token与api_key是同一凭据的两个别名self.jwt_token jwt_token or api_key认证时若两者皆空会抛出ValueError(JWT token or API key is required to authenticate)见 config.py#L26-L40 与 config.py#L78-L84。手动初始化客户端测试或高级场景from metadata.sdk import OpenMetadata, OpenMetadataConfig config OpenMetadataConfig( server_urlhttp://localhost:8585/api, jwt_tokenyour-jwt-token, ) client OpenMetadata.initialize(config)OpenMetadataConfig还提供链式builder()写法OpenMetadataConfig.builder().server_url(...).jwt_token(...).build()。另外模块级函数client()返回当前客户端、reset()关闭并重置全局状态均未配置时client()会抛出RuntimeError见init.py#L148-L159。Table 实体的完整 CRUD以Tables为例覆盖创建、查询、更新、删除的完整生命周期from metadata.generated.schema.api.data.createTable import CreateTableRequest from metadata.generated.schema.entity.data.table import Column, DataType from metadata.sdk import Tables request CreateTableRequest( nameorders, databaseSchemaservice.database.schema, columns[ Column(nameid, dataTypeDataType.BIGINT), Column(namestatus, dataTypeDataType.VARCHAR, dataLength255), ], ) table Tables.create(request) table Tables.retrieve(str(table.id.root), fields[owners, tags, columns]) table Tables.retrieve_by_name( service.database.schema.orders, fields[owners, tags, columns], ) table.description Order facts loaded from the commerce warehouse updated Tables.update(table) Tables.delete(str(updated.id.root), recursiveTrue, hard_deleteTrue)几个要点结合源码说明create(request)底层调用client.create_or_update(request)并做实体类型强制转换见 base.py#L185-L189。retrieve/retrieve_by_name支持fields按需拉取字段、以及nullable参数控制“查不到时返回 None 还是抛错”见 base.py#L191-L240。Tables.update(entity)只接收实体对象它先按entity.id取出当前版本作为 source再用client.patch(entityTable, sourcecurrent, destinationentity)把变更字段打上见 base.py#L242-L254。因此更新语义是“基于最新版本的差量补丁”而不是整体替换。delete支持recursive与hard_delete两个开关映射到底层client.delete(...)。列表查询与分页list()返回一页结果封装为EntityList包含entities、after、before三个属性from metadata.sdk import Tables page Tables.list( limit50, fields[owners, tags], filters{databaseSchema: service.database.schema}, ) for table in page.entities: print(table.fullyQualifiedName) if page.after: next_page Tables.list(limit50, afterpage.after)从 BaseEntity.list 的签名 可以看出参数默认值limit10注意不是 50、afterNone、beforeNone、fieldsNone、filtersNonefilters会以 query params 形式传给client.list_entities。需要遍历全量时使用list_all()SDK 会循环翻页直到after为空for table in Tables.list_all( batch_size100, fields[owners, tags], filters{databaseSchema: service.database.schema}, ): print(table.name)list_all的batch_size默认 100见 base.py#L300-L323。文档还特别提醒两点SDK没有TableListParams类EntityList也不暴露auto_paging_iterable()请直接用上述关键字参数完成分页。部分更新Partial Update门面类不提供patch(entity_id, json_patch)这种方法。推荐做法是取出实体 → 深拷贝 → 修改副本 → 调用update(entity)from metadata.generated.schema.type.basic import Markdown from metadata.sdk import Tables table Tables.retrieve_by_name(service.database.schema.orders) updated_table table.model_copy(deepTrue) updated_table.description Markdown(Orders curated by the analytics team) patched Tables.update(updated_table)注意description这类字段在生成的模型中是Markdown类型直接赋纯字符串在某些校验严格的路径下会失败显式包装为Markdown(...)更稳妥。对于门面未覆盖的特殊补丁流程可以穿透到 SDK 包装的底层 ingestion 客户端from metadata.generated.schema.entity.data.table import Table from metadata.sdk import client metadata client().ometa current metadata.get_by_id(entityTable, entity_idtable-id, fields[tags]) destination current.model_copy(deepTrue) destination.tags [] patched metadata.patch(entityTable, sourcecurrent, destinationdestination)Table 专属辅助操作Tables在通用 CRUD 之外提供了一批表级快捷方法内部统一走“取字段 → 改副本 → patch”的套路实现见 tables.pyfrom metadata.generated.schema.entity.data.table import TableData from metadata.sdk import Tables table Tables.add_tag(table-id, PII.Sensitive) table Tables.update_column_description( table-id, column_namestatus, descriptionCurrent order state, ) sample_data TableData(columns[id, status], rows[[1, COMPLETE]]) Tables.add_sample_data(table-id, sample_data) table_with_sample_data Tables.get_sample_data(table-id)从源码看这些方法的具体行为add_tag(table_id, tag_fqn)只拉取tags字段在副本的tags列表末尾追加{tagFQN: tag_fqn}后执行 patch见 tables.py#L29-L51所以是追加而非替换。update_column_description(table_id, column_name, description)只拉取columns按列名精确匹配并改写description后 patch见 tables.py#L53-L76。add_sample_data/get_sample_data通过model_construct构造一个轻量 Table 引用对象仅含 id / name / fqn再分别调用ingest_table_sample_data与get_sample_data见 tables.py#L88-L118。治理标签Classifications 与 Tags治理类操作同样使用复数门面。标准流程是先建分类Classification再在分类下建标签Tag最后用标签 FQN 挂到资产上from metadata.generated.schema.api.classification.createClassification import ( CreateClassificationRequest, ) from metadata.generated.schema.api.classification.createTag import CreateTagRequest from metadata.sdk import Classifications, Tables, Tags classification Classifications.create( CreateClassificationRequest( namePII, descriptionPersonally identifiable information, ) ) tag Tags.create( CreateTagRequest( classificationclassification.fullyQualifiedName.root, nameSensitive, descriptionSensitive customer data, ) ) table Tables.add_tag(table-id, tag.fullyQualifiedName.root)注意CreateTagRequest.classification需要的是分类的FQN 字符串因此源码中的FullyQualifiedEntityName需要取.root。若要整体替换某个表的标签而非追加文档给出的做法是Tables.retrieve(id, fields[tags])→ 在副本上修改tags→Tables.update(destination)。CSV 导入与导出CSV 操作返回“操作对象”链式配置后调用execute()才真正执行from metadata.sdk import Glossaries, Tables csv_text Tables.export_csv(service.database.schema.orders).execute() dry_run ( Glossaries.import_csv(BusinessGlossary) .with_data(csv_text) .set_dry_run(True) .execute() )对应源码中 CsvExportOperation / CsvImportOperation 两个 dataclass导出侧execute()调用client.export_csv(entity..., name...)导入侧通过with_data(csv_data)、set_dry_run(bool)设置参数后execute()调用client.import_csv(...)其中dry_runTrue时服务端只校验不落库适合先做导入预检。两个操作对象还支持with_async()/execute_async()的异步变体但客户端未实现export_csv_async/import_csv_async时会抛出AttributeError。Search 搜索 APImetadata.sdk.api.Search提供查询、建议、聚合与 builder 四种入口from metadata.sdk.api import Search results Search.search( querycustomer, indextable_search_index, from_0, size25, sort_fieldname.keyword, sort_orderasc, ) suggestions Search.suggest(cust, size10) aggregations Search.aggregate( query*, indextable_search_index, fielddatabase.name.keyword, ) builder_results ( Search.builder() .query(customer) .index(table_search_index) .size(25) .execute() )结合 search.py 的实现 补充几点默认值与回退逻辑Search.search的默认size10、include_aggregationsTrue见 search.py#L136-L192filters传 dict 时会被自动构造为 Elasticsearch 的bool/must/termquery_filter JSON。Search.suggest(query, fieldNone, size5)返回建议词列表。Search.aggregate中若未指定indexHTTP 回退路径会默认使用table_search_index见 search.py#L219-L243。各方法优先调用底层 ometa 客户端的同名回调如es_search_from_es、get_suggest_entities、es_aggregate不可用时回退到/search/query、/search/aggregate等 REST 端点。除search/suggest/aggregate外源码还实现了search_advanced自定义请求体、reindex(entity_type)/reindex_all触发索引重建以及对应的*_async变体可在需要时直接使用。Search.builder()返回SearchBuilder支持query/index/from_/size/sort_field/sort_order/filter(key, value)/include_aggregations链式配置execute()时若未设置 query 会抛出ValueError见 search.py#L339-L406。Lineage 血缘 APImetadata.sdk.api.Lineage覆盖按 FQN 查询、按 ID 查询、新增血缘边三类操作from metadata.generated.schema.entity.data.table import Table from metadata.sdk.api import Lineage lineage Lineage.get_lineage( service.database.schema.orders, upstream_depth1, downstream_depth1, ) lineage_by_id Lineage.get_entity_lineage( entity_typeTable, entity_idtable-id, upstream_depth2, downstream_depth1, ) Lineage.add_lineage( from_entity_idsource-table-id, from_entity_typetable, to_entity_idtarget-table-id, to_entity_typetable, descriptionCurated order facts, )从 lineage.py 的实现 可以看出get_lineage底层调用client.get_lineage_by_name并对兼容旧签名的客户端做了TypeError重试get_entity_lineage调用client.get_lineage_by_id两个查询方法的upstream_depth/downstream_depth默认均为 1返回值统一转换为EntityLineage模型add_lineage会把description包装为Markdown类型构造AddLineageRequest含fromEntity/toEntity两个EntityReference后提交见 lineage.py#L101-L144。源码中还提供了add_lineage_by_name等按 FQN 建边的便捷方法。支持的实体门面清单文档列出的当前 SDK 导出门面如下数据资产APICollections、APIEndpoints、Charts、Containers、DashboardDataModels、Dashboards、Databases、DatabaseSchemas、DataContracts、Metrics、MLModels、Pipelines、Queries、SearchIndexes、StoredProcedures、Tables服务DashboardServices、DatabaseServices、StorageServices治理Classifications、DataProducts、Domains、Glossaries、GlossaryTerms、Tags团队与用户Teams、Users数据质量TestCases、TestDefinitions、TestSuites对照 entities/init.py 的__all__与 sdk 包init当前版本还额外导出了ContextFiles、Folders、Pages、Settings四个门面且每个类都提供同名小写下划线别名如tables等价于Tables注释中标注参照 stripe 风格。若某实体尚无门面文档建议直接使用metadata.sdk.client().ometa返回的底层 ingestion 客户端或改用 ingestion 客户端 API。实体引用to_entity_reference设置 owner、domain 等引用型字段时目标字段期望的是EntityReference而非完整实体对象。to_entity_reference(entity)会读取实体的id、entityType或类名小写兜底、name、fullyQualifiedName返回包含id/type/name/fullyQualifiedName的 dict实现见 base.py#L569-L612from metadata.sdk import Teams, Users, to_entity_reference team Teams.retrieve_by_name(engineering) user Users.retrieve_by_name(john.doe) team_ref to_entity_reference(team) user_ref to_entity_reference(user)它既可作为模块级函数也可作为BaseEntity静态方法调用如Teams.to_entity_reference(team)。若实体缺少id属性会抛出ValueError。错误处理HTTP 层错误由 ingestion 客户端的APIError承载携带status_code属性适合按状态码分支处理from metadata.ingestion.ometa.client import APIError from metadata.sdk import Tables try: table Tables.retrieve(table-id) except APIError as err: if err.status_code 404: print(Table not found) elif err.status_code 401: print(Authentication failed) else: raise此外还有两类程序性错误值得注意未调用configure()就访问门面时会得到RuntimeError(SDK not configured...)见init.py#L148-L152update()传入缺少id的实体时会抛出ValueError(Entity must define an id attribute before updating)见 base.py#L245-L247。测试与运行验证SDK 自带两级测试# 从 ingestion 目录运行单元测试mock 客户端无需服务端 pytest tests/unit/sdk/ # 针对本地运行中的 OpenMetadata 服务运行集成测试 pytest tests/integration/sdk/test_sdk_integration.py -v单元测试位于 tests/unit/sdk/覆盖配置解析test_config.py、BaseEntity 行为test_base_entity.py、CSV 操作test_csv_operations.py、各实体门面test_table_entity.py、test_user_entity.py等。集成测试 test_sdk_integration.py 会真实创建 service → database → schema → table 的层级并演练 follower 管理、restore/版本流以及标签、glossary、owner、domain、data product、CSV 等元数据富化操作适合作为“端到端正确用法”的活文档参考。贡献如何新增一个实体门面文档给出了标准的五步流程与仓库实际结构一致在metadata/sdk/entities/下新建文件继承BaseEntity[实体类型, 创建请求类型]覆写entity_type()类方法在 entities/__init__.py 与 sdk __init__.py 中导出在tests/unit/sdk/下补充单元测试。示例即Tables的最小实现形态from typing import Type from metadata.generated.schema.api.data.createTable import CreateTableRequest from metadata.generated.schema.entity.data.table import Table from metadata.sdk.entities.base import BaseEntity class Tables(BaseEntity[Table, CreateTableRequest]): classmethod def entity_type(cls) - Type[Table]: return Table继承BaseEntity后create/retrieve/retrieve_by_name/update/delete/list/list_all/search/ CSV 导入导出等能力全部自动获得通常只需补充实体专属的辅助方法参考 tables.py 中的add_tag、update_column_description。小结OpenMetadata Python SDK 的价值在于用复数门面类把“生成的 Pydantic 模型 ingestion 客户端”收敛成一组类型安全、可链式的 APIconfigure()一行完成全局客户端初始化Tables.update(entity)这类方法把取数—改副本—patch 的繁琐流程封装在内部Search/Lineage则补齐了检索与血缘场景。实际开发时建议遵循文档约定门面操作用复数类、数据模型用单数生成类、部分更新走“副本 update”模式、未覆盖场景再穿透到client().ometa并配合tests/unit/sdk/与tests/integration/sdk/中的用例验证行为。该 SDK 与整个 ingestion 包一样采用 Apache License 2.0 许可见 LICENSE。【免费下载链接】OpenMetadataThe Open Context Layer for Data and AI , OpenMetadata is the open platform for building trusted data context and business semantics for humans, AI assistants, and agents.项目地址: https://gitcode.com/GitHub_Trending/op/OpenMetadata创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考