Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

1""" 

2application core constants, helper functions and base classes 

3============================================================= 

4 

5This module declares practical constants, tiny helper functions and app base classes, which are 

6reducing the code of your application (and of other ae namespace modules/portions). 

7 

8core constants 

9-------------- 

10 

11There are three debug level constants: :data:`DEBUG_LEVEL_DISABLED`, :data:`DEBUG_LEVEL_ENABLED` 

12and :data:`DEBUG_LEVEL_VERBOSE`. Short names for all debug level constants are provided by the 

13dict :data:`DEBUG_LEVELS`. The debug level of your application can be either 

14set in your code or optionally data-driven externally (using the :ref:`config-files` 

15or :ref:`config-options` of the module :mod:`.console`). 

16 

17For to use the :mod:`python logging module <logging>` in conjunction with this module 

18the constant :data:`LOGGING_LEVELS` is providing a mapping between the debug levels 

19and the python logging levels. 

20 

21The encoding of strings into byte-strings (for to output them to the console/stdout or 

22to file contents) can be tricky sometimes. For to not lose any logging output because 

23of invalid characters this module will automatically handle any :exc:`UnicodeEncodeError` 

24exception for you. Invalid characters will then automatically be converted 

25to the default encoding (specified by :data:`~ae.base.DEF_ENCODING`) with the default error 

26handling method specified by :data:`~ae.base.DEF_ENCODE_ERRORS` (both defined in the 

27:mod:`ae.base` namespace portion/module. 

28 

29 

30core helper functions 

31--------------------- 

32 

33The :func:`print_out` function, which is fully compatible to pythons 

34:func:`print`, is using the encode helpers :func:`~ae.base.force_encoding` and 

35:func:`~.ae.base.to_ascii` for to auto-correct invalid characters. 

36 

37The function :func:`hide_dup_line_prefix` is very practical if you want to remove or hide 

38redundant line prefixes in your log files, to make them better readable. 

39 

40 

41application base classes 

42------------------------ 

43 

44The classes :class:`AppBase` and :class:`SubApp` are applying logging and debugging features 

45to your application. Create in your application one instance of :class:`AppBase` for to represent 

46the main application task. If your application needs a separate logging/debugging configuration for 

47sub-threads or sub-tasks then create an :class:`SubApp` instance for each of these sub-apps. 

48 

49Sub-apps are very flexible and not tied to any fix use-case. They can be created e.g. for each sub-task or 

50application thread. You could also create a :class:`SubApp` instance for each of your external systems, 

51like a database server or for to connect your application onto different test environments 

52or to your live an your production system (e.g. for system comparison and maintenance). 

53 

54Both application classes are automatically catching and handling any exceptions and run-time 

55errors: only if any critical exception/error cannot be handled then the :meth:`~AppBase.shutdown` 

56method will make sure that all sub-apps and threads get terminated and joined. 

57Additionally all print-out buffers will be flushed for to include all the info 

58of the critical error (the last debug and error messages) into the 

59standard error/output and into any activated log files. 

60 

61 

62basic usage of an application base class 

63........................................ 

64 

65At the top of your python application main file/module create an instance of the class :class:`AppBase`:: 

66 

67 \"\"\" docstring at the top of the main module of your application \"\"\" 

68 from ae.core import AppBase 

69 

70 __version__ = '1.2.3' 

71 

72 ca = AppBase() 

73 

74In the above example the :class:`AppBase` instance will automatically use the docstring of the 

75module as application title and the string in the module variable __version___ as application version. 

76To overwrite these defaults pass your application title and version string via the 

77arguments :paramref:`~AppBase.app_title` and :paramref:`~AppBase.app_version` 

78to the instantiation of :class:`AppBase`:: 

79 

80 ca = AppBase("title of this app instance", app_version='3.2.1') 

81 

82Other automatically initialized instance attributes of :class:`AppBase` are documented underneath 

83in the :class:`class docstring <AppBase>`. They include e.g. 

84the :attr:`date and time when the instance got created <AppBase.startup_beg>`, 

85the :attr:`name/id of this application instance <AppBase.app_name>` or 

86the :attr:`application path <AppBase.app_path>`. 

87 

88 

89application class hierarchy 

90........................... 

91 

92For most use cases you will not instantiate from :class:`AppBase` directly - instead you will 

93instantiate one of the extended application classes that are inherited from this base class. 

94 

95The class :class:`~ae.console.ConsoleApp` e.g. inherits from :class:`AppBase` and is adding 

96configuration options and variables to it. So in your console application it is recommended to directly 

97use instances of :class:`~ae.console.ConsoleApp` instead of :class:`AppBase`. 

98 

99For applications with an GUI use instead one of the classes :class:`~ae.kivy_app.KivyApp`, 

100:class:`~ae.enaml_app.EnamlApp` or :class:`~ae.dabo_app.DaboApp`. 

101 

102 

103application logging 

104------------------- 

105 

106Print-outs are an essential tool for the debugging and logging of your application at run-time. In python 

107the print-outs are done with the :func:`print` function or with the python :mod:`logging` module. These 

108print-outs get per default send to the standard output and error streams of your OS and so displayed on 

109your system console/shell. The :func:`print_out` function and the :meth:`~AppBase.print_out` method of 

110this :mod:`.core` module are adding two more sophisticated ways for print-outs to the console/log-files. 

111 

112Using :class:`AppBase` is making the logging much easier and also ensures that print-outs of any 

113imported library or package will be included within your log files. This is done by redirecting the 

114standard output and error streams to your log files with the help of the :class:`_PrintingReplicator` 

115class. 

116 

117Head-less server applications like web servers are mostly not allowed to use the standard output streams. 

118For some these applications you could redirect the standard output and error streams to a log file by 

119using the OS redirection character (>):: 

120 

121 python your_application.py >log_std_out.log 2>log_std_err.log 

122 

123But because most web servers doesn't allow you to use this redirection, you can alternatively specify 

124the :paramref:`~AppBase.suppress_stdout` parameter as True in the instantiation of an :class:`AppBase` 

125instance. Additionally you can call the :meth:`~AppBase.init_logging` method for to activate a log file. 

126After that all print-outs of your application and libraries will only appear in your log file. 

127 

128Also in complex applications, where huge print-outs to the console can get lost easily, you want to use 

129a log file instead. But even a single log file can get messy to read, especially for multi-threaded 

130server applications. For that :class:`SubApp` is allowing you to create for each thread a separate 

131sub-app instance with its own log file. 

132 

133Using this module ensures that any crashes or freezes happening in your application will be fully logged. 

134Apart from the gracefully handling of :exc:`UnicodeEncodeError` exceptions, the 

135:mod:`Python faulthandler <faulthandler>` will be enabled automatically for to also catch 

136system errors and to dump a traceback of them to the console and any activated log file. 

137 

138 

139activate ae log file 

140.................... 

141 

142.. _ae-log-file: 

143 

144Ae Log Files are text files using by default the encoding of your OS console/shell. To activate the 

145redirection of your applications print-outs into a ae log file for a :class:`AppBase` instance you 

146simply specify the file name of the log file in the :meth:`~AppBase.init_logging` method call:: 

147 

148 app = AppBase() 

149 app.init_logging(log_file_name='my_log_file.log') 

150 

151 

152activate ae logging features 

153............................ 

154 

155For multi-threaded applications you can include the thread-id of the printing thread automatically 

156into your log files. For that you have to pass a True value to the 

157:paramref:`~AppBase.multi_threading` argument. For to additionally also suppress any print-outs 

158to the standard output/error streams pass True to the :paramref:`~AppBase.suppress_stdout` argument:: 

159 

160 app = AppBase(multi_threading=True, suppress_stdout=True) 

161 app.init_logging(log_file_name='my_log_file.log') 

162 

163The ae log files provided by this module are automatically rotating if the size of an log file 

164succeeds the value in MBytes defined in the :data:`LOG_FILE_MAX_SIZE`. For to adapt this value 

165to your needs you can specify the maximum log file size in MBytes with the argument 

166:paramref:`~AppBase.init_logging.log_file_size_max` in your call of :meth:`~AppBase.init_logging`:: 

167 

168 app.init_logging(log_file_name='my_log_file.log', log_file_size_max=9.) 

169 

170By using the :class:`~ae.console.ConsoleApp` class instead of :class:`AppBase` you can 

171alternatively store the logging configuration of your application within a 

172:ref:`configuration variable <config-variables>` or a 

173:ref:`configuration option <config-options>`. 

174The order of precedence for to find the appropriate logging configuration of each 

175app instance is documented :meth:`here <ae.console.ConsoleApp._init_logging>` . 

176 

177 

178using python logging module 

179........................... 

180 

181If you prefer to use instead the python logging module for the print-outs of your application, 

182then pass a :mod:`python logging configuration dictionary <logging.config>` with the individual 

183configuration of your logging handlers, files and loggers to the 

184:paramref:`~AppBase.init_logging.py_logging_params` argument of the 

185:meth:`~AppBase.init_logging` method:: 

186 

187 app.init_logging(py_logging_params=my_py_logging_config_dict) 

188 

189Passing the python logging configuration dictionary to one of the :class:`AppBase` 

190instances created by your application will automatically disable the ae log file of this 

191instance. 

192 

193 

194application debugging 

195--------------------- 

196 

197For to use the debug features of :mod:`~.core` you simple have to import the needed 

198:ref:`debug level constant <debug-level-constants>` for to pass it at instantiation of 

199your :class:`AppBase` or :class:`SubApp` class to the :paramref:`~AppBase.debug_level` argument: 

200 

201 app = AppBase(..., debug_level= :data:`DEBUG_LEVEL_ENABLED`) # same for :class:`SubApp` 

202 

203By passing :data:`DEBUG_LEVEL_ENABLED` the print-outs (and log file contents) will be more detailed, 

204and even more verbose if you use instead the debug level :data:`DEBUG_LEVEL_VERBOSE`. 

205 

206The debug level can be changed at any time in your application code by directly assigning 

207the new debug level to the :attr:`~AppBase.debug_level` property. If you prefer to change 

208the (here hard-coded) debug levels dynamically, then use the :class:`~.console.ConsoleApp` instead 

209of :class:`AppBase`, because :class:`~.console.ConsoleApp` provides this property as a 

210:ref:`configuration file variable <config-variables>` 

211and :ref:`commend line option <config-options>`. This way you can 

212specify :ref:`the actual debug level <pre-defined-config-options>` without the need 

213to change (and re-build) your application code. 

214 

215.. _debug-level-constants: 

216 

217""" 

