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[source]

Log a pydantic validation error correctly.

exception otter.util.errors.OtterError[source]

Bases: Exception

Base class for all application-specific exceptions.

exception otter.util.errors.StepInvalidError[source]

Bases: OtterError

Raise when a step is invalid somehow.

exception otter.util.errors.StepFailedError[source]

Bases: OtterError

Raise when a step fails somehow.

exception otter.util.errors.TaskAbortedError[source]

Bases: OtterError

Raise when a task is aborted.

exception otter.util.errors.TaskValidationError[source]

Bases: OtterError

Raise when a task fails validation.

exception otter.util.errors.DownloadError[source]

Bases: OtterError

Raise when an error occurs during a download.

exception otter.util.errors.NotFoundError(msg: str | None = None)[source]

Bases: OtterError

Raise when something is not found.

exception otter.util.errors.PreconditionFailedError[source]

Bases: OtterError

Raise when a precondition fails.

exception otter.util.errors.ScratchpadError[source]

Bases: OtterError

Raise when a key is not found in the scratchpad.

exception otter.util.errors.StorageError[source]

Bases: OtterError

Raise when an error occurs in a storage class.

util.fs module

File system utilities.

otter.util.fs.check_file_not_exists(path: Path, *, delete: bool = False) → None[source]

Ensure a file does not exist, optionally deleting it.

The function will make sure that a file does not exist in the given path. If delete is True, the function will delete the file if it already exists.

Warning

This function can potentially delete files!

Parameters:

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

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

Raises:

SystemExit – If the file exists and delete is False. Or if there is an error deleting the file.

Returns:

None if all checks pass.

Return type:

None

otter.util.fs.check_file_exists(path: Path) → None[source]

Ensure a file exists.

The function will make sure that a file exists in the given path.

Parameters:

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

Raises:

SystemExit – If the file does not exist.

Returns:

None if all checks pass.

Return type:

None

otter.util.fs.check_dir(path: Path) → None[source]

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) – The directory to check.

Raises:

  • SystemExit – If the directory is not writable.

  • SystemExit – If there is an error creating the directory.

Returns:

None if all checks pass.

Return type:

None

otter.util.fs.check_source(path: Path) → None[source]

Check working conditions for a file.

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

Parameters:

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

Raises:

  • SystemExit – If the file does not exist.

  • SystemExit – If the file is not readable.

Returns:

None if all checks pass.

Return type:

None

otter.util.fs.check_destination(path: Path, *, delete: bool = False) → None[source]

Check working conditions for a file and its parent directory.

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) – The path to check. Must be a file.

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

Raises:

  • SystemExit – If the file already exists.

  • SystemExit – If the parent directory is not writable.

  • SystemExit – If there is an error creating the parent directory.

Returns:

None if all checks pass.

Return type:

None

util.logger module

Configures the logger for the application.

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

Get fields from the exception record.

This function extracts the name, function and line number from an exception. 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.

Args:

record_exception: The exception record.

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

Create the log format function.

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

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

Parameters:

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

otter.util.logger.step_logging(step: Step) → Generator[None][source]

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

Parameters:

step (Step) – The step to log messages to.

class otter.util.logger.MessageQueue[source]

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[source]

Put a message in the queue.

flush() → None[source]

Dump the log messages to stdout.

otter.util.logger.init_logger(log_level: str = 'INFO') → None[source]

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.

  • message_queue (MessageQueue) – The message queue.

Module contents

Utility functions and classes.