Storage¶

commons.storage ¶

A file storage abstraction with local, SQL and Azure Blob Storage backends.

Files (Files, File): files by key, an absolute slugified path such as /cases/123/report.pdf; find, get, list (every file below a path, recursively), put and delete.
Directories (Files.list_directory, DirectoryListing, Directory): the directories and files directly below a path, e.g. for a file browser. Directories are virtual: one exists as long as a file is stored below it, so there are no empty directories and nothing to create or delete.
Metadata (File.size, File.last_modified): known by every backend without reading the file. Metadata with a domain meaning (title, owner, tags, permissions, content type) belongs to the application, linked by the key.
Backends (LocalFiles, SQLFiles, AzureFiles) and FilesWithPrefix, which puts every key below a prefix, e.g. one per tenant. SQLFiles tables created before commons 1.4 have no column last_modified and aren't migrated: drop the table or add the column by hand.

Modules:

Name	Description
`azure`
`files`
`local`
`prefix`
`sql`

Classes:

Name	Description
`AzureFile`	Lazy-loading Azure Blob Storage file.
`AzureFiles`	Azure Blob Storage implementation of Files.
`Directory`	A virtual directory: it exists as long as at least one file is stored below it.
`DirectoryListing`	The directories and files directly below a path, each sorted by key.
`File`
`Files`
`LocalFile`	Local file system implementation of File.
`LocalFiles`	Local file system implementation of Files.
`FilesWithPrefix`	Delegate that transparently prepends a path prefix to all file operations.
`SQLFile`	SQL implementation of File.
`SQLFiles`	SQL implementation of Files for storing binary files in a database table.

all `module-attribute` ¶

__all__ = [
    "AzureFile",
    "AzureFiles",
    "Directory",
    "DirectoryListing",
    "File",
    "Files",
    "FilesWithPrefix",
    "LocalFile",
    "LocalFiles",
    "SQLFile",
    "SQLFiles",
]

AzureFile ¶

AzureFile(
    key: str,
    container_client: ContainerClient,
    *,
    size: int | None = None,
    last_modified: datetime | None = None
)

Bases: File

Lazy-loading Azure Blob Storage file.

The blob bytes are not fetched until open() is called. size and last_modified come from the listing or lookup that created the file; if they weren't given, the blob's properties are fetched once on first access.

Methods:

Name	Description
`open`

Attributes:

Name	Type	Description
`key`	`str`
`size`	`int`
`last_modified`	`datetime`

Source code in src/commons/storage/azure.py

def __init__(
    self,
    key: str,
    container_client: ContainerClient,
    *,
    size: int | None = None,
    last_modified: datetime | None = None,
) -> None:
    self._key = key
    self._container_client = container_client
    self._size = size
    self._last_modified = last_modified

key `property` ¶

key: str

size `property` ¶

size: int

last_modified `property` ¶

last_modified: datetime

open ¶

open() -> BytesIO

Source code in src/commons/storage/azure.py

def open(self) -> BytesIO:
    blob = self._container_client.get_blob_client(self._key.lstrip("/"))
    return BytesIO(blob.download_blob().readall())

AzureFiles ¶

AzureFiles(
    settings: AzureStorageSettings | None = None,
    *,
    container_client: ContainerClient | None = None
)

Bases: Files

Azure Blob Storage implementation of Files.

Maps each Files key (e.g. /foo/bar.txt) to a blob name in a single Azure Storage container (e.g. foo/bar.txt). Multi-tenant prefixing is handled by :class:commons.storage.FilesWithPrefix.

Methods:

Name	Description
`find`
`list`
`list_directory`

Source code in src/commons/storage/azure.py

def __init__(
    self,
    settings: AzureStorageSettings | None = None,
    *,
    container_client: ContainerClient | None = None,
) -> None:
    if container_client is not None:
        self._container_client = container_client
        return

    if settings is None:
        settings = AzureStorageSettings.read()

    assert settings.container, "AzureStorageSettings.container must be set."
    self._container_client = _build_container_client(settings)