218import datetime 

219import faulthandler 

220import logging 

221import logging.config 

222import os 

223import sys 

224import threading 

225import traceback 

226import weakref 

227 

228from io import StringIO 

229from typing import Any, AnyStr, Dict, List, Optional, TextIO, Tuple, Union, cast 

230 

231from ae.base import DATE_TIME_ISO, DEF_ENCODE_ERRORS, force_encoding, to_ascii # type: ignore 

232from ae.paths import app_name_guess, app_data_path, app_docs_path, PATH_PLACEHOLDERS # type: ignore 

233 

234 

235__version__ = '0.1.50' #: actual version of this portion/package/module 

236 

237 

238# DON'T RE-ORDER: using module doc-string as _debug-level-constants sphinx hyperlink to following DEBUG_ constants 

239DEBUG_LEVEL_DISABLED: int = 0 #: lowest debug level - only display logging levels ERROR/CRITICAL. 

240DEBUG_LEVEL_ENABLED: int = 1 #: minimum debugging info - display logging levels WARNING or higher. 

241DEBUG_LEVEL_VERBOSE: int = 2 #: verbose debug info - display logging levels INFO/DEBUG or higher. 

242 

243DEBUG_LEVELS: Dict[int, str] = {DEBUG_LEVEL_DISABLED: 'disabled', DEBUG_LEVEL_ENABLED: 'enabled', 

244 DEBUG_LEVEL_VERBOSE: 'verbose'} 

245""" numeric ids and names of all supported debug levels. """ 

246 

247LOGGING_LEVELS: Dict[int, int] = {DEBUG_LEVEL_DISABLED: logging.WARNING, DEBUG_LEVEL_ENABLED: logging.INFO, 

248 DEBUG_LEVEL_VERBOSE: logging.DEBUG} 

249""" association between ae debug levels and python logging levels. """ 

250 

251HIDDEN_CREDENTIALS = ('password', 'token') #: credential keys that are hidden in print/repr output (not if verbose) 

252 

253 

254def hide_dup_line_prefix(last_line: str, current_line: str) -> str: 

255 """ replace duplicate characters at the begin of two strings with spaces. 

256 

257 :param last_line: last line string (e.g. the last line of text/log file). 

258 :param current_line: current line string. 

259 :return: current line string but duplicate characters at the begin are replaced by space characters. 

260 """ 

261 idx = 0 

262 min_len = min(len(last_line), len(current_line)) 

263 while idx < min_len and last_line[idx] == current_line[idx]: 

264 idx += 1 

265 return " " * idx + current_line[idx:] 

266 

267 

268MAX_NUM_LOG_FILES: int = 69 #: maximum number of :ref:`ae log files <ae-log-file>` 

