API Reference

@dataclasses.dataclass(frozen=True)
class MaildirSpool:
    """A Maildir-like directory structure: a temporary directory and a spool one."""

    tmpdir: pathlib.Path
    """The path to the temporary directory."""

    spooldir: pathlib.Path
    """The path to the spool directory."""

    @classmethod
    def from_maildir(cls, path: pathlib.Path) -> Self:
        """Use the default Maildir `tmp` and `new` subdirectory names."""
        return cls(tmpdir=path / "tmp", spooldir=path / "new")

    def create_file(
        self,
        *,
        encoding: str = "UTF-8",
        prefix: str = "",
        suffix: str = "",
    ) -> MaildirSpoolFile:
        """Create a temporary file, let the caller write to it."""
        tempf, name = create_named_file(
            dir=str(self.tmpdir),
            encoding=encoding,
            prefix=prefix,
            suffix=suffix,
        )
        return MaildirSpoolFile(
            tempf,
            name,
            delete=True,
            maildir_spool=self,
            maildir_prefix=prefix,
            maildir_suffix=suffix,
        )

    def safecat(
        self,
        *,
        contents: str,
        encoding: str = "UTF-8",
        prefix: str = "",
        suffix: str = "",
    ) -> pathlib.Path:
        """Create a file in the temp directory, move it to a unique name within the spool one."""
        with self.create_file(encoding=encoding, prefix=prefix, suffix=suffix) as tempf:
            print(contents, file=tempf, end="")
            return tempf.rename_maildir_file()

class MaildirSpoolFile(TemporaryTextFile):
    """A temporary file that keeps a reference to the Maildir-style directory it is in."""

    maildir_spool: MaildirSpool
    """The Maildir-style directory that this file was created as part of."""

    maildir_prefix: str
    """The prefix to use when moving the file to the spool directory."""

    maildir_suffix: str
    """The suffix to use when moving the file to the spool directory."""

    # ruff: ignore[too-many-arguments]
    # all the TemporaryTextFile constructor arguments and then some more
    def __init__(
        self: Self,
        file: IO[str],
        name: str,
        *,
        maildir_spool: MaildirSpool,
        delete: bool = True,
        maildir_prefix: str = "",
        maildir_suffix: str = "",
    ) -> None:
        """Create the temporary file, initialize our new fields."""
        super().__init__(file, name=name, delete=delete)
        self.maildir_spool = maildir_spool
        self.maildir_prefix = maildir_prefix
        self.maildir_suffix = maildir_suffix

    def rename_maildir_file(self) -> pathlib.Path:
        """Rename the temporary file into the spool directory."""
        inum: Final = self.path.stat().st_ino
        res: Final = (
            self.maildir_spool.spooldir / f"{self.maildir_prefix}{inum}{self.maildir_suffix}"
        )
        if res.is_symlink() or res.exists():
            raise RuntimeError(repr((self, inum, res)))

        self.flush()
        self.path.rename(res)
        self.unset_delete()
        return res

