ae.core

application core constants, helper functions and base classes

this module declares practical constants, tiny helper functions and app base classes, which are reducing the code of your application (and of other ae namespace modules/portions).

core constants

there are three debug level constants: DEBUG_LEVEL_DISABLED, DEBUG_LEVEL_ENABLED and DEBUG_LEVEL_VERBOSE. short names for all debug level constants are provided by the dict DEBUG_LEVELS. the debug level of your application can be either set in your code or optionally data-driven externally (using the config files or config options of the module ae.console).

to use the python logging module in conjunction with this module the constant LOGGING_LEVELS is providing a mapping between the debug levels and the python logging levels.

the encoding of strings into byte-strings (to output them to the console/stdout or to file contents) can be tricky sometimes. to not lose any logging output because of invalid characters this module will automatically handle any UnicodeEncodeError exception for you. invalid characters will then automatically be converted to the default encoding (specified by DEF_ENCODING) with the default error handling method specified by DEF_ENCODE_ERRORS (both defined in the ae.base namespace portion/module).

core helper functions

the print_out() function, which is fully compatible to pythons print(), is using the encode helpers force_encoding() and to_ascii() to auto-correct invalid characters.

the function hide_dup_line_prefix() is very practical if you want to remove or hide redundant line prefixes in your log files, to make them better readable.

application base classes

the classes AppBase and SubApp are applying logging and debugging features to your application. create in your application one instance of AppBase to represent the main application task. if your application needs a separate logging/debugging configuration for sub-threads or sub-tasks then create an SubApp instance for each of these sub-apps.

sub-apps are very flexible and not tied to any fix use-case. they can be created e.g. for each sub-task or application thread. you could also create a SubApp instance for each of your external systems, like a database server or to connect your application onto different test environments or to your live/production system (e.g. for system comparison and maintenance).

both application classes are automatically catching and handling any exceptions and run-time errors: only if any critical exception/error cannot be handled then the shutdown() method will make sure that all sub-apps and threads get terminated and joined. additionally all print-out buffers will be flushed to include all the info of the critical error (the last debug and error messages) into the standard error/output and into any activated log files.

basic usage of an application base class

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

"""  docstring at the top of the main module of your application  """
from ae.core import AppBase

__version__ = '1.2.3'

ca = AppBase()

in the above example the AppBase instance will automatically use the docstring title of the module as application title and the string in the module variable __version___ as application version. to overwrite these defaults pass your application title and version string via the arguments app_title and app_version to the instantiation of AppBase:

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

other automatically initialized instance attributes of AppBase are documented underneath in the class docstring. they include e.g. the date and time when the instance got created, the name/id of this application instance or the application path.

application class hierarchy

for most use cases you will not instantiate from AppBase directly - instead you will instantiate one of the extended application classes that are inherited from this base class.

the class ConsoleApp e.g. inherits from AppBase and is adding configuration options and variables to it. so in your console application it is recommended to directly use instances of ConsoleApp instead of AppBase.

for applications with an GUI use instead one of the classes KivyMainApp, EnamlMainApp or TogaMainApp.

application logging

print-outs are an essential tool for the debugging and logging of your application at run-time. in python the print-outs are done with the print() function or with the python logging module. these print-outs get per default send to the standard output and error streams of your OS and so displayed on your system console/shell. the print_out() function and the print_out() method of this core module are adding two more sophisticated ways for print-outs to the console/log-files.

using AppBase is making the logging much easier and also ensures that print-outs of any imported library or package will be included within your log files. this is done by redirecting the standard output and error streams to your log files with the help of the _PrintingReplicator class.

head-less server applications like web servers are mostly not allowed to use the standard output streams. for some these applications you could redirect the standard output and error streams to a log file by using the OS redirection character (>):

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

but because most web servers doesn’t allow you to use this redirection, you can alternatively specify the suppress_stdout parameter as True in the instantiation of an AppBase instance. additionally you can call the init_logging() method to activate a log file. after that all print-outs of your application and libraries will only appear in your log file.

