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
lowest debug level - only display logging levels ERROR/CRITICAL. |
|
minimum debugging info - display logging levels WARNING or higher. |
|
verbose debug info - display logging levels INFO/DEBUG or higher. |
|
numeric ids and names of all supported debug levels. |
|
association between ae debug levels and python logging levels. |
|
credential keys that are hidden in print/repr output (not if verbose) |
|
maximum number of ae log files |
|
max. |
|
width of rotating log file index within log file name; adding +3 to ensure index range up to factor 10^3. |
|
original sys.stdout on app startup |
|
original sys.stderr on app startup |
|
log file rotation multi-threading lock |
|
|
alias of function |
app instantiation multi-threading lock |
Functions
activate multi-threading for all app instances (normally done at main app startup). |
|
|
replace duplicate characters at the start of two strings with spaces. |
check if logging modules got initialized already and if not then do it now. |
|
determine the main instance of the |
|
|
alias of function |
|
universal/unbreakable print function - replacement for the |
determine the app names of all registered/running applications. |
Classes
|
provides easy logging and debugging for your application. |
|
separate/additional sub-app/thread/task with own/individual logging/debug configuration. |
- 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.
- 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.
- 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 (seetcp.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 thelogger
argument is specified).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 thefile
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 thesame 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_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 mainAppBase
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.
- _register_app_instance(app)[source]
register new
AppBase
instance in_APP_INSTANCES
.
- _unregister_app_instance(app_key)[source]
unregister/remove
AppBase
instance from within_APP_INSTANCES
.
- _shut_down_sub_app_instances(timeout=None)[source]
shut down all
SubApp
instances.- Parameters:
timeout¶ (
Optional
[float
]) – timeout float value in seconds used for theSubApp
shutdowns and for the acquisition of the threading locks ofthe ae log file
and theapp 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
- _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).
- _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):
app_key
id/key of this application instance.app_name
basename (without the file name extension) of the executable.app_path
file path of app executable.app_title
application title/description.app_version
application version (set via theAppBase.app_version
argument).debug_level
debug level of this instance._last_log_line_prefix
last ae log file line prefix that got print-out to the log of this app instance._log_buf_stream
ae log file buffer stream._log_file_index
index of the current rotation ae log file backup._log_file_name
path and file name of the ae log file._log_file_size_max
maximum size in MBytes of an ae log file._log_file_stream
ae log file TextIO output stream._log_with_timestamp
log timestamp line prefix if True or a non-empty strftime compatible format string.py_log_params
python logging config dictionary._nul_std_out
null stream used to prevent print-outs tostandard output
._shut_down
flag set to True if this application instance got already shutdown.startup_beg
datetime of begin of the instantiation/startup of this app instance.startup_end
datetime of end of the instantiation/startup of this application instance.suppress_stdout
flag set to True if this application does not print to stdout/console.sys_env_id
system environment id of this application instance.
- __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 attributeapp_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 attributeapp_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 attributeapp_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 attributesys_env_id
. the default value of this argument is an empty string.debug_level¶ (
int
) – default debug level to set the instance attributedebug_level
. the default value of this argument isDEBUG_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.
- __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.
- call_method(callback, *args, **kwargs)[source]
call passed callable/method with the passed args, catching and logging exceptions preventing app exit.
- Parameters:
- Return type:
- 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 withnow()
) gets printed in the format specified either by the (a) theDATE_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:
- 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:
- Return type:
- Returns:
stream passed into
curr_stream
or new/redirected stream ofcurr_stream
or None ifcurr_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 aSubApp
instance with an active log file the print-out will get redirected exclusively/only to log file of thisSubApp
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 theminimum_debug_level
argument. in this case the print-out will be delegated onto theprint_out()
.- Parameters:
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:
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 theSubApp
shutdowns and for the acquisition of the threading locks ofthe ae log file
and theapp instances
.
- _std_out_err_redirection(redirect)[source]
enable/disable the redirection of the standard output/error TextIO streams if needed.
- _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.
- _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!).
- 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.