API reference

datetime utilities

ensure_tzinfo(val: datetime, tz_or_offset: str | int | timedelta | tzinfo | ZoneInfo | timezone = 'UTC', *, is_dst: bool = False) → datetime

Creates timezone aware datetime object for val.

• if val is naive datetime, new value will be created as datetime localized in tz_or_offset timezone

• if val is already timezone aware, it will be converted to tz_or_offset timezone using datetime.datetime.astimezone

Parameters:

• val – Input value for conversion

• tz_or_offset – Anything that timezone_or_offset accepts

• is_dst – used to determine the correct timezone in the ambiguous period at the end of daylight saving time. Use is_dst=None to raise an AmbiguousTimeError for ambiguous times at the end of daylight saving time.

Returns:

Timezone aware datetime object

Raises:

ValueError – When timezone of offset can’t be parsed / determined from tz_or_offset

Note

This tries to provide safe(ish) implementation for handling naive datetime objects, but ultimate solution is to not use naive datetime objects ever/anywhere. Recommendation is to go with pip install pendulum and leave this crap behind to history.

iter_year_month(start: date | datetime, end: date | None = None, *, include_start: bool = False, include_end: bool = False)

Generates range of date(year, month, 1) from start to end.

• when start > end generated range is empty

• when start == end generated range may contain single date object depending on include_start or include_end

Parameters:

• start – begin of range

• end – end of range. If None, assumes, start == end

• include_start – include start in generated range

• include_end – include end in generated range

next_working_day(from_: date | None = None, holidays_calendar={})

Finds next work day from from_ or today.

timezone_or_offset(from_: str | int | timedelta | tzinfo | ZoneInfo | timezone | None) → ZoneInfo | timezone

Given from_ creates datetime.timezone or zoneinfo.ZoneInfo as result.

from_ can be any of following:

• str (ie. “-02:42”, None, “”, “Z”, …) which is ISO8601 offset

• str (ie. “Europe/Zagreb”) which is timezone name

• int (ie. -9000) which is total number of seconds in time offset

• datetime.timedelta

• datetime.tzinfo or something that behaves like it

benchmarking utilities

class Stopwatch

Bases: object

Simple stopwatch that measures duration of block of code in [ms]

Example

>>> import time
>>>
>>> with Stopwatch() as stopwatch:
...    time.sleep(1)
>>> assert stopwatch.duration_ms >= 1000

property duration_ms

file utilities

abspath_if_relative(relative_path: str | Path, relative_to: str | Path)

Creates absolute path from relative, but places it under other path.

Example

>>> abspath_if_relative('foo/bar/baz', relative_to='/tmp')
'/tmp/foo/bar/baz'

file_checksum(file_path: str | Path, hashlib_callable)

Given path of the file and hash function, calculates file digest

move_and_create_dest(src_path: str | Path, dst_dir: str | Path)

Moves src_path to dst_dir directory.

Expects dst_dir to be directory and if it doesn’t exits, tries to create it.

logging utilities

class PrettyFormatter(fmt: str | None = None, datefmt: str | None = None, style: logging._FormatStyle = '%', validate: bool = True, *, defaults: Mapping[str, Any] | None = None, force_single_line: bool = True, colorize: bool = False, log_colors: LogColors | None = None, secondary_log_colors: SecondaryLogColors | None = None, reset: bool = True, stream: IO | None = None)

Logging formatter for pretty logs:

• can optionally colorize output

• can optionally force output to be single line

• reformats logged tracebacks

To colorize output:

• instantiate formatter with colorize = True

• use color placeholders in format string, ie %(red)s text that will be red %(reset)s

Example YAML logging config to use it:

---
version: 1

loggers:
  my_app:
    handlers:
      - console
    level: INFO
    propagate: false

handlers:
  console:
    class: logging.StreamHandler
    filters:
      - request_id
    stream: ext://sys.stdout
    formatter: colored_multiline

formatters:
  colored_multiline:
    (): seveno_pyutil.PrettyFormatter
    force_single_line: false
    colorize: true
    format: >-
      lvl=%(log_color)s%(levelname)s%(reset)s
      ts=%(thin_white)s%(asctime)s%(reset)s
      msg=%(message)s

  colorless_single_line:
    (): seveno_pyutil.PrettyFormatter
    force_single_line: true
    colorize: false
    format: >-
      lvl=%(levelname)s
      ts=%(asctime)s
      msg=%(message)s

Some of available colors are:

[
    'black', 'bold_black', 'thin_black', 'red', 'bold_red', 'thin_red', 'green',
    'bold_green', 'thin_green', 'yellow', 'bold_yellow', 'thin_yellow', 'blue',
    'bold_blue', 'thin_blue', 'purple', 'bold_purple', 'thin_purple', 'cyan',
    'bold_cyan', 'thin_cyan', 'white', 'bold_white', 'thin_white', "..."
]