269LOG_FILE_MAX_SIZE: int = 15 #: max. size in MB of rotating :ref:`ae log files <ae-log-file>` 

270LOG_FILE_IDX_WIDTH: int = len(str(MAX_NUM_LOG_FILES)) + 3 

271""" width of rotating log file index within log file name; adding +3 to ensure index range up to factor 10^3. """ 

272 

273ori_std_out: TextIO = sys.stdout #: original sys.stdout on app startup 

274ori_std_err: TextIO = sys.stderr #: original sys.stderr on app startup 

275 

276log_file_lock: threading.RLock = threading.RLock() #: log file rotation multi-threading lock 

277 

278 

279_logger = None #: python logger for this module gets lazy/late initialized and only if requested by caller 

280 

281 

282def logger_late_init(): 

283 """ check if logging modules got initialized already and if not then do it now. """ 

284 global _logger 

285 if not _logger: 

286 _logger = logging.getLogger(__name__) 

287 

288 

289_multi_threading_activated: bool = False #: flag if threading is used in your application 

290 

291 

292def activate_multi_threading(): 

293 """ activate multi-threading for all app instances (normally done at main app startup). """ 

294 global _multi_threading_activated 

295 _multi_threading_activated = True 

296 

297 

298def _deactivate_multi_threading(): 

299 """ disable multi threading (needed for to reset app environment in unit testing). """ 

300 global _multi_threading_activated 

301 _multi_threading_activated = False 

302 

303 

304def print_out(*objects, sep: str = " ", end: str = "\n", file: Optional[TextIO] = None, flush: bool = False, 

305 encode_errors_def: str = DEF_ENCODE_ERRORS, logger: Optional['logging.Logger'] = None, 

306 app: Optional['AppBase'] = None, **kwargs): 

307 """ universal/unbreakable print function - replacement for the :func:`built-in python function print() <print>`. 

308 

309 :param objects: tuple of objects to be printed. If the first object is a string that 

310 starts with a \\\\r character then the print-out will be only sent 

311 to the standard output (and will not be added to any active log files - 

312 see also :paramref:`~print_out.end` argument). 

313 :param sep: separator character between each printed object/string (def=" "). 

314 :param end: finalizing character added to the end of this print-out (def="\\\\n"). 

315 Pass \\\\r for to suppress the print-out into :ref:`ae log file <ae-log-file>` 

316 or to any activated python logger 

317 - useful for console/shell processing animation (see :meth:`.tcp.TcpServer.run`). 

318 :param file: output stream object to be printed to (def=None which will use standard output streams). 

319 If given then the redirection to all active log files and python logging loggers 

320 will be disabled (even if the :paramref:`~print_out.logger` argument is specified). 

321 :param flush: flush stream after printing (def=False). 

322 :param encode_errors_def: default error handling for to encode (def=:data:`DEF_ENCODE_ERRORS`). 

323 :param logger: used logger for to output `objects` (def=None). Ignored if the 

324 :paramref:`print_out.file` argument gets specified/passed. 

325 :param app: the app instance from where this print-out got initiated. 

326 :param kwargs: catch unsupported kwargs for debugging (all items will be printed to all 

327 the activated logging/output streams). 

328 

329 This function is silently handling and auto-correcting string encode errors for output/log streams which are 

330 not supporting unicode. Any instance of :class:`AppBase` is providing this function as a method with the 

331 :func:`same name <AppBase.print_out>`). It is recommended to call/use this instance method instead of this function. 

332 

333 In multi-threaded applications this function prevents dismembered/fluttered print-outs from different threads. 

334 

335 This function has an alias named :func:`.po`. 

336 """ 

337 processing = end == "\r" or (objects and str(objects[0]).startswith('\r')) # True if called by Progress.next() 

338 enc = getattr(file or ori_std_out if processing else sys.stdout, 'encoding', 'utf-8') 

339 use_py_logger = False 

340 

341 main_app = main_app_instance() 

342 if main_app: 

343 file = main_app.log_file_check(file) # check if late init of logging system is needed 

344 if app and app != main_app: 

345 file = app.log_file_check(file) # check sub-app suppress_stdout/log file status and rotation 

346 else: 

347 app = main_app 

348 

349 if processing: 

350 file = ori_std_out 

351 elif logger is not None and file is None and ( 

352 app and app.py_log_params and main_app != app or main_app and main_app.py_log_params): 

353 use_py_logger = True 

354 logger_late_init() 

355 

356 if kwargs: 

357 objects += (f"\n * EXTRA KWARGS={kwargs}", ) 

358 

359 retries = 2 

360 while retries: 

361 try: 

362 print_strings = tuple(map(lambda _: str(_).encode(enc, errors=encode_errors_def).decode(enc), objects)) 

363 if use_py_logger or _multi_threading_activated: 

364 # concatenating objects also prevents fluttered log file content in multi-threading apps 

365 # .. see https://stackoverflow.com/questions/3029816/how-do-i-get-a-thread-safe-print-in-python-2-6 

366 # .. and https://stackoverflow.com/questions/50551637/end-key-in-print-not-thread-safe 

367 print_one_str = sep.join(print_strings) 

368 sep = "" 

369 if end and (not use_py_logger or end != '\n'): 

370 print_one_str += end 

371 end = "" 

372 print_strings = (print_one_str, ) 

373 

374 if use_py_logger: 

375 debug_level = app.debug_level if app else DEBUG_LEVEL_VERBOSE 

376 if logger: # mypy insists to have this extra check, although use_py_logger is including logger 

377 logger.log(level=LOGGING_LEVELS[debug_level], msg=print_strings[0]) 

378 else: 

379 print(*print_strings, sep=sep, end=end, file=file, flush=flush) 

380 break 

381 except UnicodeEncodeError: 

382 fixed_objects = list() 

383 for obj in objects: 

384 if not isinstance(obj, str) and not isinstance(obj, bytes): 

385 obj = str(obj) 

386 if retries == 2: 

387 obj = force_encoding(obj, encoding=enc) 

388 else: 

389 obj = to_ascii(obj) 

390 fixed_objects.append(obj) 

391 objects = tuple(fixed_objects) 

392 retries -= 1 

393 except (IOError, OSError, ValueError, Exception): # pragma: no cover 

394 traceback.print_exc() 

395 print("...... in ae.core.print_out(", objects, ")") 

396 break 

397 

398 

399po = print_out #: alias of function :func:`.print_out` 

400 

401 

402APP_KEY_SEP: str = '@' #: separator character used in :attr:`~AppBase.app_key` of :class:`AppBase` instance 

403 

404# Had to use type comment because the following line is throwing an error in the Sphinx docs make: 

405# _app_instances: weakref.WeakValueDictionary[str, "AppBase"] = weakref.WeakValueDictionary() 

