Altertech common Python functions library

Config helpers

Extra mods required: pyyaml, jsonschema

pyaltt2.config.choose_file(fname=None, env=None, choices=[])

Chooise existing file

Returned file path is user-expanded

Parameters:
  • fname – if specified, has top priority and others are not chechked
  • env – if specified and set, has second-top priority and choices are not inspected
  • choices – if env is not set or not specified, choose existing file from the list
Raises:

LookupError – if file doesn’t exists
pyaltt2.config.config_value(env=None, config=None, config_path=None, to_str=False, read_file='r', in_place=False, default=<class 'LookupError'>)

Get config value

Parameters:
  • env – search in system env, if specified, has top priority
  • config – config dict to process
  • config_path – config path (e.g. /some/long/key)
  • to_str – stringify value before returning
  • read_file – if value starts with ‘/’, read it from file with mode (default: ‘r’)
  • in_place – replace value in config dict
  • default – default value
Raises:

LookupError – if value is not found and no default value specified
pyaltt2.config.load_yaml(fname, schema=None)

Load config from YAML/JSON file

Parameters:
  • fname – file name to load
  • schema – JSON schema for validation

Data conversion

pyaltt2.converters.merge_dict(*args, add_keys=True)

Safely merge two dictionaries

Parameters:
  • dct0...n – dicts to merge
  • add_keys – merge dict keys (default: True)
Returns:

merged dict
pyaltt2.converters.mq_topic_match(topic, mask)

Checks if topic matches mqtt-style mask

Parameters:
  • topic – topic (string)
  • mask – mask to check
Returns:

True if matches, False if don’t
pyaltt2.converters.parse_date(val=None, return_timestamp=True, ms=False)

Parse date from string or float/integer

Input date can be either timestamp or date-time string

If input value is integer and greater than 3000, it’s considered as a timestamp, otherwise - as a year

Parameters:
  • val – value to parse
  • return_timestamp – return UNIX timestamp (default) or datetime object
  • ms – parse date from milliseconds
Returns:

UNIX timestamp (float) or datetime object. If input value is None, returns current date/time
pyaltt2.converters.parse_number(val)

Tries to parse number from any value

Valid values are:

  • any float / integer
  • 123.45
  • 123 456.899
  • 123,456.899
  • 123 456,899
  • 123.456,82
Parameters:val – value to parse
Returns:val as-is if val is integer, float or None, otherwise parsed value
Raises:ValueError – if input val can not be parsed
pyaltt2.converters.safe_int(val)

Convert string/float to integer

If input value is integer - return as-is If input value is a hexadecimal (0x00): converts hex to decimal

Parameters:val – value to convert
Raises:ValueError – if input value can not be converted
pyaltt2.converters.val_to_boolean(val)

Convert any value to boolean

Boolean: return as-is

  • Integer: 1 = True, 0 = False
  • Strings (case-insensitive): ‘1’, ‘true’, ‘t’, ‘yes’, ‘on’, ‘y’ = True
  • ‘0’, ‘false’, ‘f’, ‘no’, ‘off’, ‘n’ = False
Parameters:val – value to convert
Returns:boolean converted value, None if val is None
Raises:ValueError – if value can not be converted

Cryptography helper functions

Requires pycryptodomex

class pyaltt2.crypto.Rioja(key, bits=256)

Rioja (ˈrjoxa) is a crypto engine, similar to Fernet, but:

  • implements AES-CBC-HMAC up to AES256 (default)
  • more simple to use
Parameters:
  • key – encryption key
  • bits – key size (128, 192 or 256, default is 256)
decrypt(enc, b64=True)
Parameters:
  • enc – data to decrypt
  • b64 – decode data from base64 (default: True)
Raises:

ValueError – if HMAC auth failed
encrypt(raw, b64=True)
Parameters:
  • raw – bytes to encrypt
  • b64 – encode result in base64 (default: True)
