Skip to content

utils

utils

BASE_FORMAT_STRING module-attribute

BASE_FORMAT_STRING = "[$BOLD%(asctime)s$RESET][%(levelname)-18s]  %(message)s ($BOLD%(filename)s$RESET:%(lineno)d)"

COLOUR_FORMAT_STRING module-attribute

COLOUR_FORMAT_STRING = replace('$BOLD', BOLD_SEQ)

FORMAT_STRING module-attribute

FORMAT_STRING = replace('$BOLD', '')

ColouredFormatter

ColouredFormatter(msg, use_colour=True)

Bases: Formatter

Color logger -- only used for top-level scripts...

Source code in src/itkdb/utils.py
def __init__(self, msg, use_colour=True):
    logging.Formatter.__init__(self, msg)
    self.use_colour = use_colour

LEVELS class-attribute

LEVELS: dict[str, str] = {
    "WARNING": YELLOW,
    "INFO": WHITE,
    "DEBUG": BLUE,
    "CRITICAL": YELLOW,
    "ERROR": RED,
}

use_colour instance-attribute

use_colour = use_colour

format

format(record)
Source code in src/itkdb/utils.py
def format(self, record):
    levelname = record.levelname
    if self.use_colour and levelname in self.LEVELS:
        levelname_colour = self.LEVELS[levelname] + levelname + Colours.RESET_SEQ
        record.levelname = levelname_colour
    return logging.Formatter.format(self, record)

ColouredLogger

ColouredLogger(name, use_colour=True)

Bases: Logger

Custom logging class for coloured output. Only used for top-level scripts.

Source code in src/itkdb/utils.py
def __init__(self, name, use_colour=True):
    logging.Logger.__init__(self, name, logging.WARNING)
    colour_formatter = ColouredFormatter(
        COLOUR_FORMAT_STRING, use_colour=use_colour
    )
    console = logging.StreamHandler()
    console.setFormatter(colour_formatter)
    self.addHandler(console)

Colours

For additional colours, see: https://stackoverflow.com/questions/15580303/python-output-complex-line-with-floats-coloured-by-value

BOLD_SEQ class-attribute instance-attribute

BOLD_SEQ = '\x1b[1m'

RESET_SEQ class-attribute instance-attribute

RESET_SEQ = '\x1b[0m'

UNDERLINE_SEQ class-attribute instance-attribute

UNDERLINE_SEQ = '\x1b[4m'

get_file_components

get_file_components(files)

Parse the files keyword argument from a requests object which can be one of the following:

- {key: filepointer}
- {key: (filename, filepointer)}
- {key: (filename, filepointer, mimetype)}
- {key: (filename, filepointer, mimetype, additional headers)}

Only a single key is supported, so check if there's exactly one key or raise ValueError.

Source code in src/itkdb/utils.py
def get_file_components(files):
    """
    Parse the files keyword argument from a requests object which can be one of the following:

        - {key: filepointer}
        - {key: (filename, filepointer)}
        - {key: (filename, filepointer, mimetype)}
        - {key: (filename, filepointer, mimetype, additional headers)}

    Only a single key is supported, so check if there's exactly one key or raise ValueError.
    """
    files_list = requests.utils.to_key_val_list(files)

    if len(files_list) != 1:
        msg = f"You're creating a single attachment but you specified {len(files_list)} files."
        raise ValueError(msg)
    # now need to handle the user scenarios
    key, value = files_list[0]

    # identify:
    #   filename, filepointer, mimetype, additional headers
    fname, fpointer, ftype, fheaders = (None,) * 4
    if isinstance(value, (tuple, list)):
        if len(value) == 2:
            fname, fpointer = value
        elif len(value) == 3:
            fname, fpointer, ftype = value
        else:
            fname, fpointer, ftype, fheaders = value
    else:
        fname = requests.utils.guess_filename(value) or key
        fpointer = value

    # handle cases where we didn't get ftype/fheaders specified by user
    ftype = ftype or get_mimetype(fname, fpointer)
    fheaders = fheaders or {}

    return fname, fpointer, ftype, fheaders

get_filesize

get_filesize(fname, fpointer) -> int

Size of file in bytes.

Examples:

