ddp_utils.temp.manager¶

import ddp_utils.temp.manager

Manage namespaced temporary files and directories for application runs.

TempManager builds run-specific and persistent areas over RuntimePaths, TempDir, and TempFile and provides atomic writes and stale-run cleanup.

Examples

Use this public operation:

from ddp_utils.temp.manager import TempManager

manager = TempManager("example", isolated=True, create_now=False)

class ddp_utils.temp.manager.TempManager(namespace: str | None = None, *, isolated: bool = False, auto_cleanup_run: bool = True, create_now: bool = True)¶

Bases: object

Production-grade namespace temp manager.

The manager operates strictly inside the namespace temp area provided by ddp_utils.runtime.paths.RuntimePaths.

Layout used by this manager:

.ddp/
    <namespace>/
        temp/
            runs/
                run_<timestamp>_pid<pid>_<id>/
            persistent/
            locks/

Main concepts:

run_root: Per-process temporary working directory for the current interpreter run.
persistent_root: Temp-like storage that survives process exit until explicitly cleaned.
cleanup_stale_runs(): Deletes abandoned run directories older than a configured age.
atomic_write_*(): Uses temp staging files and os.replace() for safe writes.

Parameters

Name	Type	Description
namespace	Optional[str]	Namespace of the owning library or module.
auto_cleanup_run	bool	If `True`, the current `run_root` is removed automatically at interpreter exit.
create_now	bool	If `True`, standard temp subdirectories are created immediately.

Examples

Use this public operation:

instance = TempManager(...)

Configure shared or isolated temporary storage for the current process.

The per-run directory and its atexit cleanup hook are created lazily when run_root is first accessed. Runtime construction may process expired cleanup registrations and may create the base directory layout.

Parameters

Name	Type	Description
namespace	Optional[str]	Namespace used only when `isolated` is true.
isolated	bool	Use a namespace-specific runtime instead of shared paths.
auto_cleanup_run	bool	Register the created run directory for removal at process exit.
create_now	bool	Create the underlying runtime directories immediately.

Raises

Exception	Description
ValueError	Isolated mode is requested without a namespace.
OSError	Runtime cleanup or eager directory creation fails.

Examples

Configure lazy isolated temporary storage:

manager = TempManager("reports", isolated=True, create_now=False)

classmethod shared(*, auto_cleanup_run: bool = True, create_now: bool = True) → TempManager¶

Return the shared (non-isolated) TempManager instance, creating its directories if requested.

Parameters

Name	Type	Description
auto_cleanup_run	bool	If `True`, the current run directory is removed automatically at interpreter exit.
create_now	bool	If `True`, the standard temp subdirectories are created immediately.

Returns

Type	Description
TempManager	Shared (non-isolated) `TempManager` instance.

Examples

Use this public operation:

result = instance.shared()

classmethod isolated(namespace: str, *, auto_cleanup_run: bool = True, create_now: bool = True) → TempManager¶

Return a namespace-isolated TempManager instance, creating its directories if requested.

Parameters

Name	Type	Description
namespace	str	Namespace of the owning library or module.
auto_cleanup_run	bool	If `True`, the current run directory is removed automatically at interpreter exit.
create_now	bool	If `True`, the standard temp subdirectories are created immediately.

Returns

Type	Description
TempManager	Namespace-isolated `TempManager` instance.

Examples

Use this public operation:

result = instance.isolated(namespace=namespace_value)

property root: Path¶

Return the namespace temp root.

Returns

Path to .ddp/<namespace>/temp.

Examples

Use this public operation:

result = instance.root()

property runs_root: Path¶

Return the root directory for per-run workspaces.

Returns

Path to .ddp/<namespace>/temp/runs.

Examples

Use this public operation:

result = instance.runs_root()

property persistent_root: Path¶

Return the persistent temp root.

Returns

Path to .ddp/<namespace>/temp/persistent.

Examples

Use this public operation:

result = instance.persistent_root()

property locks_root: Path¶

Return the lock root used for temp-related lock files.

Returns

Path to .ddp/<namespace>/temp/locks.

Examples

Use this public operation:

result = instance.locks_root()

property run_root: Path¶

Return the current process run root.

The directory is created lazily on first access.

Returns

Current run directory path.

Examples

Use this public operation:

result = instance.run_root()

ensure_roots() → None¶

Ensure only the temp root exists.

Subdirectories are created lazily on first access.

Examples

Use this public operation:

result = instance.ensure_roots()

dir(*, prefix: str = 'tmp_', name: str | None = None, persistent: bool = False, auto_cleanup: bool = True) → TempDir¶

Create or reference a temporary directory.

Parameters

Name	Type	Description
prefix	str	Prefix for auto-generated directory names.
name	str \| None	Explicit directory name. If provided, no random name is generated.
persistent	bool	If `True`, place the directory under `persistent_root`. Otherwise place it under the current `run_root`.
auto_cleanup	bool	If `True`, the wrapper removes the directory on context exit.