406_app_instances = weakref.WeakValueDictionary() # type: weakref.WeakValueDictionary[str, AppBase] 

407""" dict that is weakly holding references to all :class:`AppBase` instances created at run time. 

408 

409Gets automatically initialized in :meth:`AppBase.__init__` for to allow log file split/rotation 

410and debug_level access at application thread or module level. 

411 

412The first created :class:`AppBase` instance is called the main app instance. :data:`_main_app_inst_key` 

413stores the dict key of the main instance. 

414""" 

415_main_app_inst_key: str = '' #: key in :data:`_app_instances` of main :class:`AppBase` instance 

416 

417app_inst_lock: threading.RLock = threading.RLock() #: app instantiation multi-threading lock 

418 

419 

420def main_app_instance() -> Optional['AppBase']: 

421 """ determine the main instance of the :class:`AppBase` in the current running application. 

422 

423 :return: main/first-instantiated :class:`AppBase` instance or None (if app is not fully initialized yet). 

424 """ 

425 with app_inst_lock: 

426 return _app_instances.get(_main_app_inst_key) 

427 

428 

429def registered_app_names() -> List[str]: 

430 """ determine the app names of all registered/running applications. """ 

431 with app_inst_lock: 

432 return [app.app_name for app in _app_instances.values()] 

433 

434 

435def _register_app_instance(app: 'AppBase'): 

436 """ register new :class:`AppBase` instance in :data:`_app_instances`. 

437 

438 :param app: :class:`AppBase` instance to register 

439 """ 

440 with app_inst_lock: 

441 global _app_instances, _main_app_inst_key 

442 msg = f"register_app_instance({app}) expects " 

443 assert app not in _app_instances.values(), msg + "new instance - this app got already registered" 

444 

445 key = app.app_key 

446 assert key and key not in _app_instances, \ 

447 msg + f"non-empty, unique app key (app_name+sys_env_id=={key} keys={list(_app_instances.keys())})" 

448 

449 cnt = len(_app_instances) 

450 if _main_app_inst_key: 

451 assert cnt > 0, f"No app instances registered but main app key is set to {_main_app_inst_key}" 

452 else: 

453 assert cnt == 0, f"{cnt} sub-apps {list(_app_instances.keys())} found after main app remove" 

454 _main_app_inst_key = key 

455 _app_instances[key] = app 

456 

457 

458def _unregister_app_instance(app_key: str) -> Optional['AppBase']: 

459 """ unregister/remove :class:`AppBase` instance from within :data:`_app_instances`. 

460 

461 :param app_key: app key of the instance to remove. 

462 :return: removed :class:`AppBase` instance. 

463 """ 

464 with app_inst_lock: 

465 global _app_instances, _main_app_inst_key 

466 app = _app_instances.pop(app_key, None) 

467 cnt = len(_app_instances) 

468 if app_key == _main_app_inst_key: 

469 _main_app_inst_key = '' 

470 assert cnt == 0, f"{cnt} sub-apps {list(_app_instances.keys())} found after main app {app_key}{app} remove" 

471 elif _main_app_inst_key: 

472 assert cnt > 0, f"Unregistered last app {app_key}/{app} but was not the main app {_main_app_inst_key}" 

473 return app 

474 

475 

476def _shut_down_sub_app_instances(timeout: Optional[float] = None): 

477 """ shut down all :class:`SubApp` instances. 

478 

479 :param timeout: timeout float value in seconds used for the :class:`SubApp` shutdowns and for the acquisition 

480 of the threading locks of :data:`the ae log file <log_file_lock>` and the :data:`app instances 

481 <app_inst_lock>`. 

482 """ 

483 aqc_kwargs: Dict[str, Any] = (dict(blocking=False) if timeout is None else dict(timeout=timeout)) 

484 blocked = app_inst_lock.acquire(**aqc_kwargs) 

485 main_app = main_app_instance() 

486 for app in list(_app_instances.values()): # list is needed because weak ref dict get changed in loop 

487 if app is not main_app: 

488 app.shutdown(timeout=timeout) 

489 if blocked: 

490 app_inst_lock.release() 

491 

492 

493class _PrintingReplicator: 

494 """ replacement of standard/error stream replicating print-outs to all active logging streams (log files/buffers). 

495 """ 

496 def __init__(self, sys_out_obj: TextIO = ori_std_out) -> None: 

497 """ initialise a new T-stream-object 

498 

499 :param sys_out_obj: standard output/error stream to be replicated (def=sys.stdout) 

500 """ 

501 self.sys_out_obj = sys_out_obj 

502 

503 def write(self, any_str: AnyStr) -> None: 

504 """ write string to ae logging and standard output streams. 

505 

506 Automatically suppressing UnicodeEncodeErrors if console/shell or log file has different encoding 

507 by forcing re-encoding with DEF_ENCODE_ERRORS. 

508 

509 :param any_str: string to output. 

510 """ 

511 message = cast(bytes, any_str).decode() if isinstance(any_str, bytes) else any_str 

512 app_streams: List[Tuple[Optional[AppBase], TextIO]] = list() 

513 with log_file_lock, app_inst_lock: 

514 for app in list(_app_instances.values()): 

515 stream = app.log_file_check(app.active_log_stream) # check if log rotation or buf-to-file-switch needed 

516 if stream: 

517 app_streams.append((app, stream)) 

518 if not self.sys_out_obj.closed: 

519 app_streams.append((main_app_instance(), self.sys_out_obj)) 

520 

521 if message and message[0] != '\n' and message[-1] == '\n': 

522 message = '\n' + message[:-1] 

523 log_lines = message.split('\n') 

524 for app_or_none, stream in app_streams: 

525 line_prefix = '\n' + (app_or_none.log_line_prefix() if app_or_none else '') 

526 app_msg = line_prefix.join(log_lines) 

527 try: 

528 stream.write(app_msg) 

529 except UnicodeEncodeError: 

530 stream.write(force_encoding(app_msg, encoding=stream.encoding)) 

531 

532 def __getattr__(self, attr: str) -> Any: 

533 """ get attribute value from standard output stream. 

534 

535 :param attr: name of the attribute to retrieve/return. 

536 :return: value of the attribute. 

537 """ 

538 return getattr(self.sys_out_obj, attr) 

539 

540 

541_app_threads = weakref.WeakValueDictionary() # type: weakref.WeakValueDictionary[int, threading.Thread] 

542""" weak dict for to keep the references of all application threads. Added for to prevent 

543the joining of unit testing threads in the test teardown (resetting app environment). """ 

544 

545 

546def _register_app_thread(): 

547 """ add new app thread to _app_threads if not already added. """ 

