Skip to content

type_bridge.crud.utils

utils

Shared utilities for CRUD operations.

This module re-exports functions from focused submodules for convenience. You can also import directly from the submodules:

  • type_bridge.crud.formatting: format_value, unwrap_attribute
  • type_bridge.crud.patterns: build_entity_match_pattern, build_relation_match_pattern, normalize_role_players
  • type_bridge.crud.types: is_multi_value_attribute, hydrate_attributes, etc.
  • type_bridge.crud.role_players: Role player utilities

format_value 

format_value(value)

Format a Python value for TypeQL.

Handles extraction from Attribute instances and converts Python types to their TypeQL literal representation.

Parameters:

Name Type Description Default
value Any

Python value to format (may be wrapped in Attribute instance)

 required

Returns:

Type Description
str

TypeQL-formatted string literal

Examples:

>>> format_value("hello")
'"hello"'
>>> format_value(42)
'42'
>>> format_value(True)
'true'
>>> format_value(Decimal("123.45"))
'123.45dec'
Source code in type_bridge/crud/formatting.py 
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
def format_value(value: Any) -> str:
    """Format a Python value for TypeQL.

    Handles extraction from Attribute instances and converts Python types
    to their TypeQL literal representation.

    Args:
        value: Python value to format (may be wrapped in Attribute instance)

    Returns:
        TypeQL-formatted string literal

    Examples:
        >>> format_value("hello")
        '"hello"'
        >>> format_value(42)
        '42'
        >>> format_value(True)
        'true'
        >>> format_value(Decimal("123.45"))
        '123.45dec'
    """
    # Use Rust implementation when available for performance
    if _USE_RUST and _rust_format_value is not None:
        return _rust_format_value(value)

    # Python fallback below
    # Extract value from Attribute instances first
    if hasattr(value, "value"):
        value = value.value

    if isinstance(value, str):
        # Escape special characters for TypeQL string literals (JSON-style escaping)
        # Order matters: backslashes first, then other sequences
        escaped = (
            value.replace("\\", "\\\\")  # Backslashes
            .replace('"', '\\"')  # Double quotes
            .replace("\n", "\\n")  # Newlines
            .replace("\r", "\\r")  # Carriage returns
            .replace("\t", "\\t")  # Tabs
        )
        return f'"{escaped}"'
    elif isinstance(value, bool):
        return "true" if value else "false"
    elif isinstance(value, DecimalType):
        # TypeDB decimal literals use 'dec' suffix
        return f"{value}dec"
    elif isinstance(value, (int, float)):
        return str(value)
    elif isinstance(value, datetime):
        # TypeDB datetime/datetimetz literals are unquoted ISO 8601 strings
        return value.isoformat()
    elif isinstance(value, date):
        # TypeDB date literals are unquoted ISO 8601 date strings
        return value.isoformat()
    elif isinstance(value, (IsodateDuration, timedelta)):
        # TypeDB duration literals are unquoted ISO 8601 duration strings
        return isodate.duration_isoformat(value)
    else:
        # For other types, convert to string and escape
        str_value = str(value)
        escaped = (
            str_value.replace("\\", "\\\\")  # Backslashes
            .replace('"', '\\"')  # Double quotes
            .replace("\n", "\\n")  # Newlines
            .replace("\r", "\\r")  # Carriage returns
            .replace("\t", "\\t")  # Tabs
        )
        return f'"{escaped}"'

unwrap_attribute 

unwrap_attribute(value)

Extract raw value from Attribute instance.

This utility consolidates the common pattern of extracting the underlying value from Attribute instances before processing.

Parameters:

Name Type Description Default
value Any

Value that may be an Attribute instance or raw value

 required

Returns:

Type Description
Any

The raw value (value.value if Attribute, otherwise value unchanged)

Examples:

>>> unwrap_attribute(Name("Alice"))
"Alice"
>>> unwrap_attribute("Alice")
"Alice"
>>> unwrap_attribute(42)
42
Source code in type_bridge/crud/formatting.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def unwrap_attribute(value: Any) -> Any:
    """Extract raw value from Attribute instance.

    This utility consolidates the common pattern of extracting the underlying
    value from Attribute instances before processing.

    Args:
        value: Value that may be an Attribute instance or raw value

    Returns:
        The raw value (value.value if Attribute, otherwise value unchanged)

    Examples:
        >>> unwrap_attribute(Name("Alice"))
        "Alice"
        >>> unwrap_attribute("Alice")
        "Alice"
        >>> unwrap_attribute(42)
        42
    """
    if hasattr(value, "value"):
        return value.value
    return value

