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 
 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
@dataclass_transform(kw_only_default=True, field_specifiers=(Integer, Text, Integer.fk))
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)}  # pyright:ignore[reportUnannotatedClassAttribute]

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

        super().__init_subclass__(**kwargs)

    def __init__(self, **kwargs: Any) -> None:
        """
        Minimal replication of `dataclass` behaviour.
        """
        columns: dict[str, type[Column[Any]]] = {  # pyright:ignore[reportAssignmentType]
            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].default is not None:  # pyright:ignore[reportGeneralTypeIssues]
                setattr(self, name, columns[name].default)  # pyright:ignore[reportGeneralTypeIssues]
                missing.remove(name)

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

    @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 = ",".join(columns)
        return f"""CREATE TABLE IF NOT EXISTS {cls.fqn()} ({columns_str});"""

    @classmethod
    def all(cls) -> type[SelectAll]:
        """
        Generate a Select query model that returns all the table's fields.

        ```python
        from embar.model import SelectAll
        from embar.table import Table
        class MyTable(Table): ...
        model = MyTable.all()
        assert model == SelectAll
        ```
        """
        return SelectAll

    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 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def __init__(self, **kwargs: Any) -> None:
    """
    Minimal replication of `dataclass` behaviour.
    """
    columns: dict[str, type[Column[Any]]] = {  # pyright:ignore[reportAssignmentType]
        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].default is not None:  # pyright:ignore[reportGeneralTypeIssues]
            setattr(self, name, columns[name].default)  # pyright:ignore[reportGeneralTypeIssues]
            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 
31
32
33
34
35
36
37
38
39
40
41
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)}  # pyright:ignore[reportUnannotatedClassAttribute]

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

    super().__init_subclass__(**kwargs)

all() classmethod

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

from embar.model import SelectAll
from embar.table import Table
class MyTable(Table): ...
model = MyTable.all()
assert model == SelectAll
Source code in src/embar/table.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
@classmethod
def all(cls) -> type[SelectAll]:
    """
    Generate a Select query model that returns all the table's fields.

    ```python
    from embar.model import SelectAll
    from embar.table import Table
    class MyTable(Table): ...
    model = MyTable.all()
    assert model == SelectAll
    ```
    """
    return SelectAll

ddl() classmethod

Generate a full DDL for the table.

Source code in src/embar/table.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
@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 = ",".join(columns)
    return f"""CREATE TABLE IF NOT EXISTS {cls.fqn()} ({columns_str});"""

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 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@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 
90
91
92
93
94
95
@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 
126
127
128
129
130
131
132
133
134
135
136
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