approve: 开发实施:按定稿设计继续 PBL Agent OS Phase 0/1(16 子任务 Wave 链,对齐 pbls_spec.json)

This commit is contained in:
agent.develop 2026-09-18 12:01:05 +08:00
parent c38ac15df9
commit 33a50d2c94
29 changed files with 1385 additions and 2782 deletions

View File

@ -1,132 +0,0 @@
{
"tblname": "pbl_class",
"alias": "pbl_domain_ext",
"browserfields": {
"id": {
"label": "ID",
"list": true,
"type": "number"
},
"tenant_id": {
"label": "租户ID",
"list": true,
"type": "text"
},
"class_uid": {
"label": "班级UID",
"type": "text",
"list": true
},
"name": {
"label": "名称",
"type": "select",
"list": true
},
"grade": {
"label": "年级",
"type": "select",
"list": true
},
"teacher_id": {
"label": "教师",
"type": "text",
"list": true
},
"student_count": {
"label": "学生数",
"type": "number",
"list": true
},
"roster_json": {
"label": "花名册",
"type": "text",
"list": true
},
"status": {
"label": "状态",
"type": "select",
"list": true
}
},
"params": {
"browserfields": {
"id": {
"label": "ID",
"list": true,
"type": "number"
},
"tenant_id": {
"label": "租户ID",
"list": true,
"type": "text"
},
"class_uid": {
"label": "班级UID",
"type": "text",
"list": true
},
"name": {
"label": "名称",
"type": "select",
"list": true
},
"grade": {
"label": "年级",
"type": "select",
"list": true
},
"teacher_id": {
"label": "教师",
"type": "text",
"list": true
},
"student_count": {
"label": "学生数",
"type": "number",
"list": true
},
"roster_json": {
"label": "花名册",
"type": "text",
"list": true
},
"status": {
"label": "状态",
"type": "select",
"list": true
}
},
"editable": {
"class_uid": {
"label": "班级UID",
"type": "text"
},
"name": {
"label": "名称",
"type": "select"
},
"grade": {
"label": "年级",
"type": "select"
},
"teacher_id": {
"label": "教师",
"type": "text"
},
"student_count": {
"label": "学生数",
"type": "number"
},
"roster_json": {
"label": "花名册",
"type": "text"
},
"status": {
"label": "状态",
"type": "select"
}
},
"order_by": "id DESC",
"page_size": 20
}
}

View File

@ -1,146 +0,0 @@
{
"tblname": "pbl_team",
"alias": "pbl_domain_ext",
"browserfields": {
"id": {
"label": "ID",
"list": true,
"type": "number"
},
"tenant_id": {
"label": "租户ID",
"list": true,
"type": "text"
},
"team_uid": {
"label": "团队UID",
"type": "text",
"list": true
},
"class_id": {
"label": "班级",
"type": "number",
"list": true
},
"name": {
"label": "名称",
"type": "select",
"list": true
},
"mission_id": {
"label": "Mission",
"type": "number",
"list": true
},
"members_json": {
"label": "成员",
"type": "text",
"list": true
},
"capacity": {
"label": "容量",
"type": "number",
"list": true
},
"status": {
"label": "状态",
"type": "select",
"list": true
},
"session_id": {
"label": "绑定会话",
"type": "number",
"list": true
}
},
"params": {
"browserfields": {
"id": {
"label": "ID",
"list": true,
"type": "number"
},
"tenant_id": {
"label": "租户ID",
"list": true,
"type": "text"
},
"team_uid": {
"label": "团队UID",
"type": "text",
"list": true
},
"class_id": {
"label": "班级",
"type": "number",
"list": true
},
"name": {
"label": "名称",
"type": "select",
"list": true
},
"mission_id": {
"label": "Mission",
"type": "number",
"list": true
},
"members_json": {
"label": "成员",
"type": "text",
"list": true
},
"capacity": {
"label": "容量",
"type": "number",
"list": true
},
"status": {
"label": "状态",
"type": "select",
"list": true
},
"session_id": {
"label": "绑定会话",
"type": "number",
"list": true
}
},
"editable": {
"team_uid": {
"label": "团队UID",
"type": "text"
},
"class_id": {
"label": "班级",
"type": "number"
},
"name": {
"label": "名称",
"type": "select"
},
"mission_id": {
"label": "Mission",
"type": "number"
},
"members_json": {
"label": "成员",
"type": "text"
},
"capacity": {
"label": "容量",
"type": "number"
},
"status": {
"label": "状态",
"type": "select"
},
"session_id": {
"label": "绑定会话",
"type": "number"
}
},
"order_by": "id DESC",
"page_size": 20
}
}

View File

@ -1,132 +0,0 @@
{
"tblname": "pbl_tenant",
"alias": "pbl_domain_ext",
"browserfields": {
"id": {
"label": "ID",
"list": true,
"type": "number"
},
"tenant_id": {
"label": "租户ID",
"list": true,
"type": "text"
},
"tenant_uid": {
"label": "租户UID",
"type": "text",
"list": true
},
"name": {
"label": "名称",
"type": "select",
"list": true
},
"kind": {
"label": "类型",
"type": "select",
"list": true
},
"plan": {
"label": "套餐",
"type": "select",
"list": true
},
"seat_limit": {
"label": "席位",
"type": "number",
"list": true
},
"status": {
"label": "状态",
"type": "select",
"list": true
},
"config_json": {
"label": "配置",
"type": "text",
"list": true
}
},
"params": {
"browserfields": {
"id": {
"label": "ID",
"list": true,
"type": "number"
},
"tenant_id": {
"label": "租户ID",
"list": true,
"type": "text"
},
"tenant_uid": {
"label": "租户UID",
"type": "text",
"list": true
},
"name": {
"label": "名称",
"type": "select",
"list": true
},
"kind": {
"label": "类型",
"type": "select",
"list": true
},
"plan": {
"label": "套餐",
"type": "select",
"list": true
},
"seat_limit": {
"label": "席位",
"type": "number",
"list": true
},
"status": {
"label": "状态",
"type": "select",
"list": true
},
"config_json": {
"label": "配置",
"type": "text",
"list": true
}
},
"editable": {
"tenant_uid": {
"label": "租户UID",
"type": "text"
},
"name": {
"label": "名称",
"type": "select"
},
"kind": {
"label": "类型",
"type": "select"
},
"plan": {
"label": "套餐",
"type": "select"
},
"seat_limit": {
"label": "席位",
"type": "number"
},
"status": {
"label": "状态",
"type": "select"
},
"config_json": {
"label": "配置",
"type": "text"
}
},
"order_by": "id DESC",
"page_size": 20
}
}

View File

@ -1,107 +0,0 @@
{
"summary": [
{
"name": "pbl_class",
"comment": "班级(花名册内联 roster_json",
"module": "pbl_domain_ext",
"engine": "mariadb",
"charset": "utf8mb4",
"tenant_scoped": true
}
],
"fields": [
{
"name": "tenant_id",
"comment": "租户ID(强制打头)",
"null": false,
"type": "str",
"length": 32
},
{
"name": "id",
"comment": "主键",
"null": false,
"type": "int",
"length": 20,
"unsigned": true,
"auto_increment": true
},
{
"name": "class_uid",
"comment": "班级UID",
"null": false,
"type": "str",
"length": 32
},
{
"name": "name",
"comment": "名称",
"null": false,
"type": "str",
"length": 128
},
{
"name": "grade",
"comment": "年级",
"null": false,
"type": "str",
"length": 64
},
{
"name": "teacher_id",
"comment": "教师",
"null": false,
"type": "str",
"length": 32
},
{
"name": "student_count",
"comment": "学生数",
"null": false,
"type": "int"
},
{
"name": "roster_json",
"comment": "花名册",
"null": true,
"type": "json"
},
{
"name": "status",
"comment": "状态",
"null": false,
"type": "str",
"length": 64
},
{
"name": "created_at",
"comment": "创建时间",
"null": false,
"type": "datetime"
},
{
"name": "updated_at",
"comment": "更新时间",
"null": false,
"type": "datetime"
}
],
"indexes": [
{
"name": "uk_cls_uid",
"unique": true,
"fields": [
"tenant_id",
"class_uid"
]
},
{
"name": "PRIMARY",
"unique": true,
"fields": [
"id"
]
}
],
"codes": []
}

View File

@ -0,0 +1,42 @@
{
"summary": [
"pbl_domain_ref",
"world/scene/entity 薄扩展关联表M8不改基表",
"id",
"记录复用平台 world/scene/entity 基表记录与 PBL 租户/蓝图/班级/团队的关联;基表无 tenant_idPBL 多租户隔离由本表补齐。UNIQUE(tenant_id,ref_type,ref_id)。权威 DDLdata-model.md §J1表总账pbls_spec.json tables_by_module.pbl_domain_ext=1、tables_total=36。"
],
"fields": [
{"name": "id", "type": "bigint", "primary": true, "autoincr": true, "notnull": true, "comment": "主键"},
{"name": "tenant_id", "type": "varchar(64)", "notnull": true, "comment": "租户IDPBL 多租户隔离维度,基表无此列)"},
{"name": "ref_type", "type": "varchar(32)", "notnull": true, "comment": "appcodes:pbl_domain_ref_type(world/scene/entity)"},
{"name": "ref_id", "type": "bigint", "notnull": true, "comment": "复用基表记录IDworld.id / scene.id / entity.id"},
{"name": "blueprint_id", "type": "varchar(64)", "notnull": false, "comment": "关联蓝图IDpbl_compiler apply_game_definition 落库后回写)"},
{"name": "class_id", "type": "varchar(64)", "notnull": false, "comment": "关联教学班IDpbl_governance.pbl_class"},
{"name": "team_id", "type": "varchar(64)", "notnull": false, "comment": "关联团队IDpbl_governance.pbl_team共享世界成员分组"},
{"name": "ext", "type": "json", "notnull": false, "comment": "扩展属性JSON薄扩展附加维度禁止改基表结构"},
{"name": "is_deleted", "type": "tinyint", "notnull": true, "default": "0", "comment": "软删标记0=有效 1=已解绑unbind_ref 置 1"},
{"name": "created_at", "type": "timestamp", "notnull": true, "default": "CURRENT_TIMESTAMP", "comment": "创建时间"},
{"name": "updated_at", "type": "timestamp", "notnull": true, "default": "CURRENT_TIMESTAMP", "comment": "更新时间ON UPDATE CURRENT_TIMESTAMP"}
],
"indexes": [
{"name": "PRIMARY", "fields": ["id"], "unique": true, "primary": true},
{"name": "uk_pbl_domain_ref", "fields": ["tenant_id", "ref_type", "ref_id"], "unique": true, "comment": "设计 §2UNIQUE(tenant_id,ref_type,ref_id)"},
{"name": "idx_pbl_domain_ref_bp", "fields": ["tenant_id", "blueprint_id"], "unique": false, "comment": "按蓝图查关联compiler/analytics"},
{"name": "idx_pbl_domain_ref_class", "fields": ["tenant_id", "class_id"], "unique": false, "comment": "按教学班查关联F-AS-03 班级维度)"},
{"name": "idx_pbl_domain_ref_team", "fields": ["tenant_id", "team_id"], "unique": false, "comment": "按团队查关联US-13 共享世界分组)"},
{"name": "idx_pbl_domain_ref_type", "fields": ["ref_type", "ref_id"], "unique": false, "comment": "悬挂引用一致性检查(左连接基表)"}
],
"codes": [
{
"code": "pbl_domain_ref_type",
"name": "域扩展关联类型",
"owner": "pbl_appcodes",
"items": [
{"value": "world", "label": "世界(复用基表 world"},
{"value": "scene", "label": "场景(复用基表 scene"},
{"value": "entity", "label": "实体(复用基表 entity"}
],
"note": "编码由 pbl_appcodes 模块幂等注入,本模块只读引用,不重复写入编码表"
}
]
}

View File

@ -1,30 +0,0 @@
{
"summary": "PBL 实体关联表(薄扩展):把基表 entity 绑定到 PBL 租户/班级/团队 + scene/world 父级归属,不改 entity 基表结构Q-OPEN-3",
"tblname": "pbl_entity_ref",
"fields": [
{"name": "id", "type": "bigint", "primary": true, "auto": true, "comment": "主键"},
{"name": "tenant_id", "type": "varchar(64)", "null": false, "comment": "PBL 租户ID强制打头"},
{"name": "entity_id", "type": "bigint", "null": false, "comment": "基表 entity.id只读引用不建外键"},
{"name": "scene_id", "type": "bigint", "null": false, "default": 0, "comment": "父级归属 scene.id0=未指定"},
{"name": "world_id", "type": "bigint", "null": false, "default": 0, "comment": "父级归属 world.id0=未指定"},
{"name": "class_id", "type": "bigint", "null": false, "default": 0, "comment": "归属班级 pbl_class.id"},
{"name": "team_id", "type": "bigint", "null": false, "default": 0, "comment": "归属团队 pbl_team.id"},
{"name": "visibility", "type": "varchar(16)", "null": false, "default": "team", "comment": "可见性 team/class/tenant"},
{"name": "role_code", "type": "varchar(64)", "null": false, "default": "", "comment": "角色绑定(该实体扮演的蓝图角色),空=通用"},
{"name": "source", "type": "varchar(32)", "null": false, "default": "manual", "comment": "来源 manual/compiler/import"},
{"name": "ext_json", "type": "text", "null": true, "comment": "PBL 侧扩展属性 JSON实体在 PBL 语义下的额外属性,不回写基表)"},
{"name": "created_by", "type": "varchar(64)", "null": true, "comment": "创建人"},
{"name": "created_at", "type": "datetime", "null": false, "default": "CURRENT_TIMESTAMP", "comment": "创建时间"},
{"name": "updated_at", "type": "datetime", "null": true, "comment": "更新时间"}
],
"indexes": [
{"name": "uk_pbl_entity_ref", "unique": true, "fields": ["tenant_id", "entity_id", "team_id", "role_code"], "comment": "幂等唯一键"},
{"name": "idx_pbl_entity_ref_scene", "unique": false, "fields": ["tenant_id", "scene_id"], "comment": "按场景查实体"},
{"name": "idx_pbl_entity_ref_world", "unique": false, "fields": ["tenant_id", "world_id"], "comment": "按世界查实体"},
{"name": "idx_pbl_entity_ref_team", "unique": false, "fields": ["tenant_id", "team_id"], "comment": "按团队查"}
],
"codes": {
"visibility": {"team": "团队可见", "class": "班级可见", "tenant": "租户可见"},
"source": {"manual": "人工绑定", "compiler": "Compiler 物化", "import": "批量导入"}
}
}

View File