find ¶

find(key: str) -> File | None

Source code in src/commons/storage/azure.py

def find(self, key: str) -> File | None:
    blob = self._container_client.get_blob_client(key.lstrip("/"))
    try:
        properties = blob.get_blob_properties()
    except ResourceNotFoundError:
        return None
    return AzureFile(key, self._container_client, size=properties.size, last_modified=properties.last_modified)

list ¶

list(path: str = '/') -> list[File]

Source code in src/commons/storage/azure.py

def list(self, path: str = "/") -> list[File]:
    blobs = self._container_client.list_blobs(name_starts_with=_prefix(path))
    return [AzureFile._from_properties(b, self._container_client) for b in blobs]

list_directory ¶

list_directory(path: str = '/') -> DirectoryListing

Source code in src/commons/storage/azure.py

def list_directory(self, path: str = "/") -> DirectoryListing:
    directories: list[Directory] = []
    files: list[File] = []

    # One level only: with a delimiter, the service returns the blobs below deeper paths as one prefix each.
    for item in self._container_client.walk_blobs(name_starts_with=_prefix(path), delimiter="/"):
        if isinstance(item, BlobPrefix):
            directories.append(Directory("/" + item.name.rstrip("/")))
        else:
            files.append(AzureFile._from_properties(item, self._container_client))

    return DirectoryListing(
        directories=sorted(directories, key=lambda d: d.key),
        files=sorted(files, key=lambda f: f.key),
    )

Directory `dataclass` ¶

Directory(key: str)

A virtual directory: it exists as long as at least one file is stored below it.

The key is an absolute path without a trailing slash, e.g. /foo/bar.

Attributes:

Name	Type	Description
`key`	`str`

key `instance-attribute` ¶

key: str

DirectoryListing `dataclass` ¶

DirectoryListing(
    directories: list[Directory], files: list[File]
)

The directories and files directly below a path, each sorted by key.

Attributes:

Name	Type	Description
`directories`	`list[Directory]`
`files`	`list[File]`

directories `instance-attribute` ¶

directories: list[Directory]

files `instance-attribute` ¶

files: list[File]

File ¶

Bases: ABC

Methods:

Name	Description
`get`
`open`

Attributes:

Name	Type	Description
`key`	`str`
`size`	`int`	The size of the file in bytes.
`last_modified`	`datetime`	When the file was last written, with time zone (UTC).

key `abstractmethod` `property` ¶

key: str

size `property` ¶

size: int

The size of the file in bytes.

The backends of commons know it without reading the file; this default reads the content.

last_modified `property` ¶

last_modified: datetime

When the file was last written, with time zone (UTC).

Every backend of commons provides it. It isn't abstract, so subclasses outside commons that don't implement it keep working until it is read; this default raises NotImplementedError.

get ¶

get() -> BytesIO

Source code in src/commons/storage/files.py

def get(self) -> BytesIO:
    return self.open()

open `abstractmethod` ¶

open() -> BytesIO

Source code in src/commons/storage/files.py

@abstractmethod
def open(self) -> BytesIO: ...

Files ¶

Bases: ABC

Methods:

Name	Description
`find`	Search a file by its key. Returns None if file doesn't exist.
`get`	Get a file by its key. Throws error if file does not exist.
`list`	Get all lists which are below provided path.
`list_directory`	Get the directories and files directly below the provided path, not the ones further down.
`put`	Adds or overwrites a file.
`delete`	Delete a file or files by key.

find `abstractmethod` ¶

find(key: str) -> File | None

Search a file by its key. Returns None if file doesn't exist.

Source code in src/commons/storage/files.py

@abstractmethod
def find(self, key: str) -> File | None:
    """Search a file by its key. Returns None if file doesn't exist."""
    ...

get ¶

get(key: str) -> File

Get a file by its key. Throws error if file does not exist.

Source code in src/commons/storage/files.py

def get(self, key: str) -> File:
    """Get a file by its key. Throws error if file does not exist."""
    file = self.find(key)
    assert file is not None, f"No file with key `{key}`"
    return file

