Python utils and configuration

Configuration

Constants and variables

Config files can be set via CONFIG environ var or passed directly to Config(). The config string can contain full paths to files or directories separated by ;.

Configuration from directory

If a directory is specified, all toml-files from it are taken in alphabetical order. Files with logging in name are ignored by Config.

ls /path/to/config_dir
base.toml     <-- first file for Config
config.toml   <-- second file for Config
logging.toml  <-- file for LoggingConfig

Setup environ

export CONFIG='/path/to/config_dir'

Usage in the code

from ttutils import Config

CFG = Config()  # get config path from CONFIG environ

CFG.PUBLIC_URL  # get from config files
CFG.ENV.CONFIG  # get from os env
CFG.SECRET.KEY  # get from os env and clean

If LOGGING is not set, LoggingConfig takes from CONFIG dir only files with logging in name.

from ttutils import LoggingConfig

LoggingConfig(extra_config={
    'loggers': {
        'aiohttp.access': {  # local overriding
            'level': 'ERROR',
        }
    }
})

If Config() or LoggingConfig() receive file names (not absolute paths) and CONFIG environ var points to a directory, files are loaded as $CONFIG/name:

export CONFIG='/path/to/config_dir'

from ttutils import Config, LoggingConfig

CFG = Config('base.toml')            # loads /path/to/config_dir/base.toml
CFG = LoggingConfig('logging.toml')  # loads /path/to/config_dir/logging.toml

Note: only file names and absolute paths can be mixed in one config string — relative paths like dir/file.toml are also resolved to $CONFIG/dir/file.toml.

Config file selection

Config always skips files with logging in name, regardless of source.

LoggingConfig file selection

The logging name filter applies only when config is taken from CONFIG environ var — explicit config argument or LOGGING var loads files without filter.

Configuration from files

Full paths to files separated by ; are loaded in order, later files override earlier ones.

export CONFIG='/path/to/base_config.toml;/path/to/config.toml'

from ttutils import Config

CFG = Config('/path/to/config.toml')  # or pass directly

CFG.PUBLIC_URL  # get from config files
CFG.ENV.CONFIG  # get from os env
CFG.SECRET.KEY  # get from os env and clean

Logging config files can be set via LOGGING environ var or passed directly to LoggingConfig().

export LOGGING='/path/to/logging.toml'

from ttutils import LoggingConfig

LoggingConfig('/path/to/logging.toml', extra_config={
    'loggers': {
        'aiohttp.access': {  # local overriding
            'level': 'ERROR',
        }
    }
})

Safe type convertors

from ttutils import try_int, as_bool, to_string, safe_text, text_crop, int_list, int_set

try_int('123') == 123
try_int('asd') is None

as_bool('t') is True
as_bool(1) is True
as_bool('false') is False

to_string(AClass) == '<AClass>'
to_string('text') == 'text'
to_string(b'text') == 'text'

to_bytes('text') == b'text'
to_bytes(b'text') == b'text'
to_bytes(1234567890) == b'I\x96\x02\xd2'

safe_text('<b>text</b>') == '&lt;b&gt;text&lt;/b&gt;'
safe_text('text') == 'text'

text_crop('text', 5) == 'text'
text_crop('sometext', 6) == 'some …'

int_list(['1', '2', 'a', 'b', None]) == [1, 2]
int_set(['1', '2', 'a', 'b', None]) == {1, 2}

Compress

Integer, dict integers, list integers compression/decompression functions

from ttutils import compress

compress.encode(11232423)  # 'GSiD'
compress.decode('GSi')  # 175506

compress.encode_list([12312, 34535, 12323])  # '30o-8rD-30z'
compress.decode_list('30o-8rD-30z--30C')  # [12312, 34535, 12323, 12324, 12325, 12326]

compress.encode_dict({12: [234, 453], 789: [12, 98, 99, 100, 101]})  # 'c-3G-75/cl-c-1y--1B'
compress.decode_dict('c-3G-75/cl-c-1y--1B')  # {12: [234, 453], 789: [12, 98, 99, 100, 101]}

DateTime

Datetime parse and serialize utils

from ttutils import (utcnow, utcnow_ms, utcnow_sec, parsedt, parsedt_ms,
    parsedt_sec, try_parsedt, isoformat, safe_isoformat)

utcnow()      # datetime(2022, 2, 22, 14, 28, 10, 158164, tzinfo=datetime.timezone.utc)
utcnow_ms()   # datetime(2022, 2, 22, 14, 28, 20, 824000, tzinfo=datetime.timezone.utc)
utcnow_sec()  # datetime(2022, 2, 22, 14, 28, 24, tzinfo=datetime.timezone.utc)

parsedt('2022-02-22T11:22:33.123456Z')      # datetime(2022, 2, 22, 11, 22, 33, 123456, tzinfo=datetime.timezone.utc)
parsedt_ms('2022-02-22T11:22:33.123456Z')   # datetime(2022, 2, 22, 11, 22, 33, 123000, tzinfo=datetime.timezone.utc)
parsedt_sec('2022-02-22T11:22:33.123456Z')  # datetime(2022, 2, 22, 11, 22, 33, tzinfo=datetime.timezone.utc)

try_parsedt('2022-02-22T11:22:33.123456Z')  # datetime(2022, 2, 22, 11, 22, 33, 123456, tzinfo=datetime.timezone.utc)
try_parsedt(None)  # None

isoformat(utcnow())      # '2022-02-22T14:33:51.381164Z'
try_isoformat(utcnow())  # '2022-02-22T14:33:51.381164Z'
try_isoformat(None)      # None

Concurrency

Tools for asyncio

To limit the parallelism of an asynchronous function, install a decorator

from ttutils import concurrency_limit

@concurrency_limit(2)
async def my_task(...) -> None:
    ...  # there are only 2 concurrent executions

# the queue length will be recorded in the log when the function is overloaded
log = logging.getLogger('concurrency_logger')

@concurrency_limit(2, logger=log)
async def my_task(...) -> None:
    ...  # there are only 2 concurrent executions

Stats collector

Collector предназначен для:

• сбора данных о длительности выполнения функций, методов и блоков кода,
• формированни периодических отчетов о статистике времени выполнения,
• ведении лога медленных запросов.

from ttutils.stats import Collector

stats = Collector()

@stats.atimer('k1')
async def func():
   ...

class A:
   @stats.atimer('k2')
   async def func(self):
       ...

with stats.timer('k3'):
   sync_func()
   await async_func()