hio.help package

Submodules

hio.help.helping module

hio.help.helping module

class hio.help.helping.NonStringIterable

Bases: object

Allows isinstance check for iterable that is not a string if isinstance(x, NonStringIterable):

class hio.help.helping.NonStringSequence

Bases: object

Allows isinstance check for sequence that is not a string if isinstance(x, NonStringSequence):

hio.help.helping.attributize(genie)

Decorator function:

Python generators do not support adding attributes. Adding support for attributes provides a way to pass information from a WSGI App that returns a generator to a WSGI server via the generator after the WSGI app has already started returning its body. The hio.http.Server WSGI server looks for the attributes ._status and ._headers and substitutes these if present. This allows a streaming WSGI App body iterator to later modify the headers and status taht will be returned before the body iterator began iterating. This is useful for web hooks or backend requests that are serviced by an async coroutine based WSGI app so that they may leverage the streaming support of standard WSGI but use a the coroutine based hio.http.Server as an async WSGI server.

This decorator takes a Duck Typing approach to decorating a generator function or method that returns a new function type instance that when called will return a generator like object that supports attributes. the new wrapped object acts like a generator but with attributes.

Parameters:

genie (generator function, generator method) – is either a generator function that returns a generator object a generator method that returns a generator object

If genie is a generator function then a reference to its wrapper

is injected as the first positional argument to the orginal generator function. The convention is to use the parameter ‘me’ to refer to the injected reference to the wrapper.

If genie is a generator method, that is, its first parameter is ‘self’

then a reference to its wrapper is injected as the second positional argument to the original generator method. the convention is to use the parameter ‘me’ to refer to the injected reference to the wrapper so as not to collide with the ‘self’ instance reference.

When wrapped the new type is AttributiveGenerator

Usage:

# decorated generator function
@attributize
def bar(me, req=None, rep=None):
    me._status = 400  # or copy from rep.status
    me._headers = odict(example="Hi")  # or copy from rep.headers
    yield b""
    yield b""
    yield b"Hello There"
    return b"Goodbye"

gen = bar()
msg = next(gen)  # attributes set after first next
gen._status
gen._headers

# decorated generator method
class R:
    @attributize
    def bar(self, me, req=None, rep=None):
        self.name = "Peter"
        me._status = 400  # or copy from rep.status
        me._headers = odict(example="Hi")  # or copy from rep.headers
        yield b""
        yield b""
        yield b"Hello There " + self.name.encode()
        return b"Goodbye"

r = R()
gen = r.bar()
msg = next(gen)   # attributes set after first next
gen._status
gen._headers

# use as function wrapper directly instead of as decorator
def gf(me, x):  # convention injected reference to attributed wrapper is 'me'
    me.x = 5
    me.y = 'a'
    cnt = 0
    while cnt < x:
        yield cnt
        cnt += 1

agf = attributize(gf)
ag = agf(3)
# body of gf is not run until first next call
assert isIterator(ag)
assert not hasattr(ag, 'x')
assert not hasattr(ag, 'y')
n = next(ag)  # first run here which sets up attributes
assert n == 0
assert hasattr(ag, 'x')
assert hasattr(ag, 'y')
assert ag.x == 5
assert ag.y =='a'
n = next(ag)
assert n == 1

Adding attributes to this injected reference makes them available as attributes of the resultant wrapper.

The HTTP WSGI server at hio.core.http.serving.Server uses an instance of hio.core.http.serving.Responder to generate the response for each WSGI request. The Responder instance checks its WSGI app generator for existence of the attributes ._status and ._headers. If so then it overrides its default response status with ._status and updates its default response headers with the headers in ._header. This allows a backend (webhook) to conveniently influence the response status and headers. The response body is returned by the generator itself.

Background: Unlike Python functions, Python generators do not support custom attributes and the generator locals dict at .gi_frame.f_locals dissappears once the generator is complete so its inconvenient.

Fixed attributes of generator objects:

['.__next__', '__iter__', 'close', 'gi_code', 'gi_frame', 'gi_running',
'gi_yieldfrom', 'send', 'throw']
hio.help.helping.b64ToInt(s)

Converts Base64 to int :returns: conversion s (str | bytes) to int :rtype: i (int)

Parameters:

s (str | bytes) – to be converted

hio.help.helping.codeB2ToB64(b, l)

Converts first l sextets of base2 b bytes to base64 str and returns

Returns::
code (str): conversion (encode) of l Base2 sextets from front of b