also in complex applications, where huge print-outs to the console can get lost easily, you want to use a log file instead. but even a single log file can get messy to read, especially for multithreading server applications. for that SubApp is allowing you to create for each thread a separate sub-app instance with its own log file.

using this module ensures that any crashes or freezes happening in your application will be fully logged. apart from the gracefully handling of UnicodeEncodeError exceptions, the Python faulthandler will be enabled automatically to catch system errors and to dump a traceback of them to the console and any activated log file.

activate ae log file

ae log files are text files using by default the encoding of your OS console/shell. to activate the redirection of your applications print-outs into an ae log file for a AppBase instance you simply specify the file name of the log file in the init_logging() method call:

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

activate ae logging features

for multi-threaded applications include the thread-id of the printing thread automatically into your log files by passing a True value to the multi_threading argument. to additionally also suppress any print-outs to the standard output/error streams pass True to the suppress_stdout argument:

app = AppBase(multi_threading=True, suppress_stdout=True)
app.init_logging(log_file_name='my_log_file.log')

the ae log files provided by this module are automatically rotating if the size of a log file succeeds the value in MBytes defined in the LOG_FILE_MAX_SIZE. to adapt this value to your needs you can specify the maximum log file size in MBytes with the argument log_file_size_max in your call of init_logging():

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

by using the ConsoleApp class instead of AppBase you can alternatively store the logging configuration of your application within a configuration variable or a configuration option. the order of precedence to find the appropriate logging configuration of each app instance is documented here.

using python logging module

if you prefer to use instead the python logging module for the print-outs of your application, then pass a python logging configuration dictionary with the individual configuration of your logging handlers, files and loggers to the py_logging_params argument of the init_logging() method:

app.init_logging(py_logging_params=my_py_logging_config_dict)

passing the python logging configuration dictionary to one of the AppBase instances created by your application will automatically disable the ae log file of this instance.

application debugging

to use the debug features of core you simply have to import the needed debug level constant to pass it at instantiation of your AppBase or SubApp class to the debug_level argument:

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

by passing DEBUG_LEVEL_ENABLED the print-outs (and log file contents) will be more detailed, and even more verbose if you use instead the debug level DEBUG_LEVEL_VERBOSE.

the debug level can be changed at any time in your application code by directly assigning the new debug level to the debug_level property. if you prefer to change the debug levels dynamically, then use the ConsoleApp instead of AppBase, because ConsoleApp provides this property as a configuration file variable and commend line option. this way you can specify the actual debug level without the need to change (and re-build) your application code.

Module Attributes

DEBUG_LEVEL_DISABLED

lowest debug level - only display logging levels ERROR/CRITICAL.

DEBUG_LEVEL_ENABLED

minimum debugging info - display logging levels WARNING or higher.

DEBUG_LEVEL_VERBOSE

verbose debug info - display logging levels INFO/DEBUG or higher.

DEBUG_LEVELS

numeric ids and names of all supported debug levels.

LOGGING_LEVELS

association between ae debug levels and python logging levels.

HIDDEN_CREDENTIALS

credential keys that are hidden in print/repr output (not if verbose)

MAX_NUM_LOG_FILES

maximum number of ae log files

LOG_FILE_MAX_SIZE

max.

LOG_FILE_IDX_WIDTH

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

ori_std_out

original sys.stdout on app startup

ori_std_err

original sys.stderr on app startup

log_file_lock

log file rotation multi-threading lock

po(*objects[, sep, end, file, flush, ...])

alias of function print_out()

APP_KEY_SEP

separator character used in app_key of AppBase instance

app_inst_lock

app instantiation multi-threading lock

Functions

activate_multi_threading()

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

hide_dup_line_prefix(last_line, current_line)

replace duplicate characters at the start of two strings with spaces.

logger_late_init()

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

main_app_instance()

determine the main instance of the AppBase in the current running application.