build_entity_match_pattern 

build_entity_match_pattern(model_class, var='$e', filters=None)

Build a TypeQL match pattern for an entity.

Parameters:

Name Type Description Default
model_class type[Entity]

The entity model class

 required
var str

Variable name to use (default: "$e")

 '$e'
filters dict[str, Any] | None

Attribute filters as field_name -> value

 None

Returns:

Type Description
str

TypeQL match pattern string like "$e isa person, has name "Alice""

Examples:

>>> build_entity_match_pattern(Person, filters={"name": "Alice"})
'$e isa person, has name "Alice"'
Source code in type_bridge/crud/patterns.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def build_entity_match_pattern(
    model_class: type["Entity"],
    var: str = "$e",
    filters: dict[str, Any] | None = None,
) -> str:
    """Build a TypeQL match pattern for an entity.

    Args:
        model_class: The entity model class
        var: Variable name to use (default: "$e")
        filters: Attribute filters as field_name -> value

    Returns:
        TypeQL match pattern string like "$e isa person, has name \"Alice\""

    Examples:
        >>> build_entity_match_pattern(Person, filters={"name": "Alice"})
        '$e isa person, has name "Alice"'
    """
    pattern = build_entity_match_ast(model_class, var, filters)
    return QueryCompiler().compile(pattern)

build_relation_match_pattern 

build_relation_match_pattern(model_class, var='$r', role_players=None)

Build a TypeQL match pattern for a relation.

Parameters:

Name Type Description Default
model_class type[Relation]

The relation model class

 required
var str

Variable name to use (default: "$r")

 '$r'
role_players dict[str, str] | None

Dict mapping role names to player variable names

 None

Returns:

Type Description
str

TypeQL match pattern string

Raises:

Type Description
ValueError

If a role name is not defined in the model

Examples:

>>> build_relation_match_pattern(Employment, role_players={"employee": "$p"})
'$r isa employment, (employee: $p)'
Source code in type_bridge/crud/patterns.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def build_relation_match_pattern(
    model_class: type["Relation"],
    var: str = "$r",
    role_players: dict[str, str] | None = None,
) -> str:
    """Build a TypeQL match pattern for a relation.

    Args:
        model_class: The relation model class
        var: Variable name to use (default: "$r")
        role_players: Dict mapping role names to player variable names

    Returns:
        TypeQL match pattern string

    Raises:
        ValueError: If a role name is not defined in the model

    Examples:
        >>> build_relation_match_pattern(Employment, role_players={"employee": "$p"})
        '$r isa employment, (employee: $p)'
    """
    pattern = build_relation_match_ast(model_class, var, role_players)
    return QueryCompiler().compile(pattern)

normalize_role_players 

normalize_role_players(role_players)

Normalize role players to always be lists for uniform handling.

Handles both single entities and lists of entities (for multi-cardinality roles). Also generates unique variable names for each player in the match clause.

Parameters:

Name Type Description Default
role_players dict[str, Any]

Dict mapping role_name -> entity or list of entities

 required

Returns:

Type Description
dict[str, list[Any]]

Tuple of:
dict[str, list[str]]
  • normalized_players: Dict mapping role_name -> list of entities
tuple[dict[str, list[Any]], dict[str, list[str]]]
  • var_mapping: Dict mapping role_name -> list of variable names

Examples:

>>> normalize_role_players({"employee": alice, "employer": company})
({"employee": [alice], "employer": [company]},
 {"employee": ["employee"], "employer": ["employer"]})

>>> normalize_role_players({"member": [alice, bob]})
({"member": [alice, bob]},
 {"member": ["member_0", "member_1"]})