class TemporaryDirectory(TempDir):
    """A temporary directory with a ready-made `pathlib.Path` path.

    Note that the `__enter__()` method (and consequently `with TemporaryDirectory(...) as tempd:`)
    returns a `pathlib.Path` object instead of a string!
    """

    path: pathlib.Path
    """The path to the tempfile as a `pathlib.Path` object."""

    resolved_path: pathlib.Path
    """The resolved path to the tempfile as a `pathlib.Path` object."""

    if sys.version_info >= (3, 12):

        def __init__(
            self,
            suffix: str | None = None,
            prefix: str | None = None,
            # ruff: ignore[builtin-argument-shadowing]
            # tempfile API
            dir: pathutil.PathLike | None = None,
            # ruff: ignore[boolean-type-hint-positional-argument]
            # ruff: ignore[boolean-default-value-positional-argument]
            # tempfile API
            ignore_cleanup_errors: bool = False,
            *,
            delete: bool = True,
        ) -> None:
            """Create the temporary directory, initialize our new fields."""
            super().__init__(
                suffix=suffix,
                prefix=prefix,
                dir=dir,
                ignore_cleanup_errors=ignore_cleanup_errors,
                delete=delete,
            )
            self._setup_paths()

    else:

        def __init__(
            self,
            suffix: str | None = None,
            prefix: str | None = None,
            # ruff: ignore[builtin-argument-shadowing]
            # tempfile API
            dir: pathutil.PathLike | None = None,
            # ruff: ignore[boolean-type-hint-positional-argument]
            # ruff: ignore[boolean-default-value-positional-argument]
            # tempfile API
            ignore_cleanup_errors: bool = False,
        ) -> None:
            """Create the temporary directory, initialize our new fields."""
            super().__init__(
                suffix=suffix,
                prefix=prefix,
                dir=dir,
                ignore_cleanup_errors=ignore_cleanup_errors,
            )
            self._setup_paths()

    def _setup_paths(self) -> None:
        """Initialize our new fields."""
        self.path = pathlib.Path(self.name)
        self.resolved_path = self.path.resolve()

    # ty: ignore[invalid-method-override]
    def __enter__(self) -> pathlib.Path:  # type: ignore[override]
        """Enter a context, return the path to the directory.

        Note that this method returns a `pathlib.Path` object instead of a string!
        """
        return self.path

class TemporaryTextFile(TempTextWrap):
    """A named temporary file with a ready-made `pathlib.Path` path."""

    path: pathlib.Path
    """The path to the tempfile as a `pathlib.Path` object."""

    resolved_path: pathlib.Path
    """The resolved path to the tempfile as a `pathlib.Path` object."""

    if sys.version_info >= (3, 12):

        def __init__(
            self: Self,
            file: IO[str],
            name: str,
            # ruff: ignore[boolean-type-hint-positional-argument]
            # ruff: ignore[boolean-default-value-positional-argument]
            # tempfile API
            delete: bool = True,
            # ruff: ignore[boolean-type-hint-positional-argument]
            # ruff: ignore[boolean-default-value-positional-argument]
            # tempfile API
            delete_on_close: bool = True,
        ) -> None:
            """Create the temporary file, initialize our new fields."""
            super().__init__(file, name=name, delete=delete, delete_on_close=delete_on_close)
            self._setup_paths()
    else:

        def __init__(
            self: Self,
            file: IO[str],
            name: str,
            # ruff: ignore[boolean-type-hint-positional-argument]
            # ruff: ignore[boolean-default-value-positional-argument]
            # tempfile API
            delete: bool = True,
        ) -> None:
            """Create the temporary file, initialize our new fields."""
            super().__init__(file, name=name, delete=delete)
            self._setup_paths()

    def __repr__(self) -> str:
        """Provide a Python-esque representation."""
        # Python 3.14 may evaluate `f"{self!r}"` in the constructor...
        if not hasattr(self, "path"):
            self._setup_paths()

        return (
            f"{type(self).__name__}(file={self.file!r}, name={self.name!r}, "
            f"path={self.path!r}, resolved_path={self.resolved_path!r}"
        )

    def _setup_paths(self) -> None:
        """Initialize our new fields."""
        self.path = pathlib.Path(self.name)
        self.resolved_path = self.path.resolve()

    def unset_delete(self, *, also_on_close: bool = True) -> None:
        """Unset the "delete on drop" and also possibly the "delete on close" flag.

        The caller took care of the file, e.g. renamed it, removed it, or passed it to
        another consumer to use as-is.
        """

        def handle(obj: Any) -> None:  # ruff: ignore[any-type]  # we can handle anything
            """Unset the `delete` and maybe the `delete_on_close` flags if they are there."""
            if hasattr(obj, "delete"):
                obj.delete = False
            if also_on_close and hasattr(obj, "delete_on_close"):
                obj.delete_on_close = False

        handle(self)

        match getattr(self, "_closer", None):
            case None:
                pass

            case closer:
                handle(closer)