技能名称: data-model-designer

数据模型设计器

业务场景

问题陈述

• - 跨系统数据碎片化
• 数据结构不一致
• 实体间关系缺失
• 数据集成困难

解决方案

技术实现

python
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from enum import Enum
import json

class DataType(Enum):
STRING = string
INTEGER = integer
FLOAT = float
BOOLEAN = boolean
DATE = date
DATETIME = datetime
TEXT = text
JSON = json

class RelationType(Enum):
ONETOONE = 1:1
ONETOMANY = 1:N
MANYTOMANY = N:M

class ConstraintType(Enum):
PRIMARYKEY = primarykey
FOREIGNKEY = foreignkey
UNIQUE = unique
NOTNULL = notnull

@dataclass
class Field:
name: str
data_type: DataType
nullable: bool = True
default: Any = None
description: str =
constraints: List[ConstraintType] = field(default_factory=list)

@dataclass
class Entity:
name: str
description: str
fields: List[Field] = field(default_factory=list)
primary_key: str = id

@dataclass
class Relationship:
name: str
from_entity: str
to_entity: str
relation_type: RelationType
from_field: str
to_field: str

class ConstructionDataModel:
为施工项目设计数据模型。

def init(self, project_name: str):
self.projectname = projectname
self.entities: Dict[str, Entity] = {}
self.relationships: List[Relationship] = []

def add_entity(self, entity: Entity):
向模型添加实体。
self.entities[entity.name] = entity

def add_relationship(self, relationship: Relationship):
添加实体间关系。
self.relationships.append(relationship)

def create_entity(self, name: str, description: str,
fields: List[Dict[str, Any]]) -> Entity:
根据字段定义创建实体。

entity_fields = [
Field(
name=f[name],
data_type=DataType(f.get(type, string)),
nullable=f.get(nullable, True),
default=f.get(default),
description=f.get(description, ),
constraints=[ConstraintType(c) for c in f.get(constraints, [])]
)
for f in fields
]

entity = Entity(name=name, description=description, fields=entity_fields)
self.add_entity(entity)
return entity

def createrelationship(self, fromentity: str, to_entity: str,
relation_type: str = 1:N,
from_field: str = None) -> Relationship:
创建实体间关系。

rel = Relationship(
name=f{fromentity}{to_entity},
fromentity=fromentity,
toentity=toentity,
relationtype=RelationType(relationtype),
fromfield=fromfield or f{toentity.lower()}id,
to_field=id
)
self.add_relationship(rel)
return rel

def generatesqlschema(self, dialect: str = postgresql) -> str:
生成SQL DDL语句。

sql = []
type_map = {
DataType.STRING: VARCHAR(255),
DataType.INTEGER: INTEGER,
DataType.FLOAT: DECIMAL(15,2),
DataType.BOOLEAN: BOOLEAN,
DataType.DATE: DATE,
DataType.DATETIME: TIMESTAMP,
DataType.TEXT: TEXT,
DataType.JSON: JSONB if dialect == postgresql else JSON
}

for name, entity in self.entities.items():
columns = []
for fld in entity.fields:
col = f {fld.name} {typemap.get(fld.datatype, VARCHAR(255))}
if not fld.nullable:
col += NOT NULL
if ConstraintType.PRIMARY_KEY in fld.constraints:
col += PRIMARY KEY
columns.append(col)

sql.append(fCREATE TABLE {name} (\n + ,\n.join(columns) + \n);)

for rel in self.relationships:
sql.append(fALTER TABLE {rel.from_entity}
ADD CONSTRAINT fk_{rel.name}
FOREIGN KEY ({rel.fromfield}) REFERENCES {rel.toentity}({rel.to_field});)

return \n\n.join(sql)

def generatejsonschema(self) -> Dict[str, Any]:
生成JSON Schema表示。

schemas = {}
for name, entity in self.entities.items():
properties = {}
required = []

for fld in entity.fields:
prop = {description: fld.description}
if fld.data_type == DataType.STRING:
prop[type] = string
elif fld.data_type == DataType.INTEGER:
prop[type] = integer
elif fld.data_type == DataType.FLOAT:
prop[type] = number
elif fld.data_type == DataType.BOOLEAN:
prop[type] = boolean
else:
prop[type] = string

properties[fld.name] = prop
if not fld.nullable:
required.append(fld.name)

schemas[name] = {
type: object,
title: entity.description,
properties: properties,
required: required
}
return schemas

def generateerdiagram(self) -> str:
生成Mermaid ER图。

lines = [erDiagram]
for name, entity in self.entities.items():
for fld in entity.fields[:5]:
lines.append(f {name} {{)
lines.append(f {fld.data_type.value} {fld.name})
lines.append( })

for rel in self.relationships:
rel_symbol = {
RelationType.ONETOONE: ||--||,
RelationType.ONETOMANY: ||--o{,
RelationType.MANYTOMANY: }o--o{
}.get(rel.relation_type, ||--o{)
lines.append(f {rel.fromentity} {relsymbol} {rel.to_entity} : \{rel.name}\)

return \n.join(lines)

def validate_model(self) -> List[str]:
验证数据模型是否存在问题。

issues = []
for rel in self.relationships:
if rel.from_entity not in self.entities:
issues.append(f缺少实体：{rel.from_entity})
if rel.to_entity not in self.entities:
issues.append(f缺少实体：{rel.to_entity})

for name, entity in self.entities.items():
haspk = any(ConstraintType.PRIMARYKEY in f.constraints for f in entity.fields)
if not has_pk:
issues.append(f实体{name}没有主键)

return issues

class ConstructionEntities:
标准施工数据实体。

@staticmethod
def project_entity() -> Entity:
return Entity(
name=projects,
description=施工项目,
fields=[
Field(id, DataType.INTEGER, False, constraints=[ConstraintType.PRIMARY_KEY]),
Field(code, DataType.STRING, False, constraints=[ConstraintType.UNIQUE]),
Field(name, DataType.STRING, False),
Field(status, DataType.STRING),
Field(start_date, DataType.DATE),
Field(end_date, DataType.DATE),
Field(budget, DataType.FLOAT)
]
)

@staticmethod
def activity_entity() -> Entity:
return Entity(
name=activities,
description=进度活动,
fields=[
Field(id, DataType.INTEGER, False, constraints=[ConstraintType.PRIMARY_KEY]),
Field(project_id, DataType.INTEGER, False),
Field(wbs_code, DataType.STRING),
Field(name, DataType.STRING, False),
Field(start_date, DataType.DATE),
Field(end_date, DataType.DATE),
Field(percent_complete, DataType.FLOAT)
]
)

@staticmethod
def costitementity() -> Entity:
return Entity(
name=cost_items,
description=项目成本项,