548 global _app_threads 

549 tid = threading.get_ident() 

550 if tid not in _app_threads: 

551 _app_threads[tid] = threading.current_thread() 

552 

553 

554def _join_app_threads(timeout: Optional[float] = None): 

555 """ join/finish all app threads and finally deactivate multi-threading. 

556 

557 :param timeout: timeout float value in seconds for thread joining (def=None - block/no-timeout). 

558 

559 .. note:: 

560 This function has to be called by the main app instance only. 

561 """ 

562 global _app_threads 

563 main_thread = threading.current_thread() 

564 for app_thread in list(_app_threads.values()): # threading.enumerate() also includes PyCharm/pytest threads 

565 if app_thread is not main_thread: 

566 po(f" ** joining thread id <{app_thread.ident: >6}> name={app_thread.getName()}", logger=_logger) 

567 app_thread.join(timeout) 

568 if app_thread.ident is not None: # mypy needs it because ident is Optional 

569 _app_threads.pop(app_thread.ident) 

570 _deactivate_multi_threading() 

571 

572 

573class AppBase: 

574 """ provides easy logging and debugging for your application. 

575 

576 Most applications only need a single instance of this class; apps with threads could create separate instances 

577 for each thread. 

578 

579 Instance Attributes (ordered alphabetically - ignoring underscore characters): 

580 

581 * :attr:`app_key` id/key of this application instance. 

582 * :attr:`app_name` basename (without the file name extension) of the executable. 

583 * :attr:`app_path` file path of app executable. 

584 * :attr:`app_title` application title/description. 

585 * :attr:`app_version` application version (set via the :paramref:`AppBase.app_version` argument). 

586 * :attr:`debug_level` debug level of this instance. 

587 * :attr:`_last_log_line_prefix` last ae log file line prefix that got print-out to the log of this app instance. 

588 * :attr:`_log_buf_stream` ae log file buffer stream. 

589 * :attr:`_log_file_index` index of the current rotation ae log file backup. 

590 * :attr:`_log_file_name` path and file name of the ae log file. 

591 * :attr:`_log_file_size_max` maximum size in MBytes of a ae log file. 

592 * :attr:`_log_file_stream` ae log file TextIO output stream. 

593 * :attr:`_log_with_timestamp` log timestamp line prefix if True or a non-empty strftime compatible format string. 

594 * :attr:`py_log_params` python logging config dictionary. 

595 * :attr:`_nul_std_out` null stream used for to prevent print-outs to :attr:`standard output <sys.stdout>`. 

596 * :attr:`_shut_down` flag set to True if this application instance got already shutdown. 

597 * :attr:`startup_beg` datetime of begin of the instantiation/startup of this app instance. 

598 * :attr:`startup_end` datetime of end of the instantiation/startup of this application instance. 

599 * :attr:`suppress_stdout` flag set to True if this application does not print to stdout/console. 

600 * :attr:`sys_env_id` system environment id of this application instance. 

601 """ 

602 app_title: str = "" #: title/description of this app instance 

603 app_name: str = '' #: name of this app instance 

604 app_version: str = '' #: version of this app instance 

605 _debug_level: int = DEBUG_LEVEL_VERBOSE #: debug level of this app instance 

606 sys_env_id: str = '' #: system environment id of this app instance 

607 suppress_stdout: bool = True #: flag to suppress prints to stdout 

608 startup_end: Optional[datetime.datetime] = None #: end datetime of the application startup 

609 _last_log_line_prefix: str = "" #: prefix of the last printed log line 

610 _log_buf_stream: Optional[StringIO] = None #: log file buffer stream instance 

611 _log_file_stream: Optional[TextIO] = None #: log file stream instance 

612 _log_file_index: int = 0 #: log file index (for rotating logs) 

613 _log_file_size_max: float = LOG_FILE_MAX_SIZE #: maximum log file size in MBytes (rotating log files) 

614 _log_file_name: str = "" #: log file name 

615 _log_with_timestamp: Union[bool, str] = False #: True of strftime format string to enable timestamp 

616 _nul_std_out: Optional[TextIO] = None #: logging null stream 

617 py_log_params: Dict[str, Any] = dict() #: dict of config parameters for py logging 

618 _shut_down: bool = False #: True if this app instance got shut down already 

619 

620 def __init__(self, app_title: str = '', app_name: str = '', app_version: str = '', sys_env_id: str = '', 

621 debug_level: int = DEBUG_LEVEL_DISABLED, multi_threading: bool = False, suppress_stdout: bool = False): 

622 """ initialize a new :class:`AppBase` instance. 

623 

624 :param app_title: application title/description for to set the instance attribute 

625 :attr:`~ae.core.AppBase.app_title`. 

626 

627 If not specified then the docstring of your app's main module will 

628 be used (see :ref:`example <app-title>`). 

629 

630 :param app_name: application instance name for to set the instance attribute 

631 :attr:`~ae.core.AppBase.app_name`. 

632 

633 If not specified then base name of the main module file name will be used. 

634 

635 :param app_version: application version string for to set the instance attribute 

636 :attr:`~ae.core.AppBase.app_version`. 

637 

638 If not specified then value of a global variable with the name 

639 `__version__` will be used (if declared in the actual call stack). 

640 

641 :param sys_env_id: system environment id for to set the instance attribute 

642 :attr:`~ae.core.AppBase.sys_env_id`. 

643 

644 The default value of this argument is an empty string. 

645 

646 :param debug_level: default debug level for to set the instance attribute 

647 :attr:`~ae.core.AppBase.debug_level`. 

648 

649 The default value of this argument is :data:`~ae.core.DEBUG_LEVEL_DISABLED`. 

650 

651 :param multi_threading: pass True if instance is used in multi-threading app. 

652 

653 :param suppress_stdout: pass True (for wsgi apps) for to prevent any python print outputs to stdout. 

654 """ 

655 try: 

656 from ae.inspector import stack_var # type: ignore 

657 except ImportError: # pragma: no cover 

658 def stack_var(key: str) -> str: 

659 """ get default value for app title and version (if ae.inspector is not provided/available). """ 

660 return f"AppBase.__init__(): stack_var() not imported for to determine {key} value" 

661 

662 self.startup_beg: datetime.datetime = datetime.datetime.now() #: begin of app startup datetime 

663 self.app_path: str = os.path.dirname(sys.argv[0]) #: path to folder of your main app code file 

664 

665 if not app_title: 

666 doc_str = stack_var('__doc__') 

667 app_title = doc_str.strip().split('\n')[0] if doc_str else "" 

668 if app_name: 

669 PATH_PLACEHOLDERS['app_name'] = app_name 

670 PATH_PLACEHOLDERS['app'] = app_data_path() 

