Skip to content

table

Ideally table.py would be in a table/ module but then it's impossible to import table_base.py without triggering table.py, causing a circular loop by the Many stuff (that's the reason the two were separated in the first place).

Table

Bases: TableBase

All table definitions inherit from Table.

Table is used extensively as both a class/type and as objects. - Tables/schemas are created as class MyTable(Table): ... - Table references (in where clauses, joins, FKs) refer to these types - New rows to insert into a table are created as objects

Source code in src/embar/table.py 
 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
@dataclass_transform(
    kw_only_default=True,
    field_specifiers=(
        integer,
        text,
        float_col,
        varchar,
        serial,
        boolean,
        timestamp,
        jsonb,
        smallint,
        bigint,
        smallserial,
        bigserial,
        char_col,
        numeric,
        pg_decimal,
        double_precision,
        json_col,
        time_col,
        date_col,
        interval,
        enum_col,
        vector,
    ),
)
class Table(TableBase):
    """
    All table definitions inherit from `Table`.

    Table is used extensively as both a class/type and as objects.
    - Tables/schemas are created as `class MyTable(Table): ...`
    - Table references (in where clauses, joins, FKs) refer to these types
    - New rows to insert into a table are created as objects
    """

    def __init_subclass__(cls, **kwargs: Any):
        """
        Populate `_fields` and the `embar_config` if not provided.
        """
        cls._fields = {name: attr for name, attr in cls.__dict__.items() if isinstance(attr, ColumnBase)}

        if cls.embar_config == Undefined:
            cls.embar_config: EmbarConfig = EmbarConfig()
            cls.embar_config.__set_name__(cls, "embar_config")

        cls._validate_column_annotations()
        super().__init_subclass__(**kwargs)

    @classmethod
    def _validate_column_annotations(cls) -> None:
        """
        Validate that column annotations match their field specifiers.

        Checks both the base type (e.g. Text vs Integer) and nullability
        (e.g. NullText vs Text).
        """
        annotations = cls.__annotations__
        for name, col in cls._fields.items():
            if name not in annotations:
                continue
            ann = annotations[name]
            origin = get_origin(ann)

            if origin is Null:
                # Annotation is Null[T] — column must be a Null instance
                if not isinstance(col, Null):
                    raise TypeError(
                        f"{cls.__name__}.{name}: annotation is nullable ({ann}) "
                        f"but field specifier returns non-nullable {type(col).__name__}"
                    )
                # Check that T matches the column's py_type
                ann_args = get_args(ann)
                if ann_args and col._py_type != ann_args[0]:
                    raise TypeError(
                        f"{cls.__name__}.{name}: annotation is Null[{ann_args[0].__name__}] "
                        f"but field specifier has py_type={col._py_type.__name__}"
                    )
                # Nullable annotation but not_null=True in DB — always wrong
                if col._not_null:
                    raise TypeError(
                        f"{cls.__name__}.{name}: annotation is nullable ({ann}) but field specifier has not_null=True"
                    )
            elif isinstance(ann, type) and issubclass(ann, Column):
                # Annotation is a concrete Column subclass (e.g. Text, Integer)
                if isinstance(col, Null):
                    raise TypeError(
                        f"{cls.__name__}.{name}: annotation is non-nullable ({ann.__name__}) "
                        f"but field specifier returns nullable Null"
                    )
                if not isinstance(col, ann):
                    raise TypeError(
                        f"{cls.__name__}.{name}: annotation type is {ann.__name__} "
                        f"but field specifier returns {type(col).__name__}"
                    )

    def __init__(self, **kwargs: Any) -> None:
        """
        Minimal replication of `dataclass` behaviour.
        """
        columns: dict[str, type[Column[Any]]] = {
            name: attr for name, attr in type(self).__dict__.items() if isinstance(attr, ColumnBase)
        }

        for name, value in kwargs.items():
            if name not in columns:
                raise TypeError(f"Unknown field: {name}")
            setattr(self, name, value)

        # Handle defaults for missing fields
        missing = set(columns.keys()) - set(kwargs.keys())
        for name in list(missing):
            if columns[name].has_default:
                setattr(self, name, columns[name].default)
                missing.remove(name)

        if missing:
            raise TypeError(f"Missing required fields: {missing}")

    if _PYDANTIC_AVAILABLE:

        @classmethod
        def __get_pydantic_core_schema__(
            cls,
            source_type: Any,
            handler: Any,
        ) -> "_core_schema.CoreSchema":
            return core_schema.any_schema()

    @classmethod
    def many(cls) -> ManyTable[type[Self]]:
        """
        Used to nest many of another table in a column in a model

        ```python
        from typing import Annotated
        from pydantic import BaseModel
        from embar.table import Table
        class MyTable(Table): ...
        class MyModel(BaseModel):
            messages: Annotated[list[MyTable], MyTable.many()]
        ```
        """
        return ManyTable[type[Self]](cls)

    @classmethod
    def one(cls) -> OneTable[type[Self]]:
        """
        Used to nest one of another table in a column in a model
        """
        return OneTable[type[Self]](cls)

    @classmethod
    def ddl(cls) -> str:
        """
        Generate a full DDL for the table.
        """
        columns: list[str] = []
        for attr_name, attr in cls.__dict__.items():
            if attr_name.startswith("_"):
                continue
            if isinstance(attr, ColumnBase):
                columns.append(attr.info.ddl())
        columns_str = ",\n".join(columns)
        columns_str = indent(columns_str, "    ")

        sql = f"""
CREATE TABLE IF NOT EXISTS {cls.fqn()} (
{columns_str}
);"""

        sql = dedent(sql).strip()

        return sql

    @overload
    @classmethod
    def all(cls) -> type[SelectAllPydantic]: ...
    @overload
    @classmethod
    def all(cls, use_pydantic: Literal[True]) -> type[SelectAllPydantic]: ...
    @overload
    @classmethod
    def all(cls, use_pydantic: Literal[False]) -> type[SelectAllDataclass]: ...

    @classmethod
    def all(cls, use_pydantic: bool = True) -> type[SelectAllPydantic] | type[SelectAllDataclass]:
        """
        Generate a Select query model that returns all the table's fields.

        ```python
        from embar.model import SelectAllPydantic
        from embar.table import Table
        class MyTable(Table): ...
        model = MyTable.all()
        assert model == SelectAllPydantic
        ```
        """
        if use_pydantic:
            if not _PYDANTIC_AVAILABLE:
                raise ImportError(
                    "Table.all() requires pydantic when use_pydantic=True (the default). "
                    "Either install it with: pip install 'embar[pydantic]' "
                    "or opt in to the plain-dataclass path with: MyTable.all(use_pydantic=False)"
                )
            return SelectAllPydantic
        return SelectAllDataclass

    def value_dict(self) -> dict[str, Any]:
        """
        Result is keyed to DB column names, _not_ field names.
        """
        result: dict[str, Any] = {}
        for attr_name, attr in self.__class__.__dict__.items():
            if attr_name.startswith("_"):
                continue
            if isinstance(attr, ColumnBase):
                result[attr.info.name] = getattr(self, attr_name)
        return result

