Module: migrations

class oldmemo.migrations.OwnData[source]

Bases: TypedDict

This TypedDict describes how the own data was expected to be returned by the corresponding legacy storage method.

own_bare_jid: str

own_device_id: int

class oldmemo.migrations.Trust[source]

Bases: TypedDict

This TypedDict describes how trust information was expected to be returned by the corresponding legacy storage method.

key: str

trusted: bool

class oldmemo.migrations.Session[source]

Bases: TypedDict

This TypedDict describes how session instances (more precisely ExtendedDoubleRatchet instances) were serialized in the pre-stable serialization format.

super: Mapping[str, JSONType]

other_ik: str

class oldmemo.migrations.BoundOTPK[source]

Bases: TypedDict

Used as part of the legacy state format to represent a bound pre key.

otpk: str

id: int

class oldmemo.migrations.StateSuper[source]

Bases: TypedDict

Used as part of the legacy state format to represent the super class of the X3DHDoubleRatchet.

super: Mapping[str, JSONType]

spk_id: int

spk_pub: str | None

otpk_id_counter: int

otpk_ids: Dict[str, int]

class oldmemo.migrations.State[source]

Bases: TypedDict

This TypedDict describes how the state (more precisely X3DHDoubleRatchet) was serialized in the pre-stable serialization format. Note that the pk_messages entry has been omitted from this type since it is not needed for migration. The same applies to the version field, which apparently never had any relevance.

super: StateSuper

bound_otpks: Dict[str, Dict[int, BoundOTPK]]

class oldmemo.migrations.LegacyStorage[source]

Bases: ABC

This is a slightly modified copy of the storage interface used by legacy (i.e. pre-1.0.0) python-omemo. All methods related to storing values have been removed. A few methods for deleting values have been added instead. Methods related to efficient bulk loading have been removed as well.

abstractmethod async loadOwnData()[source]

Return type:: OwnData | None
Returns:: The own data stored in this instance, if any.

abstractmethod async deleteOwnData()[source]

Delete the own data stored in this instance, if any. Do not raise if there is none.

Return type:: None

abstractmethod async loadState()[source]

Return type:: State | None
Returns:: The state stored in this instance, if any.

abstractmethod async deleteState()[source]

Delete the state stored in this instance, if any. Do not raise if there is none.

Return type:: None

abstractmethod async loadSession(bare_jid, device_id)[source]

Parameters:

bare_jid (str) – The bare JID.
device_id (int) – The device id.

Return type:

Session | None

Returns:

The session stored in this instance for the given bare JID + device id, if any.

abstractmethod async deleteSession(bare_jid, device_id)[source]

Delete the session stored in this instance for the given bare JID + device id, if any. Do not raise if there is none.

Parameters:

bare_jid (str) – The bare JID.
device_id (int) – The device id.

Return type:

None

abstractmethod async loadActiveDevices(bare_jid)[source]

Parameters:: bare_jid (str) – The bare JID.
Return type:: List[int] | None
Returns:: The list of active devices stored in this instance for the given bare JID, if any.

Note

It doesn’t matter whether you return None or an empty list of no list is stored for this bare JID.

abstractmethod async loadInactiveDevices(bare_jid)[source]

Parameters:: bare_jid (str) – The bare JID.
Return type:: Dict[int, int] | None
Returns:: A mapping of inactive devices stored in this instance for the given bare JID, if any. The mapping maps from device id to the timestamp of last activity (seconds since epoch).

Note

It doesn’t matter whether you return None or an empty dictionary of no dictionary is stored for this bare JID.

abstractmethod async deleteActiveDevices(bare_jid)[source]

Delete the list of active devices stored in this instance for the given bare JID, if any. Do not raise if there is none.

Parameters:: bare_jid (str) – The bare JID.
Return type:: None

abstractmethod async deleteInactiveDevices(bare_jid)[source]

Delete the dictionary of inactive devices stored in this instance for the given bare JID, if any. Do not raise if there is none.

Parameters:: bare_jid (str) – The bare JID.
Return type:: None

abstractmethod async loadTrust(bare_jid, device_id)[source]

Parameters:

bare_jid (str) – The bare JID.
device_id (int) – The device id.

Return type:

Trust | None

Returns:

The trust information stored in this instance for the given bare JID + device id, if any.

abstractmethod async deleteTrust(bare_jid, device_id)[source]

Delete the trust information stored in this instance for the given bare JID + device id, if any. Do not raise if there is none.

Parameters:

bare_jid (str) – The bare JID.
device_id (int) – The device id.

Return type:

None

abstractmethod async listJIDs()[source]

Return type:: List[str] | None
Returns:: A list of all bare JIDs that have associated device lists stored in the storage, if any. For a bare JID to be included in the list, it doesn’t matter if the associated device lists are empty or not. Return None if the list of bare JIDs has been deleted, do not return an empty list in that case.

abstractmethod async deleteJIDList()[source]

Delete the list of bare JIDs as returned by listJIDs(), if it exists. Do not raise if it doesn’t.

Return type:: None

async oldmemo.migrations.migrate(legacy_storage, storage, trusted_trust_level_name, undecided_trust_level_name, untrusted_trust_level_name, download_bundle)[source]

Migrate the data from a legacy storage instance to the current storage format. This function is idempotent, which means that you can run it without checking whether migrations are required or not. If migrations are not required, the function will do nothing. This function also makes sure that only safely migrated data is deleted from the legacy storage, such that migration failure at any point leaves both storages in a consistent state.

Parameters:

legacy_storage (LegacyStorage) – The legacy storage instance to migrate the data from. This assumes that the storage was used with legacy (pre-1.0.0) python-omemo in conjunction with python-omemo-backend-signal. If the storage was used with a backend other than python-omemo-backend-signal, automated migration is not possible with this function.
storage (Storage) – The storage implementation to migrate the data to.
trusted_trust_level_name (str) – The legacy storage format stored trust as a boolean, i.e. there were only trusted or untrusted devices. This is the name of the custom trust level to store when a trusted device is migrated.
undecided_trust_level_name (str) – The name of the custom trust level to store when a device without any associated trust information is migrated.
untrusted_trust_level_name (str) – The name of the custom trust level to store when an untrusted device is migrated.
download_bundle (Callable[[str, int], Awaitable[BundleImpl]]) – A function which downloads the bundle of the given bare JID + device id. May raise BundleDownloadFailed (or a subclass thereof) to indicate failure.

Return type:

None