pyaltt2.crypto.decrypt(enc, key, hmac_key=None, key_is_hash=False, b64=True, bits=256)

Decrypt encoded data with AES-CBC

Parameters:
  • enc – data to decrypt
  • key – decryption key
  • key_is_hash – consider decryption key is sha256 hash
  • hmac_key – HMAC key (optional), True or custom key
  • b64 – decode data from base64 (default: True)
  • bits – key size (128, 192 or 256, default is 256)
Raises:

ValueError – if HMAC auth failed

note: if hmac_key is True and key is hash, sha512 sum is required

pyaltt2.crypto.default_public_key = None

default per-project public key for signature verification

pyaltt2.crypto.encrypt(raw, key, hmac_key=None, key_is_hash=False, b64=True, bits=256)

Encrypt bytes with AES-CBC

Parameters:
  • raw – bytes to encrypt
  • key – encryption key
  • hmac_key – HMAC key (optional), True or custom key
  • key_is_hash – consider encryption key is sha256 hash
  • b64 – encode result in base64 (default: True)
  • bits – key size (128, 192 or 256, default is 256)
Returns:

encrypted block + 32-byte HMAC signature (if hmac_key is specified)

note: if hmac_key is True and key is hash, sha512 sum is required

pyaltt2.crypto.gen_random_str(length=32)

Generate random string (letters+numbers)

Parameters:length – string length (default: 32)
pyaltt2.crypto.sign(content, private_key, key_password=None)

Sign content with RSA key :param content: content to sign :param private_key: private RSA (PEM) key :param key_password: key password (optional)

Returns:base64-encoded RSA signature
pyaltt2.crypto.verify_signature(content, signature, public_key=None)

Verify content with RSA signature

If public key is not specified, global per-project key is used

Parameters:
  • content – content to sign
  • signature – base64-encoded RSA signature
  • public_key – public RSA (PEM) key
Returns:

base64-encoded RSA signature

Database functions

Simple thread-safe SQLAlchemy wrapper

after initialization, wrapper object contains additional params:

  • name equal to SQLAlchemy engine name
  • use_lastrowid show last row id be used (sqlite, mysql)
  • use_interval can interval columns be used (not available for sqlite, mysql)
  • parse_db_json should JSON column be parsed after selecting (sqlite, mysql)

__str__ and __repr__ methods are mapped to SQLAlchemy engine.

Extra mods required: sqlalchemy, msgpack (for KVStorage)

class pyaltt2.db.Database(dbconn=None, rq_func=None, **kwargs)

Database wrapper for SQLAlchemy

Parameters:
  • dbconn – database connection string (for SQLite - only file name is required)
  • rq_func – resource loader function (for query method)
  • kwargs – additional engine options (ignored for SQLite)
clone(**kwargs)

Clone database object

Extra kwargs (db, db_lock, g, rq_func) are assigned to object as-is

connect()

Get thread-safe db connection

create(q, *args, **kwargs)

Execute (usually INSERT) query with self.execute and return row id

row id must be in “id” field

execute(*args, _cr=False, **kwargs)

Execute SQL query

Parameters:
  • _cr – check result, raise LookupError if row count is zero
  • other – passed to SQLAlchemy connection as-is
get_engine()

Get DB engine object

list(*args, json_fields=[], **kwargs)

get self.execute result as list of dicts

Parameters:
  • json_fields – decode json fields if required
  • other – passed as-is
lookup(*args, json_fields=[], **kwargs)

Get single db row, use self.execute

Parameters:
  • json_fields – decode json fields if required
  • other – passed as-is
Returns:

single row as a dict
Raises:

LookupError – if nothing found
qcreate(q, *args, **kwargs)

Execute (usually INSERT) query with self.query and return row id

row id must be in “id” field

qlist(*args, json_fields=[], **kwargs)

get self.query result as list of dicts

Parameters:
  • json_fields – decode json fields if required
  • other – passed as-is
qlookup(*args, json_fields=[], **kwargs)

