Skip to content

graflo.architecture.vertex

Vertex configuration and management for graph databases.

This module provides classes and utilities for managing vertices in graph databases. It handles vertex configuration, field management, indexing, and filtering operations. The module supports both ArangoDB and Neo4j through the DBType enum.

Key Components
  • Vertex: Represents a vertex with its fields and indexes
  • VertexConfig: Manages vertices and their configurations
Example

vertex = Vertex(name="user", fields=["id", "name"]) config = VertexConfig(vertices=[vertex]) fields = config.fields("user") # Returns list[Field] field_names = config.fields_names("user") # Returns list[str]

Field

Bases: ConfigBaseModel

Represents a typed field in a vertex.

Field objects behave like strings for backward compatibility. They can be used in sets, as dictionary keys, and in string comparisons. The type information is preserved for databases that need it (like TigerGraph).

Attributes:

Name Type Description
name str

Name of the field
type FieldType | None

Optional type of the field. Can be FieldType enum, str, or None at construction. Strings are converted to FieldType enum by the validator. None is allowed (most databases like ArangoDB don't require types). Defaults to None.
Source code in graflo/architecture/vertex.py 
 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
class Field(ConfigBaseModel):
    """Represents a typed field in a vertex.

    Field objects behave like strings for backward compatibility. They can be used
    in sets, as dictionary keys, and in string comparisons. The type information
    is preserved for databases that need it (like TigerGraph).

    Attributes:
        name: Name of the field
        type: Optional type of the field. Can be FieldType enum, str, or None at construction.
              Strings are converted to FieldType enum by the validator.
              None is allowed (most databases like ArangoDB don't require types).
              Defaults to None.
    """

    model_config = ConfigDict(extra="forbid")

    name: str = PydanticField(
        ...,
        description="Name of the field (e.g. column or attribute name).",
    )
    type: FieldType | None = PydanticField(
        default=None,
        description="Optional field type for databases that require it (e.g. TigerGraph: INT, STRING). None for schema-agnostic backends.",
    )

    @field_validator("type", mode="before")
    @classmethod
    def normalize_type(cls, v: Any) -> FieldType | None:
        if v is None:
            return None
        if isinstance(v, FieldType):
            return v
        if isinstance(v, str):
            type_upper = v.upper()
            if type_upper not in FieldType:
                allowed_types = sorted(ft.value for ft in FieldType)
                raise ValueError(
                    f"Field type '{v}' is not allowed. "
                    f"Allowed types are: {', '.join(allowed_types)}"
                )
            return FieldType(type_upper)
        allowed_types = sorted(ft.value for ft in FieldType)
        raise ValueError(
            f"Field type must be FieldType enum, str, or None, got {type(v)}. "
            f"Allowed types are: {', '.join(allowed_types)}"
        )

    def __str__(self) -> str:
        """Return field name as string for backward compatibility."""
        return self.name

    def __repr__(self) -> str:
        """Return representation including type information."""
        if self.type:
            return f"Field(name='{self.name}', type='{self.type}')"
        return f"Field(name='{self.name}')"

    def __hash__(self) -> int:
        """Hash by name only, allowing Field objects to work in sets and as dict keys."""
        return hash(self.name)

    def __eq__(self, other: object) -> bool:
        """Compare equal to strings with same name, or other Field objects with same name."""
        if isinstance(other, Field):
            return self.name == other.name
        if isinstance(other, str):
            return self.name == other
        return False

    def __ne__(self, other: object) -> bool:
        """Compare not equal."""
        return not self.__eq__(other)

__eq__(other)

Compare equal to strings with same name, or other Field objects with same name.

Source code in graflo/architecture/vertex.py 
132
133
134
135
136
137
138
def __eq__(self, other: object) -> bool:
    """Compare equal to strings with same name, or other Field objects with same name."""
    if isinstance(other, Field):
        return self.name == other.name
    if isinstance(other, str):
        return self.name == other
    return False

__hash__()

Hash by name only, allowing Field objects to work in sets and as dict keys.

Source code in graflo/architecture/vertex.py 
128
129
130
def __hash__(self) -> int:
    """Hash by name only, allowing Field objects to work in sets and as dict keys."""
    return hash(self.name)

__ne__(other)

Compare not equal.

Source code in graflo/architecture/vertex.py 
140
141
142
def __ne__(self, other: object) -> bool:
    """Compare not equal."""
    return not self.__eq__(other)

__repr__()

Return representation including type information.

Source code in graflo/architecture/vertex.py 
122
123
124
125
126
def __repr__(self) -> str:
    """Return representation including type information."""
    if self.type:
        return f"Field(name='{self.name}', type='{self.type}')"
    return f"Field(name='{self.name}')"

__str__()

Return field name as string for backward compatibility.

Source code in graflo/architecture/vertex.py 
118
119
120
def __str__(self) -> str:
    """Return field name as string for backward compatibility."""
    return self.name

FieldType

Bases: BaseEnum

Supported field types for graph databases.

These types are primarily used for TigerGraph, which requires explicit field types. Other databases (ArangoDB, Neo4j) may use different type systems or not require types.

Attributes:

Name Type Description
INT

Integer type
UINT

Unsigned integer type
FLOAT

Floating point type
DOUBLE

Double precision floating point type
BOOL

Boolean type
STRING

String type
DATETIME

DateTime type
Source code in graflo/architecture/vertex.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
class FieldType(BaseEnum):
    """Supported field types for graph databases.

    These types are primarily used for TigerGraph, which requires explicit field types.
    Other databases (ArangoDB, Neo4j) may use different type systems or not require types.

    Attributes:
        INT: Integer type
        UINT: Unsigned integer type
        FLOAT: Floating point type
        DOUBLE: Double precision floating point type
        BOOL: Boolean type
        STRING: String type
        DATETIME: DateTime type
    """

    INT = "INT"
    UINT = "UINT"
    FLOAT = "FLOAT"
    DOUBLE = "DOUBLE"
    BOOL = "BOOL"
    STRING = "STRING"
    DATETIME = "DATETIME"

Vertex

Bases: ConfigBaseModel

Represents a vertex in the graph database.

A vertex is a fundamental unit in the graph that can have fields, indexes, and filters. Fields can be specified as strings, Field objects, or dicts. Internally, fields are stored as Field objects but behave like strings for backward compatibility.

Attributes:

Name Type Description
name str

Name of the vertex
fields list[Field]

List of field names (str), Field objects, or dicts. Will be normalized to Field objects by the validator.
indexes list[Index]

List of indexes for the vertex
filters list[FilterExpression]

List of filter expressions
dbname str | None

Optional database name (defaults to vertex name)

Examples:

>>> # Backward compatible: list of strings
>>> v1 = Vertex(name="user", fields=["id", "name"])

>>> # Typed fields: list of Field objects
>>> v2 = Vertex(name="user", fields=[
...     Field(name="id", type="INT"),
...     Field(name="name", type="STRING")
... ])

>>> # From dicts (e.g., from YAML/JSON)
>>> v3 = Vertex(name="user", fields=[
...     {"name": "id", "type": "INT"},
...     {"name": "name"}  # defaults to None type
... ])
Source code in graflo/architecture/vertex.py 
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
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
class Vertex(ConfigBaseModel):
    """Represents a vertex in the graph database.

    A vertex is a fundamental unit in the graph that can have fields, indexes,
    and filters. Fields can be specified as strings, Field objects, or dicts.
    Internally, fields are stored as Field objects but behave like strings
    for backward compatibility.

    Attributes:
        name: Name of the vertex
        fields: List of field names (str), Field objects, or dicts.
               Will be normalized to Field objects by the validator.
        indexes: List of indexes for the vertex
        filters: List of filter expressions
        dbname: Optional database name (defaults to vertex name)

    Examples:
        >>> # Backward compatible: list of strings
        >>> v1 = Vertex(name="user", fields=["id", "name"])

        >>> # Typed fields: list of Field objects
        >>> v2 = Vertex(name="user", fields=[
        ...     Field(name="id", type="INT"),
        ...     Field(name="name", type="STRING")
        ... ])

        >>> # From dicts (e.g., from YAML/JSON)
        >>> v3 = Vertex(name="user", fields=[
        ...     {"name": "id", "type": "INT"},
        ...     {"name": "name"}  # defaults to None type
        ... ])
    """

    # Allow extra keys when loading from YAML (e.g. transforms, other runtime keys)
    model_config = ConfigDict(extra="ignore")

    name: str = PydanticField(
        ...,
        description="Name of the vertex type (e.g. user, post, company).",
    )
    fields: list[Field] = PydanticField(
        default_factory=list,
        description="List of fields (names, Field objects, or dicts). Normalized to Field objects.",
    )
    indexes: list[Index] = PydanticField(
        default_factory=list,
        description="List of index definitions for this vertex. Defaults to primary index on all fields if empty.",
    )
    filters: list[FilterExpression] = PydanticField(
        default_factory=list,
        description="Filter expressions (logical formulae) applied when querying this vertex.",
    )
    dbname: str | None = PydanticField(
        default=None,
        description="Optional database collection/table name. Defaults to vertex name if not set.",
    )

    @field_validator("fields", mode="before")
    @classmethod
    def convert_to_fields(cls, v: Any) -> Any:
        if not isinstance(v, list):
            raise ValueError("fields must be a list")
        return [_normalize_fields_item(item) for item in v]

    @field_validator("indexes", mode="before")
    @classmethod
    def convert_to_indexes(cls, v: Any) -> Any:
        if not isinstance(v, list):
            return v
        result = []
        for item in v:
            if isinstance(item, dict):
                result.append(Index.model_validate(item))
            else:
                result.append(item)
        return result

    @field_validator("filters", mode="before")
    @classmethod
    def convert_to_expressions(cls, v: Any) -> Any:
        if not isinstance(v, list):
            return v
        result: list[FilterExpression] = []
        for item in v:
            if isinstance(item, FilterExpression):
                result.append(item)
            elif isinstance(item, (dict, list)):
                result.append(FilterExpression.from_dict(item))
            else:
                raise ValueError(
                    "each filter must be a FilterExpression instance or a dict/list (parsed as FilterExpression)"
                )
        return result

    @model_validator(mode="after")
    def set_dbname_and_indexes(self) -> "Vertex":
        if self.dbname is None:
            object.__setattr__(self, "dbname", self.name)
        indexes = list(self.indexes)
        if not indexes:
            object.__setattr__(
                self,
                "indexes",
                [Index(fields=[f.name for f in self.fields])],
            )
        else:
            seen_names = {f.name for f in self.fields}
            new_fields = list(self.fields)
            for idx in indexes:
                for field_name in idx.fields:
                    if field_name not in seen_names:
                        new_fields.append(Field(name=field_name, type=None))
                        seen_names.add(field_name)
            object.__setattr__(self, "fields", new_fields)
        return self

    @property
    def field_names(self) -> list[str]:
        """Get list of field names (as strings)."""
        return [field.name for field in self.fields]

    def get_fields(self) -> list[Field]:
        return self.fields

    def finish_init(self, db_flavor: DBType):
        """Complete initialization of vertex with database-specific field types.

        Args:
            db_flavor: Database flavor to use for initialization
        """
        self.fields = [
            Field(name=f.name, type=FieldType.STRING)
            if f.type is None and db_flavor == DBType.TIGERGRAPH
            else f
            for f in self.fields
        ]

field_names property

Get list of field names (as strings).

finish_init(db_flavor)

Complete initialization of vertex with database-specific field types.

Parameters:

Name Type Description Default
db_flavor DBType

Database flavor to use for initialization

 required
Source code in graflo/architecture/vertex.py 
305
306
307
308
309
310
311
312
313
314
315
316
def finish_init(self, db_flavor: DBType):
    """Complete initialization of vertex with database-specific field types.

    Args:
        db_flavor: Database flavor to use for initialization
    """
    self.fields = [
        Field(name=f.name, type=FieldType.STRING)
        if f.type is None and db_flavor == DBType.TIGERGRAPH
        else f
        for f in self.fields
    ]

VertexConfig

Bases: ConfigBaseModel

Configuration for managing vertices.

This class manages vertices, providing methods for accessing and manipulating vertex configurations.

Attributes:

Name Type Description
vertices list[Vertex]

List of vertex configurations
blank_vertices list[str]

List of blank vertex names
force_types dict[str, list]

Dictionary mapping vertex names to type lists
db_flavor DBType

Database flavor (ARANGO or NEO4J)
Source code in graflo/architecture/vertex.py 
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
class VertexConfig(ConfigBaseModel):
    """Configuration for managing vertices.

    This class manages vertices, providing methods for accessing
    and manipulating vertex configurations.

    Attributes:
        vertices: List of vertex configurations
        blank_vertices: List of blank vertex names
        force_types: Dictionary mapping vertex names to type lists
        db_flavor: Database flavor (ARANGO or NEO4J)
    """

    # Allow extra keys when loading from YAML (e.g. vertex_config wrapper key)
    model_config = ConfigDict(extra="ignore")

    vertices: list[Vertex] = PydanticField(
        ...,
        description="List of vertex type definitions (name, fields, indexes, filters).",
    )
    blank_vertices: list[str] = PydanticField(
        default_factory=list,
        description="Vertex names that may be created without explicit data (e.g. placeholders).",
    )
    force_types: dict[str, list] = PydanticField(
        default_factory=dict,
        description="Override mapping: vertex name -> list of field type names for type inference.",
    )
    db_flavor: DBType = PydanticField(
        default=DBType.ARANGO,
        description="Database flavor (ARANGO, NEO4J, TIGERGRAPH) for schema and index generation.",
    )

    _vertices_map: dict[str, Vertex] | None = PrivateAttr(default=None)
    _vertex_numeric_fields_map: dict[str, object] | None = PrivateAttr(default=None)

    @model_validator(mode="after")
    def build_vertices_map_and_validate_blank(self) -> "VertexConfig":
        object.__setattr__(
            self,
            "_vertices_map",
            {item.name: item for item in self.vertices},
        )
        object.__setattr__(self, "_vertex_numeric_fields_map", {})
        if set(self.blank_vertices) - set(self.vertex_set):
            raise ValueError(
                f" Blank vertices {self.blank_vertices} are not defined as vertices"
            )
        return self

    def _get_vertices_map(self) -> dict[str, Vertex]:
        """Return the vertices map (set by model validator)."""
        assert self._vertices_map is not None, "VertexConfig not fully initialized"
        return self._vertices_map

    @property
    def vertex_set(self):
        """Get set of vertex names.

        Returns:
            set[str]: Set of vertex names
        """
        return set(self._get_vertices_map().keys())

    @property
    def vertex_list(self):
        """Get list of vertex configurations.

        Returns:
            list[Vertex]: List of vertex configurations
        """
        return list(self._get_vertices_map().values())

    def _get_vertex_by_name_or_dbname(self, identifier: str) -> Vertex:
        """Get vertex by name or dbname.

        Args:
            identifier: Vertex name or dbname

        Returns:
            Vertex: The vertex object

        Raises:
            KeyError: If vertex is not found by name or dbname
        """
        m = self._get_vertices_map()
        # First try by name (most common case)
        if identifier in m:
            return m[identifier]

        # Try by dbname
        for vertex in m.values():
            if vertex.dbname == identifier:
                return vertex

        # Not found
        available_names = list(m.keys())
        available_dbnames = [v.dbname for v in m.values()]
        raise KeyError(
            f"Vertex '{identifier}' not found by name or dbname. "
            f"Available names: {available_names}, "
            f"Available dbnames: {available_dbnames}"
        )

    def vertex_dbname(self, vertex_name):
        """Get database name for a vertex.

        Args:
            vertex_name: Name of the vertex

        Returns:
            str: Database name for the vertex

        Raises:
            KeyError: If vertex is not found
        """
        m = self._get_vertices_map()
        try:
            value = m[vertex_name].dbname
        except KeyError as e:
            logger.error(
                f"Available vertices : {m.keys()}; vertex requested : {vertex_name}"
            )
            raise e
        return value

    def index(self, vertex_name) -> Index:
        """Get primary index for a vertex.

        Args:
            vertex_name: Name of the vertex

        Returns:
            Index: Primary index for the vertex
        """
        return self._get_vertices_map()[vertex_name].indexes[0]

    def indexes(self, vertex_name) -> list[Index]:
        """Get all indexes for a vertex.

        Args:
            vertex_name: Name of the vertex

        Returns:
            list[Index]: List of indexes for the vertex
        """
        return self._get_vertices_map()[vertex_name].indexes

    def fields(self, vertex_name: str) -> list[Field]:
        """Get fields for a vertex.

        Args:
            vertex_name: Name of the vertex or dbname

        Returns:
            list[Field]: List of Field objects
        """
        # Get vertex by name or dbname
        vertex = self._get_vertex_by_name_or_dbname(vertex_name)

        return vertex.fields

    def fields_names(
        self,
        vertex_name: str,
    ) -> list[str]:
        """Get field names for a vertex as strings.

        Args:
            vertex_name: Name of the vertex or dbname

        Returns:
            list[str]: List of field names as strings
        """
        vertex = self._get_vertex_by_name_or_dbname(vertex_name)
        return vertex.field_names

    def numeric_fields_list(self, vertex_name):
        """Get list of numeric fields for a vertex.

        Args:
            vertex_name: Name of the vertex

        Returns:
            tuple: Tuple of numeric field names

        Raises:
            ValueError: If vertex is not defined in config
        """
        if vertex_name in self.vertex_set:
            nmap = self._vertex_numeric_fields_map
            if nmap is not None and vertex_name in nmap:
                return nmap[vertex_name]
            else:
                return ()
        else:
            raise ValueError(
                " Accessing vertex numeric fields: vertex"
                f" {vertex_name} was not defined in config"
            )

    def filters(self, vertex_name) -> list[FilterExpression]:
        """Get filter clauses for a vertex.

        Args:
            vertex_name: Name of the vertex

        Returns:
            list[FilterExpression]: List of filter expressions
        """
        m = self._get_vertices_map()
        if vertex_name in m:
            return m[vertex_name].filters
        else:
            return []

    def remove_vertices(self, names: set[str]) -> None:
        """Remove vertices by name.

        Removes vertices from the configuration and from blank_vertices
        when present. Mutates the instance in place.

        Args:
            names: Set of vertex names to remove
        """
        if not names:
            return
        self.vertices[:] = [v for v in self.vertices if v.name not in names]
        m = self._get_vertices_map()
        for n in names:
            m.pop(n, None)
        self.blank_vertices[:] = [b for b in self.blank_vertices if b not in names]

    def update_vertex(self, v: Vertex):
        """Update vertex configuration.

        Args:
            v: Vertex configuration to update
        """
        self._get_vertices_map()[v.name] = v

    def __getitem__(self, key: str):
        """Get vertex configuration by name.

        Args:
            key: Vertex name

        Returns:
            Vertex: Vertex configuration

        Raises:
            KeyError: If vertex is not found
        """
        m = self._get_vertices_map()
        if key in m:
            return m[key]
        else:
            raise KeyError(f"Vertex {key} absent")

    def __setitem__(self, key: str, value: Vertex):
        """Set vertex configuration by name.

        Args:
            key: Vertex name
            value: Vertex configuration
        """
        self._get_vertices_map()[key] = value

    def finish_init(self):
        """Complete initialization of all vertices with database-specific field types.

        Uses self.db_flavor to determine database-specific initialization behavior.
        """
        for v in self.vertices:
            v.finish_init(self.db_flavor)

vertex_list property

Get list of vertex configurations.

Returns:

Type Description

list[Vertex]: List of vertex configurations

vertex_set property

Get set of vertex names.

Returns:

Type Description

set[str]: Set of vertex names

__getitem__(key)

Get vertex configuration by name.

Parameters:

Name Type Description Default
key str

Vertex name

 required

Returns:

Name Type Description
Vertex

Vertex configuration

Raises:

Type Description
KeyError

If vertex is not found
Source code in graflo/architecture/vertex.py 
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
def __getitem__(self, key: str):
    """Get vertex configuration by name.

    Args:
        key: Vertex name

    Returns:
        Vertex: Vertex configuration

    Raises:
        KeyError: If vertex is not found
    """
    m = self._get_vertices_map()
    if key in m:
        return m[key]
    else:
        raise KeyError(f"Vertex {key} absent")

__setitem__(key, value)

Set vertex configuration by name.

Parameters:

Name Type Description Default
key str

Vertex name

 required
value Vertex

Vertex configuration

 required
Source code in graflo/architecture/vertex.py 
578
579
580
581
582
583
584
585
def __setitem__(self, key: str, value: Vertex):
    """Set vertex configuration by name.

    Args:
        key: Vertex name
        value: Vertex configuration
    """
    self._get_vertices_map()[key] = value

fields(vertex_name)

Get fields for a vertex.

Parameters:

Name Type Description Default
vertex_name str

Name of the vertex or dbname

 required

Returns:

Type Description
list[Field]

list[Field]: List of Field objects
Source code in graflo/architecture/vertex.py 
467
468
469
470
471
472
473
474
475
476
477
478
479
def fields(self, vertex_name: str) -> list[Field]:
    """Get fields for a vertex.

    Args:
        vertex_name: Name of the vertex or dbname

    Returns:
        list[Field]: List of Field objects
    """
    # Get vertex by name or dbname
    vertex = self._get_vertex_by_name_or_dbname(vertex_name)

    return vertex.fields

fields_names(vertex_name)

Get field names for a vertex as strings.

Parameters:

Name Type Description Default
vertex_name str

Name of the vertex or dbname

 required

Returns:

Type Description
list[str]

list[str]: List of field names as strings
Source code in graflo/architecture/vertex.py 
481
482
483
484
485
486
487
488
489
490
491
492
493
494
def fields_names(
    self,
    vertex_name: str,
) -> list[str]:
    """Get field names for a vertex as strings.

    Args:
        vertex_name: Name of the vertex or dbname

    Returns:
        list[str]: List of field names as strings
    """
    vertex = self._get_vertex_by_name_or_dbname(vertex_name)
    return vertex.field_names

filters(vertex_name)

Get filter clauses for a vertex.

Parameters:

Name Type Description Default
vertex_name

Name of the vertex

 required

Returns:

Type Description
list[FilterExpression]

list[FilterExpression]: List of filter expressions
Source code in graflo/architecture/vertex.py 
520
521
522
523
524
525
526
527
528
529
530
531
532
533
def filters(self, vertex_name) -> list[FilterExpression]:
    """Get filter clauses for a vertex.

    Args:
        vertex_name: Name of the vertex

    Returns:
        list[FilterExpression]: List of filter expressions
    """
    m = self._get_vertices_map()
    if vertex_name in m:
        return m[vertex_name].filters
    else:
        return []

finish_init()

Complete initialization of all vertices with database-specific field types.

Uses self.db_flavor to determine database-specific initialization behavior.

Source code in graflo/architecture/vertex.py 
587
588
589
590
591
592
593
def finish_init(self):
    """Complete initialization of all vertices with database-specific field types.

    Uses self.db_flavor to determine database-specific initialization behavior.
    """
    for v in self.vertices:
        v.finish_init(self.db_flavor)

index(vertex_name)

Get primary index for a vertex.

Parameters:

Name Type Description Default
vertex_name

Name of the vertex

 required

Returns:

Name Type Description
Index Index

Primary index for the vertex
Source code in graflo/architecture/vertex.py 
445
446
447
448
449
450
451
452
453
454
def index(self, vertex_name) -> Index:
    """Get primary index for a vertex.

    Args:
        vertex_name: Name of the vertex

    Returns:
        Index: Primary index for the vertex
    """
    return self._get_vertices_map()[vertex_name].indexes[0]

indexes(vertex_name)

Get all indexes for a vertex.

Parameters:

Name Type Description Default
vertex_name

Name of the vertex

 required

Returns:

Type Description
list[Index]

list[Index]: List of indexes for the vertex
Source code in graflo/architecture/vertex.py 
456
457
458
459
460
461
462
463
464
465
def indexes(self, vertex_name) -> list[Index]:
    """Get all indexes for a vertex.

    Args:
        vertex_name: Name of the vertex

    Returns:
        list[Index]: List of indexes for the vertex
    """
    return self._get_vertices_map()[vertex_name].indexes

numeric_fields_list(vertex_name)

Get list of numeric fields for a vertex.

Parameters:

Name Type Description Default
vertex_name

Name of the vertex

 required

Returns:

Name Type Description
tuple

Tuple of numeric field names

Raises:

Type Description
ValueError

If vertex is not defined in config
Source code in graflo/architecture/vertex.py 
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
def numeric_fields_list(self, vertex_name):
    """Get list of numeric fields for a vertex.

    Args:
        vertex_name: Name of the vertex

    Returns:
        tuple: Tuple of numeric field names

    Raises:
        ValueError: If vertex is not defined in config
    """
    if vertex_name in self.vertex_set:
        nmap = self._vertex_numeric_fields_map
        if nmap is not None and vertex_name in nmap:
            return nmap[vertex_name]
        else:
            return ()
    else:
        raise ValueError(
            " Accessing vertex numeric fields: vertex"
            f" {vertex_name} was not defined in config"
        )

remove_vertices(names)

Remove vertices by name.

Removes vertices from the configuration and from blank_vertices when present. Mutates the instance in place.

Parameters:

Name Type Description Default
names set[str]

Set of vertex names to remove

 required
Source code in graflo/architecture/vertex.py 
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
def remove_vertices(self, names: set[str]) -> None:
    """Remove vertices by name.

    Removes vertices from the configuration and from blank_vertices
    when present. Mutates the instance in place.

    Args:
        names: Set of vertex names to remove
    """
    if not names:
        return
    self.vertices[:] = [v for v in self.vertices if v.name not in names]
    m = self._get_vertices_map()
    for n in names:
        m.pop(n, None)
    self.blank_vertices[:] = [b for b in self.blank_vertices if b not in names]

update_vertex(v)

Update vertex configuration.

Parameters:

Name Type Description Default
v Vertex

Vertex configuration to update

 required
Source code in graflo/architecture/vertex.py 
552
553
554
555
556
557
558
def update_vertex(self, v: Vertex):
    """Update vertex configuration.

    Args:
        v: Vertex configuration to update
    """
    self._get_vertices_map()[v.name] = v

vertex_dbname(vertex_name)

Get database name for a vertex.

Parameters:

Name Type Description Default
vertex_name

Name of the vertex

 required

Returns:

Name Type Description
str

Database name for the vertex

Raises:

Type Description
KeyError

If vertex is not found
Source code in graflo/architecture/vertex.py 
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
def vertex_dbname(self, vertex_name):
    """Get database name for a vertex.

    Args:
        vertex_name: Name of the vertex

    Returns:
        str: Database name for the vertex

    Raises:
        KeyError: If vertex is not found
    """
    m = self._get_vertices_map()
    try:
        value = m[vertex_name].dbname
    except KeyError as e:
        logger.error(
            f"Available vertices : {m.keys()}; vertex requested : {vertex_name}"
        )
        raise e
    return value