Source code in type_bridge/crud/patterns.py 
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
def normalize_role_players(
    role_players: dict[str, Any],
) -> tuple[dict[str, list[Any]], dict[str, list[str]]]:
    """Normalize role players to always be lists for uniform handling.

    Handles both single entities and lists of entities (for multi-cardinality roles).
    Also generates unique variable names for each player in the match clause.

    Args:
        role_players: Dict mapping role_name -> entity or list of entities

    Returns:
        Tuple of:
        - normalized_players: Dict mapping role_name -> list of entities
        - var_mapping: Dict mapping role_name -> list of variable names

    Examples:
        >>> normalize_role_players({"employee": alice, "employer": company})
        ({"employee": [alice], "employer": [company]},
         {"employee": ["employee"], "employer": ["employer"]})

        >>> normalize_role_players({"member": [alice, bob]})
        ({"member": [alice, bob]},
         {"member": ["member_0", "member_1"]})
    """
    normalized_players: dict[str, list[Any]] = {}
    var_mapping: dict[str, list[str]] = {}

    for role_name, entity_or_list in role_players.items():
        # Normalize to list
        entities = entity_or_list if isinstance(entity_or_list, list) else [entity_or_list]
        normalized_players[role_name] = entities

        # Generate variable names
        var_names = []
        for i in range(len(entities)):
            var_name = f"{role_name}_{i}" if len(entities) > 1 else role_name
            var_names.append(var_name)
        var_mapping[role_name] = var_names

    return normalized_players, var_mapping

build_role_player_fetch_items 

build_role_player_fetch_items(role_info)

Build fetch items for role players with their IIDs and type labels.

TypeQL 3.x doesn't allow mixing iid() with .* in the same nested block, so we fetch the IID as a separate key alongside the nested attributes. We also fetch the type label to correctly identify polymorphic role players.

Parameters:

Name Type Description Default
role_info dict[str, tuple[str, Any]]

Dict mapping role_name -> (role_var, allowed_entity_types)

 required

Returns:

Type Description
list[str]

List of fetch item strings like: '"employee_iid": iid($employee)' '"employee_type": label($employee_type)' '"employee": { $employee.* }'
The caller must add type variable bindings to the match clause:

$employee isa $employee_type;

Source code in type_bridge/crud/role_players.py 
 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
def build_role_player_fetch_items(
    role_info: dict[str, tuple[str, Any]],
) -> list[str]:
    """Build fetch items for role players with their IIDs and type labels.

    TypeQL 3.x doesn't allow mixing iid() with .* in the same nested block,
    so we fetch the IID as a separate key alongside the nested attributes.
    We also fetch the type label to correctly identify polymorphic role players.

    Args:
        role_info: Dict mapping role_name -> (role_var, allowed_entity_types)

    Returns:
        List of fetch item strings like:
            '"employee_iid": iid($employee)'
            '"employee_type": label($employee_type)'
            '"employee": { $employee.* }'

    Note: The caller must add type variable bindings to the match clause:
        $employee isa $employee_type;
    """
    fetch_items = []
    for role_name, (role_var, _) in role_info.items():
        # Fetch IID separately (can't be mixed with .* in nested block)
        fetch_items.append(f'"{role_name}_iid": iid({role_var})')
        # Fetch type label for polymorphic role player resolution
        fetch_items.append(f'"{role_name}_type": label({role_var}_type)')
        # Fetch all attributes for the role player
        fetch_items.append(f'"{role_name}": {{\n    {role_var}.*\n  }}')
    return fetch_items

build_role_player_match 

build_role_player_match(var_name, entity, entity_type_name)

Build a match clause for a role player entity.

Prefers IID-based matching when available (more precise and faster), falls back to key attribute matching, and raises a clear error if neither is available.

Used by TypeDBManager for relation CRUD operations.

Parameters:

Name Type Description Default
var_name str

The variable name to use (without $)

 required
entity Any

The entity instance

 required
entity_type_name str

The TypeDB type name for the entity

 required

Returns:

Type Description
str

A TypeQL match clause string like "$var_name isa type, iid 0x..."
str

or "$var_name isa type, has key_attr value"

Raises:

Type Description
ValueError

