Skip to content

type_bridge.crud.base

base

Base types and classes for CRUD operations.

BaseQuery 

BaseQuery(connection, model_class, filters=None)

Bases: ABC

Abstract base class for chainable query operations.

Provides shared implementation for common query methods used by both EntityQuery and RelationQuery.

Initialize base query.

Parameters:

Name Type Description Default
connection Connection

Database, Transaction, or TransactionContext

 required
model_class type[T]

Model class (Entity or Relation subclass)

 required
filters dict[str, Any] | None

Attribute filters (exact match) - optional, defaults to empty dict

 None
Source code in type_bridge/crud/base.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def __init__(
    self,
    connection: Connection,
    model_class: type[T],
    filters: dict[str, Any] | None = None,
):
    """Initialize base query.

    Args:
        connection: Database, Transaction, or TransactionContext
        model_class: Model class (Entity or Relation subclass)
        filters: Attribute filters (exact match) - optional, defaults to empty dict
    """
    self._connection = connection
    self._executor = ConnectionExecutor(connection)
    self.model_class = model_class
    self.filters = filters or {}
    self._expressions: list[Any] = []
    self._limit_value: int | None = None
    self._offset_value: int | None = None

limit 

limit(limit)

Limit number of results.

Parameters:

Name Type Description Default
limit int

Maximum number of results

 required

Returns:

Type Description
BaseQuery[T]

Self for chaining
Source code in type_bridge/crud/base.py 
58
59
60
61
62
63
64
65
66
67
68
def limit(self, limit: int) -> "BaseQuery[T]":
    """Limit number of results.

    Args:
        limit: Maximum number of results

    Returns:
        Self for chaining
    """
    self._limit_value = limit
    return self

offset 

offset(offset)

Skip number of results.

Parameters:

Name Type Description Default
offset int

Number of results to skip

 required

Returns:

Type Description
BaseQuery[T]

Self for chaining
Source code in type_bridge/crud/base.py 
70
71
72
73
74
75
76
77
78
79
80
def offset(self, offset: int) -> "BaseQuery[T]":
    """Skip number of results.

    Args:
        offset: Number of results to skip

    Returns:
        Self for chaining
    """
    self._offset_value = offset
    return self

execute abstractmethod 

execute()

Execute the query and return results.

Returns:

Type Description
list[T]

List of matching model instances
Source code in type_bridge/crud/base.py 
82
83
84
85
86
87
88
89
@abstractmethod
def execute(self) -> list[T]:
    """Execute the query and return results.

    Returns:
        List of matching model instances
    """
    ...

first 

first()

Get first matching result.

Returns:

Type Description
T | None

First result or None
Source code in type_bridge/crud/base.py 
91
92
93
94
95
96
97
98
def first(self) -> T | None:
    """Get first matching result.

    Returns:
        First result or None
    """
    results = self.limit(1).execute()
    return results[0] if results else None

count 

count()

Count matching results.

Returns:

Type Description
int

Number of matching results
Source code in type_bridge/crud/base.py 
100
101
102
103
104
105
106
def count(self) -> int:
    """Count matching results.

    Returns:
        Number of matching results
    """
    return len(self.execute())