Skip to content

Migration

mongorunway.domain.migration ¤

__all__: typing.Sequence[str] = ('Migration', 'MigrationProcess', 'MigrationReadModel') module-attribute ¤

Migration ¤

Source code in mongorunway\domain\migration.py 
 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
class Migration:
    __slots__: typing.Sequence[str] = (
        "_name",
        "_version",
        "_checksum",
        "_is_applied",
        "_description",
        "_upgrade_process",
        "_downgrade_process",
    )

    def __init__(
        self,
        *,
        name: str,
        version: int,
        checksum: str,
        is_applied: bool,
        description: str,
        upgrade_process: MigrationProcess,
        downgrade_process: MigrationProcess,
    ) -> None:
        self._name = name
        self._version = version
        self._checksum = checksum
        self._is_applied = is_applied
        self._description = description
        self._upgrade_process = upgrade_process
        self._downgrade_process = downgrade_process

    @property
    def name(self) -> str:
        return self._name

    @property
    def version(self) -> int:
        return self._version

    @property
    def checksum(self) -> str:
        return self._checksum

    @property
    def description(self) -> str:
        return self._description

    @property
    def is_applied(self) -> bool:
        return self._is_applied

    @property
    def upgrade_process(self) -> MigrationProcess:
        return self._upgrade_process

    @property
    def downgrade_process(self) -> MigrationProcess:
        return self._downgrade_process

    def set_is_applied(self, value: bool, /) -> None:
        self._is_applied = value

    def to_dict(self, *, unique: bool = False) -> typing.Dict[str, typing.Any]:
        mapping = {
            "name": self.name,
            "version": self.version,
            "checksum": self.checksum,
            "is_applied": self.is_applied,
            "description": self.description,
        }

        if unique:
            mapping["_id"] = self.version

        return mapping

__slots__: typing.Sequence[str] = ('_name', '_version', '_checksum', '_is_applied', '_description', '_upgrade_process', '_downgrade_process') instance-attribute class-attribute ¤

checksum: str property ¤

description: str property ¤

downgrade_process: MigrationProcess property ¤

is_applied: bool property ¤

name: str property ¤

upgrade_process: MigrationProcess property ¤

version: int property ¤

__init__(*, name, version, checksum, is_applied, description, upgrade_process, downgrade_process) ¤

Source code in mongorunway\domain\migration.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def __init__(
    self,
    *,
    name: str,
    version: int,
    checksum: str,
    is_applied: bool,
    description: str,
    upgrade_process: MigrationProcess,
    downgrade_process: MigrationProcess,
) -> None:
    self._name = name
    self._version = version
    self._checksum = checksum
    self._is_applied = is_applied
    self._description = description
    self._upgrade_process = upgrade_process
    self._downgrade_process = downgrade_process

set_is_applied(value) ¤

Source code in mongorunway\domain\migration.py 
97
98
def set_is_applied(self, value: bool, /) -> None:
    self._is_applied = value

to_dict(*, unique=False) ¤

Source code in mongorunway\domain\migration.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
def to_dict(self, *, unique: bool = False) -> typing.Dict[str, typing.Any]:
    mapping = {
        "name": self.name,
        "version": self.version,
        "checksum": self.checksum,
        "is_applied": self.is_applied,
        "description": self.description,
    }

    if unique:
        mapping["_id"] = self.version

    return mapping

MigrationProcess ¤

Source code in mongorunway\domain\migration.py 
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
class MigrationProcess:
    def __init__(
        self,
        commands: domain_command.AnyCommandSequence,
        migration_version: int,
        name: str,
    ) -> None:
        self._rules: domain_rule.RuleSequence = []
        self._name = name
        self._commands = commands
        self._migration_version = migration_version

    @property
    def name(self) -> str:
        return self._name

    @property
    def commands(self) -> domain_command.AnyCommandSequence:
        return self._commands

    @property
    def migration_version(self) -> int:
        return self._migration_version

    @property
    def rules(self) -> domain_rule.RuleSequence:
        return self._rules

    def has_rules(self) -> bool:
        return bool(self._rules)

    def add_rule(self: _ProcessT, rule: domain_rule.MigrationBusinessRule, /) -> _ProcessT:
        self._rules.append(rule)
        return self

commands: domain_command.AnyCommandSequence property ¤

migration_version: int property ¤

name: str property ¤

rules: domain_rule.RuleSequence property ¤

__init__(commands, migration_version, name) ¤

Source code in mongorunway\domain\migration.py 
140
141
142
143
144
145
146
147
148
149
def __init__(
    self,
    commands: domain_command.AnyCommandSequence,
    migration_version: int,
    name: str,
) -> None:
    self._rules: domain_rule.RuleSequence = []
    self._name = name
    self._commands = commands
    self._migration_version = migration_version

add_rule(rule) ¤

Source code in mongorunway\domain\migration.py 
170
171
172
def add_rule(self: _ProcessT, rule: domain_rule.MigrationBusinessRule, /) -> _ProcessT:
    self._rules.append(rule)
    return self

has_rules() ¤

Source code in mongorunway\domain\migration.py 
167
168
def has_rules(self) -> bool:
    return bool(self._rules)

MigrationReadModel dataclass ¤

Source code in mongorunway\domain\migration.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
@dataclasses.dataclass
class MigrationReadModel:
    name: str
    version: int
    checksum: str
    description: str
    is_applied: bool

    @classmethod
    def from_dict(cls, mapping: typing.MutableMapping[str, typing.Any], /) -> MigrationReadModel:
        mapping.pop("_id", None)  # For mongo records
        return cls(**mapping)

    @classmethod
    def from_migration(cls, migration: Migration, /) -> MigrationReadModel:
        return cls(
            name=migration.name,
            version=migration.version,
            checksum=migration.checksum,
            description=migration.description,
            is_applied=migration.is_applied,
        )

checksum: str instance-attribute ¤

description: str instance-attribute ¤

is_applied: bool instance-attribute ¤

name: str instance-attribute ¤

version: int instance-attribute ¤

from_dict(mapping) classmethod ¤

Source code in mongorunway\domain\migration.py 
123
124
125
126
@classmethod
def from_dict(cls, mapping: typing.MutableMapping[str, typing.Any], /) -> MigrationReadModel:
    mapping.pop("_id", None)  # For mongo records
    return cls(**mapping)

from_migration(migration) classmethod ¤

Source code in mongorunway\domain\migration.py 
128
129
130
131
132
133
134
135
136
@classmethod
def from_migration(cls, migration: Migration, /) -> MigrationReadModel:
    return cls(
        name=migration.name,
        version=migration.version,
        checksum=migration.checksum,
        description=migration.description,
        is_applied=migration.is_applied,
    )