__init__(**kwargs)

Minimal replication of dataclass behaviour.

Source code in src/embar/table.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def __init__(self, **kwargs: Any) -> None:
    """
    Minimal replication of `dataclass` behaviour.
    """
    columns: dict[str, type[Column[Any]]] = {
        name: attr for name, attr in type(self).__dict__.items() if isinstance(attr, ColumnBase)
    }

    for name, value in kwargs.items():
        if name not in columns:
            raise TypeError(f"Unknown field: {name}")
        setattr(self, name, value)

    # Handle defaults for missing fields
    missing = set(columns.keys()) - set(kwargs.keys())
    for name in list(missing):
        if columns[name].has_default:
            setattr(self, name, columns[name].default)
            missing.remove(name)

    if missing:
        raise TypeError(f"Missing required fields: {missing}")

__init_subclass__(**kwargs)

Populate _fields and the embar_config if not provided.

Source code in src/embar/table.py 
87
88
89
90
91
92
93
94
95
96
97
98
def __init_subclass__(cls, **kwargs: Any):
    """
    Populate `_fields` and the `embar_config` if not provided.
    """
    cls._fields = {name: attr for name, attr in cls.__dict__.items() if isinstance(attr, ColumnBase)}

    if cls.embar_config == Undefined:
        cls.embar_config: EmbarConfig = EmbarConfig()
        cls.embar_config.__set_name__(cls, "embar_config")

    cls._validate_column_annotations()
    super().__init_subclass__(**kwargs)