@ -1,29 +0,0 @@
{
"summary": "PBL 场景关联表(薄扩展):把基表 scene 绑定到 PBL 租户/班级/团队 + world 父级归属,不改 scene 基表结构Q-OPEN-3",
"tblname": "pbl_scene_ref",
"fields": [
{"name": "id", "type": "bigint", "primary": true, "auto": true, "comment": "主键"},
{"name": "tenant_id", "type": "varchar(64)", "null": false, "comment": "PBL 租户ID强制打头"},
{"name": "scene_id", "type": "bigint", "null": false, "comment": "基表 scene.id只读引用不建外键"},
{"name": "world_id", "type": "bigint", "null": false, "default": 0, "comment": "父级归属 world.id0=未指定"},
{"name": "class_id", "type": "bigint", "null": false, "default": 0, "comment": "归属班级 pbl_class.id"},
{"name": "team_id", "type": "bigint", "null": false, "default": 0, "comment": "归属团队 pbl_team.id"},
{"name": "visibility", "type": "varchar(16)", "null": false, "default": "team", "comment": "可见性 team/class/tenant"},
{"name": "role_code", "type": "varchar(64)", "null": false, "default": "", "comment": "角色绑定,空=通用"},
{"name": "source", "type": "varchar(32)", "null": false, "default": "manual", "comment": "来源 manual/compiler/import"},
{"name": "ext_json", "type": "text", "null": true, "comment": "PBL 侧扩展属性 JSON"},
{"name": "created_by", "type": "varchar(64)", "null": true, "comment": "创建人"},
{"name": "created_at", "type": "datetime", "null": false, "default": "CURRENT_TIMESTAMP", "comment": "创建时间"},
{"name": "updated_at", "type": "datetime", "null": true, "comment": "更新时间"}
],
"indexes": [
{"name": "uk_pbl_scene_ref", "unique": true, "fields": ["tenant_id", "scene_id", "team_id", "role_code"], "comment": "幂等唯一键"},
{"name": "idx_pbl_scene_ref_world", "unique": false, "fields": ["tenant_id", "world_id"], "comment": "按世界查场景"},
{"name": "idx_pbl_scene_ref_class", "unique": false, "fields": ["tenant_id", "class_id"], "comment": "按班级查"},
{"name": "idx_pbl_scene_ref_team", "unique": false, "fields": ["tenant_id", "team_id"], "comment": "按团队查"}
],
"codes": {
"visibility": {"team": "团队可见", "class": "班级可见", "tenant": "租户可见"},
"source": {"manual": "人工绑定", "compiler": "Compiler 物化", "import": "批量导入"}
}
}

View File

@ -1,119 +0,0 @@
{
"summary": [
{
"name": "pbl_team",
"comment": "团队(成员内联 members_json不改基表",
"module": "pbl_domain_ext",
"engine": "mariadb",
"charset": "utf8mb4",
"tenant_scoped": true
}
],
"fields": [
{
"name": "tenant_id",
"comment": "租户ID(强制打头)",
"null": false,
"type": "str",
"length": 32
},
{
"name": "id",
"comment": "主键",
"null": false,
"type": "int",
"length": 20,
"unsigned": true,
"auto_increment": true
},
{
"name": "team_uid",
"comment": "团队UID",
"null": false,
"type": "str",
"length": 32
},
{
"name": "class_id",
"comment": "班级",
"null": false,
"type": "bigint"
},
{
"name": "name",
"comment": "名称",
"null": false,
"type": "str",
"length": 128
},
{
"name": "mission_id",
"comment": "Mission",
"null": false,
"type": "bigint"
},
{
"name": "members_json",
"comment": "成员",
"null": true,
"type": "json"
},
{
"name": "capacity",
"comment": "容量",
"null": false,
"type": "int"
},
{
"name": "status",
"comment": "状态",
"null": false,
"type": "str",
"length": 64
},
{
"name": "session_id",
"comment": "绑定会话",
"null": false,
"type": "bigint"
},
{
"name": "created_at",
"comment": "创建时间",
"null": false,
"type": "datetime"
},
{
"name": "updated_at",
"comment": "更新时间",
"null": false,
"type": "datetime"
}
],
"indexes": [
{
"name": "uk_team_uid",
"unique": true,
"fields": [
"tenant_id",
"team_uid"
]
},
{
"name": "idx_team_class",
"unique": false,
"fields": [
"tenant_id",
"class_id"
]
},
{
"name": "PRIMARY",
"unique": true,
"fields": [
"id"
]
}
],
"codes": []
}

View File

@ -1,107 +0,0 @@
{
"summary": [
{
"name": "pbl_tenant",
"comment": "租户多租户边界第25章不改基表",
"module": "pbl_domain_ext",
"engine": "mariadb",
"charset": "utf8mb4",
"tenant_scoped": true
}
],
"fields": [
{
"name": "tenant_id",
"comment": "租户ID(强制打头)",
"null": false,
"type": "str",
"length": 32
},
{
"name": "id",
"comment": "主键",
"null": false,
"type": "int",
"length": 20,
"unsigned": true,
"auto_increment": true
},
{
"name": "tenant_uid",
"comment": "租户UID",
"null": false,
"type": "str",
"length": 32
},
{
"name": "name",
"comment": "名称",
"null": false,
"type": "str",
"length": 128
},
{
"name": "kind",
"comment": "类型",
"null": false,
"type": "str",
"length": 64
},
{
"name": "plan",
"comment": "套餐",
"null": false,
"type": "str",
"length": 64
},
{
"name": "seat_limit",
"comment": "席位",
"null": false,
"type": "int"
},
{
"name": "status",
"comment": "状态",
"null": false,
"type": "str",
"length": 64
},
{
"name": "config_json",
"comment": "配置",
"null": true,
"type": "json"
},
{
"name": "created_at",
"comment": "创建时间",
"null": false,
"type": "datetime"
},
{
"name": "updated_at",
"comment": "更新时间",
"null": false,
"type": "datetime"
}
],
"indexes": [
{
"name": "uk_ten_uid",
"unique": true,
"fields": [
"tenant_id",
"tenant_uid"
]
},
{
"name": "PRIMARY",
"unique": true,
"fields": [
"id"
]
}
],
"codes": []
}

View File

@ -1,28 +0,0 @@
{
"summary": "PBL 世界关联表(薄扩展):把基表 world 绑定到 PBL 租户/班级/团队,不改 world 基表结构Q-OPEN-3",
"tblname": "pbl_world_ref",
"fields": [
{"name": "id", "type": "bigint", "primary": true, "auto": true, "comment": "主键"},
{"name": "tenant_id", "type": "varchar(64)", "null": false, "comment": "PBL 租户ID强制打头所有查询必带"},
{"name": "world_id", "type": "bigint", "null": false, "comment": "基表 world.id只读引用不建外键以免锁基表"},
{"name": "class_id", "type": "bigint", "null": false, "default": 0, "comment": "归属班级 pbl_class.id0=未指定"},
{"name": "team_id", "type": "bigint", "null": false, "default": 0, "comment": "归属团队 pbl_team.id0=未指定"},
{"name": "visibility", "type": "varchar(16)", "null": false, "default": "team", "comment": "可见性 team/class/tenant"},
{"name": "role_code", "type": "varchar(64)", "null": false, "default": "", "comment": "角色绑定teacher/student/observer 或蓝图角色码),空=通用"},
{"name": "source", "type": "varchar(32)", "null": false, "default": "manual", "comment": "来源 manual/compiler/import"},
{"name": "ext_json", "type": "text", "null": true, "comment": "PBL 侧扩展属性 JSON不改基表的前提下承载额外字段"},
{"name": "created_by", "type": "varchar(64)", "null": true, "comment": "创建人"},
{"name": "created_at", "type": "datetime", "null": false, "default": "CURRENT_TIMESTAMP", "comment": "创建时间"},
{"name": "updated_at", "type": "datetime", "null": true, "comment": "更新时间"}
],
"indexes": [
{"name": "uk_pbl_world_ref", "unique": true, "fields": ["tenant_id", "world_id", "team_id", "role_code"], "comment": "幂等唯一键"},
{"name": "idx_pbl_world_ref_class", "unique": false, "fields": ["tenant_id", "class_id"], "comment": "按班级查"},
{"name": "idx_pbl_world_ref_team", "unique": false, "fields": ["tenant_id", "team_id"], "comment": "按团队查"},
{"name": "idx_pbl_world_ref_world", "unique": false, "fields": ["world_id"], "comment": "反查基表归属"}
],
"codes": {
"visibility": {"team": "团队可见", "class": "班级可见", "tenant": "租户可见"},
"source": {"manual": "人工绑定", "compiler": "Compiler 物化", "import": "批量导入"}
}
}

View File

@ -1,42 +1,34 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""pbl_domain_ext —— PBL 基础域薄扩展world/scene/entity 关联表 + 查询契约M8
"""pbl_domain_ext — world/scene/entity 薄扩展M8不改基表
对外暴露
- load_pbl_domain_ext()模块挂载入口应用 init() 调用
- api契约层pbl_*_ref_upsert / pbl_*_ref_list / pbl_domain_scope_resolve / ...
- assoc读写内核不依赖 Sage 运行时可离线单测
铁律world / scene / entity 三张基表零 ALTER零写入Q-OPEN-3
包导出三处同步注册之 init.py 中定义的 load_pbl_domain_ext
api.py 13 个契约函数必须在此导入否则 .dspy 调用报 NameError
"""
from __future__ import annotations
from .api import (CONTRACT_INTERFACES, bind_ref, bind_team_to_world,
check_ref_access, get_ref, get_team_worlds,
get_world_with_pbl_context, list_entities_by_scene, list_refs,
list_scenes_by_world, list_teams_by_class,
list_worlds_by_tenant, unbind_ref, update_ref)
from .db import (SqliteAdapter, SqlorAdapter, execute, query, set_adapter,
set_dbname, to_named_params)
from .errors import (CODE_HTTP, E_DUPLICATE, E_FORBIDDEN, E_INTERNAL,
E_NOT_FOUND, E_VALIDATION, PblError, fail, is_pbl_error, ok)
from .init import (MODULE_NAME, OWN_TABLES, READONLY_BASE_TABLES,
READONLY_FOREIGN_TABLES, get_contract_map,
load_pbl_domain_ext)
from . import api, assoc
from .init import (
API_NAMES,
MODULE_NAME,
OWN_TABLES,
READONLY_BASE_TABLES,
get_dbname,
load_cruds,
load_models,
load_module,
load_pbl_domain_ext,
)
__version__ = '1.0.0'
__all__ = [
"api",
"assoc",
"load_pbl_domain_ext",
"load_module",
"load_models",
"load_cruds",
"get_dbname",
"MODULE_NAME",
"OWN_TABLES",
"READONLY_BASE_TABLES",
"API_NAMES",
'load_pbl_domain_ext', 'get_contract_map', 'MODULE_NAME', 'OWN_TABLES',
'READONLY_BASE_TABLES', 'READONLY_FOREIGN_TABLES', 'CONTRACT_INTERFACES',
'bind_ref', 'unbind_ref', 'get_ref', 'list_refs', 'update_ref',
'list_worlds_by_tenant', 'list_scenes_by_world', 'list_entities_by_scene',
'get_world_with_pbl_context', 'check_ref_access', 'list_teams_by_class',
'bind_team_to_world', 'get_team_worlds',
'query', 'execute', 'set_adapter', 'set_dbname', 'to_named_params',
'SqlorAdapter', 'SqliteAdapter',
'ok', 'fail', 'PblError', 'is_pbl_error', 'CODE_HTTP',
'E_VALIDATION', 'E_NOT_FOUND', 'E_DUPLICATE', 'E_FORBIDDEN', 'E_INTERNAL',
'__version__',
]
__version__ = "2.0.0"

File diff suppressed because it is too large Load Diff

View File

