model.py

from typing import Optional
import itertools
from pprint import pprint
from dataclasses import dataclass
from enum import Enum

class ValidationException(Exception):
    pass

@dataclass
class ERSchema:
    entity_sets: dict[str, 'EntitySet']
    relationship_sets: dict[str, 'RelationshipSet']
    isas: dict[str, 'ISA']
    def get_entity_set(self, id) -> Optional['EntitySet']:
        return self.entity_sets.get(id, None)
    def get_relationship_set(self, id) -> Optional['RelationshipSet']:
        return self.relationship_sets.get(id, None)
    def get_isa(self, id) -> Optional['ISA']:
        return self.isas.get(id, None)
    def add_entity_set(self, ent: 'EntitySet') -> None:
        self.entity_sets[ent._id] = ent
    def add_relationship_set(self, rel: 'RelationshipSet') -> None:
        self.relationship_sets[rel._id] = rel
    def add_isa(self, isa: 'ISA') -> None:
        self.isas[isa._id] = isa
    def print(self) -> None:
        print('**** Entity Sets:')
        for ent in self.entity_sets.values():
            pprint(ent)
        print('**** Relationship Sets:')
        for rel in self.relationship_sets.values():
            pprint(rel)
        print('**** ISA Relationships:')
        for isa in self.isas.values():
            pprint(isa)

    def superclass(self, ent: 'EntitySet') -> Optional['EntitySet']:
        return next((isa.superclass for isa in self.isas.values() if isa.subclass == ent), None)

    def inheritance_chain(self, ent: 'EntitySet') -> list['EntitySet']:
        # stops at the first repeat (detected cycle):
        chain = [ent]
        supEnt = self.superclass(ent)
        while supEnt is not None:
            chain.append(supEnt)
            if supEnt in chain[:-1]:
                break
            supEnt = self.superclass(supEnt)
        return chain

    def validate(self) -> None:
        # names are unique across entity sets and relationship sets;
        # attribute names are unique within each relationship/entity set
        # (at this point we are not worried about inheritance/weak entity sets yet)
        names = {}
        for s in itertools.chain(self.entity_sets.values(), self.relationship_sets.values()):
            if s.name in names:
                raise ValidationException(f'entity/relationship sets should not have the same name "{s.name}"')
            names[s.name] = 1
            for i, attribute in enumerate(s.attributes):
                if attribute in s.attributes[:i]:
                    raise ValidationException(f'attribute "{attribute}" appears more than once inside "{s.name}"')
        # ISA: build sub2super; check no multiple inheritance
        sub2super: dict[str, EntitySet] = {}
        for isa in self.isas.values():
            if isa.subclass is None:
                raise ValidationException(f'ISA {isa._id} must specifiy a subclass')
            if isa.subclass.name in sub2super:
                raise ValidationException(f'entity set "{isa.subclass.name}" cannot inherit from multiple superclasses')
            if isa.superclass is None:
                raise ValidationException(f'ISA {isa._id} must specifiy a superclass')
            sub2super[isa.subclass.name] = isa.superclass
        # ISA: check for cycles; not the fastest algorithm but speed doesn't matter here
        for ent in self.entity_sets.values():
            chain = self.inheritance_chain(ent)
            if len(chain) > 1 and chain[-1] in chain[0:-1]:
                raise ValidationException(f'''cycle in inheritance graph detected: {
                    ' <= '.join(f'"{ent.name}"' for ent in chain[(chain.index(chain[-1])):])
                }''')
        # for each relationship set (ignoring connecting aspect):
        # if one entity set participates multiple times, the roles must be specified and distinct
        for rel in self.relationship_sets.values():
            for i, (ent, role) in enumerate(zip(rel.entity_sets, rel.roles)):
                if ent in rel.entity_sets[:i]:
                    if role is None:
                        raise ValidationException(f'in relationship set "{rel.name}", role must be specified for "{ent.name}" because it participates more than once')
                    for ent2, role2 in zip(rel.entity_sets[:i], rel.roles[:i]):
                        if ent2 == ent and role2 == role:
                            raise ValidationException(f'in relationship set "{rel.name}", distinct roles must be specified for "{ent.name}" because it participates more than once')
        # for each entity set (ignoring weak aspect):
        # attribute name shouldn't clash with inherited ones;
        # no key in subclass;
        for ent in self.entity_sets.values():
            if self.superclass(ent) is not None:
                if any(k for k in ent.is_in_key):
                    raise ValidationException(f'subclass "{ent.name}" cannot declare its own key because it should be inherited')
                for superclass in self.inheritance_chain(ent)[1:]:
                    for attr in superclass.attributes:
                        if attr in ent.attributes:
                            raise ValidationException(f'attribute "{attr}" in subclass "{ent.name}" appears already in superclass "{superclass.name}"')
        # for each connecting relationship set:
        # connecting relationship set must be binary and source must be weak (no ambiguity);
        # if connecting relationship set is exactly-one-to-exactly-one, one and only one entity set can be weak
        for rel in self.relationship_sets.values():
            if not rel.is_connecting:
                continue
            if len(rel.entity_sets) != 2:
                raise ValidationException(f'connecting relationship set "{rel.name}" must connect two entity sets, not {len(rel.entity_sets)}')
            if Multiplicity.MULT_1_1 not in rel.multiplicities:
                raise ValidationException(f'connecting relationship set "{rel.name}" must have multiplicity 1:1 going to an entity set')
            if (not rel.entity_sets[0].is_weak) and not (rel.entity_sets[1].is_weak):
                raise ValidationException(f'connecting relationship set "{rel.name}" must connect at least one weak entity set')
            if rel.multiplicities == [Multiplicity.MULT_1_1]*2:
                if rel.entity_sets[0].is_weak and rel.entity_sets[1].is_weak:
                    raise ValidationException(f'connecting relationship set "{rel.name}" is ambiguous because both entity sets are weak')
                rel.weak_entity = rel.entity_sets[0] if rel.entity_sets[0].is_weak else rel.entity_sets[1]
            else:
                rel.weak_entity = rel.entity_sets[0] if Multiplicity.MULT_1_1 == rel.multiplicities[1] else rel.entity_sets[1]
                if not rel.weak_entity.is_weak:
                    raise ValidationException(f'entity set "{rel.weak_entity.name}" should be weak because of connecting relationship set "{rel.name}"')
        # for each weak entity set:
        # it cannot have a superclass;
        # it must be the weak_entity for at least one relationship set (there can be more)
        for ent in self.entity_sets.values():
            if not ent.is_weak:
                continue
            if self.superclass(ent) is not None:
                raise ValidationException(f'entity set "{ent.name}" should not be weak because it has a superclass')
            if not any(rel for rel in self.relationship_sets.values() if rel.weak_entity == ent):
                raise ValidationException(f'entity set "{ent.name}" should not be weak because it cannot borrow key through any connecting relationship set')
        return

@dataclass
class EntitySet:
    _id: str
    name: str
    is_weak: bool
    attributes: list[str]
    is_in_key: list[bool]
    def short_repr(self) -> str:
        return f'EntitySet({self.name})'

class Multiplicity(Enum):
    MULT_0_N = '0:N'
    MULT_0_1 = '0:1'
    MULT_1_1 = '1:1'

@dataclass
class RelationshipSet:
    _id: str
    name: str
    is_connecting: bool
    entity_sets: list[EntitySet]
    multiplicities: list[Multiplicity]
    roles: list[Optional[str]]
    attributes: list[str]
    weak_entity: Optional[EntitySet] = None # set by ERSchema.validate() if is_connecting
    def add_entity_set(self, ent: EntitySet, multiplicity: Multiplicity, role: Optional[str]):
        self.entity_sets.append(ent)
        self.multiplicities.append(multiplicity)
        self.roles.append(role)

@dataclass
class ISA:
    _id: str
    superclass: EntitySet
    subclass: EntitySet