all(use_pydantic=True) classmethod 

all() -> type[SelectAllPydantic]

all(use_pydantic: Literal[True]) -> type[SelectAllPydantic]

all(
    use_pydantic: Literal[False],
) -> type[SelectAllDataclass]

Generate a Select query model that returns all the table's fields.

from embar.model import SelectAllPydantic
from embar.table import Table
class MyTable(Table): ...
model = MyTable.all()
assert model == SelectAllPydantic
Source code in src/embar/table.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
@classmethod
def all(cls, use_pydantic: bool = True) -> type[SelectAllPydantic] | type[SelectAllDataclass]:
    """
    Generate a Select query model that returns all the table's fields.

    ```python
    from embar.model import SelectAllPydantic
    from embar.table import Table
    class MyTable(Table): ...
    model = MyTable.all()
    assert model == SelectAllPydantic
    ```
    """
    if use_pydantic:
        if not _PYDANTIC_AVAILABLE:
            raise ImportError(
                "Table.all() requires pydantic when use_pydantic=True (the default). "
                "Either install it with: pip install 'embar[pydantic]' "
                "or opt in to the plain-dataclass path with: MyTable.all(use_pydantic=False)"
            )
        return SelectAllPydantic
    return SelectAllDataclass

ddl() classmethod

Generate a full DDL for the table.

Source code in src/embar/table.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
    @classmethod
    def ddl(cls) -> str:
        """
        Generate a full DDL for the table.
        """
        columns: list[str] = []
        for attr_name, attr in cls.__dict__.items():
            if attr_name.startswith("_"):
                continue
            if isinstance(attr, ColumnBase):
                columns.append(attr.info.ddl())
        columns_str = ",\n".join(columns)
        columns_str = indent(columns_str, "    ")

        sql = f"""
CREATE TABLE IF NOT EXISTS {cls.fqn()} (
{columns_str}
);"""

        sql = dedent(sql).strip()

        return sql

many() classmethod

Used to nest many of another table in a column in a model

from typing import Annotated
from pydantic import BaseModel
from embar.table import Table
class MyTable(Table): ...
class MyModel(BaseModel):
    messages: Annotated[list[MyTable], MyTable.many()]
Source code in src/embar/table.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
@classmethod
def many(cls) -> ManyTable[type[Self]]:
    """
    Used to nest many of another table in a column in a model

    ```python
    from typing import Annotated
    from pydantic import BaseModel
    from embar.table import Table
    class MyTable(Table): ...
    class MyModel(BaseModel):
        messages: Annotated[list[MyTable], MyTable.many()]
    ```
    """
    return ManyTable[type[Self]](cls)

one() classmethod

Used to nest one of another table in a column in a model

Source code in src/embar/table.py 
196
197
198
199
200
201
@classmethod
def one(cls) -> OneTable[type[Self]]:
    """
    Used to nest one of another table in a column in a model
    """
    return OneTable[type[Self]](cls)

value_dict()

Result is keyed to DB column names, not field names.

Source code in src/embar/table.py 
259
260
261
262
263
264
265
266
267
268
269
def value_dict(self) -> dict[str, Any]:
    """
    Result is keyed to DB column names, _not_ field names.
    """
    result: dict[str, Any] = {}
    for attr_name, attr in self.__class__.__dict__.items():
        if attr_name.startswith("_"):
            continue
        if isinstance(attr, ColumnBase):
            result[attr.info.name] = getattr(self, attr_name)
    return result