If entity has neither _iid nor key attributes
Source code in type_bridge/crud/role_players.py 
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
def build_role_player_match(var_name: str, entity: Any, entity_type_name: str) -> str:
    """Build a match clause for a role player entity.

    Prefers IID-based matching when available (more precise and faster),
    falls back to key attribute matching, and raises a clear error if
    neither is available.

    Used by TypeDBManager for relation CRUD operations.

    Args:
        var_name: The variable name to use (without $)
        entity: The entity instance
        entity_type_name: The TypeDB type name for the entity

    Returns:
        A TypeQL match clause string like "$var_name isa type, iid 0x..."
        or "$var_name isa type, has key_attr value"

    Raises:
        ValueError: If entity has neither _iid nor key attributes
    """
    # Prefer IID-based matching when available (more precise and faster)
    entity_iid = getattr(entity, "_iid", None)
    if entity_iid:
        return f"${var_name} isa {entity_type_name}, iid {entity_iid}"

    # Fall back to key attribute matching
    key_attrs = {
        field_name: attr_info
        for field_name, attr_info in entity.__class__.get_all_attributes().items()
        if attr_info.flags.is_key
    }

    for field_name, attr_info in key_attrs.items():
        value = getattr(entity, field_name)
        if value is not None:
            attr_class = attr_info.typ
            attr_name = attr_class.get_attribute_name()
            formatted_value = format_value(value)
            return f"${var_name} isa {entity_type_name}, has {attr_name} {formatted_value}"

    # Neither IID nor key attributes available
    raise ValueError(
        f"Role player '{var_name}' ({entity.__class__.__name__}) cannot be identified: "
        f"no _iid set and no @key attributes defined. Either fetch the entity from the "
        f"database first (to populate _iid) or add Flag(Key) to an attribute."
    )

extract_relation_attributes 

extract_relation_attributes(model_class, result)

Extract relation attributes from a query result.

Uses the unified wrap_attribute_value() helper for consistent handling of multi-value and single-value attributes.

Parameters:

Name Type Description Default
model_class type[Relation]

The relation class to extract attributes for

 required
result dict[str, Any]

The result dictionary from TypeDB fetch

 required

Returns:

Type Description
dict[str, Any]

Dictionary mapping field_name -> attribute value (ready for model constructor)
Source code in type_bridge/crud/role_players.py 
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
def extract_relation_attributes(
    model_class: type["Relation"],
    result: dict[str, Any],
) -> dict[str, Any]:
    """Extract relation attributes from a query result.

    Uses the unified wrap_attribute_value() helper for consistent handling
    of multi-value and single-value attributes.

    Args:
        model_class: The relation class to extract attributes for
        result: The result dictionary from TypeDB fetch

    Returns:
        Dictionary mapping field_name -> attribute value (ready for model constructor)
    """
    attrs: dict[str, Any] = {}
    all_attrs = model_class.get_all_attributes()

    for field_name, attr_info in all_attrs.items():
        attr_class = attr_info.typ
        attr_name = attr_class.get_attribute_name()
        if attr_name in result:
            raw_value = result[attr_name]
            # Use unified wrapping helper for consistent behavior
            attrs[field_name] = wrap_attribute_value(
                raw_value, attr_info, use_pydantic_validate=True
            )
        else:
            # For list fields (has_explicit_card), default to empty list
            # For other optional fields, explicitly set to None
            if attr_info.flags.has_explicit_card:
                attrs[field_name] = []
            else:
                attrs[field_name] = None

    return attrs

extract_role_players_from_results 

extract_role_players_from_results(result_group, role_info, multi_player_roles)

Extract and deduplicate role players from grouped query results.

For multi-player roles, collects all unique players from all rows in the group. For single-player roles, takes the first player found.

Parameters:

Name Type Description Default
result_group list[dict[str, Any]]

List of result rows with the same relation IID

 required
role_info dict[str, tuple[str, tuple[type[Entity], ...]]]

Dict mapping role_name -> (role_var, allowed_entity_classes)

 required
multi_player_roles set[str]

Set of role names that allow multiple players

 required

Returns:

Type Description
dict[str, Any]

