Skip to content

Repository

mongorunway.application.ports.repository ¤

__all__: typing.Sequence[str] = ('MigrationModelRepository') module-attribute ¤

MigrationModelRepository ¤

Bases: abc.ABC

Source code in mongorunway\application\ports\repository.py 
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
class MigrationModelRepository(abc.ABC):
    __slots__: typing.Sequence[str] = ()

    @abc.abstractmethod
    def __len__(self) -> int:
        ...

    @abc.abstractmethod
    def __contains__(self, item: typing.Any, /) -> bool:
        ...

    @abc.abstractmethod
    def has_migration(self, item: typing.Any, /) -> bool:
        ...

    @abc.abstractmethod
    def has_migration_with_version(self, migration_version: int, /) -> bool:
        ...

    @abc.abstractmethod
    def has_migrations(self) -> bool:
        ...

    @abc.abstractmethod
    def acquire_migration_model_by_version(
        self,
        migration_version: int,
    ) -> typing.Optional[domain_migration.MigrationReadModel]:
        ...

    @abc.abstractmethod
    def acquire_migration_model_by_flag(
        self, is_applied: bool
    ) -> typing.Optional[domain_migration.MigrationReadModel]:
        ...

    @abc.abstractmethod
    def acquire_all_migration_models(
        self,
        *,
        ascending_id: bool = True,
    ) -> typing.Iterator[domain_migration.MigrationReadModel]:
        ...

    @abc.abstractmethod
    def acquire_migration_models_by_flag(
        self, *, is_applied: bool
    ) -> typing.Iterator[domain_migration.MigrationReadModel]:
        ...

    @abc.abstractmethod
    def append_migration(self, migration: domain_migration.Migration, /) -> int:
        ...

    @abc.abstractmethod
    def remove_migration(self, migration_version: int, /) -> int:
        ...

    @abc.abstractmethod
    def set_applied_flag(self, migration: domain_migration.Migration, is_applied: bool) -> int:
        ...

__slots__: typing.Sequence[str] = () instance-attribute class-attribute ¤

__contains__(item) abstractmethod ¤

Source code in mongorunway\application\ports\repository.py 
39
40
41
@abc.abstractmethod
def __contains__(self, item: typing.Any, /) -> bool:
    ...

__len__() abstractmethod ¤

Source code in mongorunway\application\ports\repository.py 
35
36
37
@abc.abstractmethod
def __len__(self) -> int:
    ...

acquire_all_migration_models(*, ascending_id=True) abstractmethod ¤

Source code in mongorunway\application\ports\repository.py 
68
69
70
71
72
73
74
@abc.abstractmethod
def acquire_all_migration_models(
    self,
    *,
    ascending_id: bool = True,
) -> typing.Iterator[domain_migration.MigrationReadModel]:
    ...

acquire_migration_model_by_flag(is_applied) abstractmethod ¤

Source code in mongorunway\application\ports\repository.py 
62
63
64
65
66
@abc.abstractmethod
def acquire_migration_model_by_flag(
    self, is_applied: bool
) -> typing.Optional[domain_migration.MigrationReadModel]:
    ...

acquire_migration_model_by_version(migration_version) abstractmethod ¤

Source code in mongorunway\application\ports\repository.py 
55
56
57
58
59
60
@abc.abstractmethod
def acquire_migration_model_by_version(
    self,
    migration_version: int,
) -> typing.Optional[domain_migration.MigrationReadModel]:
    ...

acquire_migration_models_by_flag(*, is_applied) abstractmethod ¤

Source code in mongorunway\application\ports\repository.py 
76
77
78
79
80
@abc.abstractmethod
def acquire_migration_models_by_flag(
    self, *, is_applied: bool
) -> typing.Iterator[domain_migration.MigrationReadModel]:
    ...

append_migration(migration) abstractmethod ¤

Source code in mongorunway\application\ports\repository.py 
82
83
84
@abc.abstractmethod
def append_migration(self, migration: domain_migration.Migration, /) -> int:
    ...

has_migration(item) abstractmethod ¤

Source code in mongorunway\application\ports\repository.py 
43
44
45
@abc.abstractmethod
def has_migration(self, item: typing.Any, /) -> bool:
    ...

has_migration_with_version(migration_version) abstractmethod ¤

Source code in mongorunway\application\ports\repository.py 
47
48
49
@abc.abstractmethod
def has_migration_with_version(self, migration_version: int, /) -> bool:
    ...

has_migrations() abstractmethod ¤

Source code in mongorunway\application\ports\repository.py 
51
52
53
@abc.abstractmethod
def has_migrations(self) -> bool:
    ...

remove_migration(migration_version) abstractmethod ¤

Source code in mongorunway\application\ports\repository.py 
86
87
88
@abc.abstractmethod
def remove_migration(self, migration_version: int, /) -> int:
    ...

set_applied_flag(migration, is_applied) abstractmethod ¤

Source code in mongorunway\application\ports\repository.py 
90
91
92
@abc.abstractmethod
def set_applied_flag(self, migration: domain_migration.Migration, is_applied: bool) -> int:
    ...