list `abstractmethod` ¶

list(path: str = '/') -> list[File]

Get all lists which are below provided path.

Only full path segments match, e.g. we have the following files

/foo/a.txt
/foo/b.txt
/foo-bar/c.txt

list("/foo") returns

/foo/a.txt
/foo/b.txt

Source code in src/commons/storage/files.py

@abstractmethod
def list(self, path: str = "/") -> list[File]:
    """
    Get all lists which are below provided path.

    Only full path segments match, e.g. we have the following files

    ```
    /foo/a.txt
    /foo/b.txt
    /foo-bar/c.txt
    ```

    `list("/foo")` returns

    ```
    /foo/a.txt
    /foo/b.txt
    ```
    """

list_directory ¶

list_directory(path: str = '/') -> DirectoryListing

Get the directories and files directly below the provided path, not the ones further down.

Directories are virtual: a directory is listed if at least one file is stored below it, at any depth. Only full path segments match, as in list. /foo and /foo/ are the same path; a path with nothing below it (a file key or a missing key) returns an empty listing.

With the files

/foo/a.txt
/foo/bar/b.txt
/foo/bar/baz/c.txt
/foo-bar/d.txt

list_directory("/foo") returns the directory /foo/bar and the file /foo/a.txt.

Source code in src/commons/storage/files.py

def list_directory(self, path: str = "/") -> DirectoryListing:
    """
    Get the directories and files directly below the provided path, not the ones further down.

    Directories are virtual: a directory is listed if at least one file is stored below it, at any depth.
    Only full path segments match, as in `list`. `/foo` and `/foo/` are the same path; a path with nothing
    below it (a file key or a missing key) returns an empty listing.

    With the files

    ```
    /foo/a.txt
    /foo/bar/b.txt
    /foo/bar/baz/c.txt
    /foo-bar/d.txt
    ```

    `list_directory("/foo")` returns the directory `/foo/bar` and the file `/foo/a.txt`.
    """
    base = path.rstrip("/")
    directories: set[str] = set()
    files: list[File] = []

    for file in self.list(path):
        rest = file.key.removeprefix(base + "/")
        if rest == file.key:
            continue  # not below the path

        if "/" in rest:
            directories.add(base + "/" + rest.split("/", 1)[0])
        else:
            files.append(file)

    return DirectoryListing(
        directories=[Directory(key) for key in sorted(directories)],
        files=sorted(files, key=lambda f: f.key),
    )

put ¶

put(key: str, data: BytesIO) -> None

Adds or overwrites a file.

Source code in src/commons/storage/files.py

def put(self, key: str, data: BytesIO) -> None:
    """Adds or overwrites a file."""

    key_clean = string_operators.normalize_path(string_operators.ensure_path_slugified(key))
    assert key_clean == key, f"The provided key (`{key}`) is not valid. A valid key would be `{key_clean}`."
    assert len(key) > 1, "The provided key is not valid. Keys must have at least a length of two characters."

    self._put(key, data)

delete ¶

delete(key: str, recursive: bool = False) -> int

Delete a file or files by key.

Parameters:

Name	Type	Description	Default
`key`	`str`	The file key or path prefix to delete.	required
`recursive`	`bool`	If True, deletes all files under the given path prefix. If False, only deletes the exact file matching the key.	`False`

Returns:

Type	Description
`int`	The number of files deleted.

Source code in src/commons/storage/files.py

def delete(self, key: str, recursive: bool = False) -> int:
    """
    Delete a file or files by key.

    Args:
        key: The file key or path prefix to delete.
        recursive: If True, deletes all files under the given path prefix.
                   If False, only deletes the exact file matching the key.

    Returns:
        The number of files deleted.
    """
    return self._delete(key, recursive)

LocalFile ¶

LocalFile(key: str, path: Path)

Bases: File

Local file system implementation of File.

Methods:

Name	Description
`open`

Attributes:

Name	Type	Description
`key`	`str`
`size`	`int`
`last_modified`	`datetime`