Dictionary mapping role_name -> player entity or list of player entities
Source code in type_bridge/crud/role_players.py 
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
224
225
226
227
228
229
230
231
232
233
234
235
def extract_role_players_from_results(
    result_group: list[dict[str, Any]],
    role_info: dict[str, tuple[str, tuple[type["Entity"], ...]]],
    multi_player_roles: set[str],
) -> dict[str, Any]:
    """Extract and deduplicate role players from grouped query results.

    For multi-player roles, collects all unique players from all rows in the group.
    For single-player roles, takes the first player found.

    Args:
        result_group: List of result rows with the same relation IID
        role_info: Dict mapping role_name -> (role_var, allowed_entity_classes)
        multi_player_roles: Set of role names that allow multiple players

    Returns:
        Dictionary mapping role_name -> player entity or list of player entities
    """
    role_players: dict[str, Any] = {}

    for role_name, (_, allowed_entity_classes) in role_info.items():
        is_multi = role_name in multi_player_roles
        collected_players: list[Any] = []
        seen_player_keys: set[tuple[Any, ...]] = set()

        for result in result_group:
            if role_name in result and isinstance(result[role_name], dict):
                player_data = result[role_name]

                # Get the actual type label from TypeDB (fetched via label())
                type_label = result.get(f"{role_name}_type")

                # Resolve entity class from type label for polymorphic support
                entity_class = resolve_entity_class_from_label(type_label, allowed_entity_classes)

                # Hydrate player entity using shared utility
                player_iid = result.get(f"{role_name}_iid")
                player_attrs, key_values = hydrate_attributes(
                    entity_class, player_data, wrap_values=True
                )

                # Create entity instance if we have any non-None attributes
                # Note: Relations as role players may have no owned attributes
                if player_attrs == {} or any(v is not None for v in player_attrs.values()):
                    # Deduplicate players based on their attribute values
                    if key_values not in seen_player_keys:
                        seen_player_keys.add(key_values)
                        player_entity = entity_class(**player_attrs)
                        if player_iid:
                            object.__setattr__(player_entity, "_iid", player_iid)
                        collected_players.append(player_entity)

        # Store collected players
        if collected_players:
            if is_multi:
                role_players[role_name] = collected_players
            else:
                role_players[role_name] = collected_players[0]

    return role_players

group_results_by_iid 

group_results_by_iid(results)

Group query results by relation/entity IID.

TypeDB returns one row per role player combination for multi-cardinality roles. This utility groups those rows by IID so they can be merged into a single relation instance.

Parameters:

Name Type Description Default
results list[dict[str, Any]]

List of result dictionaries from TypeDB fetch

 required

Returns:

Type Description
dict[str, list[dict[str, Any]]]