671 PATH_PLACEHOLDERS['ado'] = app_docs_path() 

672 else: 

673 app_name = app_name_guess() 

674 if PATH_PLACEHOLDERS.get('main_app_name', "") in ("", 'pyTstConsAppKey', '_jb_pytest_runner'): 

675 PATH_PLACEHOLDERS['main_app_name'] = app_name 

676 if not app_version: 

677 app_version = stack_var('__version__') or "" 

678 

679 self.app_title: str = app_title #: title/description of this app instance 

680 self.app_name: str = app_name #: name of this app instance 

681 self.app_version: str = app_version #: version of this app instance 

682 self._debug_level: int = debug_level #: debug level of this app instance 

683 self.sys_env_id: str = sys_env_id #: system environment id of this app instance 

684 if multi_threading: 

685 activate_multi_threading() 

686 self.suppress_stdout: bool = suppress_stdout #: flag to suppress prints to stdout 

687 

688 self.startup_end: Optional[datetime.datetime] = None #: end datetime of the application startup 

689 

690 _register_app_thread() 

691 _register_app_instance(self) 

692 

693 def __del__(self): 

694 """ deallocate this app instance by calling :func:`AppBase.shutdown`. 

695 """ 

696 self.shutdown(exit_code=None) 

697 

698 @property 

699 def active_log_stream(self) -> Optional[Union[StringIO, TextIO]]: 

700 """ check if ae logging is active and if yes then return the currently used log stream (read-only property). 

701 

702 :return: log file or buf stream if logging is activated, else None. 

703 """ 

704 with log_file_lock: 

705 return self._log_file_stream or self._log_buf_stream 

706 

707 @property 

708 def app_key(self) -> str: 

709 """ determine the key of this application class instance (read-only property). 

710 

711 :return: application key string. 

712 """ 

713 return self.app_name + APP_KEY_SEP + self.sys_env_id 

714 

715 @property 

716 def debug_level(self) -> int: 

717 """ debug level property: 

718 

719 :getter: return the current debug level of this app instance. 

720 :setter: change the debug level of this app instance. 

721 """ 

722 return self._debug_level 

723 

724 @debug_level.setter 

725 def debug_level(self, debug_level: int): 

726 """ debug level setter (added for easier overwrite in inheriting classes). """ 

727 self._debug_level = debug_level 

728 

729 @property 

730 def debug(self) -> bool: 

731 """ True if app is in debug mode. """ 

732 return self._debug_level >= DEBUG_LEVEL_ENABLED 

733 

734 @property 

735 def verbose(self) -> bool: 

736 """ True if app is in verbose debug mode. """ 

737 return self._debug_level >= DEBUG_LEVEL_VERBOSE 

738 

739 def call_method(self, method: str, *args, **kwargs) -> Any: 

740 """ call method of this instance with the passed args, catching and logging exceptions preventing app exit. 

741 

742 :param method: name of the main app method to call. 

743 :param args: args passed to the main app method to be called. 

744 :param kwargs: kwargs passed to the main app method to be called. 

745 :return: return value of the called method or None if method throws exception or does not exist. 

746 """ 

747 event_callback = getattr(self, method, None) 

748 if event_callback is not None: 

749 assert callable(event_callback), f"AppBase.call_method: {method!r} is not callable ({args}, {kwargs})" 

750 try: 

751 return event_callback(*args, **kwargs) 

752 except (AttributeError, IndexError, LookupError, ValueError, Exception) as ex: 

753 self.po(f" *** AppBase.call_method({method}, {args}, {kwargs}): {ex}\n{traceback.format_exc()}") 

754 return None 

755 

756 def init_logging(self, py_logging_params: Optional[Dict[str, Any]] = None, log_file_name: str = "", 

757 log_file_size_max: float = LOG_FILE_MAX_SIZE, log_with_timestamp: Union[bool, str] = False, 

758 disable_buffering: bool = False): 

759 """ initialize logging system. 

760 

761 :param py_logging_params: config dict for python logging configuration. 

762 If this dict is not empty then python logging is configured with the 

763 given options in this dict and all the other kwargs are ignored. 

764 :param log_file_name: default log file name for ae logging (def='' - ae logging disabled). 

765 :param log_file_size_max: max. size in MB of ae log file (def=LOG_FILE_MAX_SIZE). 

766 :param log_with_timestamp: add timestamp prefix to each log line if True or a non-empty strftime 

767 compatible format string. 

768 :param disable_buffering: pass True to disable ae log buffering at app startup. 

769 

770 Log files and config values will be initialized as late as possible in :meth:`~AppBase.log_file_check` 

771 e.g. indirectly triggered by a request to a config variable via :meth:`~AppBase._parse_args` (like `logFile`). 

772 """ 

773 with log_file_lock: 

774 if py_logging_params: # init python logging - app is using python logging module 

775 logger_late_init() 

776 # logging.basicConfig(level=logging.DEBUG, style='{') 

777 logging.config.dictConfig(py_logging_params) # re-configure py logging module 

778 self.py_log_params = py_logging_params 

779 else: # (re-)init ae logging 

780 if self._log_file_stream: 

781 self._close_log_file() 

782 self._std_out_err_redirection(False) 

783 self._log_file_name = log_file_name 

784 self._log_file_size_max = log_file_size_max 

785 self._log_with_timestamp = log_with_timestamp 

786 if not disable_buffering: 

787 self._log_buf_stream = StringIO(initial_value="#### Log Buffer\n" if self.debug else "") 

788 

789 def log_line_prefix(self) -> str: 

790 """ compile prefix of log print-out line for this :class:`AppBase` instance. 

791 

792 The line prefix consists of (depending on the individual values of either a module variable or of an 

793 attribute this app instance): 

794 

795 * :data:`_multi_threading_activated`: if True then the thread id gets printed surrounded with 

796 angle brackets (< and >), right aligned and space padded to minimal 6 characters. 

797 * :attr:`sys_env_id`: if not empty then printed surrounded with curly brackets ({ and }), left aligned 

798 and space padded to minimal 4 characters. 

799 * :attr:`_log_with_timestamp`: if (a) True or (b) an non-empty string then the system time 

800 (determined with :meth:`~datetime.datetime.now`) gets printed in the format specified either by the 

801 (a) the :data:`~ae.base.DATE_TIME_ISO` constant or (b) by the string in this attribute. 

802 

803 This method is using the instance attribute :attr:`_last_log_line_prefix` for to keep a copy of 

804 the last printed log line prefix for to prevent the printout of duplicate characters in consecutive 

805 log lines. 

806 

807 :return: log file line prefix string including one space as separator character at the end. 

808 """ 

809 parts = list() 

