Skip to content

Backends

backends.db.abstractdb

backends.db.abstractdb

AbstractDatabase

Bases: ABC

Abstract base class for database operations.

Source code in backends/db/abstractdb.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class AbstractDatabase(ABC):
    """Abstract base class for database operations."""

    @abstractmethod
    def connect(self):
        """Establish a connection to the database."""
        pass

    @abstractmethod
    def execute_query(self, query: str, params: tuple = ()):
        """Execute a query against the database."""
        pass

    @abstractmethod
    def fetch_one(self):
        """Fetch one record from the executed query."""
        pass

    @abstractmethod
    def fetch_all(self):
        """Fetch all records from the executed query."""
        pass

    @abstractmethod
    def close(self):
        """Close the database connection."""
        pass

    def __enter__(self):
        """Enter the runtime context for this database connection."""
        self.connect()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Exit the runtime context and close the database connection."""
        self.close()
__enter__()

Enter the runtime context for this database connection.

Source code in backends/db/abstractdb.py 
32
33
34
35
def __enter__(self):
    """Enter the runtime context for this database connection."""
    self.connect()
    return self
__exit__(exc_type, exc_val, exc_tb)

Exit the runtime context and close the database connection.

Source code in backends/db/abstractdb.py 
37
38
39
def __exit__(self, exc_type, exc_val, exc_tb):
    """Exit the runtime context and close the database connection."""
    self.close()
close() abstractmethod

Close the database connection.

Source code in backends/db/abstractdb.py 
27
28
29
30
@abstractmethod
def close(self):
    """Close the database connection."""
    pass
connect() abstractmethod

Establish a connection to the database.

Source code in backends/db/abstractdb.py 
 7
 8
 9
10
@abstractmethod
def connect(self):
    """Establish a connection to the database."""
    pass
execute_query(query, params=()) abstractmethod

Execute a query against the database.

Source code in backends/db/abstractdb.py 
12
13
14
15
@abstractmethod
def execute_query(self, query: str, params: tuple = ()):
    """Execute a query against the database."""
    pass
fetch_all() abstractmethod

Fetch all records from the executed query.

Source code in backends/db/abstractdb.py 
22
23
24
25
@abstractmethod
def fetch_all(self):
    """Fetch all records from the executed query."""
    pass
fetch_one() abstractmethod

Fetch one record from the executed query.

Source code in backends/db/abstractdb.py 
17
18
19
20
@abstractmethod
def fetch_one(self):
    """Fetch one record from the executed query."""
    pass

backends.db.azsqldb

backends.db.azsqldb

AzureSQLDatabase

Bases: AbstractDatabase

Azure SQL Database implementation of the DatabaseInterface.

Source code in backends/db/azsqldb.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class AzureSQLDatabase(AbstractDatabase):
    """Azure SQL Database implementation of the DatabaseInterface."""

    def __init__(self, connection_string: str):
        self.connection_string = connection_string
        self.connection = None
        self.cursor = None

    def connect(self):
        self.connection = pyodbc.connect(self.connection_string)
        self.cursor = self.connection.cursor()

    def execute_query(self, query: str, params: tuple = ()):
        if not self.connection:
            raise ConnectionError("Database not connected.")
        self.cursor.execute(query, params)

    def fetch_one(self):
        return self.cursor.fetchone()

    def fetch_all(self):
        return self.cursor.fetchall()

    def close(self):
        if self.cursor:
            self.cursor.close()
        if self.connection:
            self.connection.close()

backends.db.sqlitedb

backends.db.sqlitedb

SQLiteDatabase

Bases: AbstractDatabase

SQLite implementation of the DatabaseInterface.

Source code in backends/db/sqlitedb.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class SQLiteDatabase(AbstractDatabase):
    """SQLite implementation of the DatabaseInterface."""

    def __init__(self, db_path: str):
        self.db_path = db_path
        self.connection = None
        self.cursor = None

    def connect(self):
        self.connection = sqlite3.connect(self.db_path)
        self.cursor = self.connection.cursor()

    def execute_query(self, query: str, params: tuple = ()):
        if not self.connection:
            raise ConnectionError("Database not connected.")
        self.cursor.execute(query, params)

    def fetch_one(self):
        return self.cursor.fetchone()

    def fetch_all(self):
        return self.cursor.fetchall()

    def close(self):
        if self.cursor:
            self.cursor.close()
        if self.connection:
            self.connection.close()

backends.filesystem.abstractfs

backends.filesystem.abstractfs

AbstractFileSystem

Bases: ABC

Abstract base class for file system operations.

Source code in backends/filesystem/abstractfs.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
class AbstractFileSystem(ABC):
    """Abstract base class for file system operations."""

    @abstractmethod
    def read(self, relative_path: str):
        """Read data from the file."""
        pass

    @abstractmethod
    def write(self, relative_path: str, data: bytes):
        """Write data to the file."""
        pass

    @abstractmethod
    def list(self, relative_dir: str):
        """List files in the specified directory."""
        pass

    @abstractmethod
    def delete(self, relative_path: str):
        """Delete the file."""
        pass
delete(relative_path) abstractmethod

Delete the file.

Source code in backends/filesystem/abstractfs.py 
22
23
24
25
@abstractmethod
def delete(self, relative_path: str):
    """Delete the file."""
    pass
list(relative_dir) abstractmethod

List files in the specified directory.

Source code in backends/filesystem/abstractfs.py 
17
18
19
20
@abstractmethod
def list(self, relative_dir: str):
    """List files in the specified directory."""
    pass
read(relative_path) abstractmethod

Read data from the file.

Source code in backends/filesystem/abstractfs.py 
 7
 8
 9
10
@abstractmethod
def read(self, relative_path: str):
    """Read data from the file."""
    pass
write(relative_path, data) abstractmethod

Write data to the file.

Source code in backends/filesystem/abstractfs.py 
12
13
14
15
@abstractmethod
def write(self, relative_path: str, data: bytes):
    """Write data to the file."""
    pass

backends.filesystem.azureblobfs

backends.filesystem.azureblobfs

AzureBlobFileSystem

Bases: AbstractFileSystem

Azure Blob Storage implementation of the FileSystemInterface.

Source code in backends/filesystem/azureblobfs.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class AzureBlobFileSystem(AbstractFileSystem):
    """Azure Blob Storage implementation of the FileSystemInterface."""

    def __init__(self, connection_string: str, container_name: str):
        self.connection_string = connection_string
        self.container_name = container_name
        self.client = BlobServiceClient.from_connection_string(connection_string)
        self.container = self.client.get_container_client(container_name)

    def read(self, relative_path: str) -> bytes:
        """Read data from an Azure Blob."""
        blob_client = self.container.get_blob_client(relative_path)
        if not blob_client.exists():
            raise FileNotFoundError(f"The file '{relative_path}' does not exist in the Azure container.")
        return blob_client.download_blob().readall()

    def write(self, relative_path: str, data: bytes):
        """Write data to an Azure Blob."""
        blob_client = self.container.get_blob_client(relative_path)
        blob_client.upload_blob(data, overwrite=True)

    def list(self, relative_dir: str) -> list[str]:
        """List files in the specified Azure Blob directory."""
        blobs = self.container.list_blobs(name_starts_with=relative_dir)
        return [blob.name for blob in blobs]

    def delete(self, relative_path: str):
        """Delete a file from Azure Blob Storage."""
        blob_client = self.container.get_blob_client(relative_path)
        if not blob_client.exists():
            raise FileNotFoundError(f"The file '{relative_path}' does not exist in the Azure container.")
        blob_client.delete_blob()
delete(relative_path)

Delete a file from Azure Blob Storage.

Source code in backends/filesystem/azureblobfs.py 
31
32
33
34
35
36
def delete(self, relative_path: str):
    """Delete a file from Azure Blob Storage."""
    blob_client = self.container.get_blob_client(relative_path)
    if not blob_client.exists():
        raise FileNotFoundError(f"The file '{relative_path}' does not exist in the Azure container.")
    blob_client.delete_blob()
list(relative_dir)

List files in the specified Azure Blob directory.

Source code in backends/filesystem/azureblobfs.py 
26
27
28
29
def list(self, relative_dir: str) -> list[str]:
    """List files in the specified Azure Blob directory."""
    blobs = self.container.list_blobs(name_starts_with=relative_dir)
    return [blob.name for blob in blobs]
read(relative_path)

Read data from an Azure Blob.

Source code in backends/filesystem/azureblobfs.py 
14
15
16
17
18
19
def read(self, relative_path: str) -> bytes:
    """Read data from an Azure Blob."""
    blob_client = self.container.get_blob_client(relative_path)
    if not blob_client.exists():
        raise FileNotFoundError(f"The file '{relative_path}' does not exist in the Azure container.")
    return blob_client.download_blob().readall()
write(relative_path, data)

Write data to an Azure Blob.

Source code in backends/filesystem/azureblobfs.py 
21
22
23
24
def write(self, relative_path: str, data: bytes):
    """Write data to an Azure Blob."""
    blob_client = self.container.get_blob_client(relative_path)
    blob_client.upload_blob(data, overwrite=True)

backends.filesystem.ftpfs

backends.filesystem.ftpfs

FTPFileSystem

Bases: AbstractFileSystem

FTP implementation of the FileSystemInterface.

Source code in backends/filesystem/ftpfs.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class FTPFileSystem(AbstractFileSystem):
    """FTP implementation of the FileSystemInterface."""

    def __init__(self, host: str, username: str, password: str, port: int = 21):
        self.ftp = FTP()
        self.ftp.connect(host, port)
        self.ftp.login(username, password)

    def read(self, relative_path: str) -> bytes:
        """Read data from an FTP file."""
        with open('tempfile', 'wb') as temp_file:
            self.ftp.retrbinary(f"RETR {relative_path}", temp_file.write)
        with open('tempfile', 'rb') as temp_file:
            return temp_file.read()

    def write(self, relative_path: str, data: bytes):
        """Write data to an FTP file."""
        with open('tempfile', 'wb') as temp_file:
            temp_file.write(data)
        with open('tempfile', 'rb') as temp_file:
            self.ftp.storbinary(f"STOR {relative_path}", temp_file)

    def list(self, relative_dir: str) -> list[str]:
        """List files in the specified FTP directory."""
        return self.ftp.nlst(relative_dir)

    def delete(self, relative_path: str):
        """Delete a file from FTP."""
        self.ftp.delete(relative_path)
delete(relative_path)

Delete a file from FTP.

Source code in backends/filesystem/ftpfs.py 
33
34
35
def delete(self, relative_path: str):
    """Delete a file from FTP."""
    self.ftp.delete(relative_path)
list(relative_dir)

List files in the specified FTP directory.

Source code in backends/filesystem/ftpfs.py 
29
30
31
def list(self, relative_dir: str) -> list[str]:
    """List files in the specified FTP directory."""
    return self.ftp.nlst(relative_dir)
read(relative_path)

Read data from an FTP file.

Source code in backends/filesystem/ftpfs.py 
15
16
17
18
19
20
def read(self, relative_path: str) -> bytes:
    """Read data from an FTP file."""
    with open('tempfile', 'wb') as temp_file:
        self.ftp.retrbinary(f"RETR {relative_path}", temp_file.write)
    with open('tempfile', 'rb') as temp_file:
        return temp_file.read()
write(relative_path, data)

Write data to an FTP file.

Source code in backends/filesystem/ftpfs.py 
22
23
24
25
26
27
def write(self, relative_path: str, data: bytes):
    """Write data to an FTP file."""
    with open('tempfile', 'wb') as temp_file:
        temp_file.write(data)
    with open('tempfile', 'rb') as temp_file:
        self.ftp.storbinary(f"STOR {relative_path}", temp_file)

SFTPFileSystem

Bases: AbstractFileSystem

SFTP implementation of the FileSystemInterface.

Source code in backends/filesystem/ftpfs.py 
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
class SFTPFileSystem(AbstractFileSystem):
    """SFTP implementation of the FileSystemInterface."""

    def __init__(self, host: str, username: str, password: str, port: int = 22):
        self.cnopts = pysftp.CnOpts()
        self.cnopts.hostkeys = None  # Disable host key verification (optional, for testing only)
        self.connection = pysftp.Connection(
            host=host, username=username, password=password, port=port, cnopts=self.cnopts
        )

    def read(self, relative_path: str) -> bytes:
        """Read data from an SFTP file."""
        with self.connection.open(relative_path, 'rb') as file:
            return file.read()

    def write(self, relative_path: str, data: bytes):
        """Write data to an SFTP file."""
        with self.connection.open(relative_path, 'wb') as file:
            file.write(data)

    def list(self, relative_dir: str) -> list[str]:
        """List files in the specified SFTP directory."""
        return self.connection.listdir(relative_dir)

    def delete(self, relative_path: str):
        """Delete a file from SFTP."""
        self.connection.remove(relative_path)
delete(relative_path)

Delete a file from SFTP.

Source code in backends/filesystem/ftpfs.py 
62
63
64
def delete(self, relative_path: str):
    """Delete a file from SFTP."""
    self.connection.remove(relative_path)
list(relative_dir)

List files in the specified SFTP directory.

Source code in backends/filesystem/ftpfs.py 
58
59
60
def list(self, relative_dir: str) -> list[str]:
    """List files in the specified SFTP directory."""
    return self.connection.listdir(relative_dir)
read(relative_path)

Read data from an SFTP file.

Source code in backends/filesystem/ftpfs.py 
48
49
50
51
def read(self, relative_path: str) -> bytes:
    """Read data from an SFTP file."""
    with self.connection.open(relative_path, 'rb') as file:
        return file.read()
write(relative_path, data)

Write data to an SFTP file.

Source code in backends/filesystem/ftpfs.py 
53
54
55
56
def write(self, relative_path: str, data: bytes):
    """Write data to an SFTP file."""
    with self.connection.open(relative_path, 'wb') as file:
        file.write(data)

backends.filesystem.gcsfs

backends.filesystem.gcsfs

GCSFileSystem

Bases: AbstractFileSystem

Google Cloud Storage implementation of the FileSystemInterface.

Source code in backends/filesystem/gcsfs.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class GCSFileSystem(AbstractFileSystem):
    """Google Cloud Storage implementation of the FileSystemInterface."""

    def __init__(self, bucket_name: str):
        self.bucket_name = bucket_name
        self.client = storage.Client()
        self.bucket = self.client.bucket(bucket_name)

    def read(self, relative_path: str) -> bytes:
        """Read data from a GCS file."""
        blob = self.bucket.blob(relative_path)
        if not blob.exists():
            raise FileNotFoundError(f"The file '{relative_path}' does not exist in the GCS bucket.")
        return blob.download_as_bytes()

    def write(self, relative_path: str, data: bytes):
        """Write data to a GCS file."""
        blob = self.bucket.blob(relative_path)
        blob.upload_from_string(data)

    def list(self, relative_dir: str) -> list[str]:
        """List files in the specified GCS directory."""
        blobs = self.client.list_blobs(self.bucket_name, prefix=relative_dir)
        return [blob.name for blob in blobs]

    def delete(self, relative_path: str):
        """Delete a file from GCS."""
        blob = self.bucket.blob(relative_path)
        if not blob.exists():
            raise FileNotFoundError(f"The file '{relative_path}' does not exist in the GCS bucket.")
        blob.delete()
delete(relative_path)

Delete a file from GCS.

Source code in backends/filesystem/gcsfs.py 
30
31
32
33
34
35
def delete(self, relative_path: str):
    """Delete a file from GCS."""
    blob = self.bucket.blob(relative_path)
    if not blob.exists():
        raise FileNotFoundError(f"The file '{relative_path}' does not exist in the GCS bucket.")
    blob.delete()
list(relative_dir)

List files in the specified GCS directory.

Source code in backends/filesystem/gcsfs.py 
25
26
27
28
def list(self, relative_dir: str) -> list[str]:
    """List files in the specified GCS directory."""
    blobs = self.client.list_blobs(self.bucket_name, prefix=relative_dir)
    return [blob.name for blob in blobs]
read(relative_path)

Read data from a GCS file.

Source code in backends/filesystem/gcsfs.py 
13
14
15
16
17
18
def read(self, relative_path: str) -> bytes:
    """Read data from a GCS file."""
    blob = self.bucket.blob(relative_path)
    if not blob.exists():
        raise FileNotFoundError(f"The file '{relative_path}' does not exist in the GCS bucket.")
    return blob.download_as_bytes()
write(relative_path, data)

Write data to a GCS file.

Source code in backends/filesystem/gcsfs.py 
20
21
22
23
def write(self, relative_path: str, data: bytes):
    """Write data to a GCS file."""
    blob = self.bucket.blob(relative_path)
    blob.upload_from_string(data)

backends.filesystem.localfs

backends.filesystem.localfs

LocalFileSystem

Bases: AbstractFileSystem

Local file system implementation of the FileSystemInterface.

Source code in backends/filesystem/localfs.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class LocalFileSystem(AbstractFileSystem):
    """Local file system implementation of the FileSystemInterface."""

    def __init__(self, root: str = '.'):
        self.root = Path(root)
        if not self.root.exists():
            self.root.mkdir(parents=True)

    def read(self, relative_path: str):
        """Read data from the local file."""
        path = self.root / relative_path
        if not path.exists():
            raise FileNotFoundError(f"The file '{relative_path}' does not exist.")
        if not path.is_file():
            raise TypeError(f"The file '{relative_path}' is not a file.")

        with open(self.root / relative_path, 'rb') as file:
            return file.read()

    def write(self, relative_path: str, data: bytes):
        """Write data to the local file."""
        with open(self.root / relative_path, 'wb') as file:
            file.write(data)

    def list(self, relative_dir: str):
        """List files in the specified directory."""
        return os.listdir(self.root / relative_dir)

    def delete(self, relative_path: str):
        """Delete the file."""
        (self.root / relative_path).unlink(missing_ok=True)
delete(relative_path)

Delete the file.

Source code in backends/filesystem/localfs.py 
35
36
37
def delete(self, relative_path: str):
    """Delete the file."""
    (self.root / relative_path).unlink(missing_ok=True)
list(relative_dir)

List files in the specified directory.

Source code in backends/filesystem/localfs.py 
31
32
33
def list(self, relative_dir: str):
    """List files in the specified directory."""
    return os.listdir(self.root / relative_dir)
read(relative_path)

Read data from the local file.

Source code in backends/filesystem/localfs.py 
15
16
17
18
19
20
21
22
23
24
def read(self, relative_path: str):
    """Read data from the local file."""
    path = self.root / relative_path
    if not path.exists():
        raise FileNotFoundError(f"The file '{relative_path}' does not exist.")
    if not path.is_file():
        raise TypeError(f"The file '{relative_path}' is not a file.")

    with open(self.root / relative_path, 'rb') as file:
        return file.read()
write(relative_path, data)

Write data to the local file.

Source code in backends/filesystem/localfs.py 
26
27
28
29
def write(self, relative_path: str, data: bytes):
    """Write data to the local file."""
    with open(self.root / relative_path, 'wb') as file:
        file.write(data)

backends.filesystem.s3fs

backends.filesystem.s3fs

S3FileSystem

Bases: AbstractFileSystem

S3 file system implementation of the FileSystemInterface.

Source code in backends/filesystem/s3fs.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
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
class S3FileSystem(AbstractFileSystem):
    """S3 file system implementation of the FileSystemInterface."""

    def __init__(self, bucket_name: str, region: str = "us-east-1"):
        self.bucket_name = bucket_name
        self.region = region
        self.s3 = boto3.client("s3", region_name=region)

    def read(self, relative_path: str) -> bytes:
        """Read data from the S3 file."""
        try:
            response = self.s3.get_object(Bucket=self.bucket_name, Key=relative_path)
            return response["Body"].read()
        except self.s3.exceptions.NoSuchKey:
            raise FileNotFoundError(
                f"The file '{relative_path}' does not exist in the S3 bucket."
            )
        except (NoCredentialsError, PartialCredentialsError):
            raise PERMISSION_ERROR

    def write(self, relative_path: str, data: bytes):
        """Write data to an S3 file."""
        try:
            self.s3.put_object(Bucket=self.bucket_name, Key=relative_path, Body=data)
        except (NoCredentialsError, PartialCredentialsError):
            raise PERMISSION_ERROR

    def list(self, relative_dir: str) -> list[str]:
        """List files in the specified S3 bucket."""
        try:
            response = self.s3.list_objects_v2(
                Bucket=self.bucket_name, Prefix=relative_dir
            )
            if "Contents" in response:
                return [item["Key"] for item in response["Contents"]]
            return []
        except (NoCredentialsError, PartialCredentialsError):
            raise PERMISSION_ERROR

    def delete(self, relative_path: str):
        """Delete the file."""
        try:
            self.s3.delete_object(Bucket=self.bucket_name, Key=relative_path)
        except self.s3.exceptions.NoSuchKey:
            raise FileNotFoundError(
                f"The file '{relative_path}' does not exist in the S3 bucket."
            )
        except (NoCredentialsError, PartialCredentialsError):
            raise PERMISSION_ERROR
delete(relative_path)

Delete the file.

Source code in backends/filesystem/s3fs.py 
48
49
50
51
52
53
54
55
56
57
def delete(self, relative_path: str):
    """Delete the file."""
    try:
        self.s3.delete_object(Bucket=self.bucket_name, Key=relative_path)
    except self.s3.exceptions.NoSuchKey:
        raise FileNotFoundError(
            f"The file '{relative_path}' does not exist in the S3 bucket."
        )
    except (NoCredentialsError, PartialCredentialsError):
        raise PERMISSION_ERROR
list(relative_dir)

List files in the specified S3 bucket.

Source code in backends/filesystem/s3fs.py 
36
37
38
39
40
41
42
43
44
45
46
def list(self, relative_dir: str) -> list[str]:
    """List files in the specified S3 bucket."""
    try:
        response = self.s3.list_objects_v2(
            Bucket=self.bucket_name, Prefix=relative_dir
        )
        if "Contents" in response:
            return [item["Key"] for item in response["Contents"]]
        return []
    except (NoCredentialsError, PartialCredentialsError):
        raise PERMISSION_ERROR
read(relative_path)

Read data from the S3 file.

Source code in backends/filesystem/s3fs.py 
17
18
19
20
21
22
23
24
25
26
27
def read(self, relative_path: str) -> bytes:
    """Read data from the S3 file."""
    try:
        response = self.s3.get_object(Bucket=self.bucket_name, Key=relative_path)
        return response["Body"].read()
    except self.s3.exceptions.NoSuchKey:
        raise FileNotFoundError(
            f"The file '{relative_path}' does not exist in the S3 bucket."
        )
    except (NoCredentialsError, PartialCredentialsError):
        raise PERMISSION_ERROR
write(relative_path, data)

Write data to an S3 file.

Source code in backends/filesystem/s3fs.py 
29
30
31
32
33
34
def write(self, relative_path: str, data: bytes):
    """Write data to an S3 file."""
    try:
        self.s3.put_object(Bucket=self.bucket_name, Key=relative_path, Body=data)
    except (NoCredentialsError, PartialCredentialsError):
        raise PERMISSION_ERROR

backends.kms.abstractkms

backends.kms.abstractkms

AbstractKMS

Bases: ABC

Abstract base class for Key Management System operations.

Source code in backends/kms/abstractkms.py 
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
class AbstractKMS(ABC):
    """
    Abstract base class for Key Management System operations.
    """

    @abstractmethod
    def generate_kek(self, description: str) -> str:
        """
        Generate a Key Encryption Key (KEK).

        Args:
            description (str): Description for the key.
        Returns:
            str: Key ID of the generated KEK.
        """
        pass

    @abstractmethod
    def encrypt_dek(self, dek: bytes, key_id: str) -> bytes:
        """
        Encrypt a Data Encryption Key (DEK).

        Args:
            dek (bytes): The plaintext DEK to encrypt.
            key_id (str): The Key ID to use for encryption.
        Returns:
            bytes: Encrypted DEK as bytes.
        """
        pass

    @abstractmethod
    def decrypt_dek(self, encrypted_dek: bytes) -> bytes:
        """
        Decrypt a Data Encryption Key (DEK).

        Args:
            encrypted_dek (bytes): The encrypted DEK to decrypt.
        Returns:
            bytes: Plaintext DEK as bytes.
        """
        pass
decrypt_dek(encrypted_dek) abstractmethod

Decrypt a Data Encryption Key (DEK).

Parameters:

Name Type Description Default
encrypted_dek bytes

The encrypted DEK to decrypt.

 required

Returns: bytes: Plaintext DEK as bytes.

Source code in backends/kms/abstractkms.py 
34
35
36
37
38
39
40
41
42
43
44
@abstractmethod
def decrypt_dek(self, encrypted_dek: bytes) -> bytes:
    """
    Decrypt a Data Encryption Key (DEK).

    Args:
        encrypted_dek (bytes): The encrypted DEK to decrypt.
    Returns:
        bytes: Plaintext DEK as bytes.
    """
    pass
encrypt_dek(dek, key_id) abstractmethod

Encrypt a Data Encryption Key (DEK).

Parameters:

Name Type Description Default
dek bytes

The plaintext DEK to encrypt.

 required
key_id str

The Key ID to use for encryption.

 required

Returns: bytes: Encrypted DEK as bytes.

Source code in backends/kms/abstractkms.py 
21
22
23
24
25
26
27
28
29
30
31
32
@abstractmethod
def encrypt_dek(self, dek: bytes, key_id: str) -> bytes:
    """
    Encrypt a Data Encryption Key (DEK).

    Args:
        dek (bytes): The plaintext DEK to encrypt.
        key_id (str): The Key ID to use for encryption.
    Returns:
        bytes: Encrypted DEK as bytes.
    """
    pass
generate_kek(description) abstractmethod

Generate a Key Encryption Key (KEK).

Parameters:

Name Type Description Default
description str

Description for the key.

 required

Returns: str: Key ID of the generated KEK.

Source code in backends/kms/abstractkms.py 
 9
10
11
12
13
14
15
16
17
18
19
@abstractmethod
def generate_kek(self, description: str) -> str:
    """
    Generate a Key Encryption Key (KEK).

    Args:
        description (str): Description for the key.
    Returns:
        str: Key ID of the generated KEK.
    """
    pass

backends.kms.awskms

backends.kms.awskms

AWSKMS

Bases: AbstractKMS

AWS KMS implementation of the AbstractKMS.

Source code in backends/kms/awskms.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
class AWSKMS(AbstractKMS):
    """
    AWS KMS implementation of the AbstractKMS.
    """

    def __init__(self):
        self.kms_client = boto3.client('kms')

    def generate_kek(self, description: str) -> str:
        response = self.kms_client.create_key(
            Description=description,
            KeyUsage='ENCRYPT_DECRYPT',
            CustomerMasterKeySpec='SYMMETRIC_DEFAULT'
        )
        return response['KeyMetadata']['KeyId']

    def encrypt_dek(self, dek: bytes, key_id: str) -> bytes:
        response = self.kms_client.encrypt(
            KeyId=key_id,
            Plaintext=dek
        )
        return response['CiphertextBlob']

    def decrypt_dek(self, encrypted_dek: bytes) -> bytes:
        response = self.kms_client.decrypt(
            CiphertextBlob=encrypted_dek
        )
        return response['Plaintext']

backends.kms.localkms

backends.kms.localkms

LocalKMS

Bases: AbstractKMS

Local implementation of the AbstractKMS for testing purposes with persistent storage.

Source code in backends/kms/localkms.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class LocalKMS(AbstractKMS):
    """
    Local implementation of the AbstractKMS for testing purposes with persistent storage.
    """
    def __init__(self, storage_file: str = "local_kms_storage.json"):
        self.storage_file = Path(storage_file)

        if not self.storage_file.exists():
            self.storage_file.parent.mkdir(parents=True, exist_ok=True)
            self.storage_file.write_text('{}', encoding='utf-8')

        self.keys = self._load_keys()
        self.counter = len(self.keys)

    def _load_keys(self) -> dict:
        if self.storage_file.exists():
            return json.loads(self.storage_file.read_text(encoding='utf-8'))
        return {}

    def _save_keys(self):
        self.storage_file.write_text(json.dumps(self.keys), encoding='utf-8')

    def generate_kek(self, description: str) -> str:
        key_id = f"local-key-{self.counter:02d}"
        self.keys[key_id] = {
            "description": description,
            "key_material": f"key-material-{self.counter}".encode().hex()
        }
        self.counter += 1
        self._save_keys()
        return key_id

    def encrypt_dek(self, dek: bytes, key_id: str) -> bytes:
        if key_id not in self.keys:
            raise ValueError("Invalid Key ID")
        return dek[::-1]  # Example encryption: reverse the bytes

    def decrypt_dek(self, encrypted_dek: bytes) -> bytes:
        return encrypted_dek[::-1]  # Reverse again to decrypt