Skip to content

Table API

The Table class provides a convenient interface for performing CRUD operations on a specific database table using the synchronous API.

Class Documentation

Table

Database table with CRUD operations.

Source code in manticore_cockroachdb/crud/table.py 
  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
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
317
318
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
class Table:
    """Database table with CRUD operations."""

    table_name = None
    schema = None

    def __init__(
        self,
        name_or_db: Union[str, Database],
        db: Optional[Database] = None,
        schema: Optional[Dict[str, str]] = None,
        if_not_exists: bool = True
    ):
        """Initialize table.

        Args:
            name_or_db: Table name or Database instance
            db: Database instance (if first arg is table name)
            schema: Column definitions {name: type}
            if_not_exists: Whether to create table only if it doesn't exist
        """
        # Handle different initialization patterns
        if isinstance(name_or_db, Database):
            self.db = name_or_db
            self.name = getattr(self, 'table_name', None)
        else:
            self.name = name_or_db
            self.db = db or Database()

        # Use schema from args or class attribute
        self._schema = schema or getattr(self, 'schema', None)
        self._initialized = False

        # Initialize table schema if provided
        if self._schema:
            self.initialize(if_not_exists=if_not_exists)

    def initialize(self, schema: Optional[Dict[str, str]] = None, if_not_exists: bool = True) -> None:
        """Initialize table schema.

        Args:
            schema: Column definitions {name: type}
            if_not_exists: Whether to create table only if it doesn't exist
        """
        # Use provided schema or instance schema
        schema_to_use = schema or self._schema

        if not schema_to_use:
            raise ValueError("No schema provided for table initialization")

        self.db.create_table(self.name, schema_to_use, if_not_exists)
        self._initialized = True

    def _check_initialized(self):
        """Check if table is initialized."""
        from .exceptions import TableNotInitializedError
        if not self._initialized:
            raise TableNotInitializedError(self.name)

    def create(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """Create a new record.

        Args:
            data: Record data

        Returns:
            Created record
        """
        self._check_initialized()
        return self.db.insert(self.name, data)

    def read(self, id: Union[str, int]) -> Optional[Dict[str, Any]]:
        """Read a record by ID.

        Args:
            id: Record ID

        Returns:
            Record data

        Raises:
            TableNotFoundError: If the record does not exist
        """
        self._check_initialized()
        results = self.db.select(self.name, where={"id": id})
        if not results:
            from .exceptions import TableNotFoundError
            raise TableNotFoundError(f"Record with id '{id}' not found in table '{self.name}'")
        return results[0]

    def get(self, id: Union[str, int]) -> Optional[Dict[str, Any]]:
        """Get a record by ID.

        Args:
            id: Record ID

        Returns:
            Record data or None if not found
        """
        self._check_initialized()
        results = self.db.select(self.name, where={"id": id})
        return results[0] if results else None

    def get_many(self, ids: List[Union[str, int]]) -> List[Dict[str, Any]]:
        """Get multiple records by their IDs.

        Args:
            ids: List of record IDs

        Returns:
            List of records matching the provided IDs
        """
        self._check_initialized()

        if not ids:
            return []

        # Build query with IN clause
        placeholders = ", ".join(["%s" for _ in ids])
        query = f'SELECT * FROM "{self.name}" WHERE "id" IN ({placeholders})'

        # Execute query
        return self.db.execute(query, tuple(ids)) or []

    def find_one(self, where: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        """Find a single record by criteria.

        Args:
            where: Filter conditions

        Returns:
            Record data or None if not found
        """
        self._check_initialized()
        results = self.db.select(self.name, where=where, limit=1)
        return results[0] if results else None

    def update(self, id: Union[str, int], data: Dict[str, Any]) -> Dict[str, Any]:
        """Update a record by ID.

        Args:
            id: Record ID
            data: Update data

        Returns:
            Updated record

        Raises:
            TableNotFoundError: If the record does not exist
        """
        self._check_initialized()
        result = self.db.update(self.name, data, {"id": id})
        if result is None:
            from .exceptions import TableNotFoundError
            raise TableNotFoundError(f"Record with id '{id}' not found in table '{self.name}'")
        return result

    def update_where(self, where: Dict[str, Any], data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        """Update records matching criteria.

        Args:
            where: Filter conditions
            data: Update data

        Returns:
            First updated record or None if none found
        """
        self._check_initialized()
        return self.db.update(self.name, data, where)

    def delete(self, id: Union[str, int]) -> bool:
        """Delete a record by ID.

        Args:
            id: Record ID

        Returns:
            True if record was deleted
        """
        self._check_initialized()
        return self.db.delete(self.name, {"id": id})

    def delete_where(self, where: Dict[str, Any]) -> bool:
        """Delete records matching criteria.

        Args:
            where: Filter conditions

        Returns:
            True if any records were deleted
        """
        self._check_initialized()
        return self.db.delete(self.name, where)

    def list(
        self,
        where: Optional[Dict[str, Any]] = None,
        order_by: Optional[str] = None,
        limit: Optional[int] = None,
        offset: Optional[int] = None
    ) -> List[Dict[str, Any]]:
        """List records.

        Args:
            where: Filter conditions
            order_by: Order by clause
            limit: Result limit
            offset: Result offset

        Returns:
            List of records
        """
        self._check_initialized()
        return self.db.select(self.name, where=where, order_by=order_by, limit=limit, offset=offset)

    def count(self, where: Optional[Dict[str, Any]] = None) -> int:
        """Count records.

        Args:
            where: Filter conditions

        Returns:
            Number of records
        """
        self._check_initialized()
        query = f'SELECT COUNT(*) as count FROM "{self.name}"'
        params = []

        if where:
            conditions = []
            for col, val in where.items():
                conditions.append(f'"{col}" = %s')
                params.append(val)
            query += f" WHERE {' AND '.join(conditions)}"

        result = self.db.execute(query, tuple(params) if params else None)
        return result[0]["count"] if result else 0

    def exists(self, id: Union[str, int]) -> bool:
        """Check if a record exists by ID.

        Args:
            id: Record ID

        Returns:
            True if record exists
        """
        self._check_initialized()
        return self.count({"id": id}) > 0

    def batch_create(self, records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Create multiple records.

        Args:
            records: Records to create

        Returns:
            Created records
        """
        self._check_initialized()
        return self.db.batch_insert(self.name, records)

    def create_many(self, records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Alias for batch_create() - create multiple records.

        Args:
            records: Records to create

        Returns:
            Created records
        """
        return self.batch_create(records)

    def batch_update(
        self,
        records: List[Dict[str, Any]],
        key_column: str = "id"
    ) -> List[Dict[str, Any]]:
        """Update multiple records.

        Args:
            records: Records to update
            key_column: Primary key column

        Returns:
            Updated records
        """
        self._check_initialized()
        return self.db.batch_update(self.name, records, key_column)

    def update_many(
        self,
        records: List[Dict[str, Any]],
        key_column: str = "id"
    ) -> List[Dict[str, Any]]:
        """Alias for batch_update() - update multiple records.

        Args:
            records: Records to update
            key_column: Primary key column

        Returns:
            Updated records
        """
        return self.batch_update(records, key_column)

    def truncate(self) -> None:
        """Delete all records from the table."""
        self._check_initialized()
        self.db.execute(f'TRUNCATE TABLE "{self.name}"', fetch=False)

    def drop(self) -> None:
        """Drop the table."""
        self._check_initialized()
        self.db.drop_table(self.name)

    def execute(self, query: str, params: Optional[tuple] = None, fetch: bool = True) -> Optional[List[Dict[str, Any]]]:
        """Execute a custom SQL query on this table.

        Args:
            query: SQL query
            params: Query parameters
            fetch: Whether to fetch results

        Returns:
            Query results or None
        """
        self._check_initialized()
        return self.db.execute(query, params, fetch)

    def __enter__(self):
        """Enter context manager."""
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Exit context manager."""
        pass  # Database connection is managed by the Database class

Functions

__init__(name_or_db, db=None, schema=None, if_not_exists=True)

Initialize table.

Parameters:

Name Type Description Default
name_or_db Union[str, Database]

Table name or Database instance

 required
db Optional[Database]

Database instance (if first arg is table name)

 None
schema Optional[Dict[str, str]]

Column definitions {name: type}

 None
if_not_exists bool

Whether to create table only if it doesn't exist

 True
Source code in manticore_cockroachdb/crud/table.py 
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
def __init__(
    self,
    name_or_db: Union[str, Database],
    db: Optional[Database] = None,
    schema: Optional[Dict[str, str]] = None,
    if_not_exists: bool = True
):
    """Initialize table.

    Args:
        name_or_db: Table name or Database instance
        db: Database instance (if first arg is table name)
        schema: Column definitions {name: type}
        if_not_exists: Whether to create table only if it doesn't exist
    """
    # Handle different initialization patterns
    if isinstance(name_or_db, Database):
        self.db = name_or_db
        self.name = getattr(self, 'table_name', None)
    else:
        self.name = name_or_db
        self.db = db or Database()

    # Use schema from args or class attribute
    self._schema = schema or getattr(self, 'schema', None)
    self._initialized = False

    # Initialize table schema if provided
    if self._schema:
        self.initialize(if_not_exists=if_not_exists)
initialize(schema=None, if_not_exists=True)

Initialize table schema.

Parameters:

Name Type Description Default
schema Optional[Dict[str, str]]

Column definitions {name: type}

 None
if_not_exists bool

Whether to create table only if it doesn't exist

 True
Source code in manticore_cockroachdb/crud/table.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def initialize(self, schema: Optional[Dict[str, str]] = None, if_not_exists: bool = True) -> None:
    """Initialize table schema.

    Args:
        schema: Column definitions {name: type}
        if_not_exists: Whether to create table only if it doesn't exist
    """
    # Use provided schema or instance schema
    schema_to_use = schema or self._schema

    if not schema_to_use:
        raise ValueError("No schema provided for table initialization")

    self.db.create_table(self.name, schema_to_use, if_not_exists)
    self._initialized = True
_check_initialized()

Check if table is initialized.

Source code in manticore_cockroachdb/crud/table.py 
61
62
63
64
65
def _check_initialized(self):
    """Check if table is initialized."""
    from .exceptions import TableNotInitializedError
    if not self._initialized:
        raise TableNotInitializedError(self.name)
create(data)

Create a new record.

Parameters:

Name Type Description Default
data Dict[str, Any]

Record data

 required

Returns:

Type Description
Dict[str, Any]

Created record
Source code in manticore_cockroachdb/crud/table.py 
67
68
69
70
71
72
73
74
75
76
77
def create(self, data: Dict[str, Any]) -> Dict[str, Any]:
    """Create a new record.

    Args:
        data: Record data

    Returns:
        Created record
    """
    self._check_initialized()
    return self.db.insert(self.name, data)
read(id)

Read a record by ID.

Parameters:

Name Type Description Default
id Union[str, int]

Record ID

 required

Returns:

Type Description
Optional[Dict[str, Any]]

Record data

Raises:

Type Description
TableNotFoundError

If the record does not exist
Source code in manticore_cockroachdb/crud/table.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def read(self, id: Union[str, int]) -> Optional[Dict[str, Any]]:
    """Read a record by ID.

    Args:
        id: Record ID

    Returns:
        Record data

    Raises:
        TableNotFoundError: If the record does not exist
    """
    self._check_initialized()
    results = self.db.select(self.name, where={"id": id})
    if not results:
        from .exceptions import TableNotFoundError
        raise TableNotFoundError(f"Record with id '{id}' not found in table '{self.name}'")
    return results[0]
get(id)

Get a record by ID.

Parameters:

Name Type Description Default
id Union[str, int]

Record ID

 required

Returns:

Type Description
Optional[Dict[str, Any]]

Record data or None if not found
Source code in manticore_cockroachdb/crud/table.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
def get(self, id: Union[str, int]) -> Optional[Dict[str, Any]]:
    """Get a record by ID.

    Args:
        id: Record ID

    Returns:
        Record data or None if not found
    """
    self._check_initialized()
    results = self.db.select(self.name, where={"id": id})
    return results[0] if results else None
get_many(ids)

Get multiple records by their IDs.

Parameters:

Name Type Description Default
ids List[Union[str, int]]

List of record IDs

 required

Returns:

Type Description
List[Dict[str, Any]]

List of records matching the provided IDs
Source code in manticore_cockroachdb/crud/table.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def get_many(self, ids: List[Union[str, int]]) -> List[Dict[str, Any]]:
    """Get multiple records by their IDs.

    Args:
        ids: List of record IDs

    Returns:
        List of records matching the provided IDs
    """
    self._check_initialized()

    if not ids:
        return []

    # Build query with IN clause
    placeholders = ", ".join(["%s" for _ in ids])
    query = f'SELECT * FROM "{self.name}" WHERE "id" IN ({placeholders})'

    # Execute query
    return self.db.execute(query, tuple(ids)) or []
find_one(where)

Find a single record by criteria.

Parameters:

Name Type Description Default
where Dict[str, Any]

Filter conditions

 required

Returns:

Type Description
Optional[Dict[str, Any]]

Record data or None if not found
Source code in manticore_cockroachdb/crud/table.py 
132
133
134
135
136
137
138
139
140
141
142
143
def find_one(self, where: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    """Find a single record by criteria.

    Args:
        where: Filter conditions

    Returns:
        Record data or None if not found
    """
    self._check_initialized()
    results = self.db.select(self.name, where=where, limit=1)
    return results[0] if results else None
update(id, data)

Update a record by ID.

Parameters:

Name Type Description Default
id Union[str, int]

Record ID

 required
data Dict[str, Any]

Update data

 required

Returns:

Type Description
Dict[str, Any]

Updated record

Raises:

Type Description
TableNotFoundError

If the record does not exist
Source code in manticore_cockroachdb/crud/table.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def update(self, id: Union[str, int], data: Dict[str, Any]) -> Dict[str, Any]:
    """Update a record by ID.

    Args:
        id: Record ID
        data: Update data

    Returns:
        Updated record

    Raises:
        TableNotFoundError: If the record does not exist
    """
    self._check_initialized()
    result = self.db.update(self.name, data, {"id": id})
    if result is None:
        from .exceptions import TableNotFoundError
        raise TableNotFoundError(f"Record with id '{id}' not found in table '{self.name}'")
    return result
update_where(where, data)

Update records matching criteria.

Parameters:

Name Type Description Default
where Dict[str, Any]

Filter conditions

 required
data Dict[str, Any]

Update data

 required

Returns:

Type Description
Optional[Dict[str, Any]]

First updated record or None if none found
Source code in manticore_cockroachdb/crud/table.py 
165
166
167
168
169
170
171
172
173
174
175
176
def update_where(self, where: Dict[str, Any], data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    """Update records matching criteria.

    Args:
        where: Filter conditions
        data: Update data

    Returns:
        First updated record or None if none found
    """
    self._check_initialized()
    return self.db.update(self.name, data, where)
delete(id)

Delete a record by ID.

Parameters:

Name Type Description Default
id Union[str, int]

Record ID

 required

Returns:

Type Description
bool

True if record was deleted
Source code in manticore_cockroachdb/crud/table.py 
178
179
180
181
182
183
184
185
186
187
188
def delete(self, id: Union[str, int]) -> bool:
    """Delete a record by ID.

    Args:
        id: Record ID

    Returns:
        True if record was deleted
    """
    self._check_initialized()
    return self.db.delete(self.name, {"id": id})
delete_where(where)

Delete records matching criteria.

Parameters:

Name Type Description Default
where Dict[str, Any]

Filter conditions

 required

Returns:

Type Description
bool

True if any records were deleted
Source code in manticore_cockroachdb/crud/table.py 
190
191
192
193
194
195
196
197
198
199
200
def delete_where(self, where: Dict[str, Any]) -> bool:
    """Delete records matching criteria.

    Args:
        where: Filter conditions

    Returns:
        True if any records were deleted
    """
    self._check_initialized()
    return self.db.delete(self.name, where)
list(where=None, order_by=None, limit=None, offset=None)

List records.

Parameters:

Name Type Description Default
where Optional[Dict[str, Any]]

Filter conditions

 None
order_by Optional[str]

Order by clause

 None
limit Optional[int]

Result limit

 None
offset Optional[int]

Result offset

 None

Returns:

Type Description
List[Dict[str, Any]]

List of records
Source code in manticore_cockroachdb/crud/table.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def list(
    self,
    where: Optional[Dict[str, Any]] = None,
    order_by: Optional[str] = None,
    limit: Optional[int] = None,
    offset: Optional[int] = None
) -> List[Dict[str, Any]]:
    """List records.

    Args:
        where: Filter conditions
        order_by: Order by clause
        limit: Result limit
        offset: Result offset

    Returns:
        List of records
    """
    self._check_initialized()
    return self.db.select(self.name, where=where, order_by=order_by, limit=limit, offset=offset)
count(where=None)

Count records.

Parameters:

Name Type Description Default
where Optional[Dict[str, Any]]

Filter conditions

 None

Returns:

Type Description
int

Number of records
Source code in manticore_cockroachdb/crud/table.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
def count(self, where: Optional[Dict[str, Any]] = None) -> int:
    """Count records.

    Args:
        where: Filter conditions

    Returns:
        Number of records
    """
    self._check_initialized()
    query = f'SELECT COUNT(*) as count FROM "{self.name}"'
    params = []

    if where:
        conditions = []
        for col, val in where.items():
            conditions.append(f'"{col}" = %s')
            params.append(val)
        query += f" WHERE {' AND '.join(conditions)}"

    result = self.db.execute(query, tuple(params) if params else None)
    return result[0]["count"] if result else 0
exists(id)

Check if a record exists by ID.

Parameters:

Name Type Description Default
id Union[str, int]

Record ID

 required

Returns:

Type Description
bool

True if record exists
Source code in manticore_cockroachdb/crud/table.py 
246
247
248
249
250
251
252
253
254
255
256
def exists(self, id: Union[str, int]) -> bool:
    """Check if a record exists by ID.

    Args:
        id: Record ID

    Returns:
        True if record exists
    """
    self._check_initialized()
    return self.count({"id": id}) > 0
batch_create(records)

Create multiple records.

Parameters:

Name Type Description Default
records List[Dict[str, Any]]

Records to create

 required

Returns:

Type Description
List[Dict[str, Any]]

Created records
Source code in manticore_cockroachdb/crud/table.py 
258
259
260
261
262
263
264
265
266
267
268
def batch_create(self, records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Create multiple records.

    Args:
        records: Records to create

    Returns:
        Created records
    """
    self._check_initialized()
    return self.db.batch_insert(self.name, records)
create_many(records)

Alias for batch_create() - create multiple records.

Parameters:

Name Type Description Default
records List[Dict[str, Any]]

Records to create

 required

Returns:

Type Description
List[Dict[str, Any]]

Created records
Source code in manticore_cockroachdb/crud/table.py 
270
271
272
273
274
275
276
277
278
279
def create_many(self, records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Alias for batch_create() - create multiple records.

    Args:
        records: Records to create

    Returns:
        Created records
    """
    return self.batch_create(records)
batch_update(records, key_column='id')

Update multiple records.

Parameters:

Name Type Description Default
records List[Dict[str, Any]]

Records to update

 required
key_column str

Primary key column

 'id'

Returns:

Type Description
List[Dict[str, Any]]

Updated records
Source code in manticore_cockroachdb/crud/table.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
def batch_update(
    self,
    records: List[Dict[str, Any]],
    key_column: str = "id"
) -> List[Dict[str, Any]]:
    """Update multiple records.

    Args:
        records: Records to update
        key_column: Primary key column

    Returns:
        Updated records
    """
    self._check_initialized()
    return self.db.batch_update(self.name, records, key_column)
update_many(records, key_column='id')

Alias for batch_update() - update multiple records.

Parameters:

Name Type Description Default
records List[Dict[str, Any]]

Records to update

 required
key_column str

Primary key column

 'id'

Returns:

Type Description
List[Dict[str, Any]]

Updated records
Source code in manticore_cockroachdb/crud/table.py 
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
def update_many(
    self,
    records: List[Dict[str, Any]],
    key_column: str = "id"
) -> List[Dict[str, Any]]:
    """Alias for batch_update() - update multiple records.

    Args:
        records: Records to update
        key_column: Primary key column

    Returns:
        Updated records
    """
    return self.batch_update(records, key_column)
truncate()

Delete all records from the table.

Source code in manticore_cockroachdb/crud/table.py 
314
315
316
317
def truncate(self) -> None:
    """Delete all records from the table."""
    self._check_initialized()
    self.db.execute(f'TRUNCATE TABLE "{self.name}"', fetch=False)
drop()

Drop the table.

Source code in manticore_cockroachdb/crud/table.py 
319
320
321
322
def drop(self) -> None:
    """Drop the table."""
    self._check_initialized()
    self.db.drop_table(self.name)
execute(query, params=None, fetch=True)

Execute a custom SQL query on this table.

Parameters:

Name Type Description Default
query str

SQL query

 required
params Optional[tuple]

Query parameters

 None
fetch bool

Whether to fetch results

 True

Returns:

Type Description
Optional[List[Dict[str, Any]]]

Query results or None
Source code in manticore_cockroachdb/crud/table.py 
324
325
326
327
328
329
330
331
332
333
334
335
336
def execute(self, query: str, params: Optional[tuple] = None, fetch: bool = True) -> Optional[List[Dict[str, Any]]]:
    """Execute a custom SQL query on this table.

    Args:
        query: SQL query
        params: Query parameters
        fetch: Whether to fetch results

    Returns:
        Query results or None
    """
    self._check_initialized()
    return self.db.execute(query, params, fetch)
__enter__()

Enter context manager.

Source code in manticore_cockroachdb/crud/table.py 
338
339
340
def __enter__(self):
    """Enter context manager."""
    return self
__exit__(exc_type, exc_val, exc_tb)

Exit context manager.

Source code in manticore_cockroachdb/crud/table.py 
342
343
344
def __exit__(self, exc_type, exc_val, exc_tb):
    """Exit context manager."""
    pass  # Database connection is managed by the Database class

Usage Examples

from manticore_cockroachdb import Database, Table

# Connect to database
db = Database(database="example_db")

# Create a table
db.create_table(
    "users",
    {
        "id": "UUID PRIMARY KEY DEFAULT gen_random_uuid()",
        "name": "TEXT NOT NULL",
        "email": "TEXT UNIQUE NOT NULL",
        "age": "INTEGER",
        "active": "BOOLEAN DEFAULT TRUE"
    },
    if_not_exists=True
)

# Create a Table instance
users = Table("users", db=db)

# Create a user
user = users.create({
    "name": "John Doe",
    "email": "john@example.com",
    "age": 30
})
print(f"Created user with ID: {user['id']}")

# Read a user
retrieved_user = users.read(user["id"])
print(f"Retrieved user: {retrieved_user['name']}")

# Update a user
updated_user = users.update(user["id"], {"age": 31})
print(f"Updated user age: {updated_user['age']}")

# List all users
all_users = users.list()
print(f"Total users: {len(all_users)}")

# Filter users
active_users = users.list(where={"active": True})
print(f"Active users: {len(active_users)}")

# Count users
user_count = users.count()
print(f"User count: {user_count}")

# Delete a user
users.delete(user["id"])
print("User deleted")

# Batch operations
batch_users = [
    {"name": "User 1", "email": "user1@example.com", "age": 21},
    {"name": "User 2", "email": "user2@example.com", "age": 22},
    {"name": "User 3", "email": "user3@example.com", "age": 23}
]

# Create multiple users in a batch
created_batch = users.batch_create(batch_users)
print(f"Created {len(created_batch)} users in batch")

# Update multiple users in a batch
updates = [
    {"id": created_batch[0]["id"], "age": 31},
    {"id": created_batch[1]["id"], "age": 32},
    {"id": created_batch[2]["id"], "age": 33}
]
updated_batch = users.batch_update(updates)
print(f"Updated {len(updated_batch)} users in batch")

# Delete multiple users in a batch
ids_to_delete = [user["id"] for user in created_batch]
users.batch_delete(ids_to_delete)
print(f"Deleted {len(ids_to_delete)} users in batch")