to Base64 chars.

One char for each of l sextets from front (left) of b. This is useful for encoding as code characters, sextets from the front of a Base2 bytes (byte string). Must provide l because of ambiguity between l=3 and l=4. Both require 3 bytes in b. Trailing pad bits are removed so returned sextets as characters are right aligned .

Parameters::

b (bytes | str): target from which to nab sextets l (int): number of sextets to convert from front of b

hio.help.helping.codeB64ToB2(s)

Convert Base64 chars in s to B2 bytes

Returns:

conversion (decode) of s Base64 chars to Base2 bytes. Where the number of total bytes returned is equal to the minimun number of chars (octet) sufficient to hold the total converted concatenated chars from s, with one sextet per each Base64 char of s. Assumes no pad chars in s.

Return type:

bs (bytes)

Sextets are left aligned with pad bits in last (rightmost) byte to support mid padding of code portion with respect to rest of primitive. This is useful for decoding as bytes, code characters from the front of a Base64 encoded string of characters.

Parameters:

s (str | bytes) – Base64 str or bytes to convert

hio.help.helping.copyfunc(f, name=None)

Copy a function in detail. To change name of func provide name parameter

functools.update_wrapper assigns and updates attributes in WRAPPER_ASSIGNMENTS (__module__, __name__, __qualname__, __doc__, __annotations__) and WRAPPER_UPDATES (__dict__).

Based on: https://stackoverflow.com/questions/6527633/how-can-i-make-a-deepcopy-of-a-function-in-python https://stackoverflow.com/questions/13503079/how-to-create-a-copy-of-a-python-function

hio.help.helping.dump(data, path)

Serialize data dict and write to file given by path where serialization is given by path’s extension of either JSON, MsgPack, or CBOR for extension .json, .mgpk, or .cbor respectively

hio.help.helping.intToB64(i, l=1)

Converts int i to at least l Base64 chars as str.

Returns Base64 conversion of i of minimum length l. If more than l chars are needed to represent i in Base64 then the string is lengthened appropriately. When fewer than l chars are needed, the string is left padded with “A”. l is the minimum number of Base64 digits, where Base64 0 is “A”.

hio.help.helping.intToB64b(i, l=1)

Converts int i to at least l Base64 chars as bytes.

Returns:

Base64 conversion of i of length minimum l. If more than

l bytes are needed to represent i in Base64 then returned bytes is extended appropriately. When less then l bytes is needed then returned bytes is prepadded with b’A’ bytes.

Return type:

b64 (bytes)

Parameters:

  • i (int) – to be converted

  • l (int) – min number of b64 digits. When empty these are left padded with Base64 0 == b’A’ digits. The length of return bytes is extended to accommodate full Base64 encoding of i regardless of l.

hio.help.helping.isIterator(obj)

Returns True if obj is an iterator object, that is,

has an __iter__ method has a __next__ method .__iter__ is callable and returns obj

Otherwise returns False

hio.help.helping.isNonStringIterable(obj)

Returns: (bool) True if obj is non-string iterable, False otherwise

Another way that is less future proof return (hasattr(x, ‘__iter__’) and not isinstance(x, (str, bytes)))

hio.help.helping.isNonStringSequence(obj)

Returns: (bool) True if obj is non-string sequence, False otherwise

hio.help.helping.isign(i)

Integer sign function :returns: 1 if i > 0, -1 if i < 0, 0 otherwise :rtype: (int)

hio.help.helping.just(n, seq, default=None)

Returns a generator of just the first n elements of seq and substitutes default (None) for any missing elements. This guarantees that a generator of exactly n elements is returned. This is to enable unpacking into n varaibles

Example:

x = (1, 2, 3, 4)
tuple(just(3, x))
(1, 2, 3)
x = (1, 2, 3)
tuple(just(3, x))
(1, 2, 3)
x = (1, 2)
tuple(just(3, x))
(1, 2, None)
x = (1, )
tuple(just(3, x))
(1, None, None)
x = ()
tuple(just(3, x))
(None, None, None)
hio.help.helping.load(path)

Return data read from file path as dict file may be either json, msgpack, or cbor given by extension .json, .mgpk, or .cbor respectively Otherwise raise IOError

hio.help.helping.nabSextets(b, l)

Nab l sextets from front of b.

Returns bytes with the first l sextets from the left of b. Length is the minimum needed to hold all l sextets. The last byte is right padded with zeros to be compatible with mid-padded codes on front of primitives. b is bytes or str, and l is the number of sextets to nab.

