Coverage for suhteita/__init__.py: 98.41%

1"""Relationships (Finnish: suhteita) maintained across distances as load test core.""" 

2 

3import datetime as dti 

4import logging 

5import os 

6import pathlib 

7import platform 

8import secrets 

9import uuid 

10from typing import Tuple, no_type_check 

11 

12# [[[fill git_describe()]]] 

13__version__ = '2025.8.25+parent.gaf1c9d0d' 

14# [[[end]]] (checksum: c114733be42dc7a51c11294c66ddbde5) 

15__version_info__ = tuple( 

16 e if '-' not in e else e.split('-')[0] for part in __version__.split('+') for e in part.split('.') if e != 'parent' 

17) 

18 

19APP_ALIAS = str(pathlib.Path(__file__).parent.name) 

20APP_ENV = APP_ALIAS.upper() 

21APP_NAME = locals()['__doc__'] 

22DEBUG = bool(os.getenv(f'{APP_ENV}_DEBUG', '')) 

23VERBOSE = bool(os.getenv(f'{APP_ENV}_VERBOSE', '')) 

24QUIET = False 

25STRICT = bool(os.getenv(f'{APP_ENV}_STRICT', '')) 

26ENCODING = 'utf-8' 

27ENCODING_ERRORS_POLICY = 'ignore' 

28DEFAULT_CONFIG_NAME = f'.{APP_ALIAS}.json' 

29 

30NODE_INDICATOR = str(uuid.uuid3(uuid.NAMESPACE_DNS, platform.node())) 

31STORE = os.getenv(f'{APP_ENV}_STORE', '') # default 'store' per argparse 

32COMMA = ',' 

33VERSION = __version__ 

34VERSION_INFO = __version_info__ 

35 

36USER = os.getenv(f'{APP_ENV}_USER', '') 

37TOKEN = os.getenv(f'{APP_ENV}_TOKEN', '') 

38BASE_URL = os.getenv(f'{APP_ENV}_BASE_URL', '') 

39IS_CLOUD = bool(os.getenv(f'{APP_ENV}_IS_CLOUD', '')) 

40PROJECT = os.getenv(f'{APP_ENV}_PROJECT', '') 

41IDENTITY = os.getenv(f'{APP_ENV}_IDENTITY', '') # default 'adhoc' per argparse 

42WORDS = os.getenv(f'{APP_ENV}_WORDS', '/usr/share/dict/words') 

43WORKFLOW_CSV = os.getenv(f'{APP_ENV}_WORKFLOW_CSV', 'to do,in progress,done') 

44 

45log = logging.getLogger() # Module level logger is sufficient 

46LOG_FOLDER = pathlib.Path('logs') 

47LOG_FILE = f'{APP_ALIAS}.log' 

48LOG_PATH = pathlib.Path(LOG_FOLDER, LOG_FILE) if LOG_FOLDER.is_dir() else pathlib.Path(LOG_FILE) 

49LOG_LEVEL = logging.INFO 

50 

51TS_FORMAT_LOG = '%Y-%m-%dT%H:%M:%S' 

52TS_FORMAT_PAYLOADS = '%Y-%m-%d %H:%M:%S.%f UTC' 

53 

54Clocking = Tuple[str, float, str] 

55 

56__all__ = [ 

57 'APP_ALIAS', 

58 'APP_ENV', 

59 'APP_NAME', 

60 'COMMA', 

61 'DEBUG', 

62 'DEFAULT_CONFIG_NAME', 

63 'ENCODING', 

64 'ENCODING_ERRORS_POLICY', 

65 'IDENTITY', 

66 'LOG_FILE', 

67 'LOG_LEVEL', 

68 'LOG_PATH', 

69 'NODE_INDICATOR', 

70 'PROJECT', 

71 'QUIET', 

72 'STORE', 

73 'STRICT', 

74 'TS_FORMAT_LOG', 

75 'TS_FORMAT_PAYLOADS', 

76 'VERBOSE', 

77 'VERSION', 

78 'VERSION_INFO', 

79 'WORDS', 

80 'WORKFLOW_CSV', 

81 'Clocking', 

82] 

83 

84 

85def two_sentences(word_count: int = 4) -> Tuple[str, str]: 

86 """DRY.""" 

87 with open(WORDS, 'rt', encoding=ENCODING) as handle: 

88 words = [word.strip() for word in handle] 

89 wun = ' '.join(secrets.choice(words) for _ in range(word_count)) 

90 two = ' '.join(secrets.choice(words) for _ in range(word_count)) 

91 del words 

92 return wun, two 

93 

94 

95@no_type_check 

96def extract_fields(data, fields): 

97 """DRY.""" 

98 return {field: data[field] for field in fields} 

99 

100 

101@no_type_check 

102def formatTime_RFC3339(self, record, datefmt=None): 

103 """HACK A DID ACK we could inject .astimezone() to localize ...""" 

104 return dti.datetime.fromtimestamp(record.created, dti.timezone.utc).isoformat() 

105 

106 

107@no_type_check 

108def init_logger(name=None, level=None): 

109 """Initialize module level logger""" 

110 global log # pylint: disable=global-statement 

111 

112 log_format = { 

113 'format': '%(asctime)s %(levelname)s [%(name)s]: %(message)s', 

114 'datefmt': TS_FORMAT_LOG, 

115 # 'filename': LOG_PATH, 

116 'level': LOG_LEVEL if level is None else level, 

117 } 

118 logging.Formatter.formatTime = formatTime_RFC3339 

119 logging.basicConfig(**log_format) 

120 log = logging.getLogger(APP_ENV if name is None else name) 

121 log.propagate = True 

122 

123 

124init_logger(name=APP_ENV, level=logging.DEBUG if DEBUG else None)