po(*objects[, sep, end, file, flush, ...])

alias of function print_out()

print_out(*objects[, sep, end, file, flush, ...])

universal/unbreakable print function - replacement for the built-in python function print().

registered_app_names()

determine the app names of all registered/running applications.

Classes

AppBase([app_title, app_name, app_version, ...])

provides easy logging and debugging for your application.

SubApp([app_title, app_name, app_version, ...])

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

DEBUG_LEVEL_DISABLED: int = 0

lowest debug level - only display logging levels ERROR/CRITICAL.

DEBUG_LEVEL_ENABLED: int = 1

minimum debugging info - display logging levels WARNING or higher.

DEBUG_LEVEL_VERBOSE: int = 2

verbose debug info - display logging levels INFO/DEBUG or higher.

DEBUG_LEVELS: Dict[int, str] = {0: 'disabled', 1: 'enabled', 2: 'verbose'}

numeric ids and names of all supported debug levels.

LOGGING_LEVELS: Dict[int, int] = {0: 30, 1: 20, 2: 10}

association between ae debug levels and python logging levels.

HIDDEN_CREDENTIALS = ('password', 'token')

credential keys that are hidden in print/repr output (not if verbose)

hide_dup_line_prefix(last_line, current_line)[source]

replace duplicate characters at the start of two strings with spaces.

Parameters:
  • last_line (str) – last line string (e.g. the last line of text/log file).

  • current_line (str) – current line string.

Return type:

str

Returns:

current line string but duplicate characters at the beginning are replaced by space chars.

MAX_NUM_LOG_FILES: int = 69

maximum number of ae log files

LOG_FILE_MAX_SIZE: int = 15

max. size in MB of rotating ae log files

LOG_FILE_IDX_WIDTH: int = 5

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

ori_std_out: TextIO = <_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>

original sys.stdout on app startup

ori_std_err: TextIO = <_io.TextIOWrapper name='<stderr>' mode='w' encoding='utf-8'>

original sys.stderr on app startup

log_file_lock: RLock = <unlocked _thread.RLock object owner=0 count=0>

log file rotation multi-threading lock

_LOGGER = None

python logger for this module gets lazy/late initialized and only if requested by caller

logger_late_init()[source]

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

_MULTI_THREADING_ACTIVATED: bool = False

flag if threading is used in your application

activate_multi_threading()[source]

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

_deactivate_multi_threading()[source]

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

print_out(*objects, sep=' ', end='\\n', file=None, flush=False, encode_errors_def='backslashreplace', logger=None, app=None, **kwargs)[source]

universal/unbreakable print function - replacement for the built-in python function print().

Parameters:
  • objects – tuple of objects to be printed. if the first object is a string that starts with a \r character then the print-out will be only sent to the standard output (and will not be added to any active log files - see also end argument).

  • sep (str) – separator character between each printed object/string (def=” “).

  • end (str) – finalizing character added to the end of this print-out (def=”\n”). pass \r to suppress the print-out into ae log file or to any activated python logger - useful for console/shell processing animation (see tcp.TcpServer.run()).

  • file (Optional[TextIO]) – output stream object to be printed to (def=None which will use standard output streams). if given then the redirection to all active log files and python logging loggers will be disabled (even if the logger argument is specified).

  • flush (bool) – flush stream after printing (def=False).

  • encode_errors_def (str) – default error handling to encode (def=:data:DEF_ENCODE_ERRORS).

  • logger (Optional[Logger]) – used logger to output objects (def=None). ignored if the file argument gets specified/passed.

  • app (Optional[AppBase]) – the app instance from where this print-out got initiated.

  • kwargs – catch unsupported kwargs for debugging (all items will be printed to all the activated logging/output streams).

this function is silently handling and autocorrecting string encode errors for output/log streams which are not supporting unicode. any instance of AppBase is providing this function as a method with the same name). it is recommended to call/use this instance method instead of this function.