810 if _multi_threading_activated: 

811 parts.append(f"<{threading.get_ident(): >6}>") 

812 if self.app_key[-1] != APP_KEY_SEP: 

813 parts.append(f"{{{self.app_key: <6}}}") 

814 if self._log_with_timestamp: 

815 format_string = DATE_TIME_ISO if isinstance(self._log_with_timestamp, bool) else self._log_with_timestamp 

816 parts.append(datetime.datetime.now().strftime(format_string)) 

817 if self.debug: 

818 parts.append(f"[{DEBUG_LEVELS[self.debug_level][0]}]") 

819 

820 prefix = "".join(parts) 

821 with log_file_lock: 

822 last_pre = self._last_log_line_prefix 

823 self._last_log_line_prefix = prefix 

824 

825 return hide_dup_line_prefix(last_pre, prefix) + " " 

826 

827 def log_file_check(self, curr_stream: Optional[TextIO] = None) -> Optional[TextIO]: 

828 """ check and possibly correct log file status and the passed currently used stream. 

829 

830 :param curr_stream: currently used stream. 

831 :return: stream passed into :paramref:`~log_file_check.curr_stream` or 

832 new/redirected stream of :paramref:`~log_file_check.curr_stream` or 

833 None if :paramref:`~log_file_check.curr_stream` is None. 

834 

835 For already opened log files check if the ae log file is big enough and if yes then do a file rotation. 

836 If log file is not opened but log file name got already set, then check if log startup buffer is active 

837 and if yes then create log file, pass log buffer content to log file and close the log buffer. 

838 """ 

839 old_stream = new_stream = None 

840 with log_file_lock: 

841 if self._log_file_stream: 

842 old_stream = self._log_file_stream 

843 self._log_file_stream.seek(0, 2) # due to non-posix-compliant Windows feature 

844 if self._log_file_stream.tell() >= self._log_file_size_max * 1024 * 1024: 

845 self._close_log_file() 

846 self._rename_log_file() 

847 self._open_log_file() 

848 new_stream = self._log_file_stream 

849 elif self._log_file_name: 

850 old_stream = self._log_buf_stream 

851 self._open_log_file() 

852 self._std_out_err_redirection(True) 

853 self._flush_and_close_log_buf() 

854 new_stream = self._log_file_stream 

855 elif self.suppress_stdout and not self._nul_std_out: # pragma: no cover/_std_out_err_redirection does it 

856 old_stream = sys.stdout 

857 sys.stdout = self._nul_std_out = new_stream = open(os.devnull, 'w') 

858 

859 if curr_stream == old_stream and new_stream: 

860 return new_stream 

861 return curr_stream 

862 

863 def print_out(self, *objects, file: Optional[TextIO] = None, **kwargs): 

864 """ app-instance-specific print-outs. 

865 

866 :param objects: objects to be printed out. 

867 :param file: output stream object to be printed to (def=None). Passing None on a main app instance 

868 will print the objects to the standard output and any active log files, but on a 

869 :class:`SubApp` instance with an active log file the print-out will get redirected 

870 exclusively/only to log file of this :class:`SubApp` instance. 

871 :param kwargs: All the other supported kwargs of this method are documented 

872 :func:`at the print_out() function of this module <print_out>`. 

873 

874 This method has an alias named :meth:`.po` 

875 """ 

876 if file is None and main_app_instance() is not self: 

877 with log_file_lock: 

878 file = self._log_buf_stream or self._log_file_stream 

879 if file: 

880 kwargs['file'] = file 

881 if 'app' not in kwargs: 

882 kwargs['app'] = self 

883 print_out(*objects, **kwargs) 

884 

885 po = print_out #: alias of method :meth:`.print_out` 

886 

887 def debug_out(self, *objects, minimum_debug_level: int = DEBUG_LEVEL_ENABLED, **kwargs): 

888 """ special debug version of :func:`builtin print() function <print>`. 

889 

890 This method will print-out the passed objects only if the :attr:`current debug level 

891 <.core.AppBase.debug_level>` of this app instance is higher than the value passed into the 

892 :paramref:`~debug_out.minimum_debug_level` argument. In this case the print-out will be 

893 delegated onto the :meth:`~.print_out`. 

894 

895 :param objects: objects to be printed out. 

896 :param minimum_debug_level: minimum debug level for to print the passed objects. 

897 :param kwargs: All the supported kwargs of this method 

898 are documented at the :func:`print_out() function <~.core.print_out>` 

899 of the :mod:`~.core` module (including the 

900 :paramref:`~.print_out.file` argument). 

901 

902 This method has an alias named :meth:`.dpo`. 

903 """ 

904 if self.debug_level >= minimum_debug_level: 

905 self.po(*objects, **kwargs) 

906 

907 dpo = debug_out #: alias of method :meth:`.debug_out` 

908 

909 def verbose_out(self, *objects, **kwargs): 

910 """ special verbose debug version of :func:`builtin print() function <print>`. 

911 

912 :param objects: objects to be printed out. 

913 :param kwargs: The :paramref:`~.core.AppBase.print_out.file` argument is documented 

914 at the :meth:`~.core.AppBase.print_out` method of the 

915 :class:`~.core.AppBase` class. All other supported kwargs of this method 

916 are documented at the :func:`print_out() function <~.core.print_out>` 

917 of the :mod:`~.core` module. 

918 

919 This method has an alias named :meth:`.vpo`. 

920 """ 

921 if self.debug_level >= DEBUG_LEVEL_VERBOSE: 

922 self.po(*objects, **kwargs) 

923 

924 vpo = verbose_out #: alias of method :meth:`.verbose_out` 

925 

926 def shutdown(self, exit_code: Optional[int] = 0, timeout: Optional[float] = None): 

927 """ shutdown this app instance and if it is the main app instance then also any created sub-app-instances. 

928 

929 :param exit_code: set application OS exit code - ignored if this is NOT the main app instance (def=0). 

930 Pass None for to prevent call of sys.exit(exit_code). 

931 :param timeout: timeout float value in seconds used for the thread termination/joining, for the 

932 :class:`SubApp` shutdowns and for the acquisition of the 

933 threading locks of :data:`the ae log file <log_file_lock>` and the :data:`app instances 

934 <app_inst_lock>`. 

935 """ 

936 if self._shut_down: 

937 return 

938 aqc_kwargs: Dict[str, Any] = dict(blocking=False) if timeout is None else dict(timeout=timeout) 

939 is_main_app_instance = main_app_instance() is self 

940 force = is_main_app_instance and exit_code # prevent deadlock on app error exit/shutdown 

941 

942 if exit_code is not None: 

943 self.po(f"#### Shutdown {self.app_name}.......... {exit_code if force else ''} {timeout}", logger=_logger) 