Get single db row, use self.query

Returns:single row as a dict
Raises:LookupError – if nothing found
query(q, qargs=[], qkwargs={}, _create=False, *args, **kwargs)

Execute SQL query by resource

Parameters:
  • q – resource name
  • qkwargs (qargs,) – format query with args/kwargs
  • other – passed as-is

Requires rq_func

class pyaltt2.db.KVStorage(db, table_name='kv')

Simple key-value database storage

Parameters:
  • db – pyaltt2.db.Database
  • table_name – storage table name (default: kv)
cleanup()

Deletes expired objects

delete(key)

Delete object in key-value storage

Parameters:key – object key
Raises:LookupError – object not found
get(key, delete=False)

Get object from key-value storage

Parameters:
  • key – object key
  • delete – delete object after getting
Raises:

LookupError – object not found
put(key=None, value=None, expires=None, override=True)

Put object to key-value storage

If no key specified, random 64-char key is generated

Parameters:
  • key – string key (1-255 chars)
  • value – value to put
  • expires – expiration either in seconds or datetime.timedelta
  • override – replace existing object
Returns:

object key
pyaltt2.db.format_condition(f, kw=None, fields=None, cond=None)
Parameters:
  • f – condition filter (dict name/value)
  • kw – kwargs query dict
  • fields – allowed fields
  • cond – initial condition
Returns:

tuple cond (string), kw (dict). String can be safely used in SQL query, dict should be set as query kwargs

JSON helper functions

JSON processing and output functions

Automatically imports rapidjson if present

pyaltt2.json.dumps(data, pretty=False, unpicklable=False, pickle_opts={}, **kwargs)

Dump to JSON

Parameters:
  • data – data to dump
  • pretty – set indent and sort keys
  • unpicklable – one-way dump for complex objects (requires jsonpickle)
  • pickle_opts – sent to jsonpickle.encode() as-is
  • **kwargs – sent to json.dumps() as-is
pyaltt2.json.jprint(data, colored=True, force_colored=False, file=None)

Pretty print JSON

Parameters:
  • data – data to encode and print
  • colored – colorize output (default: True)
  • force_colored – force colorize, even if stream is not a tty
  • file – output stream (default: sys.stdout)

Resource locking helpers

class pyaltt2.locker.Locker(mod='', timeout=5, relative=True)

Function locker decorator object

Parameters:
  • mod – module to report in logs
  • timeout – lock timeout (default: 5 sec)
  • relative – use thread-relative locking (default: True)
critical()

Function is called when lock can not be obtained

Empty by default, override or monkey-patch

Logging module and functions

Requires neotermcolor

neotermcolor styles:

  • logger:10 - debug log message
  • logger:20 - info log message
  • logger:30 - warning log message
  • logger:40 - error log message
  • logger:50 - critical log message
  • logger:exception - exceptions, printed to stdout (w/o logging)

Keeping records in memory requires neotasker library

class pyaltt2.logs.JSysLogHandler(*args, as_json=False, **kwargs)
emit(record)

Emit a record.

The record is formatted, and then sent to the syslog server. If exception information is present, it is NOT sent to the server.

class pyaltt2.logs.JWatchedFileHandler(*args, as_json=False, **kwargs)
emit(record)

Emit a record.

If underlying file has changed, reopen the file before emitting the record to it.

pyaltt2.logs.append(record=None, rd=None, **kwargs)

Append log record to memory cache

Parameters:
  • record – log record, or
  • rd – log record in dict format
  • **kwargs – passed to handle_append as-is
pyaltt2.logs.clean(**kwargs)

Clean obsolete log records from memory

Usually executed from log cleaner worker (see “start”)

pyaltt2.logs.get(level=0, t=0, n=None, pattern=None)

Get recent log records

Parameters:
  • level – minimal log level
  • t – get entries for the recent t seconds
  • n – max number of log records (default: 100)
pyaltt2.logs.handle_append(rd, **kwargs)

