Source code for betty.assets

"""
The Assets API.
"""

from __future__ import annotations

import asyncio
from collections import deque

from contextlib import suppress
from pathlib import Path
from shutil import copy2
from typing import AsyncContextManager, Sequence, AsyncIterable, TYPE_CHECKING

import aiofiles
from aiofiles.os import makedirs

from betty.fs import iterfiles

if TYPE_CHECKING:
    from aiofiles.threadpool.text import AsyncTextIOWrapper
    from types import TracebackType


class _Open:
    def __init__(self, fs: AssetRepository, file_paths: tuple[Path, ...]):
        self._fs = fs
        self._file_paths = file_paths
        self._file: AsyncContextManager[AsyncTextIOWrapper] | None = None

    async def __aenter__(self) -> AsyncTextIOWrapper:
        for file_path in map(Path, self._file_paths):
            for fs_path, fs_encoding in self._fs._paths:
                with suppress(FileNotFoundError):
                    self._file = aiofiles.open(
                        fs_path / file_path, encoding=fs_encoding
                    )
                    return await self._file.__aenter__()
        raise FileNotFoundError

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        if self._file is not None:
            await self._file.__aexit__(None, None, None)


[docs] class AssetRepository: """ Manages a set of assets. This repository unifies several directory paths on disk, overlaying them on each other. Paths added later act as fallbacks, e.g. earlier paths have priority. """
[docs] def __init__(self, *paths: tuple[Path, str | None]): self._paths = deque(paths)
def __len__(self) -> int: return len(self._paths) @property def paths(self) -> Sequence[tuple[Path, str | None]]: """ The paths to the individual layers. """ return list(self._paths)
[docs] def prepend(self, path: Path, fs_encoding: str | None = None) -> None: """ Prepend a layer path, e.g. override existing layers with the given one. """ self._paths.appendleft((path, fs_encoding))
[docs] def clear(self) -> None: """ Clear all layers from the file system. """ self._paths.clear()
[docs] def open(self, *file_paths: Path) -> _Open: """ Open a file. :param file_paths: One or more file paths within the file system. The first file path to exist will cause this function to return. Previously missing file paths will not cause errors. :raise FileNotFoundError: Raised when none of the provided paths matches an existing file. """ return _Open(self, file_paths)
[docs] async def copy2(self, source_path: Path, destination_path: Path) -> Path: """ Copy a file to a destination using :py:func:`shutil.copy2`. """ for fs_path, _ in self._paths: with suppress(FileNotFoundError): await asyncio.to_thread(copy2, fs_path / source_path, destination_path) return destination_path tried_paths = [str(fs_path / source_path) for fs_path, _ in self._paths] raise FileNotFoundError("Could not find any of %s." % ", ".join(tried_paths))
[docs] async def copytree( self, source_path: Path, destination_path: Path ) -> AsyncIterable[Path]: """ Recursively copy the files in a directory tree to another directory. """ file_destination_paths = set() for fs_path, _ in self._paths: async for file_source_path in iterfiles(fs_path / source_path): file_destination_path = destination_path / file_source_path.relative_to( fs_path / source_path ) if file_destination_path not in file_destination_paths: file_destination_paths.add(file_destination_path) await makedirs(file_destination_path.parent, exist_ok=True) await asyncio.to_thread( copy2, file_source_path, file_destination_path ) yield file_destination_path