in multithreading applications this function prevents dismembered/fluttered print-outs from different threads.

Note

this function has an alias named po().

po(*objects, sep=' ', end='\\n', file=None, flush=False, encode_errors_def='backslashreplace', logger=None, app=None, **kwargs)

alias of function print_out()

APP_KEY_SEP: str = '@'

separator character used in app_key of AppBase instance

_APP_INSTANCES: WeakValueDictionary[str, AppBase] = <WeakValueDictionary>

dict that is weakly holding references to all AppBase instances created at run time.

gets automatically initialized in AppBase.__init__() to allow log file split/rotation and debug_level access at application thread or module level.

the first created AppBase instance is called the main app instance. _MAIN_APP_INST_KEY stores the dict key of the main instance.

_MAIN_APP_INST_KEY: str = ''

key in _APP_INSTANCES of main AppBase instance

app_inst_lock: RLock = <unlocked _thread.RLock object owner=0 count=0>

app instantiation multi-threading lock

main_app_instance()[source]

determine the main instance of the AppBase in the current running application.

Return type:

Optional[AppBase]

Returns:

main/first-instantiated AppBase instance or None (if app is not fully initialized yet).

registered_app_names()[source]

determine the app names of all registered/running applications.

Return type:

List[str]

_register_app_instance(app)[source]

register new AppBase instance in _APP_INSTANCES.

Parameters:

app (AppBase) – AppBase instance to register

_unregister_app_instance(app_key)[source]

unregister/remove AppBase instance from within _APP_INSTANCES.

Parameters:

app_key (str) – app key of the instance to remove.

Return type:

Optional[AppBase]

Returns:

removed AppBase instance.

_shut_down_sub_app_instances(timeout=None)[source]

shut down all SubApp instances.

Parameters:

timeout (Optional[float]) – timeout float value in seconds used for the SubApp shutdowns and for the acquisition of the threading locks of the ae log file and the app instances.

class _PrintingReplicator(sys_out_obj=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)[source]

Bases: object

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

__init__(sys_out_obj=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>)[source]

initialise a new T-stream-object

Parameters:

sys_out_obj (TextIO) – standard output/error stream to be replicated (def=sys.stdout)

write(any_str)[source]

write string to ae logging and standard output streams.

automatically suppressing UnicodeEncodeErrors if console/shell or log file has different encoding by forcing re-encoding with DEF_ENCODE_ERRORS.

Parameters:

any_str (Union[str, bytes]) – string or bytes to output.

Return type:

None

__getattr__(attr)[source]

get attribute value from standard output stream.

Parameters:

attr (str) – name of the attribute to retrieve/return.

Return type:

Any

Returns:

value of the attribute.

_APP_THREADS: WeakValueDictionary[int, Thread] = <WeakValueDictionary>

weak dict to keep the references of all application threads. added to prevent the joining of unit testing threads in the test teardown (resetting app environment).

_register_app_thread()[source]

add new app thread to _APP_THREADS if not already added.

_join_app_threads(timeout=None)[source]

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

Parameters:

timeout (Optional[float]) – timeout float value in seconds for thread joining (def=None - block/no-timeout).

Note

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

class AppBase(app_title='', app_name='', app_version='', sys_env_id='', debug_level=0, multi_threading=False, suppress_stdout=False)[source]

Bases: object

provides easy logging and debugging for your application.

most applications only need a single instance of this class; apps with threads could create separate instances for each thread.

instance Attributes (ordered alphabetically - ignoring underscore characters):

_last_log_line_prefix: str = ''

prefix of the last printed log line

_log_buf_stream: Optional[StringIO] = None

log file buffer stream instance

_log_file_stream: Optional[TextIO] = None

log file stream instance

_log_file_index: int = 0

log file index (for rotating logs)

_log_file_size_max: float = 15

maximum log file size in MBytes (rotating log files)

_log_file_name: str = ''

log file name

_log_with_timestamp: Union[bool, str] = False

