Skip to content

migration

Migration classes for creating and running database migrations.

Ddl

Represents a DDL statement with optional constraints.

Source code in src/embar/migration.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class Ddl:
    """
    Represents a DDL statement with optional constraints.
    """

    name: str
    ddl: str
    constraints: list[str]

    def __init__(self, name: str, ddl: str, constraints: list[str] | None = None):
        """
        Create a new Ddl instance.
        """
        self.name = name
        self.ddl = ddl
        self.constraints = constraints if constraints is not None else []

__init__(name, ddl, constraints=None)

Create a new Ddl instance.

Source code in src/embar/migration.py 
21
22
23
24
25
26
27
def __init__(self, name: str, ddl: str, constraints: list[str] | None = None):
    """
    Create a new Ddl instance.
    """
    self.name = name
    self.ddl = ddl
    self.constraints = constraints if constraints is not None else []

Migration

Represents a migration that can be run against a database.

Source code in src/embar/migration.py 
 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
class Migration[Db: AllDbBase]:
    """
    Represents a migration that can be run against a database.
    """

    ddls: list[Ddl]
    _db: Db

    def __init__(self, ddls: list[Ddl], db: Db):
        """
        Create a new Migration instance.
        """
        self.ddls = ddls
        self._db = db

    @property
    def merged(self) -> str:
        """
        Get all DDL statements merged into a single string.
        """
        query = ""
        for table in self.ddls:
            query += "\n\n" + table.ddl
            for constraint in table.constraints:
                query += "\n" + constraint

        return query

    def __await__(self) -> Generator[Any, None, None]:
        """
        Run the migration asynchronously.
        """

        async def awaitable():
            db = self._db
            if isinstance(db, AsyncDbBase):
                for ddl in self.ddls:
                    await db.execute(QuerySingle(ddl.ddl))
                    for constraint in ddl.constraints:
                        await db.execute(QuerySingle(constraint))

            else:
                db = cast(DbBase, self._db)
                for ddl in self.ddls:
                    db.execute(QuerySingle(ddl.ddl))
                    for constraint in ddl.constraints:
                        db.execute(QuerySingle(constraint))

        return awaitable().__await__()

    @overload
    def run(self: Migration[DbBase]) -> None: ...
    @overload
    def run(self: Migration[AsyncDbBase]) -> Migration[Db]: ...
    def run(self) -> None | Migration[Db]:
        """
        Run the migration synchronously.
        """
        if isinstance(self._db, DbBase):
            for ddl in self.ddls:
                self._db.execute(QuerySingle(ddl.ddl))
                for constraint in ddl.constraints:
                    self._db.execute(QuerySingle(constraint))
            return
        return self

merged property

Get all DDL statements merged into a single string.

__await__()

Run the migration asynchronously.

Source code in src/embar/migration.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def __await__(self) -> Generator[Any, None, None]:
    """
    Run the migration asynchronously.
    """

    async def awaitable():
        db = self._db
        if isinstance(db, AsyncDbBase):
            for ddl in self.ddls:
                await db.execute(QuerySingle(ddl.ddl))
                for constraint in ddl.constraints:
                    await db.execute(QuerySingle(constraint))

        else:
            db = cast(DbBase, self._db)
            for ddl in self.ddls:
                db.execute(QuerySingle(ddl.ddl))
                for constraint in ddl.constraints:
                    db.execute(QuerySingle(constraint))

    return awaitable().__await__()

__init__(ddls, db)

Create a new Migration instance.

Source code in src/embar/migration.py 
54
55
56
57
58
59
def __init__(self, ddls: list[Ddl], db: Db):
    """
    Create a new Migration instance.
    """
    self.ddls = ddls
    self._db = db

run() 

run() -> None

run() -> Migration[Db]

Run the migration synchronously.

Source code in src/embar/migration.py 
100
101
102
103
104
105
106
107
108
109
110
def run(self) -> None | Migration[Db]:
    """
    Run the migration synchronously.
    """
    if isinstance(self._db, DbBase):
        for ddl in self.ddls:
            self._db.execute(QuerySingle(ddl.ddl))
            for constraint in ddl.constraints:
                self._db.execute(QuerySingle(constraint))
        return
    return self

MigrationDefs

Holds table and enum definitions for migrations.

Source code in src/embar/migration.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
class MigrationDefs:
    """
    Holds table and enum definitions for migrations.
    """

    tables: list[type[Table]]
    enums: list[type[EnumBase]]

    def __init__(self, tables: Sequence[type[Table]], enums: Sequence[type[EnumBase]] | None = None):
        """
        Create a new MigrationDefs instance.
        """
        self.tables = list(tables)
        self.enums = list(enums) if enums is not None else []

__init__(tables, enums=None)

Create a new MigrationDefs instance.

Source code in src/embar/migration.py 
38
39
40
41
42
43
def __init__(self, tables: Sequence[type[Table]], enums: Sequence[type[EnumBase]] | None = None):
    """
    Create a new MigrationDefs instance.
    """
    self.tables = list(tables)
    self.enums = list(enums) if enums is not None else []