Dictionary mapping IID -> list of result rows with that IID
Source code in type_bridge/crud/role_players.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def group_results_by_iid(results: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
    """Group query results by relation/entity IID.

    TypeDB returns one row per role player combination for multi-cardinality
    roles. This utility groups those rows by IID so they can be merged into
    a single relation instance.

    Args:
        results: List of result dictionaries from TypeDB fetch

    Returns:
        Dictionary mapping IID -> list of result rows with that IID
    """
    grouped: dict[str, list[dict[str, Any]]] = {}
    for result in results:
        iid = result.get("_iid")
        if iid:
            if iid not in grouped:
                grouped[iid] = []
            grouped[iid].append(result)
    return grouped

resolve_entity_class_from_label 

resolve_entity_class_from_label(type_label, allowed_entity_classes)

Resolve the correct Python entity class from a TypeDB type label.

This is a thin wrapper around ModelRegistry.resolve() that provides polymorphic class resolution with caching.

Parameters:

Name Type Description Default
type_label str | None

The TypeDB type label (from label() function), e.g., "person"

 required
allowed_entity_classes tuple[type[Entity], ...]

Tuple of allowed entity classes for this role

 required

Returns:

Type Description
type[Entity]

The matching Python entity class, or the first allowed class as fallback
Source code in type_bridge/crud/role_players.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def resolve_entity_class_from_label(
    type_label: str | None,
    allowed_entity_classes: tuple[type["Entity"], ...],
) -> type["Entity"]:
    """Resolve the correct Python entity class from a TypeDB type label.

    This is a thin wrapper around ModelRegistry.resolve() that provides
    polymorphic class resolution with caching.

    Args:
        type_label: The TypeDB type label (from label() function), e.g., "person"
        allowed_entity_classes: Tuple of allowed entity classes for this role

    Returns:
        The matching Python entity class, or the first allowed class as fallback
    """
    from type_bridge.models.registry import ModelRegistry

    return ModelRegistry.resolve(type_label, allowed_entity_classes)

build_metadata_fetch 

build_metadata_fetch(var)

Build a fetch clause that retrieves only IID and type metadata.

Uses TypeQL 3.8.0 built-in functions iid() and label() to fetch the internal ID and type label. This is used for queries that need to identify entities/relations without fetching all attributes.

Note: TypeQL grammar doesn't allow mixing "key": value entries with $e.* in the same fetch clause, so metadata-only fetch is separate from attribute fetch.

Parameters:

Name Type Description Default
var str

Variable name (with or without $)

 required

Returns:

Type Description
str

Fetch clause string like 'fetch { "_iid": iid($e), "_type": label($e) }'
Example

build_metadata_fetch("e") 'fetch {\n "_iid": iid($e), "_type": label($e)\n}'

Source code in type_bridge/crud/types.py 
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
def build_metadata_fetch(var: str) -> str:
    """Build a fetch clause that retrieves only IID and type metadata.

    Uses TypeQL 3.8.0 built-in functions iid() and label() to fetch
    the internal ID and type label. This is used for queries that need
    to identify entities/relations without fetching all attributes.

    Note: TypeQL grammar doesn't allow mixing "key": value entries with $e.*
    in the same fetch clause, so metadata-only fetch is separate from
    attribute fetch.

    Args:
        var: Variable name (with or without $)

    Returns:
        Fetch clause string like 'fetch { "_iid": iid($e), "_type": label($e) }'

    Example:
        >>> build_metadata_fetch("e")
        'fetch {\\n  "_iid": iid($e), "_type": label($e)\\n}'
    """
    if not var.startswith("$"):
        var = f"${var}"

    return f'fetch {{\n  "_iid": iid({var}), "_type": label({var})\n}}'

extract_entity_key 

extract_entity_key(entity)

Extract the first key attribute from an entity for matching.

This is used to build match clauses based on key attributes when IID is not available.

Parameters:

Name Type Description Default
entity Any

The entity instance

 required

Returns:

Type Description
tuple[str, str, Any] | None

Tuple of (field_name, attr_typeql_name, raw_value) if a key attribute
tuple[str, str, Any] | None

with a non-None value is found, otherwise None.

Examples:

>>> extract_entity_key(person)  # person has name as @key
("name", "name", "Alice")
Source code in type_bridge/crud/types.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def extract_entity_key(entity: Any) -> tuple[str, str, Any] | None:
    """Extract the first key attribute from an entity for matching.

    This is used to build match clauses based on key attributes when IID
    is not available.

    Args:
        entity: The entity instance

    Returns:
        Tuple of (field_name, attr_typeql_name, raw_value) if a key attribute
        with a non-None value is found, otherwise None.

    Examples:
        >>> extract_entity_key(person)  # person has name as @key
        ("name", "name", "Alice")
    """
    for field_name, attr_info in entity.__class__.get_all_attributes().items():
        if attr_info.flags.is_key:
            key_value = getattr(entity, field_name, None)
            if key_value is not None:
                attr_name = attr_info.typ.get_attribute_name()
                # Unwrap Attribute instance
                raw_value = unwrap_attribute(key_value)
                return (field_name, attr_name, raw_value)
    return None

hydrate_attributes 

hydrate_attributes(entity_class, raw_data, wrap_values=False)

Hydrate attributes from TypeDB fetch result.

Extracts attribute values from a raw TypeDB result dictionary, mapping TypeDB attribute names to Python field names. Handles multi-value attributes and collects key values for deduplication.

Parameters:

Name Type Description Default
entity_class type[Entity]

The entity class to extract attributes for

 required
raw_data dict[str, Any]

Raw attribute data from TypeDB fetch result

 required
wrap_values bool

If True, wrap attributes in Attribute instances using the unified wrap_attribute_value() helper

 False

Returns:

Type Description
dict[str, Any]

Tuple of (attrs_dict, key_values_tuple):
tuple[tuple[str, Any], ...]
  • attrs_dict: Dictionary mapping field names to values
tuple[dict[str, Any], tuple[tuple[str, Any], ...]]
  • key_values_tuple: Tuple of (attr_name, value) for key attributes

Examples:

>>> result = {"name": "Alice", "age": 30}
>>> attrs, keys = hydrate_attributes(Person, result)
>>> attrs
{"name": "Alice", "age": 30}
>>> keys
(("name", "Alice"),)  # if name is a key attribute
Source code in type_bridge/crud/types.py 
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
def hydrate_attributes(
    entity_class: type[Entity],
    raw_data: dict[str, Any],
    wrap_values: bool = False,
) -> tuple[dict[str, Any], tuple[tuple[str, Any], ...]]:
    """Hydrate attributes from TypeDB fetch result.

    Extracts attribute values from a raw TypeDB result dictionary,
    mapping TypeDB attribute names to Python field names. Handles
    multi-value attributes and collects key values for deduplication.

    Args:
        entity_class: The entity class to extract attributes for
        raw_data: Raw attribute data from TypeDB fetch result
        wrap_values: If True, wrap attributes in Attribute instances using
                     the unified wrap_attribute_value() helper

    Returns:
        Tuple of (attrs_dict, key_values_tuple):
        - attrs_dict: Dictionary mapping field names to values
        - key_values_tuple: Tuple of (attr_name, value) for key attributes

    Examples:
        >>> result = {"name": "Alice", "age": 30}
        >>> attrs, keys = hydrate_attributes(Person, result)
        >>> attrs
        {"name": "Alice", "age": 30}
        >>> keys
        (("name", "Alice"),)  # if name is a key attribute
    """
    attrs: dict[str, Any] = {}
    key_values: list[tuple[str, Any]] = []

    for field_name, attr_info in entity_class.get_all_attributes().items():
        attr_class = attr_info.typ
        attr_name = attr_class.get_attribute_name()
        is_multi = is_multi_value_attribute(attr_info.flags)

        if attr_name in raw_data:
            raw_value = raw_data[attr_name]

            if wrap_values:
                # Use unified wrapping helper for consistent behavior
                attrs[field_name] = wrap_attribute_value(
                    raw_value, attr_info, use_pydantic_validate=True
                )
            elif is_multi and isinstance(raw_value, list):
                attrs[field_name] = raw_value
            else:
                attrs[field_name] = raw_value

            # Collect key values for deduplication (convert lists to tuples for hashability)
            if attr_info.flags.is_key:
                hashable_value = tuple(raw_value) if isinstance(raw_value, list) else raw_value
                key_values.append((attr_name, hashable_value))
        else:
            # Default for missing attributes
            if is_multi or attr_info.flags.has_explicit_card:
                attrs[field_name] = []
            else:
                attrs[field_name] = None

    return attrs, tuple(sorted(key_values))

is_multi_value_attribute 

is_multi_value_attribute(flags)

Check if attribute is multi-value based on cardinality.

Multi-value attributes have either: - Unbounded cardinality (card_max is None) - Maximum cardinality > 1

Single-value attributes have: - Maximum cardinality == 1 (including 0..1 and 1..1)

Parameters:

Name Type Description Default
flags AttributeFlags

AttributeFlags instance containing cardinality information

 required

Returns:

Type Description
bool

True if multi-value (card_max is None or > 1), False if single-value

Examples:

>>> flags = AttributeFlags(card_min=0, card_max=1)
>>> is_multi_value_attribute(flags)
False
>>> flags = AttributeFlags(card_min=0, card_max=5)
>>> is_multi_value_attribute(flags)
True
>>> flags = AttributeFlags(card_min=2, card_max=None)
>>> is_multi_value_attribute(flags)
True
Source code in type_bridge/crud/types.py 
 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
def is_multi_value_attribute(flags: AttributeFlags) -> bool:
    """Check if attribute is multi-value based on cardinality.

    Multi-value attributes have either:
    - Unbounded cardinality (card_max is None)
    - Maximum cardinality > 1

    Single-value attributes have:
    - Maximum cardinality == 1 (including 0..1 and 1..1)

    Args:
        flags: AttributeFlags instance containing cardinality information

    Returns:
        True if multi-value (card_max is None or > 1), False if single-value

    Examples:
        >>> flags = AttributeFlags(card_min=0, card_max=1)
        >>> is_multi_value_attribute(flags)
        False
        >>> flags = AttributeFlags(card_min=0, card_max=5)
        >>> is_multi_value_attribute(flags)
        True
        >>> flags = AttributeFlags(card_min=2, card_max=None)
        >>> is_multi_value_attribute(flags)
        True
    """
    # Single-value: card_max == 1 (including 0..1 and 1..1)
    # Multi-value: card_max is None (unbounded) or > 1
    if flags.card_max is None:
        # Unbounded means multi-value
        return True
    return flags.card_max > 1