True of strftime format string to enable timestamp

_nul_std_out: Optional[TextIO] = None

logging null stream

py_log_params: Dict[str, Any] = {}

dict of config parameters for py logging

_shut_down: bool = False

True if this app instance got shut down already

__init__(app_title='', app_name='', app_version='', sys_env_id='', debug_level=0, multi_threading=False, suppress_stdout=False)[source]

initialize a new AppBase instance.

Parameters:
  • app_title (str) – application title/description setting the attribute app_title. if not specified then the docstring of your app’s main module will be used (see example).

  • app_name (str) – application instance name to set the attribute app_name. if not specified then base name of the main module file name will be used.

  • app_version (str) – application version string to set the attribute app_version. if not specified then value of a global variable with the name __version__ will be used (if declared in the actual call stack).

  • sys_env_id (str) – system environment id to set the instance attribute sys_env_id. the default value of this argument is an empty string.

  • debug_level (int) – default debug level to set the instance attribute debug_level. the default value of this argument is DEBUG_LEVEL_DISABLED.

  • multi_threading (bool) – pass True if instance is used in multi-threading app.

  • suppress_stdout (bool) – pass True (for wsgi apps) to prevent any python print outputs to stdout.

startup_beg: datetime

begin of app startup datetime

app_path: str

path to folder of your main app code file

app_title: str = ''

title/description of this app instance

app_name: str = ''

name of this app instance

app_version: str = ''

version of this app instance

_debug_level: int = 2

debug level of this app instance

sys_env_id: str = ''

system environment id of this app instance

suppress_stdout: bool = True

flag to suppress prints to stdout

startup_end: Optional[datetime] = None

end datetime of the application startup

__del__()[source]

deallocate this app instance by calling AppBase.shutdown().

property active_log_stream: StringIO | TextIO | None

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

Returns:

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

property app_key: str

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

Returns:

application key string.

property debug_level: int

debug level property:

Getter:

return the current debug level of this app instance.

Setter:

change the debug level of this app instance.

property debug: bool

True if app is in debug mode.

property verbose: bool

True if app is in verbose debug mode.

call_method(callback, *args, **kwargs)[source]

call passed callable/method with the passed args, catching and logging exceptions preventing app exit.

Parameters:
  • callback (Union[Callable, str]) – either a callable or the name of the main app method of this instance to call.

  • args – args passed to the main app method to be called.

  • kwargs – kwargs passed to the main app method to be called.

Return type:

Any

Returns:

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

init_logging(py_logging_params=None, log_file_name='', log_file_size_max=15, log_with_timestamp=False, disable_buffering=False)[source]

initialize logging system.

Parameters:
  • py_logging_params (Optional[Dict[str, Any]]) – config dict for python logging configuration. if this dict is not empty then python logging is configured with the given options in this dict and all the other kwargs are ignored.

  • log_file_name (str) – default log file name for ae logging (def=’’ - ae logging disabled).

  • log_file_size_max (float) – max. size in MB of ae log file (def=LOG_FILE_MAX_SIZE).

  • log_with_timestamp (Union[bool, str]) – add timestamp prefix to each log line if True or a non-empty strftime compatible format string.

  • disable_buffering (bool) – pass True to disable ae log buffering at app startup.

log files and config values will be initialized as late as possible in log_file_check(), e.g. indirectly triggered by a request to a config variable via _parse_args() (like logFile).

log_line_prefix()[source]

compile prefix of log print-out line for this AppBase instance.

the line prefix consists of (depending on the individual values of either a module variable or of an attribute this app instance):

  • _MULTI_THREADING_ACTIVATED: if True then the thread id gets printed surrounded with angle brackets (< and >), right aligned and space padded to minimal 6 characters.

  • sys_env_id: if not empty then printed surrounded with curly brackets ({ and }), left aligned and space padded to minimal 4 characters.

  • _log_with_timestamp: if (a) True or (b) a non-empty string then the system time (determined with now()) gets printed in the format specified either by the (a) the DATE_TIME_ISO constant or (b) by the string in this attribute.

