Skip to content

Schema

DataSchema

Source code in blue/data/schema.py 
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
class DataSchema:
    def __init__(self):
        """Initialize empty containers for entities and relations."""
        self.entities = {}
        self.relations = {}

    def has_entity(self, key):
        """
        Check if an entity exists in the schema.

        Parameters:
            key (str): The entity name or identifier.

        Returns:
            bool: True if the entity exists, False otherwise.
        """
        return key in self.entities

    def add_entity(self, key):
        """
        Add a new entity to the schema.

        If the entity already exists, a unique suffix is appended (e.g., "__1").

        Parameters:
            key (str): The name of the entity.

        Returns:
            str: The original entity key.
        """
        index = 0
        if key in self.entities:
            entity_obj = self.entities[key]
            index = entity_obj["index"] + 1
            entity_obj["index"] = index

        unique_key = key
        if index > 0:
            unique_key = key + "__" + str(index)

        entity_obj = {
            'name': key,
            'index': index,
            'description': '',
            'created_by': None,
            'properties': {},  # type info only
            'contents': {'attributes': {}},  # hierarchical child info  # attributes
            'icon': None,
        }

        self.entities[unique_key] = entity_obj

        return key

    def add_entity_property(self, key, attribute, type):
        """
        Add a property (attribute) to an existing entity.

        Parameters:
            key (str): The entity key.
            attribute (str): The name of the attribute.
            type (str): The type or description of the attribute.
        """
        if key not in self.entities:
            return
        entity_obj = self.entities[key]

        entity_obj['contents']['attributes'][attribute] = {
            'name': attribute,
            'info': type,
            'description': None,
        }

    def _relation_encoding(self, source, relation, target):
        """
        Create a unique encoded key for a relation.

        Parameters:
            source (str): The source entity name.
            relation (str): The relationship type.
            target (str): The target entity name.

        Returns:
            str: Encoded relation identifier string.
        """
        s = source + " " + relation + " " + target
        return s.replace(" ", "__")
        # return "(" + source + ")" + "-" + relation + "->" + "(" + target + ")"

    def has_relation(self, source, relation, target):
        """
        Check if a relation exists in the schema.

        Parameters:
            source (str): Source entity name.
            relation (str): Relation name.
            target (str): Target entity name.

        Returns:
            bool: True if the relation exists, False otherwise.
        """
        relation_encoding = self._relation_encoding(source, relation, target)
        return relation_encoding in self.relations

    def add_relation(self, source, relation, target):
        """
        Add a relation between two entities in the schema.

        If the same relation already exists, a numeric suffix is appended.

        Parameters:
            source (str): Source entity name.
            relation (str): Relation name.
            target (str): Target entity name.

        Returns:
            str: Unique key of the created relation.
        """
        key = self._relation_encoding(source, relation, target)
        index = 0
        if key in self.relations:
            relation_obj = self.relations[key]
            index = relation_obj["index"] + 1
            relation_obj["index"] = index

        unique_key = key
        if index > 0:
            unique_key = key + "__" + str(index)

        relation_obj = {}
        relation_obj['name'] = relation
        relation_obj['index'] = index
        relation_obj['source'] = source
        relation_obj['target'] = target
        relation_obj['properties'] = {}

        self.relations[unique_key] = relation_obj

        return unique_key

    def add_relation_property(self, key, property, type):
        """
        Add a property to an existing relation.

        Parameters:
            key (str): Relation key.
            property (str): Property name.
            type (str): Property type or description.
        """
        if key in self.relations:
            relation_obj = self.relations[key]
            properties_obj = relation_obj['properties']
            properties_obj[property] = type

    def get_entities(self):
        """
        Get all entities defined in the schema.

        Returns:
            dict: Mapping of entity keys to entity definitions.
        """
        return self.entities

    def get_relations(self):
        """
        Get all relations defined in the schema.

        Returns:
            dict: Mapping of relation keys to relation definitions.
        """
        return self.relations

    def to_json(self):
        """
        Convert the schema to a JSON-serializable dictionary.

        Returns:
            dict: Dictionary with 'entities' and 'relations' keys.
        """
        s = {}
        s['entities'] = self.entities.copy()
        s['relations'] = self.relations.copy()
        return s

    def __repr__(self):
        return "DataSchema()"

    def __str__(self):
        s = 'Schema:' + '\n'
        s += 'Entities: ' + '\n'
        for key in self.entities:
            entity_obj = self.entities[key]
            name = entity_obj['name']
            index = entity_obj['index']
            properties = entity_obj['properties']
            s += "key: " + key + '\n'
            s += "  name: " + name + '\n'
            s += "  index: " + str(index) + '\n'
            s += "  properties: " + '\n'
            for property in properties:
                s += "    " + property + ": " + properties[property] + '\n'

        s += 'Relations: ' + '\n'
        for key in self.relations:
            relation_obj = self.relations[key]
            name = relation_obj['name']
            index = relation_obj['index']
            source = relation_obj['source']
            target = relation_obj['target']
            properties = relation_obj['properties']
            s += "key: " + key + '\n'
            s += "  name: " + name + '\n'
            s += "  index: " + str(index) + '\n'
            s += "  source: " + source + '\n'
            s += "  target: " + target + '\n'
            s += "  properties: " + '\n'
            for property in properties:
                s += "    " + property + ": " + properties[property] + '\n'

        return s

__init__()

Initialize empty containers for entities and relations.

Source code in blue/data/schema.py 
5
6
7
8
def __init__(self):
    """Initialize empty containers for entities and relations."""
    self.entities = {}
    self.relations = {}

add_entity(key)

Add a new entity to the schema.

If the entity already exists, a unique suffix is appended (e.g., "__1").

Parameters:

Name Type Description Default
key str

The name of the entity.

 required

