Skip to content

Migration auditlog entry

mongorunway.domain.migration_auditlog_entry ¤

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

MigrationAuditlogEntry dataclass ¤

Source code in mongorunway\domain\migration_auditlog_entry.py 
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
@dataclasses.dataclass
class MigrationAuditlogEntry:
    session_id: bson.Binary
    transaction_name: str
    migration_read_model: domain_migration.MigrationReadModel
    date_fmt: str
    date: datetime.datetime = dataclasses.field(default_factory=datetime.datetime.utcnow)
    exc_name: typing.Optional[str] = None
    exc_message: typing.Optional[str] = None

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

    def is_failed(self) -> bool:
        return self.exc_name is not None or self.exc_message is not None

    def with_error(self: _SelfT, exc: BaseException, /) -> _SelfT:
        self.exc_name = type(exc).__name__
        self.exc_message = str(exc)

        return self

    def with_timezone(self: _SelfT, timezone: str) -> _SelfT:
        if timezone != "UTC":
            # Default time is utc
            try:
                self.date = self.date.astimezone(zoneinfo.ZoneInfo(timezone))
            except zoneinfo.ZoneInfoNotFoundError as exc:
                raise ModuleNotFoundError(
                    "'tzdata' module must be installed to use timezones in auditlog journals."
                ) from exc

        return self

    def format_date(self) -> str:
        return self.date.strftime(self.date_fmt)

date: datetime.datetime = dataclasses.field(default_factory=datetime.datetime.utcnow) instance-attribute class-attribute ¤

date_fmt: str instance-attribute ¤

exc_message: typing.Optional[str] = None instance-attribute class-attribute ¤

exc_name: typing.Optional[str] = None instance-attribute class-attribute ¤

migration_read_model: domain_migration.MigrationReadModel instance-attribute ¤

session_id: bson.Binary instance-attribute ¤

transaction_name: str instance-attribute ¤

format_date() ¤

Source code in mongorunway\domain\migration_auditlog_entry.py 
78
79
def format_date(self) -> str:
    return self.date.strftime(self.date_fmt)

from_dict(mapping) classmethod ¤

Source code in mongorunway\domain\migration_auditlog_entry.py 
50
51
52
53
54
55
@classmethod
def from_dict(
    cls, mapping: typing.MutableMapping[str, typing.Any], /
) -> MigrationAuditlogEntry:
    mapping.pop("_id", None)  # For mongo records
    return cls(**mapping)

is_failed() ¤

Source code in mongorunway\domain\migration_auditlog_entry.py 
57
58
def is_failed(self) -> bool:
    return self.exc_name is not None or self.exc_message is not None

with_error(exc) ¤

Source code in mongorunway\domain\migration_auditlog_entry.py 
60
61
62
63
64
def with_error(self: _SelfT, exc: BaseException, /) -> _SelfT:
    self.exc_name = type(exc).__name__
    self.exc_message = str(exc)

    return self

with_timezone(timezone) ¤

Source code in mongorunway\domain\migration_auditlog_entry.py 
66
67
68
69
70
71
72
73
74
75
76
def with_timezone(self: _SelfT, timezone: str) -> _SelfT:
    if timezone != "UTC":
        # Default time is utc
        try:
            self.date = self.date.astimezone(zoneinfo.ZoneInfo(timezone))
        except zoneinfo.ZoneInfoNotFoundError as exc:
            raise ModuleNotFoundError(
                "'tzdata' module must be installed to use timezones in auditlog journals."
            ) from exc

    return self