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
class Table:
    """Database table with CRUD operations."""

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

        Args:
            name: Table name
            db: Database instance
            schema: Column definitions {name: type}
            if_not_exists: Whether to create table only if it doesn't exist
        """
        self.name = name
        self.db = db or Database()

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

    def initialize(self, schema: 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
        """
        if schema:
            self.db.create_table(self.name, schema, if_not_exists)

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

        Args:
            data: Record data

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

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

        Args:
            id: Record ID

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

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

        Args:
            id: Record ID
            data: Update data

        Returns:
            Updated record
        """
        return self.db.update(self.name, data, {"id": id})

    def delete(self, id: str) -> bool:
        """Delete a record.

        Args:
            id: Record ID

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

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

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

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

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

        Args:
            where: Filter conditions

        Returns:
            Number of records
        """
        results = self.db.execute(
            f"SELECT COUNT(*) as count FROM {self.name}" + 
            (f" WHERE {' AND '.join(f'{k} = %s' for k in where.keys())}" if where else ""),
            tuple(where.values()) if where else None
        )
        return results[0]["count"]

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

        Args:
            records: Records to create

        Returns:
            Created records
        """
        return self.db.batch_insert(self.name, 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: Column to use as key

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

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

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Exit context."""
        # Don't close the database here as it might be shared
        pass

Functions

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

Initialize table.

Parameters:

Name Type Description Default
name str

Table name

 required
db Optional[Database]

Database instance

 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 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def __init__(
    self,
    name: str,
    db: Optional[Database] = None,
    schema: Optional[Dict[str, str]] = None,
    if_not_exists: bool = True
):
    """Initialize table.

    Args:
        name: Table name
        db: Database instance
        schema: Column definitions {name: type}
        if_not_exists: Whether to create table only if it doesn't exist
    """
    self.name = name
    self.db = db or Database()

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

Initialize table schema.

Parameters:

Name Type Description Default
schema 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 
33
34
35
36
37
38
39
40
41
def initialize(self, schema: 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
    """
    if schema:
        self.db.create_table(self.name, schema, if_not_exists)
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 
43
44
45
46
47
48
49
50
51
52
def create(self, data: Dict[str, Any]) -> Dict[str, Any]:
    """Create a new record.

    Args:
        data: Record data

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

Read a record.

Parameters:

Name Type Description Default
id str

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 
54
55
56
57
58
59
60
61
62
63
64
def read(self, id: str) -> Optional[Dict[str, Any]]:
    """Read a record.

    Args:
        id: Record ID

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

Update a record.

Parameters:

Name Type Description Default
id str

Record ID

 required
data Dict[str, Any]

Update data

 required

Returns:

Type Description
Optional[Dict[str, Any]]

Updated record
Source code in manticore_cockroachdb/crud/table.py 
66
67
68
69
70
71
72
73
74
75
76
def update(self, id: str, data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
    """Update a record.

    Args:
        id: Record ID
        data: Update data

    Returns:
        Updated record
    """
    return self.db.update(self.name, data, {"id": id})
delete(id)

Delete a record.

Parameters:

Name Type Description Default
id str

Record ID

 required

Returns:

Type Description
bool

True if record was deleted
Source code in manticore_cockroachdb/crud/table.py 
78
79
80
81
82
83
84
85
86
87
def delete(self, id: str) -> bool:
    """Delete a record.

    Args:
        id: Record ID

    Returns:
        True if record was deleted
    """
    return self.db.delete(self.name, {"id": id})
list(where=None, order_by=None, limit=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

Returns:

Type Description
List[Dict[str, Any]]

List of records
Source code in manticore_cockroachdb/crud/table.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def list(
    self,
    where: Optional[Dict[str, Any]] = None,
    order_by: Optional[str] = None,
    limit: Optional[int] = None
) -> List[Dict[str, Any]]:
    """List records.

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

    Returns:
        List of records
    """
    return self.db.select(
        self.name,
        where=where,
        order_by=order_by,
        limit=limit
    )
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 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def count(self, where: Optional[Dict[str, Any]] = None) -> int:
    """Count records.

    Args:
        where: Filter conditions

    Returns:
        Number of records
    """
    results = self.db.execute(
        f"SELECT COUNT(*) as count FROM {self.name}" + 
        (f" WHERE {' AND '.join(f'{k} = %s' for k in where.keys())}" if where else ""),
        tuple(where.values()) if where else None
    )
    return results[0]["count"]
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 
128
129
130
131
132
133
134
135
136
137
def batch_create(self, records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Create multiple records.

    Args:
        records: Records to create

    Returns:
        Created records
    """
    return self.db.batch_insert(self.name, 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

Column to use as key

 'id'

Returns:

Type Description
List[Dict[str, Any]]

Updated records
Source code in manticore_cockroachdb/crud/table.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
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: Column to use as key

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

Enter context.

Source code in manticore_cockroachdb/crud/table.py 
155
156
157
def __enter__(self):
    """Enter context."""
    return self
__exit__(exc_type, exc_val, exc_tb)

Exit context.

Source code in manticore_cockroachdb/crud/table.py 
159
160
161
162
def __exit__(self, exc_type, exc_val, exc_tb):
    """Exit context."""
    # Don't close the database here as it might be shared
    pass

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")