Source code in src/commons/storage/local.py

def __init__(self, key: str, path: Path) -> None:
    self._key = key
    self._path = path

key `property` ¶

key: str

size `property` ¶

size: int

last_modified `property` ¶

last_modified: datetime

open ¶

open() -> BytesIO

Source code in src/commons/storage/local.py

def open(self) -> BytesIO:
    return BytesIO(self._path.read_bytes())

LocalFiles ¶

LocalFiles(root: Path | str | None = None)

Bases: Files

Local file system implementation of Files.

Methods:

Name	Description
`find`
`list`
`list_directory`

Source code in src/commons/storage/local.py

def __init__(self, root: Path | str | None = None) -> None:
    if root is None:
        settings = AppSettings.read()
        root = settings.data_dir / "files"

    self._root = Path(root)
    self._root.mkdir(parents=True, exist_ok=True)

find ¶

find(key: str) -> File | None

Source code in src/commons/storage/local.py

def find(self, key: str) -> File | None:
    path = self._key_to_path(key)
    if path.is_file():
        return LocalFile(key, path)
    return None

list ¶

list(path: str = '/') -> list[File]

Source code in src/commons/storage/local.py

def list(self, path: str = "/") -> list[File]:
    normalized_path = path.rstrip("/")
    files: list[File] = []

    if not self._root.exists():
        return files

    for file_path in self._root.rglob("*"):
        if file_path.is_file():
            key = self._path_to_key(file_path)
            if key.startswith(normalized_path + "/") or (normalized_path == "" and key.startswith("/")):
                files.append(LocalFile(key, file_path))

    return files

list_directory ¶

list_directory(path: str = '/') -> DirectoryListing

Source code in src/commons/storage/local.py

def list_directory(self, path: str = "/") -> DirectoryListing:
    folder = self._key_to_path(path.rstrip("/"))
    directories: list[Directory] = []
    files: list[File] = []

    # Keys with `..` must not list folders outside the root.
    if not folder.is_dir() or not folder.resolve().is_relative_to(self._root.resolve()):
        return DirectoryListing(directories=directories, files=files)

    for entry in folder.iterdir():
        key = self._path_to_key(entry)
        if entry.is_file():
            files.append(LocalFile(key, entry))
        elif entry.is_dir() and any(p.is_file() for p in entry.rglob("*")):
            directories.append(Directory(key))

    return DirectoryListing(
        directories=sorted(directories, key=lambda d: d.key),
        files=sorted(files, key=lambda f: f.key),
    )

FilesWithPrefix ¶

FilesWithPrefix(inner: Files, prefix: str)

Bases: Files

Delegate that transparently prepends a path prefix to all file operations.

Methods:

Name	Description
`find`
`list`
`list_directory`

Source code in src/commons/storage/prefix.py

def __init__(self, inner: Files, prefix: str) -> None:
    prefix = string_operators.normalize_path(string_operators.ensure_path_slugified(prefix))
    assert len(prefix) > 1, "The prefix must have at least two characters (e.g. '/tenant')."
    self._inner = inner
    self._prefix = prefix

find ¶

find(key: str) -> File | None

Source code in src/commons/storage/prefix.py

def find(self, key: str) -> File | None:
    result = self._inner.find(self._add_prefix(key))
    if result is None:
        return None
    return self._strip_prefix(result)

list ¶

list(path: str = '/') -> list[File]

Source code in src/commons/storage/prefix.py

def list(self, path: str = "/") -> list[File]:
    results = self._inner.list(self._add_prefix(path))
    return [self._strip_prefix(f) for f in results]

list_directory ¶

list_directory(path: str = '/') -> DirectoryListing

Source code in src/commons/storage/prefix.py

def list_directory(self, path: str = "/") -> DirectoryListing:
    listing = self._inner.list_directory(self._add_prefix(path))
    return DirectoryListing(
        directories=[Directory(d.key.removeprefix(self._prefix)) for d in listing.directories],
        files=[self._strip_prefix(f) for f in listing.files],
    )

