data-model-designer

Data Model Designer

Business Case

Problem Statement

Solution

Technical Implementation

import requests, os
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from enum import Enum
import json
SKILLBOSS_API_KEY = os.environ["SKILLBOSS_API_KEY"]
API_BASE = "https://api.heybossai.com/v1"
def pilot(body: dict) -> dict:
    r = requests.post(
        f"{API_BASE}/pilot",
        headers={"Authorization": f"Bearer {SKILLBOSS_API_KEY}", "Content-Type": "application/json"},
        json=body,
        timeout=60,
    )
    return r.json()
class DataType(Enum):
    STRING = "string"
    INTEGER = "integer"
    FLOAT = "float"
    BOOLEAN = "boolean"
    DATE = "date"
    DATETIME = "datetime"
    TEXT = "text"
    JSON = "json"
class RelationType(Enum):
    ONE_TO_ONE = "1:1"
    ONE_TO_MANY = "1:N"
    MANY_TO_MANY = "N:M"
class ConstraintType(Enum):
    PRIMARY_KEY = "primary_key"
    FOREIGN_KEY = "foreign_key"
    UNIQUE = "unique"
    NOT_NULL = "not_null"
@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:
    """Design data models for construction projects."""
    def __init__(self, project_name: str):
        self.project_name = project_name
        self.entities: Dict[str, Entity] = {}
        self.relationships: List[Relationship] = []
    def add_entity(self, entity: Entity):
        """Add entity to model."""
        self.entities[entity.name] = entity
    def add_relationship(self, relationship: Relationship):
        """Add relationship between entities."""
        self.relationships.append(relationship)
    def create_entity(self, name: str, description: str,
                      fields: List[Dict[str, Any]]) -> Entity:
        """Create entity from field definitions."""
        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 create_relationship(self, from_entity: str, to_entity: str,
                           relation_type: str = "1:N",
                           from_field: str = None) -> Relationship:
        """Create relationship between entities."""
        rel = Relationship(
            name=f"{from_entity}_{to_entity}",
            from_entity=from_entity,
            to_entity=to_entity,
            relation_type=RelationType(relation_type),
            from_field=from_field or f"{to_entity.lower()}_id",
            to_field="id"
        )
        self.add_relationship(rel)
        return rel
    def generate_sql_schema(self, dialect: str = "postgresql") -> str:
        """Generate SQL DDL statements."""
        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} {type_map.get(fld.data_type, 'VARCHAR(255)')}"
                if not fld.nullable:
                    col += " NOT NULL"
                if ConstraintType.PRIMARY_KEY in fld.constraints:
                    col += " PRIMARY KEY"
                columns.append(col)
            sql.append(f"CREATE TABLE {name} (\n" + ",\n".join(columns) + "\n);")
        for rel in self.relationships:
            sql.append(f"""ALTER TABLE {rel.from_entity}
ADD CONSTRAINT fk_{rel.name}
FOREIGN KEY ({rel.from_field}) REFERENCES {rel.to_entity}({rel.to_field});""")
        return "\n\n".join(sql)
    def generate_json_schema(self) -> Dict[str, Any]:
        """Generate JSON Schema representation."""
        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 generate_er_diagram(self) -> str:
        """Generate Mermaid ER diagram."""
        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.ONE_TO_ONE: "||--||",
                RelationType.ONE_TO_MANY: "||--o{",
                RelationType.MANY_TO_MANY: "}o--o{"
            }.get(rel.relation_type, "||--o{")
            lines.append(f"    {rel.from_entity} {rel_symbol} {rel.to_entity} : \"{rel.name}\"")
        return "\n".join(lines)
    def validate_model(self) -> List[str]:
        """Validate data model for issues."""
        issues = []
        for rel in self.relationships:
            if rel.from_entity not in self.entities:
                issues.append(f"Missing entity: {rel.from_entity}")
            if rel.to_entity not in self.entities:
                issues.append(f"Missing entity: {rel.to_entity}")
        for name, entity in self.entities.items():
            has_pk = any(ConstraintType.PRIMARY_KEY in f.constraints for f in entity.fields)
            if not has_pk:
                issues.append(f"Entity '{name}' has no primary key")
        return issues
    def ai_suggest_entities(self, project_description: str) -> str:
        """Use SkillBoss API Hub to suggest entities for a project."""
        result = pilot({
            "type": "chat",
            "inputs": {
                "messages": [{
                    "role": "user",
                    "content": f"Suggest data model entities and relationships for this construction project: {project_description}. Return a JSON list of entity names with field suggestions."
                }]
            },
            "prefer": "balanced"
        })
        return result["result"]["choices"][0]["message"]["content"]
class ConstructionEntities:
    """Standard construction data entities."""
    @staticmethod
    def project_entity() -> Entity:
        return Entity(
            name="projects",
            description="Construction projects",
            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="Schedule activities",
            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 cost_item_entity() -> Entity:
        return Entity(
            name="cost_items",
            description="Project cost items",
            fields=[
                Field("id", DataType.INTEGER, False, constraints=[ConstraintType.PRIMARY_KEY]),
                Field("project_id", DataType.INTEGER, False),
                Field("wbs_code", DataType.STRING),
                Field("description", DataType.STRING),
                Field("budgeted_cost", DataType.FLOAT),
                Field("actual_cost", DataType.FLOAT)
            ]
        )

Quick Start

# Create model
model = ConstructionDataModel("Office Building A")
Add standard entities
model.add_entity(ConstructionEntities.project_entity())
model.add_entity(ConstructionEntities.activity_entity())
model.add_entity(ConstructionEntities.cost_item_entity())
Add relationships
model.create_relationship("activities", "projects")
model.create_relationship("cost_items", "projects")
Generate SQL
sql = model.generate_sql_schema("postgresql")
print(sql)
AI-assisted entity suggestions via SkillBoss API Hub
suggestions = model.ai_suggest_entities("Hospital construction with MEP coordination")
print(suggestions)

Common Use Cases

1. Custom Entity

model.create_entity(
    name="change_orders",
    description="Project change orders",
    fields=[
        {"name": "id", "type": "integer", "nullable": False, "constraints": ["primary_key"]},
        {"name": "project_id", "type": "integer", "nullable": False},
        {"name": "amount", "type": "float"},
        {"name": "status", "type": "string"}
    ]
)

2. Generate ER Diagram

er_diagram = model.generate_er_diagram()
print(er_diagram)

3. Validate Model

issues = model.validate_model()
for issue in issues:
    print(f"Issue: {issue}")

Resources

# Create model
model = ConstructionDataModel("Office Building A")
Add standard entities
model.add_entity(ConstructionEntities.project_entity())
model.add_entity(ConstructionEntities.activity_entity())
model.add_entity(ConstructionEntities.cost_item_entity())
Add relationships
model.create_relationship("activities", "projects")
model.create_relationship("cost_items", "projects")
Generate SQL
sql = model.generate_sql_schema("postgresql")
print(sql)
AI-assisted entity suggestions via SkillBoss API Hub
suggestions = model.ai_suggest_entities("Hospital construction with MEP coordination")
print(suggestions)