Others can be retrieved via:

colorlog.ColoredFormatter()._escape_code_map("DEBUG").keys()

or check docs for colorlog.

format(record: LogRecord) → str

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

formatException(ei) → str

Format and return the specified exception information as a string.

This default implementation just uses traceback.print_exception()

class SQLFilter(*args, colorize_queries=False, multiline_queries=False, shorten_logs=True, **kwargs)

Filter for SQLAlchemy SQL loggers. Optionally reformats and colorizes queries.

To use it with Flask and SQLAlchemy:

• configure Flask app

• configure SQLAlchemy engine events (call :meth:register_sqlalchemy_logging_events)

• configure logging

All three steps are presented in example below. After it has been configured, provides following logging placeholders:

placeholder	description
%(sql)s	Formatted SQL statement that was executed
%(sql_duration)s	Formatted duration of SQL execution

Parameters:

• colorize_queries (bool) – Should apply shell coloring escape sequences to formatted SQL?

• multiline_queries (bool) – Should emit SQL as indented, multiline of single line log statements? In development it is usually nice to have it be True. In production environments, multiline logs are pain and should be avoided.

Example:

import logging
from logging.config import dictConfig

import flask
from seveno_pyutil import FlaskSQLStats, SQLFilter

dictConfig(
    {
        "version": 1,
        "disable_existing_loggers": False,
        "formatters": {
            "app": {"format": "lvl=%(levelname)s msg=%(message)s"},
            "app.db": {
                "format": "lvl=%(levelname)s, sqld=%(sql_duration)s sql=%(sql)s msg=%(message)s"
            },
        },
        "filters": {
            "colored_sql": {
                "()": "seveno_pyutil.SQLFilter",
                "colorize_queries": True,
                "multiline_queries": True,
            }
        },
        "handlers": {
            "console": {
                "class": "logging.StreamHandler",
                "level": "DEBUG",
                "formatter": "app",
                "stream": "ext://sys.stdout",
            },
            "console_sqlalchemy": {
                "class": "logging.StreamHandler",
                "level": "DEBUG",
                "formatter": "db",
                "filters": ["colored_sql"],
                "stream": "ext://sys.stdout",
            },
        },
        "loggers": {
            "myapp": {
                "level": "INFO",
                "propagate": False,
                "handlers": ["console"],
            },
            "myapp.db": {
                "level": "DEBUG",
                "propagate": False,
                "handlers": ["console_sqlalchemy"],
            },
        },
    }
)


class RequestLoggingMiddleware:
    def __init__(self, sql_logger_name, app=None):
        self.sql_logger_name = sql_logger_name
        if app:
            self.init_app(app)

    def init_app(self, app: flask.Flask):
        SQLFilter.register_sqlalchemy_logging_events(self.sql_logger_name)

        @app.before_request
        def log_current_request():
            FlaskSQLStats.open()

            @flask.after_this_request
            def log_current_response(response):
                logger.info("Some HTTP request was processed :)")
                FlaskSQLStats.close()
                return response


logger = logging.getLogger(__name__)
app = flask.Flask(__name__)
RequestLoggingMiddleware("myapp.db").init_app(app)

filter(record: LogRecord)

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

class StandardMetadataFilter(name='')

Filter that adds few more attributes to log records.

placeholder	description
%(hostname)s	hostname
%(isotime)s	Local time represented as ISO8601
%(isotime_utc)s	local time converted to UTC and represented as ISO8601 string

filter(record)

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

log_to_console_for(logger_name: str)

Sometimes, usually during development, we want to quickly see output of some particular package logger. This method will configure such logger to spit stuff out to console.

log_to_tmp_file_for(logger_name: str, file_path: str | Path = '/tmp/seveno_pyutil.log')

Quick setup for given logger directing it to /tmp/seveno_pyutil.log This is of course mainly used during development, especially when playing with things in Python console.

silence_logger(logger: Logger)

For given logger, replaces all its handlers with logging.NullHandler.

metaprogramming helpers

class LeafSubclassRetriever(base_class)

Bases: object

http://code.activestate.com/recipes/577858-concrete-class-finder/

value()

all_subclasses(klass)

getval(src: Mapping | object, attr: object, default=None)

Companion of dict.get and getattr which ensures default value even when original method would had returned None

Example:

d1 = {}
d2 = {"foo": None}
d3 = {"foo": "bar"}

d1.get("foo", 42)  # => 42
d2.get("foo", 42)  # => None
d3.get("foo", 42)  # => "bar"

getval(d1, "foo", 42)  # => 42
getval(d2, "foo", 42)  # => 42
getval(d3, "foo", 42)  # => "bar"

# and also

getval({"a": ""}, "a", None) is None  # => True

# and with objects other than dicts

class Foo:
    def __init__(self, foo=None):
        self.foo = foo