this method is using the instance attribute _last_log_line_prefix to keep a copy of the last printed log line prefix to prevent the printout of duplicate characters in consecutive log lines.

Return type:

str

Returns:

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

log_file_check(curr_stream=None)[source]

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

Parameters:

curr_stream (Optional[TextIO]) – currently used stream.

Return type:

Optional[TextIO]

Returns:

stream passed into curr_stream or new/redirected stream of curr_stream or None if curr_stream is None.

for already opened log files check if the ae log file is big enough and if yes then do a file rotation. if log file is not opened but log file name got already set, then check if log startup buffer is active and if yes then create log file, pass log buffer content to log file and close the log buffer.

print_out(*objects, file=None, **kwargs)[source]

app-instance-specific print-outs.

Parameters:
  • objects – objects to be printed out.

  • file (Optional[TextIO]) – output stream object to be printed to (def=None). passing None on a main app instance will print the objects to the standard output and any active log files, but on a SubApp instance with an active log file the print-out will get redirected exclusively/only to log file of this SubApp instance.

  • kwargs – all the other supported kwargs of this method are documented at the print_out() function of this module.

this method has an alias named po()

po(*objects, file=None, **kwargs)

alias of method print_out()

debug_out(*objects, minimum_debug_level=1, **kwargs)[source]

special debug version of builtin print() function.

this method will print out the passed objects only if the current debug level of this app instance is higher than the value passed into the minimum_debug_level argument. in this case the print-out will be delegated onto the print_out().

Parameters:
  • objects – objects to be printed out.

  • minimum_debug_level (int) – minimum debug level to print the passed objects.

  • kwargs – all the supported kwargs of this method are documented at the print_out() function of the core module (including the file argument).

this method has an alias named dpo().

dpo(*objects, minimum_debug_level=1, **kwargs)

alias of method debug_out()

verbose_out(*objects, **kwargs)[source]

special verbose debug version of builtin print() function.

Parameters:
  • objects – objects to be printed out.

  • kwargs – the file argument is documented at the print_out() method of the AppBase class. all other supported kwargs of this method are documented at the print_out() function of the core module.

this method has an alias named vpo().

vpo(*objects, **kwargs)

alias of method verbose_out()

shutdown(exit_code=0, timeout=None)[source]

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

Parameters:
  • exit_code (Optional[int]) – set application OS exit code - ignored if this is NOT the main app instance (def=0). pass None to prevent call of sys.exit(exit_code).

  • timeout (Optional[float]) – timeout float value in seconds used for the thread termination/joining, for the SubApp shutdowns and for the acquisition of the threading locks of the ae log file and the app instances.

_std_out_err_redirection(redirect)[source]

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

Parameters:

redirect (bool) – pass True to enable or False to disable the redirection.

_append_eof_and_flush_file(stream_file, stream_name)[source]

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

Parameters:
  • stream_file (TextIO) – file stream.

  • stream_name (str) – name of the file stream (only used for debugging/error messages).

_flush_and_close_log_buf()[source]

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

_open_log_file()[source]

open the ae log file with path and file name specified by _log_file_name.

tries to create a log sub-folder - if specified in _log_file_name and the folder does not exist (folder creation is limited to one folder level).

Note

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

_close_log_file()[source]

close the ae log file.

_rename_log_file()[source]

rename rotating log file while keeping first/startup log and log file count below MAX_NUM_LOG_FILE.

class SubApp(app_title='', app_name='', app_version='', sys_env_id='', debug_level=0, multi_threading=False, suppress_stdout=False)[source]

Bases: AppBase

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

create an instance of this class for every extra thread and task where your application needs separate logging and/or debug configuration - additional to the main app instance.

all members of this class are documented at the AppBase class.

startup_beg: datetime

begin of app startup datetime

app_path: str

path to folder of your main app code file