hio.help.helping.nonStringIterable(obj)

Returns: (bool) True if obj is non-string iterable, False otherwise

Another way that is less future proof return (hasattr(x, ‘__iter__’) and not isinstance(x, (str, bytes)))

hio.help.helping.nonStringSequence(obj)

Returns: (bool) True if obj is non-string sequence, False otherwise

hio.help.helping.ocfn(path, mode='r+', perm=384)

Atomically open or create file from filepath.

If file already exists, open with mode. Otherwise create a new file using write/update mode (binary if “b” in mode). Returns file object. perm uses mode bits like stat.S_IRUSR | stat.S_IWUSR (0o600).

hio.help.helping.repack(n, seq, default=None)

Repacks seq into a generator of len n and returns the generator. The purpose is to enable unpacking into n variables. The first n-1 elements of seq are returned as the first n-1 elements of the generator and any remaining elements are returned in a tuple as the last element of the generator default (None) is substituted for missing elements when len(seq) < n

Example:

x = (1, 2, 3, 4)
tuple(repack(3, x))
(1, 2, (3, 4))

x = (1, 2, 3)
tuple(repack(3, x))
(1, 2, (3,))

x = (1, 2)
tuple(repack(3, x))
(1, 2, ())

x = (1, )
tuple(repack(3, x))
(1, None, ())

x = ()
tuple(repack(3, x))
(None, None, ())
hio.help.helping.sceil(r)

Symmetric ceiling function Returns the symmetric ceiling of r away from zero.

Because int() provides a symmetric floor towards zero, increment int(r) by 1 when r - int(r) > 0 (r positive), -1 when r - int(r) < 0 (r negative), and 0 when r - int(r) == 0 (r integral already).

hio.help.ogling module

hio.help.ogling module

Provides python stdlib logging module support

class hio.help.ogling.Ogler(name='main', level=40, temp=False, prefix=None, headDirPath=None, reopen=False, clear=False, consoled=True, syslogged=True, filed=True, when='H', interval=1, count=48)

Bases: object

Olger instances provide loggers as global logging facility Only need one Ogler per application Uses python stdlib logging module, logging.getLogger(name). Multiple calls to .getLogger() with the same name will always return a reference to the same Logger object.

name

usage specific component used in file name

Type:

str

level

logging severity level

Type:

int

temp

True means use /tmp directory

Type:

bool

prefix

application specific path prefix and formating

Type:

str

headDirPath

head of path

Type:

str

dirPath

full directory path

Type:

str

path

full file path

Type:

str

opened

True means file is opened, False not opened

Type:

bool

consoled

True means log to console (stderr), False do not

Type:

bool

syslogged

True means log to syslog, False do not

Type:

bool

filed

True means log to rotating file at .path, False do not

Type:

bool

when

interval type for timed rotating file handler

Type:

str

interval

length of interval of type when

Type:

int

count

backup count number of backups to keep

Type:

int

AltHeadDirPath = '/home/docs'
HeadDirPath = '/usr/local/var'
Prefix = 'hio'
TailDirPath = 'logs'
TempHeadDir = '/tmp'
TempPrefix = 'test_'
TempSuffix = '_temp'
__init__(name='main', level=40, temp=False, prefix=None, headDirPath=None, reopen=False, clear=False, consoled=True, syslogged=True, filed=True, when='H', interval=1, count=48)

Init Loggery factory instance

Parameters:

  • name (str) – application specific log file name

  • level (int) –

    logging level from logging. Higher is more restrictive. This sets the minimum level of the baseLogger and failLogger relative to the global level. Logs will output if action level is at or above set level.

    Level Numeric value CRITICAL 50 ERROR 40 WARNING 30 INFO 20 DEBUG 10 NOTSET 0

  • temp (bool) – True means use /tmp directory and clear on close False means use headDirpath

  • prefix (str) – application specific path prefix and logging template

  • headDirPath (str) – custom headDirPath for log file

  • reopen (str)

  • clear (bool) – True means clear .dirPath when closing in reopen

  • consoled (bool) – True means log to console (stderr), False do not

  • syslogged (bool) – True means log to syslog, False do not

  • filed (bool) – True means log to rotating file in .dirPath, False do not

clearDirPath()

Remove logfile directory at .dirPath

close(clear=False)

Set .opened to False and remove directory at .path :param clear is boolean: :param True means clear directory:

getLogger(name='hio.help.ogling', level=None)

Returns Basic Logger default is to name logger after module

reopen(name=None, temp=None, headDirPath=None, clear=False)

Create file handler log. Use or Create if not preexistent, directory path .dirPath for file .path First closes .path if already opened. If clear is True then also clears .path before reopening

Parameters:

  • name (name is optional) –

    if None or unchanged then ignore otherwise recreate path

    When recreating path, If not provided use .name

  • boolean (temp is optional) –

    If None ignore Otherwise

    Assign to .temp If True then open temporary directory, If False then open persistent directory

  • database (headDirPath is optional str head directory pathname of main) –

    if None or unchanged then ignore otherwise recreate path

    When recreating path, If not provided use default .HeadDirpath

  • closing (clear is Boolean True means clear .path when)

resetLevel(name='hio.help.ogling', level=None, globally=False)

Resets the level of preexisting loggers to level. If level is None then use .level

hio.help.ogling.initOgler(level=50, **kwa)

Initialize the ogler global instance once.

Usage: assign ogler = hio.help.ogling.initOgler() at module top level. Critical is most severe to restrict logging by default. force is Boolean True to reinit even if global ogler is not None; level is default logging level. Call in package .__init__ to ensure that global ogler is defined by default. Users may reset level and reopen log file before calling ogler.getLoggers().

hio.help.ogling.openOgler(cls=None, name='test', temp=True, **kwa)

Context manager wrapper Ogler instances. Defaults to temporary file logs. Context ‘with’ statements call .close on exit of ‘with’ block

Parameters:

  • instance (cls is Class instance of subclass)

  • oglers (name is str name of ogler instance for filename so can have multiple) – at different paths thar each use different log file directories

  • Boolean (temp is) – Otherwise open in persistent directory, do not clear on close

  • directory (True means open in temporary) – Otherwise open in persistent directory, do not clear on close

  • close (clear on) – Otherwise open in persistent directory, do not clear on close

Usage:

with openOgler(name=”bob”) as ogler:

logger = ogler.getLogger() ….

with openOgler(name=”eve”, cls=SubclassedOgler)

hio.help.timing module

hio.help.timing module

class hio.help.timing.AsyncTimer(duration=0.0, start=None, **kwa)

Bases: Timer

Class to manage real elaspsed time using asyncio event loop time. Namely asyncio.get_event_loop().time()

._start is start tyme in seconds
._stop  is stop tyme in seconds
Properties:

.duration is float time duration in seconds of timer from ._start to ._stop .elaspsed is float time elasped in seconds since ._start .remaining is float time remaining in seconds until ._stop .expired is boolean, True if expired, False otherwise, i.e. time >= ._stop

- start

start timer at current time

- restart

restart timer at last ._stop so no time lost

__init__(duration=0.0, start=None, **kwa)

Initialization method for instance. :param duration is float duration of timer in seconds: :type duration is float duration of timer in seconds: fractional :param start is float optional start time in seconds allows starting before: or after current time

property duration

duration property getter, .duration = ._stop - ._start .duration is float duration tyme

property elapsed

elapsed time property getter, Returns elapsed time in seconds (fractional) since ._start.

property expired

expired property

Returns True if timer has expired, False otherwise. time.time() >= ._stop,

property remaining

remaining time property getter, Returns remaining time in seconds (fractional) before ._stop.

restart(duration=None)

Lossless restart of Timer at start = ._stop for duration if provided, Otherwise current duration. No time lost. Useful to extend Timer so no time lost

start(duration=None, start=None)

Starts Timer of duration secs at start time start secs. If duration not provided then uses current duration If start not provided then starts at current time.time()

class hio.help.timing.MonoTimer(duration=0.0, start=None, retro=True)

Bases: Timer

Class to manage real elaspsed time using time module but with monotonically increating time guarantee in spite of system time being retrograded.

If the system clock is retrograded (moved back in time) while the timer is running then time.time() could move to before the start time. MonoTimer detects this retrograde and if retro is True then retrogrades the start and stop times back Otherwise it raises a TimerRetroError. MonoTimer is not able to detect a prograded clock (moved forward in time)

._start is start time in seconds
._stop  is stop time in seconds
._last is last measured time in seconds with retrograde handling
.retro is boolean If True retrograde ._start and ._stop when time is retrograded.
Properties:

.duration is float time duration in seconds of timer from ._start to ._stop .elaspsed is float time elasped in seconds since ._start .remaining is float time remaining in seconds until ._stop .expired is boolean True if expired, False otherwise, i.e. time >= ._stop .latest is float latest measured time in seconds with retrograte handling

- start

start timer at current time and return start time

- restart

restart timer at last ._stop with no time lost, returns start time

__init__(duration=0.0, start=None, retro=True)

Initialization method for instance. duration is in seconds (fractional). start is an optional start time in seconds allowing start before or after current time. retro is boolean. True means automatically shift timer whenever a retrograded clock is detected, otherwise raise RetroTimerError.

property elapsed

elapsed time property getter, Returns elapsed time in seconds (fractional) since ._start.

property expired

Returns True if timer has expired, False otherwise. .latest >= ._stop,

property latest

latest measured time property getter, Returns latest measured time in seconds adjusted for retrograded system time.

property remaining

remaining time property getter, Returns remaining time in seconds (fractional) before ._stop.

exception hio.help.timing.RetroTimerError

Bases: TimerError

Error due to real time being retrograded before start time of timer. Usage: raise RetroTimerError(“error message”).

class hio.help.timing.Timer(duration=0.0, start=None, **kwa)

Bases: Mixin

Class to manage real elaspsed time using time module.

Attribute notes:

  • ._start is start tyme in seconds

  • ._stop is stop tyme in seconds

Property notes:

  • .duration is float time duration in seconds of timer from ._start to ._stop

  • .elaspsed is float time elasped in seconds since ._start

  • .remaining is float time remaining in seconds until ._stop

  • .expired is boolean, True if expired, False otherwise, i.e. time >= ._stop

Method notes:

  • start() starts timer at current time

  • restart() restarts timer at last ._stop so no time lost

__init__(duration=0.0, start=None, **kwa)

Initialization method for instance. :param duration is float duration of timer in seconds: :type duration is float duration of timer in seconds: fractional :param start is float optional start time in seconds allows starting before: or after current time

property duration

duration property getter, .duration = ._stop - ._start .duration is float duration tyme

property elapsed

elapsed time property getter, Returns elapsed time in seconds (fractional) since ._start.

property expired

expired property

Returns True if timer has expired, False otherwise. time.time() >= ._stop,

property remaining

remaining time property getter, Returns remaining time in seconds (fractional) before ._stop.

restart(duration=None)

Lossless restart of Timer at start = ._stop for duration if provided, Otherwise current duration. No time lost. Useful to extend Timer so no time lost

start(duration=None, start=None)

Starts Timer of duration secs at start time start secs. If duration not provided then uses current duration If start not provided then starts at current time.time()

exception hio.help.timing.TimerError

Bases: HioError

Generic Timer Errors. Usage: raise TimerError(“error message”).

hio.help.timing.fromIso8601(dts)

Returns datetime object from RFC-3339 profile of ISO 8601 format str or bytes. Converts dts from ISO 8601 format to datetime object

YYYY-MM-DDTHH:MM:SS.ffffff+HH:MM[:SS[.ffffff]] .strftime(‘%Y-%m-%dT%H:%M:%S.%f%z’) ‘2020-08-22T17:50:09.988921+00:00’ Assumes TZ aware For nanosecond use instead attotime or datatime64 in pandas or numpy

hio.help.timing.nowIso8601()

Returns timezone aware datetime of current UTC time in RFC-3339 profile of ISO 8601 format. Uses now(timezone.utc)

YYYY-MM-DDTHH:MM:SS.ffffff+HH:MM[:SS[.ffffff]] .strftime(‘%Y-%m-%dT%H:%M:%S.%f%z’) ‘2020-08-22T17:50:09.988921+00:00’ Assumes TZ aware For nanosecond use instead attotime or datatime64 in pandas or numpy

hio.help.timing.nowUTC()

Returns timezone aware datetime of current UTC time Convenience function that allows monkeypatching in tests to mock time

hio.help.timing.toIso8601(dt=None)

Returns str datetime dt in RFC-3339 profile of ISO 8601 format. Converts datetime object dt to ISO 8601 formt If dt is missing use now(timezone.utc)

YYYY-MM-DDTHH:MM:SS.ffffff+HH:MM[:SS[.ffffff]] .strftime(‘%Y-%m-%dT%H:%M:%S.%f%z’) ‘2020-08-22T17:50:09.988921+00:00’ Assumes TZ aware For nanosecond use instead attotime or datatime64 in pandas or numpy

Module contents

hio.help package