Skip to content

constraint

Table constraints like indexes and unique constraints.

Index

Creates a database index on one or more columns.

from embar.column.common import Integer
from embar.config import EmbarConfig
from embar.constraint import Index
from embar.table import Table
class MyTable(Table):
    embar_config: EmbarConfig = EmbarConfig(
        constraints=[Index("my_idx").on(lambda: MyTable.id)]
    )
    id: Integer = Integer()
Source code in src/embar/constraint.py 
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
class Index:
    """
    Creates a database index on one or more columns.

    ```python
    from embar.column.common import Integer
    from embar.config import EmbarConfig
    from embar.constraint import Index
    from embar.table import Table
    class MyTable(Table):
        embar_config: EmbarConfig = EmbarConfig(
            constraints=[Index("my_idx").on(lambda: MyTable.id)]
        )
        id: Integer = Integer()
    ```
    """

    name: str

    def __init__(self, name: str):
        """
        Create a new Index instance.
        """
        self.name = name

    def on(self, *columns: Callable[[], ColumnBase]) -> IndexReady:
        """
        Specify the columns this index should be created on.
        """
        return IndexReady(self.name, False, *columns)

__init__(name)

Create a new Index instance.

Source code in src/embar/constraint.py 
32
33
34
35
36
def __init__(self, name: str):
    """
    Create a new Index instance.
    """
    self.name = name

on(*columns)

Specify the columns this index should be created on.

Source code in src/embar/constraint.py 
38
39
40
41
42
def on(self, *columns: Callable[[], ColumnBase]) -> IndexReady:
    """
    Specify the columns this index should be created on.
    """
    return IndexReady(self.name, False, *columns)

IndexReady

Bases: Constraint

A fully configured index ready to generate SQL.

Source code in src/embar/constraint.py 
 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
class IndexReady(Constraint):
    """
    A fully configured index ready to generate SQL.
    """

    unique: bool
    name: str
    columns: tuple[Callable[[], ColumnBase], ...]
    _where_clause: Callable[[], WhereClause] | None = None

    def __init__(self, name: str, unique: bool, *columns: Callable[[], ColumnBase]):
        """
        Create a new IndexReady instance.
        """
        self.name = name
        self.unique = unique
        self.columns = columns

    def where(self, where_clause: Callable[[], WhereClause]) -> Self:
        """
        Add a WHERE clause to create a partial index.
        """
        self._where_clause = where_clause
        return self

    @override
    def sql(self) -> QuerySingle:
        """
        Generate the CREATE INDEX SQL statement.
        """
        # Not so sure about this, seems a bit brittle to just get the name as a string?
        table_names = [c().info.table_name for c in self.columns]
        if len(set(table_names)) > 1:
            raise ValueError(f"Index {self.name}: all columns must be in the same table")
        table_name = table_names[0]

        cols = ", ".join(f'"{c().info.name}"' for c in self.columns)
        unique = " UNIQUE " if self.unique else ""
        params: dict[str, PyType] = {}

        where_sql = ""
        if self._where_clause:
            count = -1

            def get_count() -> int:
                nonlocal count
                count += 1
                return count

            where = self._where_clause().sql(get_count)
            where_sql = f" WHERE {where.sql}"
            params = {**params, **where.params}

        query = f'CREATE {unique} INDEX "{self.name}" ON "{table_name}"({cols}){where_sql};'

        return QuerySingle(query, params)

__init__(name, unique, *columns)

Create a new IndexReady instance.

Source code in src/embar/constraint.py 
75
76
77
78
79
80
81
def __init__(self, name: str, unique: bool, *columns: Callable[[], ColumnBase]):
    """
    Create a new IndexReady instance.
    """
    self.name = name
    self.unique = unique
    self.columns = columns

sql()

Generate the CREATE INDEX SQL statement.

Source code in src/embar/constraint.py 
 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
@override
def sql(self) -> QuerySingle:
    """
    Generate the CREATE INDEX SQL statement.
    """
    # Not so sure about this, seems a bit brittle to just get the name as a string?
    table_names = [c().info.table_name for c in self.columns]
    if len(set(table_names)) > 1:
        raise ValueError(f"Index {self.name}: all columns must be in the same table")
    table_name = table_names[0]

    cols = ", ".join(f'"{c().info.name}"' for c in self.columns)
    unique = " UNIQUE " if self.unique else ""
    params: dict[str, PyType] = {}

    where_sql = ""
    if self._where_clause:
        count = -1

        def get_count() -> int:
            nonlocal count
            count += 1
            return count

        where = self._where_clause().sql(get_count)
        where_sql = f" WHERE {where.sql}"
        params = {**params, **where.params}

    query = f'CREATE {unique} INDEX "{self.name}" ON "{table_name}"({cols}){where_sql};'

    return QuerySingle(query, params)

where(where_clause)

Add a WHERE clause to create a partial index.

Source code in src/embar/constraint.py 
83
84
85
86
87
88
def where(self, where_clause: Callable[[], WhereClause]) -> Self:
    """
    Add a WHERE clause to create a partial index.
    """
    self._where_clause = where_clause
    return self

UniqueIndex

Creates a unique database index on one or more columns.

Source code in src/embar/constraint.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class UniqueIndex:
    """
    Creates a unique database index on one or more columns.
    """

    name: str

    def __init__(self, name: str):
        """
        Create a new UniqueIndex instance.
        """
        self.name = name

    def on(self, *columns: Callable[[], ColumnBase]) -> IndexReady:
        """
        Specify the columns this unique index should be created on.
        """
        return IndexReady(self.name, True, *columns)

__init__(name)

Create a new UniqueIndex instance.

Source code in src/embar/constraint.py 
52
53
54
55
56
def __init__(self, name: str):
    """
    Create a new UniqueIndex instance.
    """
    self.name = name

on(*columns)

Specify the columns this unique index should be created on.

Source code in src/embar/constraint.py 
58
59
60
61
62
def on(self, *columns: Callable[[], ColumnBase]) -> IndexReady:
    """
    Specify the columns this unique index should be created on.
    """
    return IndexReady(self.name, True, *columns)