Returns:

Name Type Description
str

The original entity key.
Source code in blue/data/schema.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def add_entity(self, key):
    """
    Add a new entity to the schema.

    If the entity already exists, a unique suffix is appended (e.g., "__1").

    Parameters:
        key (str): The name of the entity.

    Returns:
        str: The original entity key.
    """
    index = 0
    if key in self.entities:
        entity_obj = self.entities[key]
        index = entity_obj["index"] + 1
        entity_obj["index"] = index

    unique_key = key
    if index > 0:
        unique_key = key + "__" + str(index)

    entity_obj = {
        'name': key,
        'index': index,
        'description': '',
        'created_by': None,
        'properties': {},  # type info only
        'contents': {'attributes': {}},  # hierarchical child info  # attributes
        'icon': None,
    }

    self.entities[unique_key] = entity_obj

    return key

add_entity_property(key, attribute, type)

Add a property (attribute) to an existing entity.

Parameters:

Name Type Description Default
key str

The entity key.

 required
attribute str

The name of the attribute.

 required
type str

The type or description of the attribute.

 required
Source code in blue/data/schema.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def add_entity_property(self, key, attribute, type):
    """
    Add a property (attribute) to an existing entity.

    Parameters:
        key (str): The entity key.
        attribute (str): The name of the attribute.
        type (str): The type or description of the attribute.
    """
    if key not in self.entities:
        return
    entity_obj = self.entities[key]

    entity_obj['contents']['attributes'][attribute] = {
        'name': attribute,
        'info': type,
        'description': None,
    }

add_relation(source, relation, target)

Add a relation between two entities in the schema.

If the same relation already exists, a numeric suffix is appended.

Parameters:

Name Type Description Default
source str

Source entity name.

 required
relation str

Relation name.

 required
target str

Target entity name.

 required

Returns:

Name Type Description
str

Unique key of the created relation.
Source code in blue/data/schema.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def add_relation(self, source, relation, target):
    """
    Add a relation between two entities in the schema.

    If the same relation already exists, a numeric suffix is appended.

    Parameters:
        source (str): Source entity name.
        relation (str): Relation name.
        target (str): Target entity name.

    Returns:
        str: Unique key of the created relation.
    """
    key = self._relation_encoding(source, relation, target)
    index = 0
    if key in self.relations:
        relation_obj = self.relations[key]
        index = relation_obj["index"] + 1
        relation_obj["index"] = index

    unique_key = key
    if index > 0:
        unique_key = key + "__" + str(index)

    relation_obj = {}
    relation_obj['name'] = relation
    relation_obj['index'] = index
    relation_obj['source'] = source
    relation_obj['target'] = target
    relation_obj['properties'] = {}

    self.relations[unique_key] = relation_obj

    return unique_key

add_relation_property(key, property, type)

Add a property to an existing relation.

Parameters:

Name Type Description Default
key str

Relation key.

 required
property str

Property name.

 required
type str

Property type or description.

 required
Source code in blue/data/schema.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
def add_relation_property(self, key, property, type):
    """
    Add a property to an existing relation.

    Parameters:
        key (str): Relation key.
        property (str): Property name.
        type (str): Property type or description.
    """
    if key in self.relations:
        relation_obj = self.relations[key]
        properties_obj = relation_obj['properties']
        properties_obj[property] = type

get_entities()

Get all entities defined in the schema.

Returns:

Name Type Description
dict

Mapping of entity keys to entity definitions.
Source code in blue/data/schema.py 
158
159
160
161
162
163
164
165
def get_entities(self):
    """
    Get all entities defined in the schema.

    Returns:
        dict: Mapping of entity keys to entity definitions.
    """
    return self.entities

get_relations()

Get all relations defined in the schema.

Returns:

Name Type Description
dict

Mapping of relation keys to relation definitions.
Source code in blue/data/schema.py 
167
168
169
170
171
172
173
174
def get_relations(self):
    """
    Get all relations defined in the schema.

    Returns:
        dict: Mapping of relation keys to relation definitions.
    """
    return self.relations

has_entity(key)

Check if an entity exists in the schema.

Parameters:

Name Type Description Default
key str

The entity name or identifier.

 required

Returns:

Name Type Description
bool

True if the entity exists, False otherwise.
Source code in blue/data/schema.py 
10
11
12
13
14
15
16
17
18
19
20
def has_entity(self, key):
    """
    Check if an entity exists in the schema.

    Parameters:
        key (str): The entity name or identifier.

    Returns:
        bool: True if the entity exists, False otherwise.
    """
    return key in self.entities

has_relation(source, relation, target)

Check if a relation exists in the schema.

Parameters:

Name Type Description Default
source str

Source entity name.

 required
relation str

Relation name.

 required
target str

Target entity name.

 required

Returns:

Name Type Description
bool

True if the relation exists, False otherwise.
Source code in blue/data/schema.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def has_relation(self, source, relation, target):
    """
    Check if a relation exists in the schema.

    Parameters:
        source (str): Source entity name.
        relation (str): Relation name.
        target (str): Target entity name.

    Returns:
        bool: True if the relation exists, False otherwise.
    """
    relation_encoding = self._relation_encoding(source, relation, target)
    return relation_encoding in self.relations

to_json()

Convert the schema to a JSON-serializable dictionary.

Returns:

Name Type Description
dict

Dictionary with 'entities' and 'relations' keys.
Source code in blue/data/schema.py 
176
177
178
179
180
181
182
183
184
185
186
def to_json(self):
    """
    Convert the schema to a JSON-serializable dictionary.

    Returns:
        dict: Dictionary with 'entities' and 'relations' keys.
    """
    s = {}
    s['entities'] = self.entities.copy()
    s['relations'] = self.relations.copy()
    return s
Last update: 2025-10-09