@ -1,735 +0,0 @@
"""PBL 基础域薄扩展 —— world/scene/entity 关联表读写内核M8
设计约束Q-OPEN-3 决议 projects/pbls/docs/01-design/modules/pbl_domain_ext.md
1. **不改基表**world / scene / entity 三张基表结构与数据一律不动本模块只新增
三张 PBL 侧关联表pbl_world_ref / pbl_scene_ref / pbl_entity_ref承载归属关系
2. **租户强制打头**任何读写都必须带 tenant_id缺失/为空立即 fail-closed 抛错
不做默认租户兜底
3. **薄扩展**关联表只存归属tenant/class/team+ 可见性 + 角色绑定 + 扩展 JSON
业务状态仍留在基表与 scense_runtime不在本模块复制
4. **SQL 白名单**表名/列名一律经 _safe_col 校验值一律走占位符杜绝拼接注入
本文件不依赖 Sage 运行时sqlor/ServerEnv 仅在 SqlorDb 适配器里延迟导入
因此可被 tests/ 下的内存 DB 直接单测
"""
from __future__ import annotations
import json
import re
from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple
MODULE_NAME = "pbl_domain_ext"
# ---------------------------------------------------------------- 关联表常量
WORLD_REF = "pbl_world_ref"
SCENE_REF = "pbl_scene_ref"
ENTITY_REF = "pbl_entity_ref"
REF_TABLES: Tuple[str, ...] = (WORLD_REF, SCENE_REF, ENTITY_REF)
# 基表(只读引用,本模块绝不写入 / 绝不 DDL
BASE_TABLES: Tuple[str, ...] = ("world", "scene", "entity")
REF_SCHEMA: Dict[str, Dict[str, Any]] = {
WORLD_REF: {"key": "world_id", "parents": (), "base": "world"},
SCENE_REF: {"key": "scene_id", "parents": ("world_id",), "base": "scene"},
ENTITY_REF: {"key": "entity_id", "parents": ("scene_id", "world_id"), "base": "entity"},
}
# ---------------------------------------------------------------- 可见性 / 来源
VISIBILITY_TEAM = "team"
VISIBILITY_CLASS = "class"
VISIBILITY_TENANT = "tenant"
VISIBILITY_VALUES: Tuple[str, ...] = (VISIBILITY_TEAM, VISIBILITY_CLASS, VISIBILITY_TENANT)
SOURCE_MANUAL = "manual"
SOURCE_COMPILER = "compiler"
SOURCE_IMPORT = "import"
SOURCE_VALUES: Tuple[str, ...] = (SOURCE_MANUAL, SOURCE_COMPILER, SOURCE_IMPORT)
# ---------------------------------------------------------------- 错误码
ERR_TENANT_REQUIRED = "PBL_DOMAIN_EXT_TENANT_REQUIRED"
ERR_PARAM_REQUIRED = "PBL_DOMAIN_EXT_PARAM_REQUIRED"
ERR_BAD_VISIBILITY = "PBL_DOMAIN_EXT_BAD_VISIBILITY"
ERR_NOT_FOUND = "PBL_DOMAIN_EXT_NOT_FOUND"
ERR_CONFLICT = "PBL_DOMAIN_EXT_CONFLICT"
ERR_DB_UNAVAILABLE = "PBL_DOMAIN_EXT_DB_UNAVAILABLE"
MAX_LIMIT = 200
DEFAULT_LIMIT = 50
_COL_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
_SOURCE_RE = re.compile(r"^[A-Za-z0-9_\-]{1,32}$")
# role_code 允许蓝图角色码常见字符;显式禁掉引号/分号/空白,纵深防御(值本身仍走占位符)
_ROLE_RE = re.compile(r"^[A-Za-z0-9_\-.:@]{0,64}$")
class DomainExtError(Exception):
"""PBL 基础域薄扩展统一异常fail-closed带机器可读 code"""
def __init__(self, code: str, message: str, **detail: Any) -> None:
super().__init__(message)
self.code = code
self.message = message
self.detail: Dict[str, Any] = dict(detail)
def to_dict(self) -> Dict[str, Any]:
payload: Dict[str, Any] = {"code": self.code, "message": self.message}
if self.detail:
payload["detail"] = self.detail
return payload
# ================================================================ DB 适配层
class DbPort(object):
"""关联表读写所需的最小 DB 契约(便于单测注入内存实现)。"""
def select(self, sql: str, params: Sequence[Any] = ()) -> List[Dict[str, Any]]:
raise NotImplementedError
def insert(self, sql: str, params: Sequence[Any] = ()) -> int:
raise NotImplementedError
def execute(self, sql: str, params: Sequence[Any] = ()) -> int:
raise NotImplementedError
class SqlorDb(DbPort):
"""sqlor 适配器:库名走 ServerEnv().get_module_dbname禁止硬编码 DBNAME。"""
def __init__(self, dbname: Optional[str] = None) -> None:
self._dbname = dbname
self._sor: Any = None
@property
def sor(self) -> Any:
if self._sor is None:
try:
import sqlor as _sqlor # type: ignore
except Exception as exc: # pragma: no cover - 运行环境缺失
raise DomainExtError(ERR_DB_UNAVAILABLE, "sqlor 不可用: %s" % (exc,))
self._sor = _sqlor
return self._sor
@property
def dbname(self) -> str:
if self._dbname:
return self._dbname
try:
from appbase import ServerEnv # type: ignore
except Exception: # pragma: no cover
ServerEnv = None # type: ignore
if ServerEnv is not None:
try:
name = ServerEnv().get_module_dbname(MODULE_NAME)
if name:
return str(name)
except Exception: # pragma: no cover
pass
raise DomainExtError(
ERR_DB_UNAVAILABLE,
"无法解析模块库名ServerEnv().get_module_dbname('%s')" % MODULE_NAME,
)
def _run(self, sql: str, params: Sequence[Any]) -> Any:
args = tuple(params or ())
try:
return self.sor.sqlExe(self.dbname, sql, args)
except TypeError:
return self.sor.sqlExe(sql, args)
def select(self, sql: str, params: Sequence[Any] = ()) -> List[Dict[str, Any]]:
return _normalize_rows(self._run(sql, params))
def insert(self, sql: str, params: Sequence[Any] = ()) -> int:
result = self._run(sql, params)
if isinstance(result, (list, tuple)) and result:
head = result[0]
if isinstance(head, dict):
for key in ("lastrowid", "insert_id", "id"):
if key in head:
return int(head[key] or 0)
try:
return int(head)
except (TypeError, ValueError):
return 0
if isinstance(result, dict):
for key in ("lastrowid", "insert_id", "id"):
if key in result:
return int(result[key] or 0)
try:
return int(result or 0)
except (TypeError, ValueError):
return 0
def execute(self, sql: str, params: Sequence[Any] = ()) -> int:
result = self._run(sql, params)
if isinstance(result, (list, tuple)):
return len(result)
try:
return int(result or 0)
except (TypeError, ValueError):
return 0
def _normalize_rows(result: Any) -> List[Dict[str, Any]]:
"""把 sqlor 各种返回形态统一成 list[dict]。"""
if result is None:
return []
if isinstance(result, dict):
for key in ("rows", "data", "result", "items"):
if key in result:
return _normalize_rows(result[key])
return [result]
if isinstance(result, (list, tuple)):
rows: List[Dict[str, Any]] = []
for item in result:
if isinstance(item, dict):
rows.append(item)
elif isinstance(item, (list, tuple)):
rows.append({str(index): value for index, value in enumerate(item)})
return rows
return []
def default_db() -> DbPort:
"""缺省 DB生产路径sqlor + ServerEnv 库名映射。"""
return SqlorDb()
# ================================================================ 归一化 / 校验
def _norm_tenant(tenant_id: Any) -> str:
if tenant_id is None:
raise DomainExtError(ERR_TENANT_REQUIRED, "缺少 tenant_idPBL 全域强制租户打头)")
text = str(tenant_id).strip()
if not text:
raise DomainExtError(ERR_TENANT_REQUIRED, "tenant_id 不能为空字符串")
if len(text) > 64:
raise DomainExtError(ERR_TENANT_REQUIRED, "tenant_id 超长(>64", length=len(text))
return text
def _norm_id(value: Any, field: str, required: bool = True) -> int:
if value is None or (isinstance(value, str) and not value.strip()):
if required:
raise DomainExtError(ERR_PARAM_REQUIRED, "缺少必填参数 %s" % field, field=field)
return 0
try:
ivalue = int(str(value).strip())
except (TypeError, ValueError):
raise DomainExtError(ERR_PARAM_REQUIRED, "参数 %s 必须为整数,实际 %r" % (field, value), field=field)
if ivalue < 0:
raise DomainExtError(ERR_PARAM_REQUIRED, "参数 %s 不能为负数,实际 %d" % (field, ivalue), field=field)
if ivalue == 0 and required:
raise DomainExtError(ERR_PARAM_REQUIRED, "参数 %s 必须为正整数" % field, field=field)
return ivalue
def _norm_visibility(value: Any) -> str:
text = str(value or "").strip().lower()
if not text:
return VISIBILITY_TEAM
if text not in VISIBILITY_VALUES:
raise DomainExtError(
ERR_BAD_VISIBILITY,
"visibility 非法: %r,允许值 %s" % (value, "/".join(VISIBILITY_VALUES)),
)
return text
def _norm_role(role_code: Any) -> str:
text = str(role_code or "").strip()
if len(text) > 64:
raise DomainExtError(ERR_PARAM_REQUIRED, "role_code 超长(>64", length=len(text))
if text and not _ROLE_RE.match(text):
raise DomainExtError(
ERR_PARAM_REQUIRED,
"role_code 含非法字符: %r(限 [A-Za-z0-9_-.:@]≤64" % (role_code,),
)
return text
def _norm_source(value: Any) -> str:
text = str(value or SOURCE_MANUAL).strip()
if not _SOURCE_RE.match(text):
raise DomainExtError(ERR_PARAM_REQUIRED, "source 非法: %r(限 [A-Za-z0-9_-]{1,32}" % (value,))
return text
def _auto_visibility(class_id: int, team_id: int) -> str:
if team_id > 0:
return VISIBILITY_TEAM
if class_id > 0:
return VISIBILITY_CLASS
return VISIBILITY_TENANT
def _check_scope(visibility: str, class_id: int, team_id: int) -> None:
if visibility == VISIBILITY_TEAM and team_id <= 0:
raise DomainExtError(ERR_PARAM_REQUIRED, "visibility=team 时必须提供 team_id>0")
if visibility == VISIBILITY_CLASS and class_id <= 0:
raise DomainExtError(ERR_PARAM_REQUIRED, "visibility=class 时必须提供 class_id>0")
def _dump_ext(ext: Any) -> str:
if ext is None:
return ""
if isinstance(ext, str):
text = ext.strip()
if not text:
return ""
try:
json.loads(text)
except ValueError as exc:
raise DomainExtError(ERR_PARAM_REQUIRED, "ext 必须是合法 JSON 字符串: %s" % (exc,))
return text
try:
return json.dumps(ext, ensure_ascii=False, sort_keys=True)
except (TypeError, ValueError) as exc:
raise DomainExtError(ERR_PARAM_REQUIRED, "ext 无法序列化为 JSON: %s" % (exc,))
def _safe_col(name: Any) -> str:
text = str(name)
if not _COL_RE.match(text):
raise DomainExtError(ERR_PARAM_REQUIRED, "非法标识符(表名/列名): %r" % (name,))
return text
def _clamp_limit(limit: Any) -> int:
try:
value = int(limit)
except (TypeError, ValueError):
return DEFAULT_LIMIT
if value <= 0:
return DEFAULT_LIMIT
return min(value, MAX_LIMIT)
def _clamp_offset(offset: Any) -> int:
try:
value = int(offset)
except (TypeError, ValueError):
return 0
return max(0, min(value, 1000000))
def _columns(table: str) -> List[str]:
schema = REF_SCHEMA[table]
cols = ["tenant_id", schema["key"]]
cols.extend(schema["parents"])
cols.extend(["class_id", "team_id", "visibility", "role_code", "source", "ext_json", "created_by"])
return cols
# ================================================================ SQL 构造(受限子集)
def _build_where(conds: Sequence[Tuple[str, Any]]) -> Tuple[str, List[Any]]:
clauses: List[str] = []
params: List[Any] = []
for name, value in conds:
clauses.append("%s=%%s" % _safe_col(name))
params.append(value)
where = (" WHERE " + " AND ".join(clauses)) if clauses else ""
return where, params
def _select(db: DbPort, table: str, conds: Sequence[Tuple[str, Any]],
order_by: str = "id", limit: Any = DEFAULT_LIMIT, offset: Any = 0,
cols: str = "*") -> List[Dict[str, Any]]:
where, params = _build_where(conds)
sql = "SELECT %s FROM %s%s ORDER BY %s LIMIT %d OFFSET %d" % (
cols, _safe_col(table), where, _safe_col(order_by), _clamp_limit(limit), _clamp_offset(offset),
)
return db.select(sql, tuple(params))
def _count(db: DbPort, table: str, conds: Sequence[Tuple[str, Any]]) -> int:
where, params = _build_where(conds)
sql = "SELECT COUNT(*) AS cnt FROM %s%s" % (_safe_col(table), where)
rows = db.select(sql, tuple(params))
if not rows:
return 0
first = rows[0]
for key in ("cnt", "count", "COUNT(*)"):
if key in first:
return int(first[key] or 0)
values = list(first.values())
return int(values[0] or 0) if values else 0
def _require_table(table: str) -> str:
name = _safe_col(table)
if name not in REF_SCHEMA:
raise DomainExtError(ERR_PARAM_REQUIRED, "未知关联表: %s(允许 %s" % (table, "/".join(REF_TABLES)))
return name
# ================================================================ 通用读写
def _find_ref(db: DbPort, table: str, tenant: str, key_id: int, team_id: int, role_code: str) -> Optional[Dict[str, Any]]:
key_col = REF_SCHEMA[table]["key"]
conds = [("tenant_id", tenant), (key_col, key_id), ("team_id", team_id), ("role_code", role_code)]
rows = _select(db, table, conds, order_by="id", limit=1, offset=0)
return rows[0] if rows else None
def upsert_ref(db: DbPort, table: str, tenant_id: Any, key_value: Any,
parents: Optional[Dict[str, Any]] = None, class_id: Any = 0, team_id: Any = 0,
visibility: Any = None, role_code: Any = "", source: Any = SOURCE_MANUAL,
ext: Any = None, created_by: Any = "") -> Dict[str, Any]:
"""关联表幂等写入:唯一键 (tenant_id, 业务主键, team_id, role_code)。
存在则更新归属/可见性/扩展不存在则插入返回 action=created|updated
"""
name = _require_table(table)
schema = REF_SCHEMA[name]
tenant = _norm_tenant(tenant_id)
key_col = schema["key"]
key_id = _norm_id(key_value, key_col)
given = dict(parents or {})
parent_values: Dict[str, int] = {}
for pname in schema["parents"]:
parent_values[pname] = _norm_id(given.get(pname, 0), pname, required=False)
for extra in given:
if extra not in schema["parents"]:
raise DomainExtError(ERR_PARAM_REQUIRED, "%s 不支持父级列 %s" % (name, extra))
cid = _norm_id(class_id, "class_id", required=False)
tid = _norm_id(team_id, "team_id", required=False)
vis = _norm_visibility(visibility if visibility is not None else _auto_visibility(cid, tid))
_check_scope(vis, cid, tid)
role = _norm_role(role_code)
src = _norm_source(source)
ext_text = _dump_ext(ext)
existing = _find_ref(db, name, tenant, key_id, tid, role)
base_result: Dict[str, Any] = {
"table": name,
"tenant_id": tenant,
"class_id": cid,
"team_id": tid,
"visibility": vis,
"role_code": role,
"source": src,
}
base_result[key_col] = key_id
base_result.update(parent_values)
if existing is not None:
assigns = ["class_id=%s", "visibility=%s", "source=%s", "ext_json=%s", "updated_at=NOW()"]
for pname in schema["parents"]:
assigns.append("%s=%%s" % _safe_col(pname))
sql = "UPDATE %s SET %s WHERE tenant_id=%%s AND id=%%s" % (name, ", ".join(assigns))
params: List[Any] = [cid, vis, src, ext_text]
for pname in schema["parents"]:
params.append(parent_values[pname])
params.append(tenant)
params.append(int(existing.get("id", 0)))
db.execute(sql, tuple(params))
base_result["action"] = "updated"
base_result["id"] = int(existing.get("id", 0))
return base_result
cols = _columns(name)
sql = "INSERT INTO %s (%s) VALUES (%s)" % (name, ", ".join(cols), ", ".join(["%s"] * len(cols)))
params = [tenant, key_id]
for pname in schema["parents"]:
params.append(parent_values[pname])
params.extend([cid, tid, vis, role, src, ext_text, str(created_by or "")])
new_id = db.insert(sql, tuple(params))
base_result["action"] = "created"
base_result["id"] = int(new_id or 0)
return base_result
def list_refs(db: DbPort, table: str, tenant_id: Any, filters: Optional[Dict[str, Any]] = None,
limit: Any = DEFAULT_LIMIT, offset: Any = 0) -> Dict[str, Any]:
"""关联表分页查询tenant_id 强制打头,过滤字段走白名单。"""
name = _require_table(table)
schema = REF_SCHEMA[name]
tenant = _norm_tenant(tenant_id)
conds: List[Tuple[str, Any]] = [("tenant_id", tenant)]
allowed = set(schema["parents"]) | {"class_id", "team_id", "visibility", "role_code", "source", schema["key"]}
for field, value in dict(filters or {}).items():
if field not in allowed:
raise DomainExtError(ERR_PARAM_REQUIRED, "不支持的过滤字段: %s(允许 %s" % (field, sorted(allowed)))
if value is None or (isinstance(value, str) and not value.strip()):
continue
conds.append((field, value))
total = _count(db, name, conds)
items = _select(db, name, conds, order_by="id", limit=limit, offset=offset)
return {
"table": name,
"tenant_id": tenant,
"total": total,
"limit": _clamp_limit(limit),
"offset": _clamp_offset(offset),
"items": items,
}
def get_ref(db: DbPort, table: str, tenant_id: Any, key_value: Any, team_id: Any = 0,
role_code: Any = "", strict: bool = True) -> Optional[Dict[str, Any]]:
name = _require_table(table)
tenant = _norm_tenant(tenant_id)
key_col = REF_SCHEMA[name]["key"]
key_id = _norm_id(key_value, key_col)
tid = _norm_id(team_id, "team_id", required=False)
role = _norm_role(role_code)
row = _find_ref(db, name, tenant, key_id, tid, role)
if row is None and strict:
raise DomainExtError(ERR_NOT_FOUND, "关联记录不存在: %s.%s=%d" % (name, key_col, key_id), table=name)
return row
def delete_ref(db: DbPort, table: str, tenant_id: Any, ref_id: Any) -> int:
name = _require_table(table)
tenant = _norm_tenant(tenant_id)
rid = _norm_id(ref_id, "id")
sql = "DELETE FROM %s WHERE tenant_id=%%s AND id=%%s" % name
return int(db.execute(sql, (tenant, rid)) or 0)
# ================================================================ 三表薄封装
def upsert_world_ref(db: DbPort, tenant_id: Any, world_id: Any, class_id: Any = 0, team_id: Any = 0,
visibility: Any = None, role_code: Any = "", source: Any = SOURCE_MANUAL,
ext: Any = None, created_by: Any = "") -> Dict[str, Any]:
return upsert_ref(db, WORLD_REF, tenant_id, world_id, None, class_id, team_id,
visibility, role_code, source, ext, created_by)
def upsert_scene_ref(db: DbPort, tenant_id: Any, scene_id: Any, world_id: Any = 0, class_id: Any = 0,
team_id: Any = 0, visibility: Any = None, role_code: Any = "",
source: Any = SOURCE_MANUAL, ext: Any = None, created_by: Any = "") -> Dict[str, Any]:
return upsert_ref(db, SCENE_REF, tenant_id, scene_id, {"world_id": world_id}, class_id, team_id,
visibility, role_code, source, ext, created_by)
def upsert_entity_ref(db: DbPort, tenant_id: Any, entity_id: Any, scene_id: Any = 0, world_id: Any = 0,
class_id: Any = 0, team_id: Any = 0, visibility: Any = None, role_code: Any = "",
source: Any = SOURCE_MANUAL, ext: Any = None, created_by: Any = "") -> Dict[str, Any]:
return upsert_ref(db, ENTITY_REF, tenant_id, entity_id, {"scene_id": scene_id, "world_id": world_id},
class_id, team_id, visibility, role_code, source, ext, created_by)
def list_world_refs(db: DbPort, tenant_id: Any, filters: Optional[Dict[str, Any]] = None,
limit: Any = DEFAULT_LIMIT, offset: Any = 0) -> Dict[str, Any]:
return list_refs(db, WORLD_REF, tenant_id, filters, limit, offset)
def list_scene_refs(db: DbPort, tenant_id: Any, filters: Optional[Dict[str, Any]] = None,
limit: Any = DEFAULT_LIMIT, offset: Any = 0) -> Dict[str, Any]:
return list_refs(db, SCENE_REF, tenant_id, filters, limit, offset)
def list_entity_refs(db: DbPort, tenant_id: Any, filters: Optional[Dict[str, Any]] = None,
limit: Any = DEFAULT_LIMIT, offset: Any = 0) -> Dict[str, Any]:
return list_refs(db, ENTITY_REF, tenant_id, filters, limit, offset)
def domain_stats(db: DbPort, tenant_id: Any) -> Dict[str, Any]:
"""租户维度的关联表计数(供治理看板 / QC 核对)。"""
tenant = _norm_tenant(tenant_id)
stats: Dict[str, Any] = {"tenant_id": tenant}
for name in REF_TABLES:
stats[name] = _count(db, name, [("tenant_id", tenant)])
stats["total"] = sum(int(stats[name]) for name in REF_TABLES)
return stats
# ================================================================ 作用域解析
def _absorb(target: Dict[str, Any], row: Dict[str, Any]) -> None:
"""把关联行的归属信息吸收进解析结果(已确定的值不被覆盖:越具体越优先)。"""
for field in ("team_id", "class_id"):
if not target.get(field) and row.get(field) is not None:
try:
value = int(row.get(field) or 0)
except (TypeError, ValueError):
value = 0
if value > 0:
target[field] = value
if not target.get("visibility") and row.get("visibility"):
target["visibility"] = str(row["visibility"])
def resolve_scope(db: DbPort, tenant_id: Any, world_id: Any = None, scene_id: Any = None,
entity_id: Any = None) -> Dict[str, Any]:
"""由 entity → scene → world 逐级回溯,解析出 PBL 归属作用域。
返回 {tenant_id, world, scene, entity, world_id, scene_id, entity_id,
team_id, class_id, visibility, resolved_from}三级都查不到即 fail-closed
"""
tenant = _norm_tenant(tenant_id)
wid = _norm_id(world_id, "world_id", required=False)
sid = _norm_id(scene_id, "scene_id", required=False)
eid = _norm_id(entity_id, "entity_id", required=False)
if wid <= 0 and sid <= 0 and eid <= 0:
raise DomainExtError(ERR_PARAM_REQUIRED, "world_id / scene_id / entity_id 至少提供一个正整数")
result: Dict[str, Any] = {
"tenant_id": tenant, "world": None, "scene": None, "entity": None,
"world_id": wid, "scene_id": sid, "entity_id": eid,
"team_id": 0, "class_id": 0, "visibility": "", "resolved_from": "",
}
if eid > 0:
row = get_ref(db, ENTITY_REF, tenant, eid, strict=False)
if row is not None:
result["entity"] = row
result["resolved_from"] = "entity"
_absorb(result, row)
if sid <= 0:
sid = _norm_id(row.get("scene_id"), "scene_id", required=False)
result["scene_id"] = sid
if wid <= 0:
wid = _norm_id(row.get("world_id"), "world_id", required=False)
result["world_id"] = wid
if sid > 0:
row = get_ref(db, SCENE_REF, tenant, sid, strict=False)
if row is not None:
result["scene"] = row
if not result["resolved_from"]:
result["resolved_from"] = "scene"
_absorb(result, row)
if wid <= 0:
wid = _norm_id(row.get("world_id"), "world_id", required=False)
result["world_id"] = wid
if wid > 0:
row = get_ref(db, WORLD_REF, tenant, wid, strict=False)
if row is not None:
result["world"] = row
if not result["resolved_from"]:
result["resolved_from"] = "world"
_absorb(result, row)
if not result["resolved_from"]:
raise DomainExtError(
ERR_NOT_FOUND,
"未找到任何 PBL 关联记录world=%s scene=%s entity=%s" % (wid, sid, eid),
world_id=wid, scene_id=sid, entity_id=eid,
)
return result
# ================================================================ Game Definition 物化
_ID_KEYS: Dict[str, Tuple[str, ...]] = {
WORLD_REF: ("world_id", "worldId", "world"),
SCENE_REF: ("scene_id", "sceneId", "scene", "scenes"),
ENTITY_REF: ("entity_id", "entityId", "entity", "entities"),
}
def _walk(node: Any) -> Iterable[Any]:
if isinstance(node, dict):
yield node
for value in node.values():
for sub in _walk(value):
yield sub
elif isinstance(node, (list, tuple)):
for item in node:
for sub in _walk(item):
yield sub
def _as_id_list(value: Any) -> List[int]:
out: List[int] = []
if isinstance(value, (list, tuple)):
for item in value:
out.extend(_as_id_list(item))
return out
if isinstance(value, dict):
for key in ("id", "world_id", "scene_id", "entity_id"):
if key in value:
out.extend(_as_id_list(value[key]))
break
return out
if isinstance(value, bool) or value is None:
return out
try:
ivalue = int(str(value).strip())
except (TypeError, ValueError):
return out
if ivalue > 0:
out.append(ivalue)
return out
def collect_ids(game_definition: Any, table: str) -> List[int]:
"""从 Compiler 产出的 Game Definition 里递归抽取某类基表 id去重保序"""
name = _require_table(table)
if isinstance(game_definition, (str, bytes)):
game_definition = json.loads(game_definition)
ids: List[int] = []
seen = set()
for node in _walk(game_definition):
for key in _ID_KEYS[name]:
if key in node:
for cand in _as_id_list(node[key]):
if cand not in seen:
seen.add(cand)
ids.append(cand)
return ids
def bind_game_definition(db: DbPort, tenant_id: Any, game_definition: Any, class_id: Any = 0,
team_id: Any = 0, source: Any = SOURCE_COMPILER,
created_by: Any = "") -> Dict[str, Any]:
"""把 Compiler 产出的 Game Definition 物化成三张关联表的归属记录(幂等)。
只写 PBL 关联表不回写 world/scene/entity 基表
"""
tenant = _norm_tenant(tenant_id)
if isinstance(game_definition, (str, bytes)):
text = game_definition.decode("utf-8") if isinstance(game_definition, bytes) else game_definition
if not text.strip():
raise DomainExtError(ERR_PARAM_REQUIRED, "game_definition 为空")
try:
game_definition = json.loads(text)
except ValueError as exc:
raise DomainExtError(ERR_PARAM_REQUIRED, "game_definition 不是合法 JSON: %s" % (exc,))
if not isinstance(game_definition, dict):
raise DomainExtError(ERR_PARAM_REQUIRED, "game_definition 必须是对象或 JSON 字符串")
cid = _norm_id(class_id, "class_id", required=False)
tid = _norm_id(team_id, "team_id", required=False)
vis = _auto_visibility(cid, tid)
src = _norm_source(source)
world_ids = collect_ids(game_definition, WORLD_REF)
scene_ids = collect_ids(game_definition, SCENE_REF)
entity_ids = collect_ids(game_definition, ENTITY_REF)
summary: Dict[str, Any] = {
"tenant_id": tenant, "class_id": cid, "team_id": tid, "visibility": vis, "source": src,
"world_refs": 0, "scene_refs": 0, "entity_refs": 0, "created": 0, "updated": 0, "skipped": [],
}
if not world_ids and not scene_ids and not entity_ids:
summary["skipped"].append("game_definition 中未发现 world/scene/entity id")
return summary
primary_world = world_ids[0] if world_ids else 0
if len(world_ids) > 1:
summary["skipped"].append("发现多个 world_id=%s,父级归属取首个 %d" % (world_ids, primary_world))
primary_scene = scene_ids[0] if len(scene_ids) == 1 else 0
if len(scene_ids) > 1:
summary["skipped"].append("发现多个 scene_id=%sentity 的 scene 父级留空(0)待人工确认" % (scene_ids,))
for wid in world_ids:
res = upsert_world_ref(db, tenant, wid, cid, tid, vis, "", src, None, created_by)
summary["world_refs"] += 1
summary[res["action"]] = summary.get(res["action"], 0) + 1
for sid in scene_ids:
res = upsert_scene_ref(db, tenant, sid, primary_world, cid, tid, vis, "", src, None, created_by)
summary["scene_refs"] += 1
summary[res["action"]] = summary.get(res["action"], 0) + 1
for eid in entity_ids:
res = upsert_entity_ref(db, tenant, eid, primary_scene, primary_world, cid, tid, vis, "", src, None, created_by)
summary["entity_refs"] += 1
summary[res["action"]] = summary.get(res["action"], 0) + 1
return summary