class Bar:
    pass

o1 = Bar()
o2 = Foo()
o3 = Foo("bar")

getattr(o1, "foo", 42)  # => 42
getattr(o2, "foo", 42)  # => None
getattr(o3, "foo", 42)  # => "bar"

getval(o1, "foo", 42)  # => 42
getval(o2, "foo", 42)  # => 42
getval(o3, "foo", 42)  # => "bar"

import_string(dotted_path)

Import a dotted module path and return the attribute/class designated by the last name in the path. Raise ImportError if the import failed.

leaf_subclasses(klass)

Returns all leaf subclasses of given klass

os utilities

current_user()

Queries OS for current user username.

current_user_home()

Queries OS for path to current user home directory.

string utilities

is_blank(obj: Any) → bool

True if obj is empty string, None, string that contains only spaces and space like characters, or iterable that contains only these kinds of strings/objects

collections utilities

in_batches(iterable: Iterable[T], of_size: int = 1) → Generator[Iterable[T]]

Generator that yields generator slices of iterable.

Since it is elegant and working flawlessly, it is shameless C/P from https://stackoverflow.com/questions/8991506/iterate-an-iterator-by-chunks-of-n-in-python/8998040#8998040

Warning

Each returned batch should be completely consumed before next batch is yielded. See example below to better understand what that means.

Example:

from seveno_pyutil import in_batches

g = (o for o in range(10))
for batch in in_batches(g, of_size=3):
    print(list(batch))
# [0, 1, 2]
# [3, 4, 5]
# [6, 7, 8]
# [9]

# And this happens if whole batch is not consumed before yielding another one...

g = list(range(10))
for batch in in_batches(g, of_size=3):
    print( [next(batch), next(batch)] )
# [0, 1]
# [2, 3]
# [4, 5]
# [6, 7]
# [8, 9]

error utilities

class ExceptionsAsErrors(errors_store: Mapping | object, subkey=None)

Bases: object

Context manager that swallows exceptions and stores them as structured error dict that can later be added to marshmallow.ValidationError.

It uses add_error_to to update provided errors_store with caught exceptions.

Example:

errors = {}
with ExceptionsAsErrors(errors) as e:
    raise RuntimeError("ZOMG!")
errors == {'_schema': ['ZOMG!']}

errors = {}
with ExceptionsAsErrors(errors, subkey="some_name") as e:
    raise RuntimeError("ZOMG!")
errors == {'some_name': ['ZOMG!']}

add_error_to(errors_store: Mapping | object, error: str | Sequence | object | Mapping | Exception)

Updates error store, merging messages from error

Example:

errors_store = {
    "person": {
        "email": ["is not an email"],
        "name": ["is too long"],
        "date_of_birth": ["is not a date"]
    },
    "job": ["is not from allowed values list"]
}

add_error_to(errors_store, {"person": {"email": "is from illegal domain"}})

errors_store == {
    "person": {
        "email": ["is not an email", "is from illegal domain"],
        "name": ["is too long"],
        "date_of_birth": ["is not a date"]
    },
    "job": ["is not from allowed values list"]
}

But, trying to do this:

add_error_to(errors_store, {"person": "is illegally formed"})

will add/update _schema key because person has child keys and replacing them with ["is illegally formed"] would loose that data. Thus, errors_store will look like this now:

errors_store == {
    "person": {
        "email": ["is not an email", "is from illegal domain"],
        "name": ["is too long"],
        "date_of_birth": ["is not a date"],
        "_schema": ["is illegally formed"]
    },
    "job": ["is not from allowed values list"]
}

Another example of this behavior:

add_error_to(errors_store, {"job": {"title": "can't be blank"}})

will result in:

errors_store == {
    "person": {
        "email": ["is not an email", "is from illegal domain"],
        "name": ["is too long"],
        "date_of_birth": ["is not a date"],
        "_schema": ["is illegally formed"]
    },
    "job": {
        "_schema": ["is not from allowed values list"],
        "title": ["can't be blank"]
    }
}

Also, method is smart enough to correctly update errors store with either one or a sequence of messages. So both of these are valid:

add_error_to(errors_store, {"job": {"title": "is overpaid"}})
add_error_to(errors_store, {"job": {"title": ["is forbiden", "doesn't exist"]}})

errors_store == {
    "person": {
        "email": ["is not an email", "is from illegal domain"],
        "name": ["is too long"],
        "date_of_birth": ["is not a date"],
        "_schema": ["is illegally formed"]
    },
    "job": {
        "_schema": ["is not from allowed values list"],
        "title": [
            "can't be blank", "is overpaid", "is forbiden", "doesn't exist"
        ]
    }
}

Parameters:

errors_store – either a dict into which errors will be added or an object. If object we expect it to have attribute errors and that these are a dict. If no such attribute exists on given object, we will attach our own.