Called after record is appended

Parameters:
  • rd – log record in dict format
  • **kwargs – got from append as-is
pyaltt2.logs.init(**kwargs)

Initialize logger

Parameters:
  • name – software product name
  • host – custom host name
  • log_file – file to log to
  • log_stdout – 0 - do not log, 1 - log, 2 - log auto (if no more log hdlrs)
  • syslog – True for /dev/log, socket path or host[:port]
  • level – log level (default: 20)
  • tracebacks – log tracebacks (default: False)
  • ignore – use “ignore” symbol - memory hdlr ignores records starting with
  • ignore_mods – list of modules to ignore
  • omit_ignore_for_level – omit ignore props for >= level
  • stdout_ignore – use “ignore” symbol in stdout logger as well
  • keep_logmem – keep log records in memory for the specified time (seconds)
  • keep_exceptions – keep number of recent exceptions
  • colorize – colorize stdout if possible
  • formatter – log formatter
  • syslog_formatter – if defined, use custom formatter for syslog
  • log_json – true/false
  • syslog_json – true/false
pyaltt2.logs.log_traceback(display=False, use_ignore=False, force=False, e=None, critical=False)

Log exception traceback

Parameters:
  • display – display traceback instead of logging
  • use_ignore – use ignore symbol for traceback string
  • force – force log, even if tracebacks are disabled
  • e – exception or exc_info to log (optional)
pyaltt2.logs.serialize()

Get dict with internal data

pyaltt2.logs.set_debug(debug=False)

Set debug mode ON/OFF

Parameters:debug – True = ON, False = OFF
pyaltt2.logs.start(loop=None)

Start log cleaner

Requires neotasker module, task supervisor must be started before

Parameters:loop – neotasker async loop to execute cleaner worker in
pyaltt2.logs.stop()

Optional method to stop log cleaner

Language processing

pyaltt2.lp.parse_func_str

Parse function call from string

Returns:tuple fn, args, kwargs
Raises:ValueError – if string is invalid

Resource storage

Similar to setuptools.pkg_resources, but with extended functionality

class pyaltt2.res.ResourceStorage(resource_dir='.', mod=None)

Resource storage loader

Usage example:

from pyaltt.res import ResourceStorage from functools import partial

rs = ResourceStorage(mod=mymod)

rq = partial(rs.get, resource_subdir=’sql’, ext=’sql’)

rq(‘object.select.data’) - will load resource from (will try all variations until file is found):

  • sql/object.select.data.sql
  • sql/object/select.data.sql
  • sql/object/select/data.sql

Resource is loaded once and cached forever.

init resource storage for module

If module is specified, set directory to module_dir/resources

Parameters:
  • resource_dir – resource directory or
  • mod – module name
Raises:

LookupError – if module name is specified, but module is not found
get

Get resource

Loads resource from resource storage directory

Parameters:
  • resource_id – resource id. dots are replaced with “/” automatically until resource is found
  • resource_subdir – resource sub directory
  • ext – resource extension
  • mode – file open mode (default: “r”, use “rb” for binary data)
  • default – return default value if resource is not found
Raises:

LookupError – if resource is not found and no default value provided

Network helper functions

Extra mods required: netaddr

pyaltt2.network.generate_netacl(source, default=[])

Generates IP ACL

Doesn’t work with IPv6

Parameters:
  • source – string with IP/network or iterable
  • default – returns if source is empty
pyaltt2.network.netacl_match(ip, acl)

Check if IP mathches network ACL

Doesn’t work with IPv6

Parameters:
  • ip – IP address to check
  • acl – list of netadd.IPNetwork objects
Returns:

True if ACL matches, False if not
pyaltt2.network.parse_host_port(hp, default_port=0)

Parse host/port from string

Doesn’t work with IPv6

Parameters:
  • hp – host/port string to parse
  • default_port – default port if no port is specified (default: 0)
Returns:

tuple (host, port)