>>> fname = itkdb.data / "1x1.jpg"
>>> with fname.open("rb") as fpointer:
...     get_filesize(fname, fpointer)
...
125
Source code in src/itkdb/utils.py
def get_filesize(fname, fpointer) -> int:
    """
    Size of file in bytes.

    Examples:

        >>> fname = itkdb.data / "1x1.jpg"
        >>> with fname.open("rb") as fpointer:
        ...     get_filesize(fname, fpointer)
        ...
        125
    """
    try:
        size = Path(fname).stat().st_size
    except FileNotFoundError:
        fpointer.seek(0, os.SEEK_END)
        size = fpointer.tell()
        fpointer.seek(0)

    return size

get_mimetype

get_mimetype(fname, fpointer) -> str

Mimetype of file.

Examples:

>>> fname = itkdb.data / "1x1.jpg"
>>> with fname.open("rb") as fpointer:
...     get_mimetype(fname, fpointer)
...
'image/jpeg'
Source code in src/itkdb/utils.py
def get_mimetype(fname, fpointer) -> str:
    """
    Mimetype of file.

    Examples:

        >>> fname = itkdb.data / "1x1.jpg"
        >>> with fname.open("rb") as fpointer:
        ...     get_mimetype(fname, fpointer)
        ...
        'image/jpeg'

    """
    try:
        ftype = magic.from_file(str(fname), mime=True)
    except FileNotFoundError:
        ftype = magic.from_buffer(fpointer.read(2048), mime=True)
        fpointer.seek(0)
    return ftype

is_eos_uploadable

is_eos_uploadable(fname, fpointer) -> bool

Decision on whether a file should be uploaded to EOS or not.

Examples:

>>> fname = itkdb.data / "1x1.sh"
>>> with fname.open("rb") as fpointer:
...     is_eos_uploadable(fname, fpointer)
...
False

>>> fname = itkdb.data / "1x1.jpg"
>>> with fname.open("rb") as fpointer:
...     is_eos_uploadable(fname, fpointer)
...
True

Added in version 0.4.0

Source code in src/itkdb/utils.py
def is_eos_uploadable(fname, fpointer) -> bool:
    """
    Decision on whether a file should be uploaded to EOS or not.

    Examples:

        >>> fname = itkdb.data / "1x1.sh"
        >>> with fname.open("rb") as fpointer:
        ...     is_eos_uploadable(fname, fpointer)
        ...
        False

        >>> fname = itkdb.data / "1x1.jpg"
        >>> with fname.open("rb") as fpointer:
        ...     is_eos_uploadable(fname, fpointer)
        ...
        True

    !!! note "Added in version 0.4.0"

    """
    return (
        is_image(fname, fpointer)
        or is_root(fname, fpointer)
        or is_largefile(fname, fpointer)
    )

is_image

is_image(fname, fpointer) -> bool

Whether file is an image or not.

Examples:

>>> fname = itkdb.data / "1x1.sh"
>>> with fname.open("rb") as fpointer:
...     is_image(fname, fpointer)
...
False

>>> fname = itkdb.data / "1x1.jpg"
>>> with fname.open("rb") as fpointer:
...     is_image(fname, fpointer)
...
True
Source code in src/itkdb/utils.py
def is_image(fname, fpointer) -> bool:
    """
    Whether file is an image or not.

    Examples:

        >>> fname = itkdb.data / "1x1.sh"
        >>> with fname.open("rb") as fpointer:
        ...     is_image(fname, fpointer)
        ...
        False

        >>> fname = itkdb.data / "1x1.jpg"
        >>> with fname.open("rb") as fpointer:
        ...     is_image(fname, fpointer)
        ...
        True

    """
    return get_mimetype(fname, fpointer).startswith("image/")

is_largefile

is_largefile(fname, fpointer, limit=64 * 1000) -> bool

Whether file is a large file or not.

Examples:

>>> fname = itkdb.data / "1x1.jpg"
>>> with fname.open("rb") as fpointer:
...     is_largefile(fname, fpointer)
...
False

>>> fname = itkdb.data / "1x1.jpg"
>>> with fname.open("rb") as fpointer:
...     is_largefile(fname, fpointer, limit=100) # (1)!
...
True
  1. Set the lower limit for filesize to 100 bytes.
