util package

util.errors module

Custom exceptions for the pis package.

exception pis.util.errors.DownloadError(src: str, error: Exception)[source]

Bases: PISError

Raise when an error occurs during a download.

exception pis.util.errors.HelperError(msg: str)[source]

Bases: PISError

Raise when an error occurs in a helper.

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

Bases: PISError

Raise when something is not found.

exception pis.util.errors.PISCriticalError(msg: str)[source]

Bases: PISError

Raise when a critical error occurs.

exception pis.util.errors.PISError[source]

Bases: Exception

Base class for all exceptions in the PIS package.

exception pis.util.errors.PreconditionFailedError[source]

Bases: PISError

Raise when a precondition fails.

exception pis.util.errors.ScratchpadError(sentinel: str | Path)[source]

Bases: PISError

Raise when a key is not found in the scratchpad.

exception pis.util.errors.StepFailedError(step: str = '', func: str = '')[source]

Bases: PISError

Raise when a step fails.

exception pis.util.errors.StorageError(msg: str)[source]

Bases: PISError

Raise when an error occurs in a storage class.

exception pis.util.errors.TaskAbortedError[source]

Bases: PISError

Raise when a task is aborted.

exception pis.util.errors.ValidationError[source]

Bases: PISError

Raise when a validation fails.

util.fs module

File system utilities.

pis.util.fs.absolute_path(path: Path | str) → Path[source]

Get the full path of a file or directory.

The function will try to figure if the path contains the work directory. If not, it will be prepended.

Parameters:: path (Path | str) – The path to the file or directory.
Returns:: The full path of the file or directory.
Return type:: Path

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

Check working conditions for a path.

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 path to check. Must be a directory.

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

pis.util.fs.check_file(path: Path) → None[source]

Check working conditions for a file.

The function will make sure that a file does not exist in the given path.

Warning

The function will delete the file if it already exists.

Parameters:: path (Path) – The path to check. Must be a file.
Raises:: SystemExit – If there is an error deleting the file.
Returns:: None if all checks pass.
Return type:: None

pis.util.fs.check_fs(path: Path) → 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.

Warning

The function will delete the file if it already exists.

Parameters:

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

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.download module

Helper that downloads files from various sources.

class pis.util.download.AbortableStreamWrapper(stream, *, abort: Event)[source]

A wrapper around a stream that can be aborted.

This class wraps a stream and in every read operation, it will raise a pis.util.errors.TaskAbortedError if the abort event is set. This is useful to abort downloads when another task fails.

read(*args, **kwargs) → bytes[source]

Read from the stream and raise TaskAbortedError if the abort event is set.

Returns:: The data read from the stream.
Return type:: bytes
Raises:: TaskAbortedError – If the abort event is set.

class pis.util.download.DownloadHelper[source]

Helper that downloads files from various sources.

download(src: str, dst: Path | str, abort: Event | None = None) → Path[source]: Download a file.

class pis.util.download.Downloader[source]

Base class for downloaders.

download(src: str, dst: Path, *, abort: Event | None = None) → Path[source]

Download a file from a source to a destination.

This method is a no-op and should be overridden by subclasses.

Parameters:

src (str) – The source URL.
dst (Path) – The destination path.
abort (Event | None, optional) – An event that can be set to abort the download, defaults to None

Returns:

The destination path.

Return type:

Path

class pis.util.download.GoogleSheetsDownloader[source]

Bases: Downloader

Downloader for Google Sheets URLs.

download(src: str, dst: Path, *, abort: Event | None = None) → Path[source]: Download a Google Sheet.

class pis.util.download.GoogleStorageDownloader[source]

Bases: Downloader

Downloader for Google Storage URLs.

download(src: str, dst: Path, *, abort: Event | None = None) → Path[source]: Download a file from Google Storage.

class pis.util.download.HttpDownloader[source]

Bases: Downloader

Downloader for HTTP and HTTPS URLs.

download(src: str, dst: Path, *, abort: Event | None = None) → Path[source]: Download a file from an HTTP or HTTPS URL.

pis.util.download.download(src: str, dst: Path | str, *, abort: Event | None = None) → Path[source]: Instantiate a DownloadHelper and download a file.

util.logger module

Configures the logger for the application.

pis.util.logger.get_exception_info(record_exception) → 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.

pis.util.logger.get_format_log(include_task: bool = True) → Callable[[...], str][source]: Get the format for the log messages.

pis.util.logger.init_logger(log_level: str) → None[source]

Initializes the logger.

PIS uses two handlers by default, one for the console and one for the log file.

Args:: log_level (str): The log level to use.

pis.util.logger.task_logging(task: Task)[source]

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

Args:: task (Task): The task to log messages to.
Yields:: None

util.misc module

Misc utility functions.

pis.util.misc.date_str(d: datetime | None = None) → str[source]

Return a string representation of a datetime object.

Parameters:: d (datetime | None, optional) – datetime object to convert to string.
Returns:: string representation of the datetime object.
Return type:: str

pis.util.misc.list_str(a_list: Iterable[Any], dict_values: bool = False) → str[source]

Return a string representation of a list or dictionary.

Parameters:

a_list (Iterable[Any]) – The list or dictionary to convert to a string.
dict_values (bool, optional) – Whether to include the values of a dictionary.

Returns:

A string representation of the list or dictionary.

Return type:

str

pis.util.misc.real_name(t: Task | BaseTaskDefinition) → str[source]

Return the real name of a task.

The naming convention is snake_case for task names.

The real name of a task is the first word of the name. It determines the type of task that will be spawned when the task is run, so the name should exist in the task registry (a task with that name must exist in the tasks module).

Parameters:: t (Task | BaseTaskDefinition) – Task or TaskDefinition object to get the real name of.
Returns:: The real name of the task.
Return type:: str

Module contents

PIS Utilities.