178
pbl_domain_ext/base.py Normal file
View File

@ -0,0 +1,178 @@
"""pbl_domain_ext.base — 复用基表world/scene/entity**只读**投影层。
设计约束modules/pbl_domain_ext.md §4 / pbls_spec.json referenced_extension_rule
- 三禁止新增表改既有列语义删除列 本模块对基表** ALTER零写入**
- Phase 0/1 执行口径零改表PBL 维度全部落 pbl_domain_ref
- 优先经复用模块既有 read 契约取数ServerEnv 上的 world/scene/entity 函数
宿主未挂载时退化为只读 SELECT仍然零写入
悬挂引用一致性§4 风险基表记录被复用模块删除后 pbl_domain_ref 可能悬挂
本层提供 ``filter_existing_ids`` list_refs 左连接过滤失效 ref
"""
from . import db
from .errors import E_INTERNAL, raise_error
WORLD_TABLE = 'world'
SCENE_TABLE = 'scene'
ENTITY_TABLE = 'entity'
TABLE_BY_REF_TYPE = {
'world': WORLD_TABLE,
'scene': SCENE_TABLE,
'entity': ENTITY_TABLE,
}
# 基表父子关联列候选(不同版本复用模块列名可能不同,运行时探测,禁止猜死)
SCENE_PARENT_CANDIDATES = ('world_id', 'worldid', 'parent_id')
ENTITY_PARENT_CANDIDATES = ('scene_id', 'sceneid', 'parent_id')
WORLD_NAME_CANDIDATES = ('name', 'world_name', 'title')
_column_cache = {}
_read_func_cache = {}
def ref_type_to_table(ref_type):
table = TABLE_BY_REF_TYPE.get(ref_type)
if not table:
raise_error(E_INTERNAL, '未知 ref_type无法映射复用基表', ref_type=ref_type)
return table
async def table_columns(table):
"""探测基表列名MySQL information_schema / sqlite PRAGMA结果缓存。"""
if table in _column_cache:
return _column_cache[table]
cols = []
try:
rows = await db.query(
"SELECT COLUMN_NAME AS col FROM information_schema.COLUMNS "
"WHERE TABLE_NAME=${tbl}$", {'tbl': table})
cols = [str(r.get('col') or r.get('COLUMN_NAME') or '') for r in rows]
except Exception:
cols = []
if not cols:
try:
rows = await db.query("PRAGMA table_info(%s)" % table)
cols = [str(r.get('name', '')) for r in rows]
except Exception as exc:
raise_error(E_INTERNAL, '无法探测复用基表列(基表可能未部署)',
table=table, reason=str(exc))
cols = [c for c in cols if c]
_column_cache[table] = cols
return cols
async def resolve_column(table, candidates, required=True):
"""在候选列名中挑出基表真实存在的第一列(禁止凭记忆写死列名)。"""
cols = await table_columns(table)
lowered = {c.lower(): c for c in cols}
for cand in candidates:
hit = lowered.get(cand.lower())
if hit:
return hit
if required:
raise_error(E_INTERNAL, '复用基表缺少预期关联列', table=table,
candidates=list(candidates), actual=cols)
return None
def clear_cache():
_column_cache.clear()
_read_func_cache.clear()
def _env_read_func(names):
"""探测宿主 ServerEnv 上复用模块的既有 read 契约(只读函数)。"""
key = tuple(names)
if key in _read_func_cache:
return _read_func_cache[key]
func = None
try:
from ahserver.serverenv import ServerEnv # type: ignore
env = ServerEnv()
for name in names:
candidate = getattr(env, name, None)
if callable(candidate):
func = candidate
break
except Exception:
func = None
_read_func_cache[key] = func
return func
async def fetch_rows_by_ids(ref_type, ids):
"""按 id 批量只读取基表行;返回 {id: row}。优先走复用模块 read 契约。"""
ids = [int(i) for i in ids if i is not None]
if not ids:
return {}
table = ref_type_to_table(ref_type)
reader = _env_read_func(('list_%ss' % ref_type, 'get_%s' % ref_type))
if callable(reader):
try:
data = await reader({'ids': ids}) if _takes_dict(reader) else await reader(ids)
rows = _coerce_rows(data)
if rows:
return {int(r['id']): r for r in rows if r.get('id') is not None}
except Exception:
pass # 退化到只读 SELECT
marks = ','.join('${id%d}$' % i for i in range(len(ids)))
params = {'id%d' % i: int(v) for i, v in enumerate(ids)}
rows = await db.query(
"SELECT * FROM %s WHERE id IN (%s)" % (table, marks), params)
return {int(r['id']): r for r in rows if r.get('id') is not None}
async def fetch_row_by_id(ref_type, ref_id):
got = await fetch_rows_by_ids(ref_type, [ref_id])
return got.get(int(ref_id))
async def base_exists(ref_type, ref_id):
"""bind_ref 前置存在性校验§3.1 PBL_E_NOT_FOUND 分支)。"""
table = ref_type_to_table(ref_type)
rows = await db.query(
"SELECT id FROM %s WHERE id=${rid}$ LIMIT 1" % table, {'rid': int(ref_id)})
return bool(rows)
async def filter_existing_ids(ref_type, ids):
"""悬挂引用过滤:返回基表中仍存在的 id 集合。"""
got = await fetch_rows_by_ids(ref_type, ids)
return set(got.keys())
async def fetch_children(ref_type, parent_id):
"""按父记录取子基表行world→scene、scene→entity只读。"""
if ref_type == 'scene':
table = SCENE_TABLE
col = await resolve_column(SCENE_TABLE, SCENE_PARENT_CANDIDATES)
elif ref_type == 'entity':
table = ENTITY_TABLE
col = await resolve_column(ENTITY_TABLE, ENTITY_PARENT_CANDIDATES)
else:
raise_error(E_INTERNAL, 'fetch_children 仅支持 scene/entity', ref_type=ref_type)
return await db.query(
"SELECT * FROM %s WHERE %s=${pid}$" % (table, col), {'pid': int(parent_id)})
def _takes_dict(func):
try:
import inspect
sig = inspect.signature(func)
params = list(sig.parameters.values())
return len(params) == 1 and params[0].name in ('filters', 'params', 'ns', 'data')
except Exception:
return False
def _coerce_rows(data):
if isinstance(data, dict):
for key in ('items', 'rows', 'data', 'list'):
if isinstance(data.get(key), list):
return data[key]
return [data]
if isinstance(data, list):
return data
return []

201
pbl_domain_ext/db.py Normal file
View File