Source code in src/itkdb/utils.py
def is_largefile(fname, fpointer, limit=64 * 1000) -> bool:
    """
    Whether file is a large file or not.

    Examples:

        >>> fname = itkdb.data / "1x1.jpg"
        >>> with fname.open("rb") as fpointer:
        ...     is_largefile(fname, fpointer)
        ...
        False

        >>> fname = itkdb.data / "1x1.jpg"
        >>> with fname.open("rb") as fpointer:
        ...     is_largefile(fname, fpointer, limit=100) # (1)!
        ...
        True

    1. :material-file: Set the lower `limit` for filesize to 100 bytes.

    """
    return get_filesize(fname, fpointer) > limit

is_root

is_root(_, fpointer) -> bool

Whether file is a ROOT file or not.

Examples:

>>> fname = itkdb.data / "1x1.jpg"
>>> with fname.open("rb") as fpointer:
...     is_root(fname, fpointer)
...
False

>>> fname = itkdb.data / "tiny.root"
>>> with fname.open("rb") as fpointer:
...     is_root(fname, fpointer)
...
True

Added in version 0.4.0

Source code in src/itkdb/utils.py
def is_root(_, fpointer) -> bool:
    """
    Whether file is a ROOT file or not.

    Examples:

        >>> fname = itkdb.data / "1x1.jpg"
        >>> with fname.open("rb") as fpointer:
        ...     is_root(fname, fpointer)
        ...
        False

        >>> fname = itkdb.data / "tiny.root"
        >>> with fname.open("rb") as fpointer:
        ...     is_root(fname, fpointer)
        ...
        True

    !!! note "Added in version 0.4.0"

    """
    data = fpointer.read(4)
    fpointer.seek(0)
    return data == b"root"

merge_url_query_params

merge_url_query_params(
    url: str, additional_params: dict
) -> str

Merge a url with the specified query parameters.

Source code in src/itkdb/utils.py
def merge_url_query_params(url: str, additional_params: dict) -> str:
    """
    Merge a url with the specified query parameters.
    """
    url_components = urlparse(url)
    original_params = parse_qs(url_components.query)
    # Before Python 3.5 you could update original_params with
    # additional_params, but here all the variables are immutable.
    merged_params = {**original_params, **additional_params}
    updated_query = urlencode(merged_params, doseq=True)
    # _replace() is how you can create a new NamedTuple with a changed field
    return url_components._replace(query=updated_query).geturl()

pretty_print

pretty_print(req)

Pretty-print requests.Request object.

Source code in src/itkdb/utils.py
def pretty_print(req):
    """
    Pretty-print requests.Request object.
    """
    request = req.prepare() if isinstance(req, requests.Request) else req
    headers = "\r\n".join(
        f"{k}: {v if k != 'Authorization' else 'Bearer <TOKEN>'}"
        for k, v in request.headers.items()
    )
    try:
        body = (
            ""
            if request.body is None
            else (
                request.body.decode()
                if isinstance(request.body, bytes)
                else request.body
            )
        )
    except UnicodeDecodeError:
        body = "<Decoding error>"

    return f"Host: {urlparse(request.url).netloc}\r\n{request.method} {request.path_url} HTTP/1.1\r\n{headers}\r\n\r\n{body}"

sizeof_fmt

sizeof_fmt(num, suffix='B') -> str

Return human-readable of bytes.

Examples:

>>> sizeof_fmt(2**8)
'256.0B'
>>> sizeof_fmt(2**16) # (1)!
'64.0KiB'
>>> sizeof_fmt(2**32)
'4.0GiB'
  1. Don't forget that the KiB suffix indicates 1024 bytes, rather than the metric 1000 bytes.

Added in version 0.4.0

Source code in src/itkdb/utils.py
def sizeof_fmt(num, suffix="B") -> str:
    """
    Return human-readable of bytes.

    Examples:

        >>> sizeof_fmt(2**8)
        '256.0B'
        >>> sizeof_fmt(2**16) # (1)!
        '64.0KiB'
        >>> sizeof_fmt(2**32)
        '4.0GiB'

    1. :material-alert: Don't forget that the `KiB` suffix indicates 1024 bytes, rather than the metric 1000 bytes.

    !!! note "Added in version 0.4.0"

    """
    for unit in ["", "Ki", "Mi", "Gi", "Ti", "Pi", "Ei", "Zi"]:
        if abs(num) < 1024.0:
            return f"{num:3.1f}{unit}{suffix}"
        num /= 1024.0
    return f"{num:.1f}Yi{suffix}"