Skip to content

EncryptionKeyStorageStrategy

Interface for key management strategies used in event encryption.

Implementations are responsible for storing, retrieving, and deleting encryption keys for specific data subjects. Supports multi-tenancy via scoped_for_tenant. Used by the Encryption service for key management and crypto-shredding.

Source code in event_sourcery/_event_store/event/encryption.py 
 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
class EncryptionKeyStorageStrategy:
    """
    Interface for key management strategies used in event encryption.

    Implementations are responsible for storing, retrieving, and deleting encryption
    keys for specific data subjects. Supports multi-tenancy via scoped_for_tenant.
    Used by the Encryption service for key management and crypto-shredding.
    """

    def get(self, subject_id: str) -> bytes | None:
        """
        Retrieves the encryption key for the given subject identifier.

        Args:
            subject_id (str): The subject identifier.

        Returns:
            bytes | None: The encryption key, or None if not found.
        """
        raise NotImplementedError()

    def store(self, subject_id: str, key: bytes) -> None:
        """
        Stores the encryption key for the given subject identifier.

        Args:
            subject_id (str): The subject identifier.
            key (bytes): The encryption key to store.
        """
        raise NotImplementedError()

    def delete(self, subject_id: str) -> None:
        """
        Deletes the encryption key for the given subject identifier.

        Args:
            subject_id (str): The subject identifier whose key should be deleted.
        """
        raise NotImplementedError()

    def scoped_for_tenant(self, tenant_id: TenantId) -> Self:
        """
        Returns a key storage strategy instance scoped for the given tenant.

        Args:
            tenant_id (TenantId): The tenant identifier.

        Returns:
            Self: The key storage strategy instance for the tenant.
        """
        raise NotImplementedError()

delete(subject_id)

Deletes the encryption key for the given subject identifier.

Parameters:

Name Type Description Default
subject_id str

The subject identifier whose key should be deleted.

 required
Source code in event_sourcery/_event_store/event/encryption.py 
82
83
84
85
86
87
88
89
def delete(self, subject_id: str) -> None:
    """
    Deletes the encryption key for the given subject identifier.

    Args:
        subject_id (str): The subject identifier whose key should be deleted.
    """
    raise NotImplementedError()

get(subject_id)

Retrieves the encryption key for the given subject identifier.

Parameters:

Name Type Description Default
subject_id str

The subject identifier.

 required

Returns:

Type Description
bytes | None

bytes | None: The encryption key, or None if not found.
Source code in event_sourcery/_event_store/event/encryption.py 
60
61
62
63
64
65
66
67
68
69
70
def get(self, subject_id: str) -> bytes | None:
    """
    Retrieves the encryption key for the given subject identifier.

    Args:
        subject_id (str): The subject identifier.

    Returns:
        bytes | None: The encryption key, or None if not found.
    """
    raise NotImplementedError()

scoped_for_tenant(tenant_id)

Returns a key storage strategy instance scoped for the given tenant.

Parameters:

Name Type Description Default
tenant_id TenantId

The tenant identifier.

 required

Returns:

Name Type Description
Self Self

The key storage strategy instance for the tenant.
Source code in event_sourcery/_event_store/event/encryption.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def scoped_for_tenant(self, tenant_id: TenantId) -> Self:
    """
    Returns a key storage strategy instance scoped for the given tenant.

    Args:
        tenant_id (TenantId): The tenant identifier.

    Returns:
        Self: The key storage strategy instance for the tenant.
    """
    raise NotImplementedError()

store(subject_id, key)

Stores the encryption key for the given subject identifier.

Parameters:

Name Type Description Default
subject_id str

The subject identifier.

 required
key bytes

The encryption key to store.

 required
Source code in event_sourcery/_event_store/event/encryption.py 
72
73
74
75
76
77
78
79
80
def store(self, subject_id: str, key: bytes) -> None:
    """
    Stores the encryption key for the given subject identifier.

    Args:
        subject_id (str): The subject identifier.
        key (bytes): The encryption key to store.
    """
    raise NotImplementedError()