@ -0,0 +1,201 @@
"""pbl_domain_ext.db — 数据访问适配层M8 薄扩展)。
职责
1. 统一 SQL 占位符风格模块内 SQL 一律写 sqlor 风格 ``${name}$``
测试/异构后端可注入适配器sqlite3 ``:name``转换在适配器内完成
2. 统一取库名``ServerEnv().get_module_dbname('pbl_domain_ext')``禁止硬编码 DBNAME
3. 宿主可注入适配器``set_adapter``保证模块 host-agnosticmodule-development-spec
铁律本模块对 world/scene/entity 三张复用基表**只读**SELECT ALTER零写入
所有写入只落自有表 pbl_domain_ref1 data-model.md §J1
"""
import re
from .errors import E_INTERNAL, raise_error
MODULE_NAME = 'pbl_domain_ext'
_PLACEHOLDER_RE = re.compile(r'\$\{(\w+)\}\$')
_adapter = None
_dbname_override = None
def to_named_params(sql):
"""sqlor 风格 ``${name}$`` → DB-API 命名风格 ``:name``(异构后端/测试用)。"""
return _PLACEHOLDER_RE.sub(lambda m: ':' + m.group(1), sql)
def set_adapter(adapter):
"""注入数据访问适配器(宿主/测试。adapter 需实现 async query/execute。"""
global _adapter
_adapter = adapter
return adapter
def get_adapter():
return _adapter
def set_dbname(dbname):
"""覆盖库名(仅测试/特殊宿主使用;生产走 get_module_dbname"""
global _dbname_override
_dbname_override = dbname
def resolve_dbname():
"""按 module-development-spec 取库名ServerEnv().get_module_dbname(模块名)。"""
if _dbname_override:
return _dbname_override
last_err = None
try:
from ahserver.serverenv import ServerEnv # type: ignore
env = ServerEnv()
getter = getattr(env, 'get_module_dbname', None)
if callable(getter):
name = getter(MODULE_NAME)
if name:
return name
except Exception as exc: # pragma: no cover - 宿主未挂载时退化
last_err = exc
try: # 部分宿主把 get_module_dbname 注入为内置全局
import builtins
getter = getattr(builtins, 'get_module_dbname', None)
if callable(getter):
name = getter(MODULE_NAME)
if name:
return name
except Exception as exc: # pragma: no cover
last_err = exc
raise_error(E_INTERNAL, '无法解析模块库名get_module_dbname 不可用)',
module=MODULE_NAME, reason=str(last_err) if last_err else 'no provider')
class SqlorAdapter:
"""生产适配器sqlor 上下文sor.R 查询 / sor.sqlExe 执行)。
只用 sqlor 标准 APIsor.C/U/D/R/I/sqlExe不编造 save/list/insert
注意 pitfall``return`` 不能写在 ``async with`` 块内会静默返回 None
因此本类先在块内收集结果退出块后再 return
"""
def __init__(self, dbname=None):
self._dbname = dbname
@property
def dbname(self):
return self._dbname or resolve_dbname()
def _context(self):
"""按可用性依次探测 sqlor 上下文提供者。"""
errors = []
try:
from sqlor import DBPools # type: ignore
pools = DBPools()
if hasattr(pools, 'sqlorContext'):
return pools.sqlorContext(self.dbname)
except Exception as exc:
errors.append('sqlor.DBPools: %s' % exc)
try:
from ahserver.serverenv import ServerEnv # type: ignore
env = ServerEnv()
for attr in ('sqlorContext', 'sqlor_context'):
if hasattr(env, attr):
return getattr(env, attr)(self.dbname)
db = getattr(env, 'db', None)
if db is not None and hasattr(db, 'sqlorContext'):
return db.sqlorContext(self.dbname)
except Exception as exc:
errors.append('ServerEnv: %s' % exc)
try:
import builtins
ctx = getattr(builtins, 'sqlorContext', None)
if callable(ctx):
return ctx(self.dbname)
except Exception as exc:
errors.append('builtin sqlorContext: %s' % exc)
raise_error(E_INTERNAL, '无可用 sqlor 上下文(宿主未挂载数据库层)',
module=MODULE_NAME, dbname=self.dbname, probes=errors)
async def query(self, sql, params=None):
params = dict(params or {})
rows = []
async with self._context() as sor:
reader = getattr(sor, 'R', None)
if not callable(reader):
raise_error(E_INTERNAL, 'sqlor 上下文缺少 R() 查询接口')
rows = await reader(sql, params)
return _normalize_rows(rows)
async def execute(self, sql, params=None):
params = dict(params or {})
affected = 0
async with self._context() as sor:
runner = getattr(sor, 'sqlExe', None) or getattr(sor, 'I', None)
if not callable(runner):
raise_error(E_INTERNAL, 'sqlor 上下文缺少 sqlExe()/I() 执行接口')
affected = await runner(sql, params)
return affected if isinstance(affected, int) else 0
class SqliteAdapter:
"""测试/离线适配器sqlite3同步驱动包一层 async 接口)。
用于 tests/ 下跑真实 SQL 逻辑过滤/分页/唯一约束/软删/连接
不依赖生产 MySQL 与宿主挂载
"""
def __init__(self, conn):
self._conn = conn
self._conn.row_factory = _dict_factory
async def query(self, sql, params=None):
cur = self._conn.execute(to_named_params(sql), dict(params or {}))
rows = cur.fetchall()
cur.close()
return [dict(r) for r in rows]
async def execute(self, sql, params=None):
cur = self._conn.execute(to_named_params(sql), dict(params or {}))
self._conn.commit()
affected = cur.rowcount
cur.close()
return affected if affected and affected > 0 else 0
def _dict_factory(cursor, row):
return {col[0]: row[idx] for idx, col in enumerate(cursor.description)}
def _normalize_rows(rows):
"""sqlor 返回值可能是 list[dict] / list[list] / DictObject统一成 list[dict]。"""
if not rows:
return []
out = []
for row in rows:
if isinstance(row, dict):
out.append(dict(row))
elif hasattr(row, '_asdict'):
out.append(dict(row._asdict()))
elif hasattr(row, 'keys'):
try:
out.append({k: row[k] for k in row.keys()})
continue
except Exception:
out.append(dict(row))
else:
out.append(dict(row))
return out
async def query(sql, params=None):
"""模块内统一查询入口。"""
adapter = _adapter or SqlorAdapter()
return await adapter.query(sql, params)
async def execute(sql, params=None):
"""模块内统一执行入口(写操作只允许落 pbl_domain_ref"""
adapter = _adapter or SqlorAdapter()
return await adapter.execute(sql, params)

108
pbl_domain_ext/errors.py Normal file
View File

@ -0,0 +1,108 @@
"""pbl_domain_ext.errors — 错误码与异常M8 薄扩展)。
优先复用 pbl_common 的统一错误码/异常PBL_E_*pbl_common 不可用时
独立单测宿主未挂载退化为本地等价实现保证模块 host-agnostic
错误码语义与设计 projects/pbls/docs/01-design/modules/pbl_domain_ext.md §3 一致
PBL_E_VALIDATION 入参非法ref_type 不在 world/scene/entityref_id 非正整数tenant_id 缺失
PBL_E_NOT_FOUND 关联记录不存在 / 复用基表记录不存在
PBL_E_DUPLICATE 已绑定UNIQUE(tenant_id,ref_type,ref_id) 冲突
PBL_E_FORBIDDEN 跨租户/跨班级/跨团队越权访问US-21一律 403/404
PBL_E_INTERNAL 基础设施异常取不到 sqlor 上下文等
"""
try: # pragma: no cover - 依赖宿主是否挂载 pbl_common
from pbl_common.errors import PblError as _CommonPblError # type: ignore
from pbl_common.errors import error_payload as _common_error_payload # type: ignore
_HAVE_COMMON = True
except Exception: # pragma: no cover
_CommonPblError = None
_common_error_payload = None
_HAVE_COMMON = False
E_VALIDATION = 'PBL_E_VALIDATION'
E_NOT_FOUND = 'PBL_E_NOT_FOUND'
E_DUPLICATE = 'PBL_E_DUPLICATE'
E_FORBIDDEN = 'PBL_E_FORBIDDEN'
E_INTERNAL = 'PBL_E_INTERNAL'
# 错误码 → HTTP 语义dspy 契约层据此回填 statusUS-21 要求越权 403/404
CODE_HTTP = {
E_VALIDATION: 400,
E_NOT_FOUND: 404,
E_DUPLICATE: 409,
E_FORBIDDEN: 403,
E_INTERNAL: 500,
}
class PblDomainExtError(Exception):
"""本模块统一异常pbl_common.PblError 不可用时的等价实现)。"""
def __init__(self, code, message, detail=None):
super().__init__(message)
self.code = code
self.message = message
self.detail = detail or {}
def to_payload(self):
return {
'success': False,
'error_code': self.code,
'message': self.message,
'detail': self.detail,
'http_status': CODE_HTTP.get(self.code, 500),
}
if _HAVE_COMMON and _CommonPblError is not None:
# 复用 pbl_common 异常类型,保证跨模块错误处理一致
PblError = _CommonPblError
else:
PblError = PblDomainExtError
def raise_error(code, message, **detail):
"""抛出统一异常(兼容 pbl_common.PblError 的构造签名)。"""
try:
raise PblError(code, message, detail)
except TypeError:
# pbl_common.PblError 签名不同时退化为本模块异常
raise PblDomainExtError(code, message, detail)
def ok(data=None, **extra):
"""成功响应包体(契约层统一出口)。"""
payload = {'success': True, 'error_code': None, 'message': 'ok', 'data': data}
payload.update(extra)
return payload
def fail(code, message, **detail):
"""失败响应包体(契约层统一出口,不抛异常场景使用)。"""
payload = {
'success': False,
'error_code': code,
'message': message,
'detail': detail,
'http_status': CODE_HTTP.get(code, 500),
}
return payload
def is_pbl_error(exc):
return isinstance(exc, (PblDomainExtError,)) or (
_CommonPblError is not None and isinstance(exc, _CommonPblError)
)
def error_code_of(exc):
return getattr(exc, 'code', None) or E_INTERNAL
def error_message_of(exc):
return getattr(exc, 'message', None) or str(exc)
def error_detail_of(exc):
return getattr(exc, 'detail', None) or {}

View File