944 

945 a_blocked = (False if force else app_inst_lock.acquire(**aqc_kwargs)) 

946 if is_main_app_instance: 

947 _shut_down_sub_app_instances(timeout=timeout) 

948 if _multi_threading_activated: 

949 _join_app_threads(timeout=timeout) 

950 

951 l_blocked = (False if force else log_file_lock.acquire(**aqc_kwargs)) 

952 

953 self._flush_and_close_log_buf() 

954 self._close_log_file() 

955 if self._log_file_index: 

956 self._rename_log_file() 

957 

958 if self._nul_std_out: 

959 if not self._nul_std_out.closed: 

960 self._append_eof_and_flush_file(self._nul_std_out, "NUL stdout") 

961 self._nul_std_out.close() 

962 self._nul_std_out = None 

963 

964 if self.py_log_params: 

965 logging.shutdown() 

966 

967 self._std_out_err_redirection(False) 

968 

969 if l_blocked: 

970 log_file_lock.release() 

971 

972 _unregister_app_instance(self.app_key) 

973 if a_blocked: 

974 app_inst_lock.release() 

975 self._shut_down = True 

976 if is_main_app_instance and exit_code is not None: 

977 sys.exit(exit_code) # pragma: no cover (would break/cancel test run) 

978 

979 def _std_out_err_redirection(self, redirect: bool): 

980 """ enable/disable the redirection of the standard output/error TextIO streams if needed. 

981 

982 :param redirect: pass True to enable or False to disable the redirection. 

983 """ 

984 is_main_app_instance = main_app_instance() is self 

985 if redirect: 

986 if not isinstance(sys.stdout, _PrintingReplicator): # sys.stdout==ori_std_out not works with pytest/capsys 

987 if not self.suppress_stdout: 

988 std_out = ori_std_out 

989 elif self._nul_std_out and not self._nul_std_out.closed: 

990 std_out = self._nul_std_out # pragma: no cover - should never happen 

991 else: 

992 std_out = self._nul_std_out = open(os.devnull, 'w') 

993 sys.stdout = cast(TextIO, _PrintingReplicator(sys_out_obj=std_out)) 

994 sys.stderr = cast(TextIO, _PrintingReplicator(sys_out_obj=ori_std_err)) 

995 else: 

996 if is_main_app_instance: 

997 sys.stderr = ori_std_err 

998 sys.stdout = ori_std_out 

999 

1000 if is_main_app_instance: 

1001 if redirect: 

1002 faulthandler.enable(file=sys.stdout) 

1003 elif faulthandler.is_enabled(): 

1004 faulthandler.disable() # pragma: no cover (badly testable - would cancel/break test runs) 

1005 

1006 def _append_eof_and_flush_file(self, stream_file: TextIO, stream_name: str): 

1007 """ add special end-of-file marker and flush the internal buffers to the file stream. 

1008 

1009 :param stream_file: file stream. 

1010 :param stream_name: name of the file stream (only used for debugging/error messages). 

1011 """ 

1012 try: 

1013 try: 

1014 # cannot use print_out() here because of recursions on log file rotation, so use built-in print() 

1015 print(file=stream_file) 

1016 if self.debug: 

1017 print('EoF', file=stream_file) 

1018 except Exception as ex: # pragma: no cover - pylint: disable=broad-except 

1019 self.po(f"Ignorable {stream_name} end-of-file marker exception={ex}", logger=_logger) 

1020 

1021 stream_file.flush() 

1022 

1023 except Exception as ex: # pylint: disable=broad-except 

1024 self.po(f"Ignorable {stream_name} flush exception={ex}", logger=_logger) 

1025 

1026 def _flush_and_close_log_buf(self): 

1027 """ flush and close ae log buffer and pass content to log stream if opened. 

1028 """ 

1029 stream = self._log_buf_stream 

1030 if stream: 

1031 if self._log_file_stream: 

1032 self._append_eof_and_flush_file(stream, "ae log buf") 

1033 buf = stream.getvalue() + ("\n#### End Of Log Buffer" if self.debug else "") 

1034 self._log_file_stream.write(buf) 

1035 self._log_buf_stream = None 

1036 stream.close() 

1037 

1038 def _open_log_file(self): 

1039 """ open the ae log file with path and file name specified by :attr:`_log_file_name`. 

1040 

1041 Tries to create a log sub-folder - if specified in :attr:`_log_file_name` and 

1042 the folder does not exists (folder creation is limited to one folder level). 

1043 

1044 .. note:: 

1045 A already existing file with the same file name will be overwritten (file contents get lost!). 

1046 """ 

1047 log_dir = os.path.dirname(self._log_file_name) 

1048 if log_dir and not os.path.exists(log_dir): 

1049 os.mkdir(log_dir) 

1050 self._log_file_stream = open(self._log_file_name, "w", errors=DEF_ENCODE_ERRORS) 

1051 

1052 def _close_log_file(self): 

1053 """ close the ae log file. 

1054 """ 

1055 if self._log_file_stream: 

1056 stream = self._log_file_stream 

1057 self._append_eof_and_flush_file(stream, "ae log file") 

1058 self._log_file_stream = None 

1059 stream.close() 

1060 

1061 def _rename_log_file(self): 

1062 """ rename rotating log file while keeping first/startup log and log file count below :data:`MAX_NUM_LOG_FILE`. 

1063 """ 

1064 file_base, file_ext = os.path.splitext(self._log_file_name) 

1065 dfn = f"{file_base}-{self._log_file_index:0>{LOG_FILE_IDX_WIDTH}}{file_ext}" 

1066 if os.path.exists(dfn): 

1067 os.remove(dfn) # remove old log file from previous app run 

1068 if os.path.exists(self._log_file_name): # prevent errors after log file error or unit test cleanup 

1069 os.rename(self._log_file_name, dfn) 

1070 

1071 self._log_file_index += 1 

1072 if self._log_file_index > MAX_NUM_LOG_FILES: # use > instead of >= for to always keep first/startup log file 

1073 first_idx = self._log_file_index - MAX_NUM_LOG_FILES 

1074 dfn = f"{file_base}-{first_idx:0>{LOG_FILE_IDX_WIDTH}}{file_ext}" 

1075 if os.path.exists(dfn): 

1076 os.remove(dfn) 

1077 

1078 

1079class SubApp(AppBase): 

1080 """ separate/additional sub-app/thread/task with own/individual logging/debug configuration. 

1081 

1082 Create an instance of this class for every extra thread and task where your application needs separate 

1083 logging and/or debug configuration - additional to the main app instance. 

1084 

1085 All members of this class are documented at the :class:`AppBase` class. 

1086 """