SQLFile ¶

SQLFile(
    key: str,
    data: bytes,
    *,
    last_modified: datetime | None = None
)

Bases: File

SQL implementation of File.

Methods:

Name	Description
`open`

Attributes:

Name	Type	Description
`key`	`str`
`size`	`int`
`last_modified`	`datetime`

Source code in src/commons/storage/sql.py

def __init__(self, key: str, data: bytes, *, last_modified: datetime | None = None) -> None:
    self._key = key
    self._data = data
    self._last_modified = last_modified

key `property` ¶

key: str

size `property` ¶

size: int

last_modified `property` ¶

last_modified: datetime

open ¶

open() -> BytesIO

Source code in src/commons/storage/sql.py

def open(self) -> BytesIO:
    return BytesIO(self._data)

SQLFiles ¶

SQLFiles(engine: Engine | None = None)

Bases: Files

SQL implementation of Files for storing binary files in a database table.

The table files has the columns key, data and last_modified. A table created by a version of commons before 1.4 has no column last_modified; it isn't migrated, so drop it or add the column by hand.

Methods:

Name	Description
`find`
`list`

Source code in src/commons/storage/sql.py

def __init__(self, engine: Engine | None = None) -> None:
    if engine is None:
        engine = create_engine()

    self._engine = engine
    _FilePE.metadata.create_all(self._engine, checkfirst=True)
    self._create_session = sessionmaker(bind=self._engine)

find ¶

find(key: str) -> File | None

Source code in src/commons/storage/sql.py

def find(self, key: str) -> File | None:
    with self._create_session() as session:
        stmt = select(_FilePE).where(_FilePE.key == key)
        result = session.execute(stmt).scalar_one_or_none()

        if result:
            return SQLFile(result.key, result.data, last_modified=_as_utc(result.last_modified))
        return None

list ¶

list(path: str = '/') -> list[File]

Source code in src/commons/storage/sql.py

def list(self, path: str = "/") -> list[File]:
    normalized_path = path.rstrip("/")

    with self._create_session() as session:
        stmt = select(_FilePE)
        results = session.execute(stmt).scalars().all()

        files: list[File] = []
        for result in results:
            if result.key.startswith(normalized_path + "/") or (
                normalized_path == "" and result.key.startswith("/")
            ):
                files.append(SQLFile(result.key, result.data, last_modified=_as_utc(result.last_modified)))

        return files

Storage¶

commons.storage ¶

__all__ module-attribute ¶

AzureFile ¶

key property ¶

size property ¶

last_modified property ¶

open ¶

AzureFiles ¶

find ¶

list ¶

list_directory ¶

Directory dataclass ¶

key instance-attribute ¶

DirectoryListing dataclass ¶

directories instance-attribute ¶

files instance-attribute ¶

File ¶

key abstractmethod property ¶

size property ¶

last_modified property ¶

get ¶

open abstractmethod ¶

Files ¶

find abstractmethod ¶

get ¶

list abstractmethod ¶

list_directory ¶

put ¶

delete ¶

LocalFile ¶

key property ¶

size property ¶

last_modified property ¶

open ¶

LocalFiles ¶

find ¶

list ¶

list_directory ¶

FilesWithPrefix ¶

find ¶

list ¶

list_directory ¶

SQLFile ¶

key property ¶

size property ¶

last_modified property ¶

open ¶

SQLFiles ¶

find ¶

list ¶

all `module-attribute` ¶

key `property` ¶

size `property` ¶

last_modified `property` ¶

Directory `dataclass` ¶

key `instance-attribute` ¶

DirectoryListing `dataclass` ¶

directories `instance-attribute` ¶

files `instance-attribute` ¶

key `abstractmethod` `property` ¶

size `property` ¶

last_modified `property` ¶

open `abstractmethod` ¶

find `abstractmethod` ¶

list `abstractmethod` ¶

key `property` ¶

size `property` ¶

last_modified `property` ¶

key `property` ¶

size `property` ¶

last_modified `property` ¶