Returns

Type	Description
TempDir	`TempDir` wrapper.

Examples

Use this public operation:

result = instance.dir()

file(*, suffix: str = '', prefix: str = 'tmp_', name: str | None = None, persistent: bool = False, auto_cleanup: bool = True) → TempFile¶

Create or reference a temporary file.

Parameters

Name	Type	Description
suffix	str	File suffix, such as `".json"`.
prefix	str	Prefix for auto-generated filenames.
name	str \| None	Explicit file name. If provided, no random name is generated.
persistent	bool	If `True`, place the file under `persistent_root`. Otherwise place it under the current `run_root`.
auto_cleanup	bool	If `True`, the wrapper removes the file on context exit.

Returns

Type	Description
TempFile	`TempFile` wrapper.

Examples

Use this public operation:

result = instance.file()

temp_path(*, suffix: str = '', prefix: str = 'tmp_', persistent: bool = False) → Path¶

Create a unique temp file path without automatic cleanup.

Parameters

Name	Type	Description
suffix	str	File suffix.
prefix	str	File prefix.
persistent	bool	If `True`, allocate under `persistent_root`.

Returns

Type	Description
Path	Unique filesystem path.

Examples

Use this public operation:

result = instance.temp_path()

from_text(text: str, *, suffix: str = '.txt', prefix: str = 'tmp_', persistent: bool = False, auto_cleanup: bool = True, encoding: str = 'utf-8') → TempFile¶

Create a temp file from text content.

Parameters

Name	Type	Description
text	str	Text content.
suffix	str	File suffix.
prefix	str	File prefix.
persistent	bool	If `True`, place under `persistent_root`.
auto_cleanup	bool	If `True`, remove on context exit.
encoding	str	Text encoding.

Returns

Type	Description
TempFile	`TempFile` wrapper.

Examples

Use this public operation:

result = instance.from_text(text=text_value)

from_bytes(data: bytes, *, suffix: str = '', prefix: str = 'tmp_', persistent: bool = False, auto_cleanup: bool = True) → TempFile¶

Create a temp file from binary content.

Parameters

Name	Type	Description
data	bytes	Binary content.
suffix	str	File suffix.
prefix	str	File prefix.
persistent	bool	If `True`, place under `persistent_root`.
auto_cleanup	bool	If `True`, remove on context exit.

Returns

Type	Description
TempFile	`TempFile` wrapper.

Examples

Use this public operation:

result = instance.from_bytes(data=data_value)

copy_in(source: Path | str, *, name: str | None = None, persistent: bool = False, auto_cleanup: bool = True) → TempFile¶

Copy an existing file into temp storage.

Parameters

Name	Type	Description
source	Path \| str	Source file path.
name	str \| None	Optional target file name. Defaults to source file name.
persistent	bool	If `True`, place under `persistent_root`.
auto_cleanup	bool	If `True`, remove on context exit.

Returns

Type	Description
TempFile	`TempFile` wrapper for the copied file.

Examples

Use this public operation:

result = instance.copy_in(source=source_value)

atomic_write_bytes(target: Path | str, data: bytes) → Path¶

Atomically write bytes to a target file.

The method writes to a staged temp file first and then replaces the target using os.replace().

Parameters

Name	Type	Description
target	Path \| str	Target file path.
data	bytes	Binary payload.

Returns

Type	Description
Path	Final target path.

Examples

Use this public operation:

result = instance.atomic_write_bytes(target=target_value, data=data_value)

atomic_write_text(target: Path | str, data: str, *, encoding: str = 'utf-8') → Path¶

Atomically write text to a target file.

The method writes to a staged temp file first and then replaces the target using os.replace().

Parameters

Name	Type	Description
target	Path \| str	Target file path.
data	str	Text payload.
encoding	str	Text encoding.

Returns

Type	Description
Path	Final target path.

Examples

Use this public operation:

result = instance.atomic_write_text(target=target_value, data=data_value)

cleanup_run() → None¶

Remove the current run directory, if it exists.

Examples

Use this public operation:

result = instance.cleanup_run()

cleanup_persistent() → None¶

Remove all persistent temp data and recreate the directory.

Examples

Use this public operation:

result = instance.cleanup_persistent()

cleanup_all() → None¶

Remove all temp data for the namespace and recreate standard roots.

Examples

Use this public operation:

result = instance.cleanup_all()

cleanup_stale_runs(older_than_seconds: int) → int¶

Remove stale run directories older than the given age.

The current active run_root is never removed by this method.

Parameters

Name	Type	Description
older_than_seconds	int	Age threshold in seconds.

Returns

Type	Description
int	Number of removed run directories.

Examples

Use this public operation:

result = instance.cleanup_stale_runs(older_than_seconds=older_than_seconds_value)