Python utils and configuration

Configuration

Constants and variables

Config files can be set via CONFIG environ var or passed directly to Config() (argument takes priority). Config string contains absolute paths to files separated by ;, later files override earlier ones. Relative paths are not supported (use DirConfig for directories).

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

from ttutils import Config

CFG = Config()  # get config path from CONFIG environ
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 configuration

Logging config files can be set via LOGGING environ var or passed directly to LoggingConfig() (argument takes priority). Config string contains absolute paths to files separated by ;. Relative paths are not supported. extra_config updates sections after files, apply_config (default True) applies config via dictConfig.

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

from ttutils import LoggingConfig

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

Config file selection

LoggingConfig file selection

Configuration from directory

DirConfig loads configs from a directory specified by CONFIG environ var (must be an existing directory). Relative file names are resolved against it.

export CONFIG='/path/to/config_dir'

ls /path/to/config_dir
base.toml     <-- loaded by DirConfig (alphabetical order)
logging.toml  <-- applied via LoggingConfig

from ttutils import DirConfig

CFG = DirConfig()                      # all toml-files except `logging` in name; `*logging*` applied via LoggingConfig
CFG = DirConfig('db.toml;redis.toml')  # only specified files, `*logging*` still applied
CFG = DirConfig('', 'access.toml')     # only specified log config

DirConfig file selection

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.UTC)
utcnow_ms()   # datetime(2022, 2, 22, 14, 28, 20, 824000, tzinfo=datetime.UTC)
utcnow_sec()  # datetime(2022, 2, 22, 14, 28, 24, tzinfo=datetime.UTC)

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

try_parsedt('2022-02-22T11:22:33.123456Z')  # datetime(2022, 2, 22, 11, 22, 33, 123456, tzinfo=datetime.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()