Utilities

The otter.util package contains utility modules for Otter, incluing errors used throughout the application.

util.errors module

Custom exceptions.

otter.util.errors.log_pydantic(e: ValidationError) → str

Log a pydantic validation error correctly.

exception otter.util.errors.OtterError

Bases: Exception

Base class for all application-specific exceptions.

exception otter.util.errors.StepInvalidError

Bases: OtterError

Raise when a step is invalid somehow.

exception otter.util.errors.StepFailedError

Bases: OtterError

Raise when a step fails somehow.

exception otter.util.errors.TaskRunError

Bases: OtterError

Raise when a task fails to run.

exception otter.util.errors.TaskAbortedError

Bases: OtterError

Raise when a task is aborted.

exception otter.util.errors.TaskValidationError

Bases: OtterError

Raise when a task fails validation.

exception otter.util.errors.DownloadError

Bases: OtterError

Raise when an error occurs during a download.

exception otter.util.errors.UploadError

Bases: OtterError

Raise when an error occurs during an upload.

exception otter.util.errors.CopyError

Bases: OtterError

Raise when an error occurs during a copy operation.

exception otter.util.errors.NotFoundError(msg: str | None = None, thing: str | None = None)

Bases: OtterError

Raise when something is not found.

exception otter.util.errors.PreconditionFailedError

Bases: OtterError

Raise when a precondition fails.

exception otter.util.errors.ScratchpadError

Bases: OtterError

Raise when a key is not found in the scratchpad.

exception otter.util.errors.StorageError

Bases: OtterError

Raise when an error occurs in a storage class.

exception otter.util.errors.ManifestError

Bases: OtterError

Raise when an error occurs in the manifest management.

exception otter.util.errors.FSError

Bases: OtterError

Raise when an error occurs in the filesystem operations.

util.fs module

Local file system utilities.

otter.util.fs.check_dir(path: Path | str) → None

Check working conditions for a directory.

The function will make sure that the directory exists and is writable. If it does not exist, the function will attempt to create it.

Parameters:

path (Path | str) – The directory to check.

Raises:

FSError – If the path exists but is not a directory or is not writable; or if the directory does not exist and cannot be created.

otter.util.fs.check_source(path: Path | str) → None

Check working conditions for a file.

The function will make sure that the file exists and is readable.

Parameters:

path (Path | str) – The path to check. Must be a file.

Raises:

FSError – If the path is not a file, does not exist, or is not readable.

otter.util.fs.check_destination(path: Path | str, *, delete: bool = False) → None

Check working conditions for a destination path.

The function will make sure that the file does not exist and that the parent directory exists and is writable. If the parent directory does not exist, the function will attempt to create it.

If delete is True, the function will delete the file if it already exists.

Warning

This function can potentially delete files!

Parameters:

• path (Path | str) – The path to check. Must be a file.

• delete (bool) – Whether to delete the file if it already exists.

Raises:

FSError – If the parent directory checks fail, or if the file exists and delete is False.

util.logger module

Configures the logger for the application.

otter.util.logger.get_exception_info(record_exception: loguru.RecordException | None) → tuple[str, str, str, str, str]

Get fields from the exception record.

This function extracts the name, function and line number from an exception. It will also return the exception type and message.

It will go back in the stack to the first frame originated inside the app, that way it will make sure the error is meaningful in the logs. If we don’t do this, the error will be logged as raising from the in the report decorator, which is not very useful.

Parameters:

record_exception (loguru.RecordException | None) – The exception record to extract the information from.

otter.util.logger.get_format_log(include_task: bool = True) → Callable[..., str]

Create the log format function.

otter.util.logger.task_logging(task: Task) → Generator[None]

Context manager that appends log messages to the task’s manifest.

We check if there is already a stdout logger, and if not, we add one. This is added to make sure spawned processes also log to stdout. We’ve seen that sometimes the logger configuration is not inherited by child processes for some unknown reason. Probably related to:

https://github.com/Delgan/loguru/issues/912

Parameters:

task (Task) – The task to log messages to.

class otter.util.logger.MessageQueue

Bases: object

A queue for log messages.

This class is used to hold log messages until the logger is configured.

put(message: loguru.Message) → None

Put a message in the queue.

flush() → None

Dump the log messages to stdout.

otter.util.logger.init_logger_early() → None

Initialize early logging.

otter.util.logger.init_logger(log_level: str = 'INFO', app_name: str | None = None) → None

Initialize the logger.

Once the logger is set up, dumps the log messages held in the queue.

Parameters:

• log_level (str) – The log level to use.

• app_name (str | None) – The name of the Otter application, defaults to otter.

Module contents

Utility functions and classes.