@ -1,124 +1,158 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""`load_pbl_domain_ext()` —— pbl_domain_ext 模块唯一挂载入口M8
"""pbl_domain_ext.init — 模块挂载入口load_pbl_domain_ext
三处同步注册module-development-spec
表定义 models/*.jsonpbl_tenant/pbl_class/pbl_team + pbl_world_ref/pbl_scene_ref/pbl_entity_ref
CRUD 定义 json/*.json
契约端点 wwwroot/api/*.dspy 本文件把 api.py 的契约函数挂到 ServerEnv
M8 薄扩展world/scene/entity 三模块的 PBL 侧关联叠加**不改基表**
自有表 1 pbl_domain_refdata-model.md §J1UNIQUE(tenant_id,ref_type,ref_id)
铁律
- **不改基表**world / scene / entity 三张基表零 ALTER零写入Q-OPEN-3 决议
PBL 归属关系只落在本模块自有关联表
- 库名走 ServerEnv().get_module_dbname('pbl_domain_ext')禁止硬编码 DBNAME
- 所有契约 tenant_id 强制打头缺失即 fail-closed
对外契约 = 设计 modules/pbl_domain_ext.md §3 13 个接口 wwwroot/api/*.dspy 一一对应
§3.1 关联管理 bind_ref / unbind_ref / get_ref / list_refs / update_ref
api/pbl_domain_ref_bind.dspy / _unbind / _get / _list / _update
§3.2 租户隔离封装 list_worlds_by_tenant / list_scenes_by_world / list_entities_by_scene
/ get_world_with_pbl_context / check_ref_access
api/pbl_world_list_by_tenant.dspy / pbl_scene_list_by_world.dspy
/ pbl_entity_list_by_scene.dspy / pbl_world_get_context.dspy
/ pbl_domain_ref_check_access.dspy
§3.3 团队/班级维度 list_teams_by_class / bind_team_to_world / get_team_worlds
api/pbl_team_list_by_class.dspy / pbl_team_bind_world.dspy
/ pbl_team_world_list.dspy
本文件对 ahserver 的导入是**延迟且容错** Sage 运行时离线单测 / QC 静态核验
load_pbl_domain_ext() 仍返回挂载摘要并把错误记入 summary['env_error']不抛异常
OWN_TABLES models/ 严格一一对应QC #4仅 pbl_domain_ref 1 张;
pbl_tenant/pbl_class/pbl_team pbl_governance 模块本模块只读引用不声明不建表
"""
from __future__ import annotations
from . import api
from . import base
from . import db
from .errors import (CODE_HTTP, E_DUPLICATE, E_FORBIDDEN, E_INTERNAL, E_NOT_FOUND,
E_VALIDATION, PblError, fail, is_pbl_error, ok)
from .api import (CONTRACT_INTERFACES, bind_ref, bind_team_to_world,
check_ref_access, get_ref, get_team_worlds,
get_world_with_pbl_context, list_entities_by_scene, list_refs,
list_scenes_by_world, list_teams_by_class,
list_worlds_by_tenant, unbind_ref, update_ref)
import json
import os
from typing import Any, Dict, List, Optional
MODULE_NAME = 'pbl_domain_ext'
from . import api, assoc
# 自有表(与 models/*.json、sql/pbl_domain_ext.sql 一一对应,共 1 张)
OWN_TABLES = ['pbl_domain_ref']
MODULE_NAME = assoc.MODULE_NAME
MODULE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# 只读引用的复用基表(零 ALTER、零写入
READONLY_BASE_TABLES = ['world', 'scene', 'entity']
# 本模块自有表(归属主数据 + 三张关联表)
OWN_TABLES: List[str] = [
"pbl_tenant",
"pbl_class",
"pbl_team",
assoc.WORLD_REF,
assoc.SCENE_REF,
assoc.ENTITY_REF,
]
# 只读引用的基表(本模块绝不 DDL / 绝不写入)
READONLY_BASE_TABLES: List[str] = list(assoc.BASE_TABLES)
# 契约清单(与 wwwroot/api/*.dspy 一一对应)
API_NAMES: List[str] = list(api.API_NAMES)
# 只读引用的他模块表pbl_governance 所有,本模块不建表不写入)
READONLY_FOREIGN_TABLES = ['pbl_team_member']
def get_module_dir() -> str:
return MODULE_DIR
def load_models() -> List[Dict[str, Any]]:
"""读取 models/*.json 表定义(供应用建表 / QC 核对字段)。"""
models: List[Dict[str, Any]] = []
model_dir = os.path.join(MODULE_DIR, "models")
if not os.path.isdir(model_dir):
return models
for fname in sorted(os.listdir(model_dir)):
if not fname.endswith(".json"):
continue
with open(os.path.join(model_dir, fname), "r", encoding="utf-8") as fp:
models.append(json.load(fp))
return models
def load_cruds() -> List[Dict[str, Any]]:
"""读取 json/*.json CRUD 定义。"""
cruds: List[Dict[str, Any]] = []
crud_dir = os.path.join(MODULE_DIR, "json")
if not os.path.isdir(crud_dir):
return cruds
for fname in sorted(os.listdir(crud_dir)):
if not fname.endswith(".json"):
continue
with open(os.path.join(crud_dir, fname), "r", encoding="utf-8") as fp:
cruds.append(json.load(fp))
return cruds
def get_dbname() -> str:
"""模块库名ServerEnv 映射优先,禁止硬编码。"""
return assoc.SqlorDb().dbname
def load_pbl_domain_ext(env: Optional[Any] = None) -> Dict[str, Any]:
"""挂载入口:应用 init() 里 `load_pbl_domain_ext()` 调用,返回挂载摘要。
env 缺省时尝试取 ahserver.ServerEnv()取不到离线环境则只返回摘要不抛错
"""
summary: Dict[str, Any] = {
"module": MODULE_NAME,
"module_dir": MODULE_DIR,
"own_tables": list(OWN_TABLES),
"readonly_base_tables": list(READONLY_BASE_TABLES),
"apis": list(API_NAMES),
"models": len(load_models()),
"cruds": len(load_cruds()),
"registered": [],
def get_contract_map():
"""契约接口 → (实现位置, dspy 端点) 映射,供 QC/PM 核对需求覆盖。"""
return {
'bind_ref': ('pbl_domain_ext/api.py:bind_ref', 'api/pbl_domain_ref_bind.dspy'),
'unbind_ref': ('pbl_domain_ext/api.py:unbind_ref', 'api/pbl_domain_ref_unbind.dspy'),
'get_ref': ('pbl_domain_ext/api.py:get_ref', 'api/pbl_domain_ref_get.dspy'),
'list_refs': ('pbl_domain_ext/api.py:list_refs', 'api/pbl_domain_ref_list.dspy'),
'update_ref': ('pbl_domain_ext/api.py:update_ref', 'api/pbl_domain_ref_update.dspy'),
'list_worlds_by_tenant': ('pbl_domain_ext/api.py:list_worlds_by_tenant',
'api/pbl_world_list_by_tenant.dspy'),
'list_scenes_by_world': ('pbl_domain_ext/api.py:list_scenes_by_world',
'api/pbl_scene_list_by_world.dspy'),
'list_entities_by_scene': ('pbl_domain_ext/api.py:list_entities_by_scene',
'api/pbl_entity_list_by_scene.dspy'),
'get_world_with_pbl_context': ('pbl_domain_ext/api.py:get_world_with_pbl_context',
'api/pbl_world_get_context.dspy'),
'check_ref_access': ('pbl_domain_ext/api.py:check_ref_access',
'api/pbl_domain_ref_check_access.dspy'),
'list_teams_by_class': ('pbl_domain_ext/api.py:list_teams_by_class',
'api/pbl_team_list_by_class.dspy'),
'bind_team_to_world': ('pbl_domain_ext/api.py:bind_team_to_world',
'api/pbl_team_bind_world.dspy'),
'get_team_worlds': ('pbl_domain_ext/api.py:get_team_worlds',
'api/pbl_team_world_list.dspy'),
}
target = env
if target is None:
def load_pbl_domain_ext(env=None):
"""挂载模块:把 13 个契约函数注册到 ServerEnv三处同步注册之 ③)。
env 缺省时自动取宿主 ServerEnv返回 env 便于链式挂载
"""
if env is None:
try:
from ahserver.serverenv import ServerEnv # type: ignore
target = ServerEnv()
except Exception as exc: # pragma: no cover - 离线环境
summary["env_error"] = "ahserver.ServerEnv 不可用: %s" % (exc,)
env = ServerEnv()
except Exception:
env = _FallbackEnv()
if target is not None:
for name in API_NAMES:
setattr(target, name, getattr(api, name))
summary["registered"].append(name)
# §3.1 关联管理
env.pbl_bind_ref = bind_ref
env.pbl_unbind_ref = unbind_ref
env.pbl_get_ref = get_ref
env.pbl_list_refs = list_refs
env.pbl_update_ref = update_ref
# §3.2 租户隔离查询封装
env.pbl_list_worlds_by_tenant = list_worlds_by_tenant
env.pbl_list_scenes_by_world = list_scenes_by_world
env.pbl_list_entities_by_scene = list_entities_by_scene
env.pbl_get_world_with_pbl_context = get_world_with_pbl_context
env.pbl_check_ref_access = check_ref_access
# §3.3 团队/班级维度
env.pbl_list_teams_by_class = list_teams_by_class
env.pbl_bind_team_to_world = bind_team_to_world
env.pbl_get_team_worlds = get_team_worlds
try:
summary["dbname"] = get_dbname()
except assoc.DomainExtError as exc:
summary["dbname"] = None
summary["dbname_error"] = exc.to_dict()
# 设计原名(无前缀)同步注册,供其他 pbl_* 模块直接按设计接口名调用
env.bind_ref = bind_ref
env.unbind_ref = unbind_ref
env.get_ref = get_ref
env.list_refs = list_refs
env.update_ref = update_ref
env.list_worlds_by_tenant = list_worlds_by_tenant
env.list_scenes_by_world = list_scenes_by_world
env.list_entities_by_scene = list_entities_by_scene
env.get_world_with_pbl_context = get_world_with_pbl_context
env.check_ref_access = check_ref_access
env.list_teams_by_class = list_teams_by_class
env.bind_team_to_world = bind_team_to_world
env.get_team_worlds = get_team_worlds
return summary
# 模块元信息(供宿主/巡检读取)
env.pbl_domain_ext_module_info = {
'module': MODULE_NAME,
'milestone': 'M8',
'own_tables': list(OWN_TABLES),
'readonly_base_tables': list(READONLY_BASE_TABLES),
'readonly_foreign_tables': list(READONLY_FOREIGN_TABLES),
'contracts': list(CONTRACT_INTERFACES),
'contract_map': get_contract_map(),
'base_table_altered': False,
}
return env
# 兼容旧命名
load_module = load_pbl_domain_ext
class _FallbackEnv:
"""无宿主 ServerEnv 时的最小注册容器(保证模块可独立单测)。"""
def __init__(self):
self._attrs = {}
def __setattr__(self, name, value):
if name.startswith('_'):
object.__setattr__(self, name, value)
else:
self._attrs[name] = value
def __getattr__(self, name):
attrs = object.__getattribute__(self, '_attrs')
if name in attrs:
return attrs[name]
raise AttributeError(name)
__all__ = [
'load_pbl_domain_ext', 'get_contract_map', 'MODULE_NAME', 'OWN_TABLES',
'READONLY_BASE_TABLES', 'READONLY_FOREIGN_TABLES', 'api', 'base', 'db',
'bind_ref', 'unbind_ref', 'get_ref', 'list_refs', 'update_ref',
'list_worlds_by_tenant', 'list_scenes_by_world', 'list_entities_by_scene',
'get_world_with_pbl_context', 'check_ref_access', 'list_teams_by_class',
'bind_team_to_world', 'get_team_worlds', 'CONTRACT_INTERFACES',
'ok', 'fail', 'PblError', 'is_pbl_error', 'CODE_HTTP',
'E_VALIDATION', 'E_NOT_FOUND', 'E_DUPLICATE', 'E_FORBIDDEN', 'E_INTERNAL',
]

View File

@ -1,51 +1,30 @@
-- pbl_domain_ext 表 DDL自动生成与 apps/pbls/scripts/ddl/pbls_tables.sql 同源)
CREATE TABLE IF NOT EXISTS `pbl_tenant` (
`tenant_id` VARCHAR(32) NOT NULL NOT NULL COMMENT ID(),
`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT ,
`tenant_uid` VARCHAR(32) NOT NULL COMMENT "租户UID",
`name` VARCHAR(128) NOT NULL COMMENT "名称",
`kind` VARCHAR(64) NOT NULL COMMENT "类型",
`plan` VARCHAR(64) NOT NULL COMMENT "套餐",
`seat_limit` INT NOT NULL DEFAULT 0 COMMENT "席位",
`status` VARCHAR(64) NOT NULL COMMENT "状态",
`config_json` LONGTEXT NULL COMMENT "配置",
`created_at` DATETIME NOT NULL DEFAULT '1970-01-01 00:00:00' COMMENT "创建时间(应用层写入)",
`updated_at` DATETIME NOT NULL DEFAULT '1970-01-01 00:00:00' COMMENT "更新时间(应用层写入)",
PRIMARY KEY (`id`),
UNIQUE KEY `uk_ten_uid` (`tenant_id`, `tenant_uid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT="租户多租户边界第25章不改基表";
-- =====================================================================
-- pbl_domain_extM8— world/scene/entity 薄扩展
-- 权威 DDLprojects/pbls/docs/01-design/data-model.md §JJ1. pbl_domain_ref
-- 表总账projects/pbls/pbls_spec.json tables_by_module.pbl_domain_ext = 1
-- 铁律:不改 world / scene / entity 三张复用基表(零 ALTER、零侵入模块代码
-- =====================================================================
CREATE TABLE IF NOT EXISTS `pbl_class` (
`tenant_id` VARCHAR(32) NOT NULL NOT NULL COMMENT ID(),
`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT ,
`class_uid` VARCHAR(32) NOT NULL COMMENT "班级UID",
`name` VARCHAR(128) NOT NULL COMMENT "名称",
`grade` VARCHAR(64) NOT NULL COMMENT "年级",
`teacher_id` VARCHAR(32) NOT NULL COMMENT "教师",
`student_count` INT NOT NULL DEFAULT 0 COMMENT "学生数",
`roster_json` LONGTEXT NULL COMMENT "花名册",
`status` VARCHAR(64) NOT NULL COMMENT "状态",
`created_at` DATETIME NOT NULL DEFAULT '1970-01-01 00:00:00' COMMENT "创建时间(应用层写入)",
`updated_at` DATETIME NOT NULL DEFAULT '1970-01-01 00:00:00' COMMENT "更新时间(应用层写入)",
PRIMARY KEY (`id`),
UNIQUE KEY `uk_cls_uid` (`tenant_id`, `class_uid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT="班级(花名册内联 roster_json";
CREATE TABLE IF NOT EXISTS `pbl_team` (
`tenant_id` VARCHAR(32) NOT NULL NOT NULL COMMENT ID(),
`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT ,
`team_uid` VARCHAR(32) NOT NULL COMMENT "团队UID",
`class_id` BIGINT UNSIGNED NOT NULL COMMENT "班级",
`name` VARCHAR(128) NOT NULL COMMENT "名称",
`mission_id` BIGINT UNSIGNED NOT NULL COMMENT "Mission",
`members_json` LONGTEXT NULL COMMENT "成员",
`capacity` INT NOT NULL DEFAULT 0 COMMENT "容量",
`status` VARCHAR(64) NOT NULL COMMENT "状态",
`session_id` BIGINT UNSIGNED NOT NULL COMMENT "绑定会话",
`created_at` DATETIME NOT NULL DEFAULT '1970-01-01 00:00:00' COMMENT "创建时间(应用层写入)",
`updated_at` DATETIME NOT NULL DEFAULT '1970-01-01 00:00:00' COMMENT "更新时间(应用层写入)",
PRIMARY KEY (`id`),
UNIQUE KEY `uk_team_uid` (`tenant_id`, `team_uid`),
KEY `idx_team_class` (`tenant_id`, `class_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT="团队(成员内联 members_json不改基表";
-- J1. pbl_domain_ref — world/scene/entity 薄扩展关联ref_type + ref_id 唯一)
CREATE TABLE IF NOT EXISTS pbl_domain_ref (
id BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键',
tenant_id VARCHAR(64) NOT NULL COMMENT '租户ID基表无租户列PBL 多租户隔离由本表补齐)',
ref_type VARCHAR(32) NOT NULL COMMENT 'appcodes:pbl_domain_ref_type(world/scene/entity)',
ref_id BIGINT NOT NULL COMMENT '复用基表记录IDworld.id / scene.id / entity.id',
blueprint_id VARCHAR(64) DEFAULT NULL COMMENT '关联蓝图IDpbl_blueprintcompiler apply 后回写)',
class_id VARCHAR(64) DEFAULT NULL COMMENT '关联教学班IDpbl_governance.pbl_class',
team_id VARCHAR(64) DEFAULT NULL COMMENT '关联团队IDpbl_governance.pbl_team共享世界分组',
ext JSON DEFAULT NULL COMMENT '扩展属性JSON薄扩展附加维度不改基表',
is_deleted TINYINT NOT NULL DEFAULT 0 COMMENT '软删标记0=有效 1=已解绑unbind_ref 置 1',
created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
PRIMARY KEY (id),
UNIQUE KEY uk_pbl_domain_ref (tenant_id, ref_type, ref_id),
KEY idx_pbl_domain_ref_bp (tenant_id, blueprint_id),
KEY idx_pbl_domain_ref_class (tenant_id, class_id),
KEY idx_pbl_domain_ref_team (tenant_id, team_id),
KEY idx_pbl_domain_ref_type (ref_type, ref_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='world/scene/entity 薄扩展关联表M8不改基表';
-- 编码ref_type 取值受 appcodes:pbl_domain_ref_type 约束world/scene/entity
-- 由 pbl_appcodes 模块幂等注入,本模块不重复写入编码表。

View File

@ -1,69 +0,0 @@
-- ============================================================
-- pbl_domain_ext —— M8 基础域薄扩展 DDLworld/scene/entity 关联表)
-- 铁律:不修改 world / scene / entity 三张基表Q-OPEN-3 决议)
-- 关联表不建外键(避免锁基表 / 跨库),只存 id 引用
-- 方言MariaDB / MySQLBIGINT AUTO_INCREMENT
-- ============================================================
CREATE TABLE IF NOT EXISTS pbl_world_ref (
id BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键',
tenant_id VARCHAR(64) NOT NULL COMMENT 'PBL 租户ID强制打头',
world_id BIGINT NOT NULL COMMENT '基表 world.id只读引用',
class_id BIGINT NOT NULL DEFAULT 0 COMMENT '归属班级 pbl_class.id',
team_id BIGINT NOT NULL DEFAULT 0 COMMENT '归属团队 pbl_team.id',
visibility VARCHAR(16) NOT NULL DEFAULT 'team' COMMENT '可见性 team/class/tenant',
role_code VARCHAR(64) NOT NULL DEFAULT '' COMMENT '角色绑定,空=通用',
source VARCHAR(32) NOT NULL DEFAULT 'manual' COMMENT '来源 manual/compiler/import',
ext_json TEXT NULL COMMENT 'PBL 侧扩展属性 JSON',
created_by VARCHAR(64) NULL COMMENT '创建人',
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
updated_at DATETIME NULL COMMENT '更新时间',
PRIMARY KEY (id),
UNIQUE KEY uk_pbl_world_ref (tenant_id, world_id, team_id, role_code),
KEY idx_pbl_world_ref_class (tenant_id, class_id),
KEY idx_pbl_world_ref_team (tenant_id, team_id),
KEY idx_pbl_world_ref_world (world_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='PBL 世界关联表(薄扩展,不改 world 基表)';
CREATE TABLE IF NOT EXISTS pbl_scene_ref (
id BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键',
tenant_id VARCHAR(64) NOT NULL COMMENT 'PBL 租户ID强制打头',
scene_id BIGINT NOT NULL COMMENT '基表 scene.id只读引用',
world_id BIGINT NOT NULL DEFAULT 0 COMMENT '父级归属 world.id',
class_id BIGINT NOT NULL DEFAULT 0 COMMENT '归属班级 pbl_class.id',
team_id BIGINT NOT NULL DEFAULT 0 COMMENT '归属团队 pbl_team.id',
visibility VARCHAR(16) NOT NULL DEFAULT 'team' COMMENT '可见性 team/class/tenant',
role_code VARCHAR(64) NOT NULL DEFAULT '' COMMENT '角色绑定,空=通用',
source VARCHAR(32) NOT NULL DEFAULT 'manual' COMMENT '来源 manual/compiler/import',
ext_json TEXT NULL COMMENT 'PBL 侧扩展属性 JSON',
created_by VARCHAR(64) NULL COMMENT '创建人',
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
updated_at DATETIME NULL COMMENT '更新时间',
PRIMARY KEY (id),
UNIQUE KEY uk_pbl_scene_ref (tenant_id, scene_id, team_id, role_code),
KEY idx_pbl_scene_ref_world (tenant_id, world_id),
KEY idx_pbl_scene_ref_class (tenant_id, class_id),
KEY idx_pbl_scene_ref_team (tenant_id, team_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='PBL 场景关联表(薄扩展,不改 scene 基表)';
CREATE TABLE IF NOT EXISTS pbl_entity_ref (
id BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键',
tenant_id VARCHAR(64) NOT NULL COMMENT 'PBL 租户ID强制打头',
entity_id BIGINT NOT NULL COMMENT '基表 entity.id只读引用',
scene_id BIGINT NOT NULL DEFAULT 0 COMMENT '父级归属 scene.id',
world_id BIGINT NOT NULL DEFAULT 0 COMMENT '父级归属 world.id',
class_id BIGINT NOT NULL DEFAULT 0 COMMENT '归属班级 pbl_class.id',
team_id BIGINT NOT NULL DEFAULT 0 COMMENT '归属团队 pbl_team.id',
visibility VARCHAR(16) NOT NULL DEFAULT 'team' COMMENT '可见性 team/class/tenant',
role_code VARCHAR(64) NOT NULL DEFAULT '' COMMENT '角色绑定,空=通用',
source VARCHAR(32) NOT NULL DEFAULT 'manual' COMMENT '来源 manual/compiler/import',
ext_json TEXT NULL COMMENT 'PBL 侧扩展属性 JSON',
created_by VARCHAR(64) NULL COMMENT '创建人',
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
updated_at DATETIME NULL COMMENT '更新时间',
PRIMARY KEY (id),
UNIQUE KEY uk_pbl_entity_ref (tenant_id, entity_id, team_id, role_code),
KEY idx_pbl_entity_ref_scene (tenant_id, scene_id),
KEY idx_pbl_entity_ref_world (tenant_id, world_id),
KEY idx_pbl_entity_ref_team (tenant_id, team_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='PBL 实体关联表(薄扩展,不改 entity 基表)';

View File

@ -1,514 +0,0 @@
"""M8 基础域薄扩展离线单测(不依赖 Sage 运行时 / 不连库)。
运行cd modules/pbl_domain_ext && python3 -m pytest tests/ -q
python3 tests/test_assoc.py pytest 时直接跑
覆盖租户强制打头 / 幂等 upsert / 可见性推导与校验 / 过滤白名单 /
作用域回溯解析 / Game Definition 物化 / 不改基表铁律 / 契约层错误码
"""
from __future__ import annotations
import json
import os
import sys
import unittest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from fake_db import FakeDb # noqa: E402 (tests/ 目录内导入)
from pbl_domain_ext import api, assoc # noqa: E402
from pbl_domain_ext.assoc import DomainExtError # noqa: E402
TENANT = "T-DEMO-001"
def new_db() -> FakeDb:
return FakeDb([assoc.WORLD_REF, assoc.SCENE_REF, assoc.ENTITY_REF])
class TestTenantGuard(unittest.TestCase):
"""租户强制打头:缺失/空/超长一律 fail-closed。"""
def test_missing_tenant_raises(self):
db = new_db()
for bad in (None, "", " "):
with self.assertRaises(DomainExtError) as ctx:
assoc.upsert_world_ref(db, bad, 101)
self.assertEqual(ctx.exception.code, assoc.ERR_TENANT_REQUIRED)
def test_overlong_tenant_raises(self):
db = new_db()
with self.assertRaises(DomainExtError) as ctx:
assoc.upsert_world_ref(db, "T" * 65, 101)
self.assertEqual(ctx.exception.code, assoc.ERR_TENANT_REQUIRED)
def test_missing_key_raises(self):
db = new_db()
with self.assertRaises(DomainExtError) as ctx:
assoc.upsert_world_ref(db, TENANT, None)
self.assertEqual(ctx.exception.code, assoc.ERR_PARAM_REQUIRED)
def test_non_integer_key_raises(self):
db = new_db()
with self.assertRaises(DomainExtError):
assoc.upsert_scene_ref(db, TENANT, "abc")
def test_negative_key_raises(self):
db = new_db()
with self.assertRaises(DomainExtError):
assoc.upsert_entity_ref(db, TENANT, -3)
def test_list_without_tenant_raises(self):
db = new_db()
with self.assertRaises(DomainExtError) as ctx:
assoc.list_world_refs(db, None)
self.assertEqual(ctx.exception.code, assoc.ERR_TENANT_REQUIRED)
class TestUpsertIdempotent(unittest.TestCase):
def test_create_then_update_same_row(self):
db = new_db()
first = assoc.upsert_world_ref(db, TENANT, 101, class_id=7, team_id=9, role_code="teacher")
self.assertEqual(first["action"], "created")
self.assertGreater(first["id"], 0)
second = assoc.upsert_world_ref(db, TENANT, 101, class_id=8, team_id=9, role_code="teacher")
self.assertEqual(second["action"], "updated")
self.assertEqual(second["id"], first["id"])
self.assertEqual(len(db.tables[assoc.WORLD_REF]), 1)
self.assertEqual(db.tables[assoc.WORLD_REF][0]["class_id"], 8)
def test_different_role_is_different_row(self):
db = new_db()
assoc.upsert_world_ref(db, TENANT, 101, team_id=9, role_code="teacher")
assoc.upsert_world_ref(db, TENANT, 101, team_id=9, role_code="student")
self.assertEqual(len(db.tables[assoc.WORLD_REF]), 2)
def test_different_tenant_isolated(self):
db = new_db()
assoc.upsert_world_ref(db, TENANT, 101, team_id=9)
assoc.upsert_world_ref(db, "T-OTHER", 101, team_id=9)
self.assertEqual(len(db.tables[assoc.WORLD_REF]), 2)
page = assoc.list_world_refs(db, TENANT)
self.assertEqual(page["total"], 1)
self.assertEqual(page["items"][0]["tenant_id"], TENANT)
def test_scene_and_entity_parents_persisted(self):
db = new_db()
assoc.upsert_scene_ref(db, TENANT, 202, world_id=101, team_id=9)
assoc.upsert_entity_ref(db, TENANT, 303, scene_id=202, world_id=101, team_id=9)
scene_row = db.tables[assoc.SCENE_REF][0]
entity_row = db.tables[assoc.ENTITY_REF][0]
self.assertEqual(scene_row["world_id"], 101)
self.assertEqual(entity_row["scene_id"], 202)
self.assertEqual(entity_row["world_id"], 101)
def test_unknown_parent_column_rejected(self):
db = new_db()
with self.assertRaises(DomainExtError):
assoc.upsert_ref(db, assoc.SCENE_REF, TENANT, 202, {"entity_id": 1})
def test_ext_json_serialized(self):
db = new_db()
assoc.upsert_world_ref(db, TENANT, 101, team_id=9, ext={"stage": 2, "tags": ["a", "b"]})
stored = db.tables[assoc.WORLD_REF][0]["ext_json"]
self.assertEqual(json.loads(stored)["stage"], 2)
def test_ext_string_json_passthrough(self):
db = new_db()
assoc.upsert_world_ref(db, TENANT, 102, team_id=9, ext='{"k": 1}')
self.assertEqual(json.loads(db.tables[assoc.WORLD_REF][0]["ext_json"])["k"], 1)
def test_ext_bad_json_rejected(self):
db = new_db()
with self.assertRaises(DomainExtError):
assoc.upsert_world_ref(db, TENANT, 101, team_id=9, ext="{not json")
def test_bad_source_rejected(self):
db = new_db()
with self.assertRaises(DomainExtError):
assoc.upsert_world_ref(db, TENANT, 101, team_id=9, source="drop table;")
class TestVisibility(unittest.TestCase):
def test_auto_visibility_by_scope(self):
db = new_db()
self.assertEqual(assoc.upsert_world_ref(db, TENANT, 1, team_id=5)["visibility"], "team")
self.assertEqual(assoc.upsert_world_ref(db, TENANT, 2, class_id=5)["visibility"], "class")
self.assertEqual(assoc.upsert_world_ref(db, TENANT, 3)["visibility"], "tenant")
def test_explicit_visibility_must_match_scope(self):
db = new_db()
with self.assertRaises(DomainExtError) as ctx:
assoc.upsert_world_ref(db, TENANT, 4, visibility="team")
self.assertEqual(ctx.exception.code, assoc.ERR_PARAM_REQUIRED)
with self.assertRaises(DomainExtError):
assoc.upsert_world_ref(db, TENANT, 5, visibility="class")
def test_bad_visibility_value(self):
db = new_db()
with self.assertRaises(DomainExtError) as ctx:
assoc.upsert_world_ref(db, TENANT, 6, visibility="public")
self.assertEqual(ctx.exception.code, assoc.ERR_BAD_VISIBILITY)
class TestListAndFilter(unittest.TestCase):
def setUp(self):
self.db = new_db()
for i in range(1, 8):
assoc.upsert_world_ref(self.db, TENANT, 100 + i, class_id=7, team_id=9)
assoc.upsert_world_ref(self.db, "T-OTHER", 999, team_id=1)
def test_pagination(self):
page = assoc.list_world_refs(self.db, TENANT, limit=3, offset=0)
self.assertEqual(page["total"], 7)
self.assertEqual(len(page["items"]), 3)
page2 = assoc.list_world_refs(self.db, TENANT, limit=3, offset=6)
self.assertEqual(len(page2["items"]), 1)
def test_limit_clamped(self):
page = assoc.list_world_refs(self.db, TENANT, limit=99999)
self.assertEqual(page["limit"], assoc.MAX_LIMIT)
def test_filter_whitelist(self):
page = assoc.list_world_refs(self.db, TENANT, {"class_id": 7})
self.assertEqual(page["total"], 7)
with self.assertRaises(DomainExtError):
assoc.list_world_refs(self.db, TENANT, {"password": "x"})
def test_empty_filter_value_ignored(self):
page = assoc.list_world_refs(self.db, TENANT, {"role_code": "", "team_id": 9})
self.assertEqual(page["total"], 7)
def test_scene_filter_by_world(self):
assoc.upsert_scene_ref(self.db, TENANT, 201, world_id=101, team_id=9)
assoc.upsert_scene_ref(self.db, TENANT, 202, world_id=102, team_id=9)
page = assoc.list_scene_refs(self.db, TENANT, {"world_id": 101})
self.assertEqual(page["total"], 1)
self.assertEqual(page["items"][0]["scene_id"], 201)
class TestGetDelete(unittest.TestCase):
def test_get_strict_and_lenient(self):
db = new_db()
assoc.upsert_entity_ref(db, TENANT, 303, scene_id=202, world_id=101, team_id=9)
row = assoc.get_ref(db, assoc.ENTITY_REF, TENANT, 303, team_id=9)
self.assertEqual(row["scene_id"], 202)
with self.assertRaises(DomainExtError) as ctx:
assoc.get_ref(db, assoc.ENTITY_REF, TENANT, 404, team_id=9)
self.assertEqual(ctx.exception.code, assoc.ERR_NOT_FOUND)
self.assertIsNone(assoc.get_ref(db, assoc.ENTITY_REF, TENANT, 404, team_id=9, strict=False))
def test_delete_only_own_tenant(self):
db = new_db()
res = assoc.upsert_world_ref(db, TENANT, 101, team_id=9)
assoc.upsert_world_ref(db, "T-OTHER", 101, team_id=9)
affected = assoc.delete_ref(db, assoc.WORLD_REF, TENANT, res["id"])
self.assertEqual(affected, 1)
self.assertEqual(len(db.tables[assoc.WORLD_REF]), 1)
self.assertEqual(db.tables[assoc.WORLD_REF][0]["tenant_id"], "T-OTHER")
class TestScopeResolve(unittest.TestCase):
def setUp(self):
self.db = new_db()
assoc.upsert_world_ref(self.db, TENANT, 101, class_id=7, team_id=9)
assoc.upsert_scene_ref(self.db, TENANT, 202, world_id=101, class_id=7, team_id=9)
assoc.upsert_entity_ref(self.db, TENANT, 303, scene_id=202, world_id=101, team_id=9)
def test_resolve_from_entity_backfills_parents(self):
scope = assoc.resolve_scope(self.db, TENANT, entity_id=303)
self.assertEqual(scope["resolved_from"], "entity")
self.assertEqual(scope["scene_id"], 202)
self.assertEqual(scope["world_id"], 101)
self.assertEqual(scope["team_id"], 9)
self.assertEqual(scope["class_id"], 7)
self.assertEqual(scope["visibility"], "team")
self.assertIsNotNone(scope["world"])
def test_resolve_from_scene_only(self):
scope = assoc.resolve_scope(self.db, TENANT, scene_id=202)
self.assertEqual(scope["resolved_from"], "scene")
self.assertEqual(scope["world_id"], 101)
def test_resolve_from_world_only(self):
scope = assoc.resolve_scope(self.db, TENANT, world_id=101)
self.assertEqual(scope["resolved_from"], "world")
self.assertEqual(scope["class_id"], 7)
def test_resolve_requires_one_id(self):
with self.assertRaises(DomainExtError) as ctx:
assoc.resolve_scope(self.db, TENANT)
self.assertEqual(ctx.exception.code, assoc.ERR_PARAM_REQUIRED)
def test_resolve_not_found_fail_closed(self):
with self.assertRaises(DomainExtError) as ctx:
assoc.resolve_scope(self.db, TENANT, entity_id=999)
self.assertEqual(ctx.exception.code, assoc.ERR_NOT_FOUND)
def test_resolve_cross_tenant_not_leaked(self):
with self.assertRaises(DomainExtError):
assoc.resolve_scope(self.db, "T-OTHER", entity_id=303)
def test_resolve_entity_without_scene_parent(self):
assoc.upsert_entity_ref(self.db, TENANT, 404, world_id=101, team_id=9)
scope = assoc.resolve_scope(self.db, TENANT, entity_id=404)
self.assertEqual(scope["resolved_from"], "entity")
self.assertEqual(scope["world_id"], 101)
self.assertIsNone(scope["scene"])
class TestMaterialize(unittest.TestCase):
DEFINITION = {
"game": {
"world_id": 101,
"scenes": [{"scene_id": 202, "entities": [{"entity_id": 303}, {"entityId": 304}]}],
},
"meta": {"worldId": 101},
}
def test_collect_ids_dedup(self):
ids = assoc.collect_ids(self.DEFINITION, assoc.WORLD_REF)
self.assertEqual(ids, [101])
self.assertEqual(assoc.collect_ids(self.DEFINITION, assoc.SCENE_REF), [202])
self.assertEqual(sorted(assoc.collect_ids(self.DEFINITION, assoc.ENTITY_REF)), [303, 304])
def test_collect_ids_from_json_string(self):
ids = assoc.collect_ids(json.dumps(self.DEFINITION), assoc.WORLD_REF)
self.assertEqual(ids, [101])
def test_bind_creates_all_refs(self):
db = new_db()
summary = assoc.bind_game_definition(db, TENANT, self.DEFINITION, class_id=7, team_id=9)
self.assertEqual(summary["world_refs"], 1)
self.assertEqual(summary["scene_refs"], 1)
self.assertEqual(summary["entity_refs"], 2)
self.assertEqual(summary["created"], 4)
self.assertEqual(summary["source"], "compiler")
self.assertEqual(len(db.tables[assoc.ENTITY_REF]), 2)
def test_bind_is_idempotent(self):
db = new_db()
assoc.bind_game_definition(db, TENANT, self.DEFINITION, team_id=9)
second = assoc.bind_game_definition(db, TENANT, json.dumps(self.DEFINITION), team_id=9)
self.assertEqual(second["updated"], 4)
self.assertEqual(second["created"], 0)
self.assertEqual(len(db.tables[assoc.WORLD_REF]), 1)
def test_bind_rejects_empty_and_bad_json(self):
db = new_db()
with self.assertRaises(DomainExtError):
assoc.bind_game_definition(db, TENANT, " ")
with self.assertRaises(DomainExtError):
assoc.bind_game_definition(db, TENANT, "{bad json")
with self.assertRaises(DomainExtError):
assoc.bind_game_definition(db, TENANT, [1, 2, 3])
def test_bind_without_ids_reports_skip(self):
db = new_db()
summary = assoc.bind_game_definition(db, TENANT, {"game": {"name": "x"}})
self.assertEqual(summary["world_refs"], 0)
self.assertTrue(summary["skipped"])
def test_multiple_scenes_leave_parent_zero(self):
db = new_db()
definition = {"scenes": [{"scene_id": 1}, {"scene_id": 2}], "entities": [{"entity_id": 5}]}
summary = assoc.bind_game_definition(db, TENANT, definition, team_id=9)
self.assertEqual(summary["scene_refs"], 2)
self.assertEqual(db.tables[assoc.ENTITY_REF][0]["scene_id"], 0)
self.assertTrue(any("多个 scene_id" in s for s in summary["skipped"]))
def test_multiple_worlds_take_first(self):
db = new_db()
definition = {"worlds": [{"world_id": 11}, {"world_id": 22}], "scenes": [{"scene_id": 5}]}
summary = assoc.bind_game_definition(db, TENANT, definition, team_id=9)
self.assertEqual(summary["world_refs"], 2)
self.assertEqual(db.tables[assoc.SCENE_REF][0]["world_id"], 11)
self.assertTrue(any("多个 world_id" in s for s in summary["skipped"]))
class TestBaseTableUntouched(unittest.TestCase):
"""铁律:任何路径都不得写 world/scene/entity 基表。"""
def test_no_base_table_write(self):
db = new_db()
assoc.upsert_world_ref(db, TENANT, 101, team_id=9)
assoc.upsert_scene_ref(db, TENANT, 202, world_id=101, team_id=9)
assoc.upsert_entity_ref(db, TENANT, 303, scene_id=202, world_id=101, team_id=9)
assoc.bind_game_definition(db, TENANT, TestMaterialize.DEFINITION, team_id=9)
assoc.resolve_scope(db, TENANT, entity_id=303)
self.assertEqual(db.base_table_writes, [])
for sql, _params in db.executed:
lowered = " ".join(sql.lower().split())
if lowered.startswith(("insert", "update", "delete")):
for base in ("world", "scene", "entity"):
self.assertFalse(lowered.startswith("insert into %s " % base), sql)
self.assertFalse(lowered.startswith("update %s " % base), sql)
self.assertFalse(lowered.startswith("delete from %s " % base), sql)
def test_all_writes_target_ref_tables_only(self):
db = new_db()
assoc.bind_game_definition(db, TENANT, TestMaterialize.DEFINITION, team_id=9)
write_tables = set()
for sql, _params in db.executed:
lowered = " ".join(sql.lower().split())
if lowered.startswith("insert into"):
write_tables.add(lowered.split()[2])
elif lowered.startswith("update"):
write_tables.add(lowered.split()[1])
elif lowered.startswith("delete from"):
write_tables.add(lowered.split()[2])
self.assertTrue(write_tables)
self.assertTrue(write_tables.issubset(set(assoc.REF_TABLES)), write_tables)
def test_table_name_injection_guard(self):
db = new_db()
with self.assertRaises(DomainExtError):
assoc.list_refs(db, "pbl_world_ref; DROP TABLE world", TENANT)
with self.assertRaises(DomainExtError):
assoc.upsert_ref(db, "world", TENANT, 1)
def test_filter_column_injection_guard(self):
db = new_db()
with self.assertRaises(DomainExtError):
assoc.list_refs(db, assoc.WORLD_REF, TENANT, {"1=1 OR tenant_id": "x"})
def test_values_always_parameterized(self):
db = new_db()
evil = "x'; DROP TABLE pbl_world_ref; --"
assoc.upsert_world_ref(db, TENANT, 101, team_id=9, role_code=evil)
for sql, params in db.executed:
self.assertNotIn("DROP TABLE", sql)
self.assertEqual(db.tables[assoc.WORLD_REF][0]["role_code"], evil)
self.assertEqual(len(db.tables[assoc.WORLD_REF]), 1)
class TestApiContract(unittest.TestCase):
def setUp(self):
self.db = new_db()
def call(self, name, **params):
params["_db"] = self.db
return api.call_api(name, params)
def test_upsert_and_list_ok(self):
res = self.call("pbl_world_ref_upsert", tenant_id=TENANT, world_id=101, class_id=7, team_id=9)
self.assertTrue(res["ok"], res)
self.assertEqual(res["data"]["action"], "created")
page = self.call("pbl_world_ref_list", tenant_id=TENANT, team_id=9)
self.assertTrue(page["ok"])
self.assertEqual(page["data"]["total"], 1)
def test_scene_and_entity_upsert_ok(self):
scene = self.call("pbl_scene_ref_upsert", tenant_id=TENANT, scene_id=202, world_id=101, team_id=9)
entity = self.call("pbl_entity_ref_upsert", tenant_id=TENANT, entity_id=303, scene_id=202,
world_id=101, team_id=9, ext={"hp": 100})
self.assertTrue(scene["ok"], scene)
self.assertTrue(entity["ok"], entity)
self.assertEqual(entity["data"]["visibility"], "team")
def test_missing_tenant_returns_error_code(self):
res = self.call("pbl_world_ref_list", world_id=101)
self.assertFalse(res["ok"])
self.assertEqual(res["code"], assoc.ERR_TENANT_REQUIRED)
def test_unknown_api(self):
res = api.call_api("pbl_not_exist", {"tenant_id": TENANT})
self.assertFalse(res["ok"])
self.assertEqual(res["code"], assoc.ERR_PARAM_REQUIRED)
def test_unknown_table_alias(self):
res = self.call("pbl_domain_ref_get", table="payroll", tenant_id=TENANT, id_value=1)
self.assertFalse(res["ok"])
def test_ref_get_and_delete(self):
created = self.call("pbl_scene_ref_upsert", tenant_id=TENANT, scene_id=202, world_id=101, team_id=9)
rid = created["data"]["id"]
got = self.call("pbl_domain_ref_get", table="scene", tenant_id=TENANT, scene_id=202, team_id=9)
self.assertTrue(got["ok"])
self.assertTrue(got["found"])
deleted = self.call("pbl_domain_ref_delete", table="pbl_scene_ref", tenant_id=TENANT, id=rid)
self.assertEqual(deleted["data"]["deleted"], 1)
missing = self.call("pbl_domain_ref_get", table="scene", tenant_id=TENANT, scene_id=202,
team_id=9, strict=False)
self.assertTrue(missing["ok"])
self.assertFalse(missing["found"])
def test_ref_get_strict_not_found(self):
res = self.call("pbl_domain_ref_get", table="world", tenant_id=TENANT, world_id=777)
self.assertFalse(res["ok"])
self.assertEqual(res["code"], assoc.ERR_NOT_FOUND)
def test_scope_resolve_and_stats(self):
self.call("pbl_world_ref_upsert", tenant_id=TENANT, world_id=101, class_id=7, team_id=9)
self.call("pbl_scene_ref_upsert", tenant_id=TENANT, scene_id=202, world_id=101, team_id=9)
self.call("pbl_entity_ref_upsert", tenant_id=TENANT, entity_id=303, scene_id=202,
world_id=101, team_id=9)
scope = self.call("pbl_domain_scope_resolve", tenant_id=TENANT, entity_id=303)
self.assertTrue(scope["ok"])
self.assertEqual(scope["data"]["resolved_from"], "entity")
self.assertEqual(scope["data"]["world_id"], 101)
stats = self.call("pbl_domain_stats", tenant_id=TENANT)
self.assertEqual(stats["data"]["total"], 3)
self.assertEqual(stats["data"][assoc.ENTITY_REF], 1)
def test_materialize_contract(self):
res = self.call("pbl_domain_materialize_game_definition", tenant_id=TENANT,
game_definition=TestMaterialize.DEFINITION, class_id=7, team_id=9)
self.assertTrue(res["ok"], res)
self.assertEqual(res["data"]["world_refs"], 1)
self.assertEqual(res["data"]["entity_refs"], 2)
bad = self.call("pbl_domain_materialize_game_definition", tenant_id=TENANT,
game_definition="{oops")
self.assertFalse(bad["ok"])
self.assertEqual(bad["code"], assoc.ERR_PARAM_REQUIRED)
def test_materialize_missing_definition(self):
res = self.call("pbl_domain_materialize_game_definition", tenant_id=TENANT)
self.assertFalse(res["ok"])
self.assertEqual(res["code"], assoc.ERR_PARAM_REQUIRED)
class TestInitRegistration(unittest.TestCase):
def test_module_metadata(self):
from pbl_domain_ext import init as init_mod
self.assertEqual(init_mod.MODULE_NAME, "pbl_domain_ext")
for table in (assoc.WORLD_REF, assoc.SCENE_REF, assoc.ENTITY_REF):
self.assertIn(table, init_mod.OWN_TABLES)
self.assertEqual(init_mod.READONLY_BASE_TABLES, ["world", "scene", "entity"])
self.assertIn("pbl_domain_scope_resolve", init_mod.API_NAMES)
def test_models_and_cruds_loadable(self):
from pbl_domain_ext import init as init_mod
models = init_mod.load_models()
names = {m.get("tblname") for m in models}
for table in (assoc.WORLD_REF, assoc.SCENE_REF, assoc.ENTITY_REF):
self.assertIn(table, names)
self.assertGreaterEqual(len(init_mod.load_cruds()), 3)
def test_load_entry_returns_summary_without_runtime(self):
from pbl_domain_ext import init as init_mod
summary = init_mod.load_pbl_domain_ext()
self.assertEqual(summary["module"], "pbl_domain_ext")
self.assertIsNone(summary["dbname"]) # 无 ServerEnv 时不抛,只记录错误
self.assertIn("dbname_error", summary)
def test_load_entry_registers_apis_on_env(self):
from pbl_domain_ext import init as init_mod
class FakeEnv(object):
def __init__(self):
self.apis = {}
def register_api(self, name, handler):
self.apis[name] = handler
env = FakeEnv()
init_mod.load_pbl_domain_ext(env)
self.assertIn("pbl_world_ref_list", env.apis)
if __name__ == "__main__":
unittest.main(verbosity=2)

View File

@ -1,4 +0,0 @@
# pbl_domain_ext/api/pbl_class_list.dspy —— 契约端点(自动生成,勿手改:改 spec 后跑 gen_artifacts.py
debug('pbl_domain_ext/api/pbl_class_list.dspy: START params_kw={dict(params_kw)}')
data = await pbl_class_list(**params_kw)
return data

View File

@ -1,4 +0,0 @@
# pbl_domain_ext/api/pbl_class_save.dspy —— 契约端点(自动生成,勿手改:改 spec 后跑 gen_artifacts.py
debug('pbl_domain_ext/api/pbl_class_save.dspy: START params_kw={dict(params_kw)}')
data = await pbl_class_save(**params_kw)
return data

View File

@ -1,4 +0,0 @@
# pbl_domain_ext/api/pbl_domain_materialize_game_definition.dspy —— 契约端点(自动生成,勿手改:改 spec 后跑 gen_artifacts.py
debug('pbl_domain_ext/api/pbl_domain_materialize_game_definition.dspy: START params_kw={dict(params_kw)}')
data = await pbl_domain_materialize_game_definition(**params_kw)
return data

View File

@ -0,0 +1,29 @@
# api/pbl_domain_ref_bind.dspy — 契约端点bind_ref设计 §3.1 接口 1
# 实现pbl_domain_ext/api.py:bind_ref经 init.py 注册为 env.pbl_bind_ref
# 入参ref_type(world/scene/entity) ref_id blueprint_id class_id team_id ext tenant_id
# 说明ext 原样透传dict 或 JSON 字符串均可,由 api._dump_ext 统一校验/序列化)
debug('pbl_domain_ref_bind.dspy: START params_kw=%s' % dict(params_kw))
_HTTP = {'PBL_E_VALIDATION': 400, 'PBL_E_NOT_FOUND': 404, 'PBL_E_DUPLICATE': 409,
'PBL_E_FORBIDDEN': 403, 'PBL_E_INTERNAL': 500}
try:
_data = await pbl_bind_ref(
params_kw.get('ref_type'),
params_kw.get('ref_id'),
blueprint_id=params_kw.get('blueprint_id'),
class_id=params_kw.get('class_id'),
team_id=params_kw.get('team_id'),
ext=params_kw.get('ext'),
tenant_id=params_kw.get('tenant_id'),
)
debug('pbl_domain_ref_bind.dspy: OK ref_type=%s ref_id=%s'
% (params_kw.get('ref_type'), params_kw.get('ref_id')))
return {'success': True, 'error_code': None, 'message': 'ok', 'data': _data}
except Exception as exc:
_code = getattr(exc, 'code', None) or 'PBL_E_INTERNAL'
_msg = getattr(exc, 'message', None) or str(exc)
_detail = getattr(exc, 'detail', None) or {}
debug('pbl_domain_ref_bind.dspy: FAIL code=%s msg=%s' % (_code, _msg))
return {'success': False, 'error_code': _code, 'message': _msg,
'detail': _detail, 'http_status': _HTTP.get(_code, 500)}

View File

@ -0,0 +1,22 @@
# api/pbl_domain_ref_get.dspy — 契约端点get_ref设计 §3.1 接口 3
# 实现pbl_domain_ext/api.py:get_refenv.pbl_get_ref
debug('pbl_domain_ref_get.dspy: START params_kw=%s' % dict(params_kw))
_HTTP = {'PBL_E_VALIDATION': 400, 'PBL_E_NOT_FOUND': 404, 'PBL_E_DUPLICATE': 409,
'PBL_E_FORBIDDEN': 403, 'PBL_E_INTERNAL': 500}
try:
_data = await pbl_get_ref(
params_kw.get('ref_type'),
params_kw.get('ref_id'),
tenant_id=params_kw.get('tenant_id'),
)
debug('pbl_domain_ref_get.dspy: OK id=%s' % (_data or {}).get('id'))
return {'success': True, 'error_code': None, 'message': 'ok', 'data': _data}
except Exception as exc:
_code = getattr(exc, 'code', None) or 'PBL_E_INTERNAL'
_msg = getattr(exc, 'message', None) or str(exc)
_detail = getattr(exc, 'detail', None) or {}
debug('pbl_domain_ref_get.dspy: FAIL code=%s msg=%s' % (_code, _msg))
return {'success': False, 'error_code': _code, 'message': _msg,
'detail': _detail, 'http_status': _HTTP.get(_code, 500)}

View File

@ -0,0 +1,22 @@
# api/pbl_domain_ref_unbind.dspy — 契约端点unbind_ref设计 §3.1 接口 2
# 实现pbl_domain_ext/api.py:unbind_refenv.pbl_unbind_ref软删 is_deleted=1
debug('pbl_domain_ref_unbind.dspy: START params_kw=%s' % dict(params_kw))
_HTTP = {'PBL_E_VALIDATION': 400, 'PBL_E_NOT_FOUND': 404, 'PBL_E_DUPLICATE': 409,
'PBL_E_FORBIDDEN': 403, 'PBL_E_INTERNAL': 500}
try:
_data = await pbl_unbind_ref(
params_kw.get('ref_type'),
params_kw.get('ref_id'),
tenant_id=params_kw.get('tenant_id'),
)
debug('pbl_domain_ref_unbind.dspy: OK result=%s' % _data)
return {'success': True, 'error_code': None, 'message': 'ok', 'data': _data}
except Exception as exc:
_code = getattr(exc, 'code', None) or 'PBL_E_INTERNAL'
_msg = getattr(exc, 'message', None) or str(exc)
_detail = getattr(exc, 'detail', None) or {}
debug('pbl_domain_ref_unbind.dspy: FAIL code=%s msg=%s' % (_code, _msg))
return {'success': False, 'error_code': _code, 'message': _msg,
'detail': _detail, 'http_status': _HTTP.get(_code, 500)}

View File

@ -1,4 +0,0 @@
# pbl_domain_ext/api/pbl_team_list.dspy —— 契约端点(自动生成,勿手改:改 spec 后跑 gen_artifacts.py
debug('pbl_domain_ext/api/pbl_team_list.dspy: START params_kw={dict(params_kw)}')
data = await pbl_team_list(**params_kw)
return data

View File

@ -1,4 +0,0 @@
# pbl_domain_ext/api/pbl_team_save.dspy —— 契约端点(自动生成,勿手改:改 spec 后跑 gen_artifacts.py
debug('pbl_domain_ext/api/pbl_team_save.dspy: START params_kw={dict(params_kw)}')
data = await pbl_team_save(**params_kw)
return data

View File

@ -1,4 +0,0 @@
# pbl_domain_ext/api/pbl_tenant_upsert.dspy —— 契约端点(自动生成,勿手改:改 spec 后跑 gen_artifacts.py
debug('pbl_domain_ext/api/pbl_tenant_upsert.dspy: START params_kw={dict(params_kw)}')
data = await pbl_tenant_upsert(**params_kw)
return data