xtd package

Introduction

What is XTD ?

XTD is a library that provides a very high levels set of tools designed to efficiently develop industrial-level python softwares.

Features

  • Static application
    • Unified command line & file configuration
    • Ready to use logging facility
    • Crash-persistent memory data
    • Statistic measurements and output
  • Web application
    • Everything included in static application
    • Ready to use web server (cherrypy based)
    • HTTP api to access
      • logs
      • statistics
      • persistent parameters

Compatibility

Warning

Python 3.x

Installation

sudo pip3 install xtd

Get Started

Basic Application

XTD is designed to develop your software by inheriting the main Application object.

Example :

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
import xtd.core.application

class MergeApplication(xtd.core.application.Application):
  def __init__(self):
    super().__init__("MyAppName")
    # register input options

  def initialize(self):
    super().initialize()
    # do some initialization suff

  def process(self):
    l_exitCode = 0
    # do some stuff
    return l_exitCode, True

MergeApplication().execute()

Input options

XTD provides a way to declare and read command-line and file configuration options in a unified way. An option:

  • is registered with an unique name and belongs to a section
  • is attached to zero-or-more checks that will validate input values

Command-line & File config

User can provides values to options in 3 different ways :

  • from internal default value
  • from command line with --<section>-<name> VALUE option
  • from program’s json configuration file with {"<section>": {"<name>": VALUE}}

When multiple values are available, they are taken with the following order of priority (from lowest to highest) :

  1. registered default value
  2. value in configuration file
  3. value on command-line

Registering options

Arguments are registered with the register() and register_section() methods of the ConfigManager (singleton) object

This ConfigManager is accessible via the config() method of your Application or directly from the singleton.

Note

Full option documentation available at Option

Standard check functions checkers

The following code:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import xtd.core.application
import xtd.core.config
import xtd.core.logger
import xtd.core.config.checkers

class MergeApplication(xtd.core.application.Application):
  def __init__(self):
    super().__init__("MergeApplication")

    self.m_config.register_section("input", "Input settings", [{
      "name"        : "directory",
      "default"     : "./work_to_do/",
      "description" : "Read input files from given directory",
      "checks"      : xtd.core.config.checkers.is_dir(p_write=True)
    },{
      "name"        : "count",
      "default"     : 1,
      "description" : "Number of file to read in directory",
      "checks"      : xtd.core.config.checkers.is_int(p_min=0)
    }])

    self.m_config.register_section("output", "Output settings", [{
      "name"        : "file",
      "description" : "Destination file",
      "checks"      : xtd.core.config.checkers.is_file(p_write=True)
    }])


if __name__ == "__main__":
  l_app = MergeApplication()
  l_app.execute()

Produces :

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
$ python3 test.py --help

Usage: MergeApplication [options]

Options:
  --help                       show this help message and exit

  General Settings:
    --config-file=ARG          use FILE as configuration file [default:MergeApplication/MergeApplication.json]

  Input settings:
    --input-directory=ARG      Read input files from given directory [default:./work_to_do/]
    --input-count=ARG          Number of file to read in directory [default:1]

  Logging Settings:
    --log-config=ARG           Logging configuration [default:{}]
    --log-override=ARG         override part of logging configuration [default:{}]

  Output settings:
    --output-file              Destination file [default:False]

  Persitent Param Settings:
    --param-directory=ARG      Destination directory for admin persistent parameters [default:/tmp/snmp/MergeApplication/admin]

  Stats Settings:
    --stat-writters=ARG        Enabled given stat output writter. Possibles values :
                               * disk : write counters to --stat-disk-directory path each --stat-disk-interval seconds
                               * http : post counters in json format to --stat-http-url url each --stat-http-interval seconds
                               Can specify a comma separated combinaison of theses values
                               [default:['disk']]
    --stat-disk-directory=ARG  Destination directory for counter disk writter [default:/tmp/snmp/MergeApplication/stat/]
    --stat-disk-interval=ARG   Interval in second between two disk outputs [default:50]
    --stat-http-url=ARG        Destination POST url for http stat writter [default:http://localhost/counter]
    --stat-http-interval=ARG   Interval in second between two http outputs [default:50]

Reading options

Nothing simpler than reading option values.

From your Application object:

import xtd.core.application

class MyApp(xtd.core.application.Application):
  def process(self):
    l_inputDir   = self.config().get("input", "directory")
    l_count      = self.config().get("input", "count")
    l_outputFile = self.config().get("output", "file")

Or, from anywhere in your code :

from xtd.core import config

def my_function(self):
  l_inputDir   = config.get("input", "directory")
  l_count      = config.get("input", "count")
  l_outputFile = config.get("output", "file")

Note

The xtd.core.application.Application.initialize() method has to be called before you attempt to read options values.

Logging

Features

  • standard python logging module compliant
  • logger.<level>(<module>, <message>) primitives in addition to the standard logging.getLogger(<module>).<level>(<message>) functions
  • rich default configuration including:
    • standard SysLogHandler bound to /dev/log socket
    • standard RotatingFileHandler bound to ./out.log
    • standard StreamHandler bound to sys.stdout This particular handler is setup with a special Formatter LocationFormatter who pads fields to help with vertical reading and colorizes record fields depending on their value

Configuration

By default, Application defines two unified options that changes the logging behavior:

  • --log-config: full logging configuration in json format. Details about configuration format is available in object LogManager
  • --log-override: partial logging configuration that will be merged on top of full configuration. This option respect the same format as the full format except that you may only specify parts. This is useful when you want to override the log level for a specific module on the command line while the rest of the configuration is in the configuration file

Note

Even we use command line options, keep in mind that they always have their configuration file equivalent

Example:

$ # activate log-level 10 (debug) for module "a.b.c"
$ python myapp.py --log-override='{"handler" : {"a.b.c" : { "level" : 10 } } }'

Persistent data

Todo

some doc

Statistics

Todo

some doc

Subpackages

xtd.core package

Subpackages

xtd.core.config package
xtd.core.config.get(p_section, p_name)[source]
xtd.core.config.set(p_section, p_name, p_value)[source]
xtd.core.config.sections()[source]
xtd.core.config.section_exists(p_section)[source]
xtd.core.config.options(p_section)[source]
xtd.core.config.option_exists(p_section, p_name)[source]
Submodules
xtd.core.config.checkers module
xtd.core.config.checkers.check_file(p_section, p_name, p_value, p_read=False, p_write=False, p_execute=False)[source]

check that given config parameter is a valid file with given rwx attributes

If p_value does not exists and only write attribute is requested, the function checks that the file can be created in its parent directory

Parameters:
  • p_section (str) – parameter section name
  • p_name (str) – parameter name
  • p_value (str) – parameter value
  • p_read (str) – target file should be readable, default False
  • p_write (str) – target file should be writable, default False
  • p_execute (str) – target file should be executable, default False
Returns:

file absolute path
Return type:

bool
Raises:
  • ConfigValueFileError – p_value is a directory
  • ConfigValueFileModeError – p_value dosen’t meet requested rwx attributes
xtd.core.config.checkers.check_dir(p_section, p_name, p_value, p_read=False, p_write=False, p_execute=False)[source]

check that given value is a valid directory for given rwx attributes

Parameters:
  • p_section (str) – parameter section name
  • p_name (str) – parameter name
  • p_value (str) – target directory path
  • p_read (str) – target file should be readable, default False
  • p_write (str) – target file should be writable, default False
  • p_execute (str) – target file should be executable, default False
Returns:

directory absolute path
Return type:

bool
Raises:
  • ConfigValueDirError – p_value is not a directory
  • ConfigValueDirModeError – directory doesn’t meet requested rwx attributes
xtd.core.config.checkers.check_int(p_section, p_name, p_value, p_min=None, p_max=None)[source]

check that given value is a valid integer

If not None, the function will insure that value fits the requested minimum and maximum parameters

Parameters:
  • p_section (str) – parameter section name
  • p_name (str) – parameter name
  • p_value (str) – target value
  • p_min (int) – minimum accepted value, default : None
  • p_max (int) – maximum accepted value, default : None
Returns:

integer converted value
Return type:

int
Raises:
  • ConfigValueTypeError – value is not an integer, nor a int-convertible string
  • ConfigValueLimitsError – value doesn’t match requested min and max constraints
xtd.core.config.checkers.check_float(p_section, p_name, p_value, p_min=None, p_max=None)[source]

check that given value is a valid float

Same as check_int()

Parameters:
  • p_section (str) – parameter section name
  • p_name (str) – parameter name
  • p_value (str) – target value
  • p_min (float) – minimum accepted value, default : None
  • p_max (float) – maximum accepted value, default : None
Raises:
  • ConfigValueTypeError – value is not an integer, nor a int-convertible string
  • ConfigValueLimitsError – value doesn’t match requested min and max constraints
Returns:

float converted value
Return type:

float
xtd.core.config.checkers.check_bool(p_section, p_name, p_value)[source]

check that given value is a valid boolean

Valid boolean are :
  • native bool object
  • str object in the following list : ["on", "yes", "true", "off", "no", "false"] (case insensitive)
Parameters:
  • p_section (str) – parameter section name
  • p_name (str) – parameter name
  • p_value (str) – target value
Raises:

ConfigValueTypeError – invalid input boolean
Returns:

converted value
Return type:

bool
xtd.core.config.checkers.check_enum(p_section, p_name, p_value, p_values)[source]

check that given value matches a set of possible values

Parameters:
  • p_section (str) – parameter section name
  • p_name (str) – parameter name
  • p_value (str) – input value
  • p_values (list) – set of possible authorized values
Raises:

ConfigValueEnumError – value not found in possible values
Returns:

input value
Return type:

bool
xtd.core.config.checkers.check_mail(p_section, p_name, p_value)[source]

check that given value is a syntactical valid mail address

Parameters:
  • p_section (str) – parameter section name
  • p_name (str) – parameter name
  • p_value (str) – input value
Raises:

ConfigValueValueError – value not an email
Returns:

input value
Return type:

bool
xtd.core.config.checkers.check_array(p_section, p_name, p_value, p_check=None, p_delim=', ')[source]

check that given value is convertible to array

A str value is converted to array by splitting each comma separated elements

Additionally, the function checks that each elements meets p_check function requirement.

Example:

l_value = "1,2,3,4,5"
l_value = check_array(l_value, check_float)
print(l_value)
# [1.0, 2.0, 3.0, 4.0]

l_value = "on,on;off,no;true,false"
l_value = check_array(l_value, is_array(p_check=check_bool), p_delim=";")
print(l_value)
# [[True, True], [False, False], [True, False]]
Parameters:
  • p_section (str) – parameter section name
  • p_name (str) – parameter name
  • p_value (str) – input value
  • p_check (function) – sub-checker to call on each elements, default None
  • p_delim (str) – used as delimiter to split given str value
Raises:

ConfigValueValueError – value not an email
Returns:

array-converted value
Return type:

list
xtd.core.config.checkers.check_host(p_section, p_name, p_value)[source]

check that value is locally resolvable hostname

Parameters:
  • p_section (str) – parameter section name
  • p_name (str) – parameter name
  • p_value (str) – input value
Raises:

ConfigValueValueError – value not an hostname
Returns:

input value
Return type:

str
xtd.core.config.checkers.check_json(p_section, p_name, p_value)[source]

check that value is a valid json

if value is str, performs a :py:func:json.loads

Parameters:
  • p_section (str) – parameter section name
  • p_name (str) – parameter name
  • p_value (str) – input value
Raises:

ConfigValueValueError – value not a json
Returns:

dict-converted value
Return type:

dict
xtd.core.config.checkers.check_socket(p_section, p_name, p_value, p_schemes=None, p_checkUnix=False)[source]

check that value is a valid socket url

Parameters:
  • p_section (str) – parameter section name
  • p_name (str) – parameter name
  • p_value (str) – input value
  • p_scheme (list) – valid url schemes
  • p_checkUnix (bool) – authorize “unix+//<file>/path” based url
Raises:

ConfigValueValueError – value a valid socket url
Returns:

input value
Return type:

str
xtd.core.config.checkers.is_file(*p_args, **p_kwds)[source]

Currified version of check_file()

xtd.core.config.checkers.is_dir(*p_args, **p_kwds)[source]

Currified version of check_dir()

xtd.core.config.checkers.is_int(*p_args, **p_kwds)[source]

Currified version of check_int()

xtd.core.config.checkers.is_float(*p_args, **p_kwds)[source]

Currified version of check_float()

xtd.core.config.checkers.is_bool(*p_args, **p_kwds)[source]

Currified version of check_bool()

xtd.core.config.checkers.is_enum(*p_args, **p_kwds)[source]

Currified version of check_enum()

xtd.core.config.checkers.is_mail(*p_args, **p_kwds)[source]

Currified version of check_mail()

xtd.core.config.checkers.is_array(*p_args, **p_kwds)[source]

Currified version of check_array()

xtd.core.config.checkers.is_host(*p_args, **p_kwds)[source]

Currified version of check_host()

xtd.core.config.checkers.is_json(*p_args, **p_kwds)[source]

Currified version of check_json()

xtd.core.config.checkers.is_socket(*p_args, **p_kwds)[source]

Currified version of check_socket()

xtd.core.config.formatter module
class xtd.core.config.formatter.IndentedHelpFormatterWithNL[source]

Bases: optparse.IndentedHelpFormatter

__init__()[source]
format_description(description)[source]
format_option(p_option)[source]
__module__ = 'xtd.core.config.formatter'
xtd.core.config.manager module
class xtd.core.config.manager.Option(p_section, p_name, p_prop=None)[source]

Bases: object

Option object for ConfigManager

Available option properties:

config
Allow option to be read from configuration file, default True
cmdline
Allow option to be read from command line, default True
default
Internal default value for option, default None
valued
Expects a value for option. Default True if default value profived. For non-valued options, default value is False and reading them from command line will store a True value
description
Option description to display on usage message
checks
Array of functions to validate option value. You may provide a single function. Default []. See xtd.core.config.checkers for standard check functions.
longopt
Override long option name. Long options has be be unique. Default --<section>-<name>.
mandatory
Option is mandatory on command line, often used with non-valued options. Default False

Note

Provided check callback must respect the following signature :

def function(p_section, p_section, p_value)

They must return the input p_value (possible possibly trans-typed) and raise ConfigError if value is rejected

See xtd.core.config.checkers for standard check functions.

Parameters:
  • p_section (str) – option’s section name
  • p_name (str) – option’s name
  • p_props (dict) – option definition
Raises:

xtd.core.error.ConfigError – encountered unknown property
__init__(p_section, p_name, p_prop=None)[source]
validate(p_value)[source]
__dict__ = mappingproxy({'__weakref__': <attribute '__weakref__' of 'Option' objects>, '_update': <function Option._update at 0x7f24e41c5158>, '__dict__': <attribute '__dict__' of 'Option' objects>, '__doc__': " Option object for :py:class:`ConfigManager`\n\n Available option properties:\n\n config\n Allow option to be read from configuration file, default ``True``\n\n cmdline\n Allow option to be read from command line, default ``True``\n\n default\n Internal default value for option, default ``None``\n\n valued\n Expects a value for option. Default ``True`` if default value profived.\n For non-valued options, default value is ``False`` and reading them\n from command line will store a ``True`` value\n\n description\n Option description to display on usage message\n\n checks\n Array of functions to validate option value. You may provide a single\n function. Default ``[]``. See :py:mod:`xtd.core.config.checkers` for\n standard check functions.\n\n longopt\n Override long option name. Long options has be be unique. Default ``--<section>-<name>``.\n\n mandatory\n Option is mandatory on command line, often used with non-valued options. Default\n ``False``\n\n Note:\n\n Provided check callback must respect the following signature :\n\n .. code-block:: python\n\n def function(p_section, p_section, p_value)\n\n They must return the input ``p_value`` (possible possibly trans-typed)\n and raise :py:exc:`~xtd.core.error.ConfigError` if value is\n rejected\n\n See :py:mod:`xtd.core.config.checkers` for standard check functions.\n\n Args:\n p_section (str): option's section name\n p_name (str): option's name\n p_props (dict) : option definition\n\n\n Raises:\n xtd.core.error.ConfigError: encountered unknown property\n\n ", '__init__': <function Option.__init__ at 0x7f24e423dea0>, '__module__': 'xtd.core.config.manager', 'validate': <function Option.validate at 0x7f24e41c51e0>})
__module__ = 'xtd.core.config.manager'
__weakref__

list of weak references to the object (if defined)

class xtd.core.config.manager.ConfigManager[source]

Bases: object

Unified command-line & file config option manager

The main user methods are :

Main documentation for option definition : Option

__metaclass__

xtd.core.mixin.Singleton

makes this object a singleton

__init__()[source]
register_section(p_section, p_title, p_options)[source]

Register a set of options to a given section

See Option for full documentation of option properties

Parameters:
  • p_section (str) – section tag
  • p_title (str) – the section title in the command-line usage
  • p_options (list of dict) – options definition
Returns:

self
Return type:

ConfigManager
Raises:

xtd.core.error.ConfigError – invalid option definition
register(p_section, p_name, p_props)[source]

Register an option in a specific section

See Option for full documentation of option properties

Parameters:
  • p_name (str) – option name
  • p_section (str) – section name
  • p_props (dict) – option properties
Returns:

self
Return type:

ConfigManager
sections()[source]

Get sections tags

Returns:array of str of all section names
Return type:(list)
section_exists(p_section)[source]

Indicates if specified section has been registered

Parameters:p_section (str) – section name
Returns:true is p_section is registered
Return type:bool
options(p_section)[source]

Get the list of all registered option names for specefic a section

Parameters:p_section (str) – section name
Raises:xtd.core.error.ConfigError – p_section not registered
Returns:array of str of option names
Return type:list
option_exists(p_section, p_name)[source]

Indicates if specified option has been registered in section

Parameters:
  • p_section (str) – section name
  • p_option (str) – option name
Returns:

true is p_section is registered and contains p_option
Return type:

bool
get(p_section, p_name)[source]

Get option value

Parameters:
  • p_section (str) – section name
  • p_option (str) – option name
Raises:

xtd.core.error.ConfigValueError – section/option not found
Returns:

current option value
Return type:

(undefined)
set(p_section, p_name, p_value)[source]

set option value

Warning

This method stores the input value immediately without validating it against option’s checks.

Parameters:
  • p_section (str) – section name
  • p_option (str) – option name
Raises:

xtd.core.error.ConfigValueError – section/option not found
help(p_file=None)[source]

Display command line help message

Parameters:p_file (file) – output stream, defaults to sys.stdout
initialize()[source]

Initializes object

Usually called by Application object.

parse(p_argv=None)[source]

Parses command line and file options

Usually called by Application object.

Parameters:p_argv (list of str) – list of command line arguments
get_name()[source]

Get parsed application name sys.argv[0]

Returns:program’s sys.argv[0]
Return type:str
get_args()[source]

Get command line post-parse remaining options

Returns:unparsed command line options
Return type:list
set_usage(p_usage)[source]

Set command line usage message

See optparse.OptionParser

Parameters:p_usage (str) – usage string
option_cmdline_given(p_section, p_option)[source]
__dict__ = mappingproxy({'get': <function ConfigManager.get at 0x7f24e41c56a8>, 'option_exists': <function ConfigManager.option_exists at 0x7f24e41c5620>, 'sections': <function ConfigManager.sections at 0x7f24e41c5488>, '__doc__': 'Unified command-line & file config option manager\n\n The main user methods are :\n\n * :py:meth:`register_section`\n * :py:meth:`get`\n * :py:meth:`set_usage`\n\n\n Main documentation for option definition : :py:class:`Option`\n\n Attributes:\n __metaclass__ (:py:class:`xtd.core.mixin.Singleton`) : makes this object a singleton\n ', 'help': <function ConfigManager.help at 0x7f24e41c57b8>, 'get_args': <function ConfigManager.get_args at 0x7f24e41c59d8>, '_cmd_attribute_name': <staticmethod object at 0x7f24e421aef0>, '__init__': <function ConfigManager.__init__ at 0x7f24e41c5268>, 'get_name': <function ConfigManager.get_name at 0x7f24e41c5950>, 'section_exists': <function ConfigManager.section_exists at 0x7f24e41c5510>, 'options': <function ConfigManager.options at 0x7f24e41c5598>, 'parse': <function ConfigManager.parse at 0x7f24e41c58c8>, 'set_usage': <function ConfigManager.set_usage at 0x7f24e41c5a60>, '_load_data': <function ConfigManager._load_data at 0x7f24e41c5b70>, '__module__': 'xtd.core.config.manager', 'option_cmdline_given': <function ConfigManager.option_cmdline_given at 0x7f24e41c5d90>, 'initialize': <function ConfigManager.initialize at 0x7f24e41c5840>, 'set': <function ConfigManager.set at 0x7f24e41c5730>, 'register': <function ConfigManager.register at 0x7f24e41c5400>, '_cmd_parser_create': <function ConfigManager._cmd_parser_create at 0x7f24e41c5c80>, '__weakref__': <attribute '__weakref__' of 'ConfigManager' objects>, 'register_section': <function ConfigManager.register_section at 0x7f24e41c5378>, '__dict__': <attribute '__dict__' of 'ConfigManager' objects>, '_file_parser_load': <function ConfigManager._file_parser_load at 0x7f24e41c5e18>, '_validate': <function ConfigManager._validate at 0x7f24e41c5ea0>, '_get_option': <function ConfigManager._get_option at 0x7f24e41c5ae8>, '_cmd_parser_load': <function ConfigManager._cmd_parser_load at 0x7f24e41c5d08>})
__module__ = 'xtd.core.config.manager'
__weakref__

list of weak references to the object (if defined)

xtd.core.logger package
Submodules
xtd.core.logger.filter module
class xtd.core.logger.filter.FieldFilter(fields=None)[source]

Bases: logging.Filter

__init__(fields=None)[source]
__module__ = 'xtd.core.logger.filter'
filter(p_record)[source]
xtd.core.logger.formatter module
class xtd.core.logger.formatter.LocationFormatter(fmt='%(asctime)s (%(name)s) [%(levelname)s] : %(message)s %(location)s', datefmt='%Y-%m-%d %H:%M:%S', locfmt='at %(pathname)s:%(lineno)s -> %(funcName)s', locstyle=None)[source]

Bases: logging.Formatter

__init__(fmt='%(asctime)s (%(name)s) [%(levelname)s] : %(message)s %(location)s', datefmt='%Y-%m-%d %H:%M:%S', locfmt='at %(pathname)s:%(lineno)s -> %(funcName)s', locstyle=None)[source]
__module__ = 'xtd.core.logger.formatter'
format(p_record)[source]
xtd.core.logger.manager module
xtd.core.logger.manager.DEFAULT_CONFIG = {'handlers': {'rotfile': {'backupCount': 3, 'maxBytes': 15728640, 'filters': [], 'filename': 'out.log', 'class': 'logging.handlers.RotatingFileHandler', 'formatter': 'default'}, 'syslog': {'class': 'logging.handlers.SysLogHandler', 'address': '/dev/log', 'filters': [], 'formatter': 'default'}, 'stdout': {'class': 'logging.StreamHandler', 'filters': ['colored'], 'stream': 'stdout', 'formatter': 'location'}}, 'loggers': {'root': {'handlers': ['stdout', 'rotfile', 'syslog'], 'level': 40}}, 'filters': {'colored': {'class': 'xtd.core.logger.filter.FieldFilter', 'fields': {'name': {'pad': 'left', 'styles': {'default': {'colors': ['red'], 'attrs': ['bold']}}}, 'levelname': {'pad': 'left', 'styles': {'EXCEPTION': {'colors': ['magenta'], 'attrs': ['bold']}, 'ERROR': {'colors': ['red'], 'attrs': ['bold']}, 'INFO': {'colors': ['yellow'], 'attrs': ['bold']}, 'WARNING': {'colors': ['red'], 'attrs': []}, 'default': {'colors': ['yellow'], 'attrs': ['bold']}, 'DEBUG': {'colors': ['yellow'], 'attrs': []}}}}}}, 'formatters': {'location': {'class': 'xtd.core.logger.formatter.LocationFormatter', 'locstyle': {'colors': ['grey'], 'attrs': ['bold']}, 'fmt': '%(asctime)s (%(name)s) [%(levelname)s] : %(message)s %(location)s', 'locfmt': 'at %(pathname)s:%(lineno)s -> %(funcName)s', 'datefmt': '%Y-%m-%d %H:%M:%S'}, 'default': {'class': 'logging.Formatter', 'fmt': '%(asctime)s (%(name)s) [%(levelname)s] : %(message)s', 'datefmt': '%a %d %b %Y at %H-%M'}}}

Default logging configuration

Todo

some doc

class xtd.core.logger.manager.WrapperLogger(p_name)[source]

Bases: logging.Logger

__init__(p_name)[source]
findCaller(p_stack=False)[source]
handle(p_record)[source]
__module__ = 'xtd.core.logger.manager'
class xtd.core.logger.manager.LogManager[source]

Bases: object

Todo

some

__init__()[source]
add_formatter(p_name, p_obj)[source]
add_filter(p_name, p_obj)[source]
add_handler(p_name, p_obj)[source]
get_formatter(p_name)[source]
get_filter(p_name)[source]
get_handler(p_name)[source]
load_config(p_config=None, p_override=None)[source]
__dict__ = mappingproxy({'_load_formatters': <function LogManager._load_formatters at 0x7f24e4228b70>, 'add_handler': <function LogManager.add_handler at 0x7f24e42287b8>, '__doc__': '\n .. todo:: some\n ', 'get_filter': <function LogManager.get_filter at 0x7f24e42288c8>, 'get_handler': <function LogManager.get_handler at 0x7f24e4228950>, '__module__': 'xtd.core.logger.manager', 'initialize': <function LogManager.initialize at 0x7f24e4228d08>, '_load_filters': <function LogManager._load_filters at 0x7f24e4228ae8>, '__dict__': <attribute '__dict__' of 'LogManager' objects>, '_get_class': <staticmethod object at 0x7f24e42269e8>, '__init__': <function LogManager.__init__ at 0x7f24e4228598>, 'load_config': <function LogManager.load_config at 0x7f24e4228a60>, '__weakref__': <attribute '__weakref__' of 'LogManager' objects>, 'get_formatter': <function LogManager.get_formatter at 0x7f24e4228840>, '_load_handlers': <function LogManager._load_handlers at 0x7f24e4228bf8>, 'add_formatter': <function LogManager.add_formatter at 0x7f24e42286a8>, 'add_filter': <function LogManager.add_filter at 0x7f24e4228730>, '_load_loggers': <function LogManager._load_loggers at 0x7f24e4228c80>})
__module__ = 'xtd.core.logger.manager'
__weakref__

list of weak references to the object (if defined)

initialize(p_config=None, p_override=None)[source]
xtd.core.logger.tools module
xtd.core.logger.tools.get(p_module=None)[source]
xtd.core.logger.tools.debug(p_module, p_msg, *p_args, **p_kwds)[source]
xtd.core.logger.tools.info(p_module, p_msg, *p_args, **p_kwds)[source]
xtd.core.logger.tools.warning(p_module, p_msg, *p_args, **p_kwds)[source]
xtd.core.logger.tools.error(p_module, p_msg, *p_args, **p_kwds)[source]
xtd.core.logger.tools.critical(p_module, p_msg, *p_args, **p_kwds)[source]
xtd.core.logger.tools.exception(p_module, p_msg, *p_args, **p_kwds)[source]
xtd.core.logger.tools.log(p_level, p_module, p_msg, *p_args, **p_kwds)[source]
xtd.core.param package
Submodules
xtd.core.param.manager module

module doc top

class xtd.core.param.manager.Param(p_name, p_value, p_callbacks=None)[source]

Bases: object

Object that holds an JSON-serializable value

Parameters:
  • p_name (str) – Name of the parameter
  • p_value (json-serializable) – Initial parameter value
  • p_callbacks (Optional[function]) – List of function to call whenever when the value changes Can be an array of functions or a single function

When changing value with the set() method, all registered listerners are called and if none of them raises an error, the value in stored.

New callbacks can be registered with listen

Note

Each callback must respect the following prototype : function(p_parameter, p_oldValue, p_newValue)

  • p_parameter (Param): the modified Param object
  • p_oldValue (json-serializable): parameter’s old value
  • p_newvalue (json-serializable): parameter’s new value

Callback must raise xtd.core.error.XtdError is new value is not acceptable

__init__(p_name, p_value, p_callbacks=None)[source]
listen(p_handler)[source]
get()[source]
set(p_value)[source]
__dict__ = mappingproxy({'__weakref__': <attribute '__weakref__' of 'Param' objects>, 'get': <function Param.get at 0x7f24e4242598>, 'set': <function Param.set at 0x7f24e4242620>, 'listen': <function Param.listen at 0x7f24e4242510>, '__dict__': <attribute '__dict__' of 'Param' objects>, '__doc__': "Object that holds an JSON-serializable value\n\n Args:\n p_name (str) : Name of the parameter\n p_value (json-serializable) : Initial parameter value\n p_callbacks (Optional[function]) : List of function to call whenever when the value changes Can be an array of functions or a single function\n\n When changing value with the :meth:`set` method, all registered listerners are called\n and if none of them raises an error, the value in stored.\n\n New callbacks can be registered with :obj:`listen`\n\n Note:\n Each callback must respect the following prototype :\n ``function(p_parameter, p_oldValue, p_newValue)``\n\n - **p_parameter** (Param): the modified Param object\n - **p_oldValue** (json-serializable): parameter's old value\n - **p_newvalue** (json-serializable): parameter's new value\n\n Callback must raise :obj:`xtd.core.error.XtdError` is new value is not acceptable\n ", '__init__': <function Param.__init__ at 0x7f24e4242488>, '__module__': 'xtd.core.param.manager'})
__module__ = 'xtd.core.param.manager'
__weakref__

list of weak references to the object (if defined)

class xtd.core.param.manager.ParamManager(p_adminDir)[source]

Bases: object

Constructor

Parameters:p_adminDir (str) – directory to dump-to/load-from synced parameters
Raises:xtd.core.error.XtdError – p_adminDir is not writable
__init__(p_adminDir)[source]

Constructor

Parameters:p_adminDir (str) – directory to dump-to/load-from synced parameters
Raises:xtd.core.error.XtdError – p_adminDir is not writable
register(p_name, p_value, p_callbacks=None, p_sync=False)[source]
register_param(p_param, p_sync=False)[source]
get_names()[source]
get_param(p_name)[source]
get(p_name)[source]
set(p_name, p_value)[source]
__dict__ = mappingproxy({'get': <function ParamManager.get at 0x7f24e4242b70>, 'set': <function ParamManager.set at 0x7f24e4242bf8>, '_create_dir': <staticmethod object at 0x7f24e41c76d8>, 'listen': <function ParamManager.listen at 0x7f24e4242c80>, '__dict__': <attribute '__dict__' of 'ParamManager' objects>, '__doc__': 'Stores in memory global parameters\n ', 'get_names': <function ParamManager.get_names at 0x7f24e4242a60>, '__init__': <function ParamManager.__init__ at 0x7f24e42426a8>, '_load': <function ParamManager._load at 0x7f24e42428c8>, '__weakref__': <attribute '__weakref__' of 'ParamManager' objects>, 'register_param': <function ParamManager.register_param at 0x7f24e42429d8>, 'register': <function ParamManager.register at 0x7f24e4242950>, 'get_param': <function ParamManager.get_param at 0x7f24e4242ae8>, '_write': <function ParamManager._write at 0x7f24e4242840>, '__module__': 'xtd.core.param.manager'})
__module__ = 'xtd.core.param.manager'
__weakref__

list of weak references to the object (if defined)

listen(p_name, p_listener)[source]
xtd.core.stat package
xtd.core.stat.get(p_ns, p_name)[source]
Submodules
xtd.core.stat.counter module

Inheritance diagram of xtd.core.stat.counter

class xtd.core.stat.counter.BaseCounter(p_name)[source]

Bases: object

Abstract base counter

The almost-empty shell insure base methods are protected by a lock.

Parameters:p_name (str) – object name
__init__(p_name)[source]
visit(p_visitor)[source]

Visit object tree with visitor

update()[source]

Update object

Generally called by user just before visiting the object in order to gather “fresh” data

__dict__ = mappingproxy({'_visit_safe': <function BaseCounter._visit_safe at 0x7f24e422aa60>, 'update': <function BaseCounter.update at 0x7f24e422a9d8>, '__dict__': <attribute '__dict__' of 'BaseCounter' objects>, '__doc__': ' Abstract base counter\n\n The almost-empty shell insure base methods are protected\n by a lock.\n\n Args:\n p_name (str): object name\n ', '__init__': <function BaseCounter.__init__ at 0x7f24e422a8c8>, '__weakref__': <attribute '__weakref__' of 'BaseCounter' objects>, '__module__': 'xtd.core.stat.counter', 'visit': <function BaseCounter.visit at 0x7f24e422a950>, '_update_safe': <function BaseCounter._update_safe at 0x7f24e422aae8>})
__module__ = 'xtd.core.stat.counter'
__weakref__

list of weak references to the object (if defined)

class xtd.core.stat.counter.Value(p_name, p_value=None, p_type='i')[source]

Bases: xtd.core.stat.counter.BaseCounter

Thread-safe numeric value holder

Parameters:
  • p_name (str) – Object’s name
  • p_value (numeric) – Internal value, type depends on p_type
  • p_type (str) – One character type spec, see multiprocessing.Value
Raises:

Visitors

Value visitor must follow the following prototype: function(p_name, p_value)

  • p_name (str): name of visited Value
  • p_value (numeric|str): internal value or NaN is unset
__init__(p_name, p_value=None, p_type='i')[source]
unset()[source]

Make the current value undefined

val

(Property) internal value

If set to None, the current value is undefined

Returns:current internal value, None if unset
Return type:(numeric)
Raises:TypeError – affected value dosen’t match constructor type
incr(p_val=1)[source]

Increments the current value

Parameters:p_val (numeric) – add p_val to current internal value
Raises:TypeError – given value dosen’t match constructor type
decr(p_val=1)[source]

Decrements the current value

Parameters:p_val (numeric) – subtract p_val to current internal value
Raises:TypeError – given value dosen’t match constructor type
__module__ = 'xtd.core.stat.counter'
class xtd.core.stat.counter.Int32(p_name, p_value=None)[source]

Bases: xtd.core.stat.counter.Value

Value specialization for signed 32 bits integer

TYPE = 'i'

multiprocessing.Value type spec

__init__(p_name, p_value=None)[source]
__module__ = 'xtd.core.stat.counter'
class xtd.core.stat.counter.Int64(p_name, p_value=None)[source]

Bases: xtd.core.stat.counter.Value

Value specialization for signed 64 bits integer

TYPE = 'l'

multiprocessing.Value type spec

__init__(p_name, p_value=None)[source]
__module__ = 'xtd.core.stat.counter'
class xtd.core.stat.counter.UInt32(p_name, p_value=None)[source]

Bases: xtd.core.stat.counter.Value

Value specialization for unsigned 32 bits integer

TYPE = 'I'

multiprocessing.Value type spec

__init__(p_name, p_value=None)[source]
__module__ = 'xtd.core.stat.counter'
class xtd.core.stat.counter.UInt64(p_name, p_value=None)[source]

Bases: xtd.core.stat.counter.Value

Value specialization for unsigned 64 bits integer

TYPE = 'L'

multiprocessing.Value type spec

__init__(p_name, p_value=None)[source]
__module__ = 'xtd.core.stat.counter'
class xtd.core.stat.counter.Float(p_name, p_value=None)[source]

Bases: xtd.core.stat.counter.Value

Value specialization for float

TYPE = 'f'

multiprocessing.Value type spec

__init__(p_name, p_value=None)[source]
__module__ = 'xtd.core.stat.counter'
class xtd.core.stat.counter.Double(p_name, p_value=None)[source]

Bases: xtd.core.stat.counter.Value

Value specialization for double

TYPE = 'd'

multiprocessing.Value type spec

__init__(p_name, p_value=None)[source]
__module__ = 'xtd.core.stat.counter'
class xtd.core.stat.counter.Composed(p_name)[source]

Bases: xtd.core.stat.counter.BaseCounter

Manage a collection child counters

__init__(p_name)[source]
register(p_counter)[source]

Register a child counter

Current object name is prepend to registered child name with the following format : <parent-name>.<child-name>

Parameters:p_counter (BaseCounter) – child counter to add
__module__ = 'xtd.core.stat.counter'
class xtd.core.stat.counter.TimedSample(p_name, p_timeMs=10000, p_maxSamples=20000, p_type='i')[source]

Bases: xtd.core.stat.counter.Composed

Holds the min, max and average value of collected items over a fixed period of time

When no items are available for the last p_timeMs, the 3 sub counters are undefined, thus, collected by visitors as NaN.

Parameters:
  • p_name (str) – counter name
  • p_timeMs (int) – maximum amount of time (millisecond) to keep collected values
  • p_maxSamples (int) – maximum amount of values to keep
  • p_type (str) – internal type representation, see multiprocessing.Value
__init__(p_name, p_timeMs=10000, p_maxSamples=20000, p_type='i')[source]
push(p_val)[source]

Add a value in collection

p_val (int): value to add

__module__ = 'xtd.core.stat.counter'
class xtd.core.stat.counter.Perf(p_name, p_timeMs=10000, p_maxSamples=20000)[source]

Bases: xtd.core.stat.counter.TimedSample

Designed to monitor the min, max and average time of an event

At event start call work_begin() to store the current time. At event end, call work_end() to calculate the time detla and add it the base class samples that monitors the min max and average values

The time resolution is the microsecond (10^-6 second).

Note

Events beginnings and ends can’t be interleaved in the same thread.

__module__ = 'xtd.core.stat.counter'
__init__(p_name, p_timeMs=10000, p_maxSamples=20000)[source]
work_begin()[source]

Record begin time of an event

Raises:CounterError – called twice from same thread with no work_end() between them
work_end()[source]

Record end time of an event and push it into base class

Raises:CounterError – called without calling work_begin() first from same thread
exception xtd.core.stat.counter.CounterError(p_module, p_name, p_msg, *p_args, **p_kwds)[source]

Bases: xtd.core.error.XtdError

Generic counter error class

Parameters:
  • p_module (str) – module name
  • p_name (str) – counter name
  • p_msg (str) – error message
  • p_args (tuple) – generic message format arguments
  • p_kwds (dict) – generic message format keywords
m_name

str

name of counter that raised the error

__module__ = 'xtd.core.stat.counter'
__init__(p_module, p_name, p_msg, *p_args, **p_kwds)[source]
xtd.core.stat.handler module
class xtd.core.stat.handler.BaseHandler(p_name, p_interval=50, p_fetcher=None)[source]

Bases: xtd.core.tools.thread.SafeThread

Abstract statistic handler

Spawns a dedicated thread that outputs a set counters every p_interval seconds.

User can provide a custom function to get the counters to output. By default the object gets all counters registered in the StatManager singleton.

User should inherit this class and define write() method.

Note

The functor must take no parameter and return something suitable for write()

Parameters:
  • p_name (str) – handler name
  • p_interval (int) – interval, in second, between two outputs
  • p_fetcher (function) – functor that retrieves data counters
__init__(p_name, p_interval=50, p_fetcher=None)[source]
write(p_counters)[source]

Ouput available counter data (abstract)

Parameters:p_counters (dict) – dictionary holding all available counters assiciated with their namespace. { "<namespace>" : [ <counter1>, <counter2>, ... ] }
work()[source]

Output available counter data

Implementation of xtd.core.tools.thread.SafeThread.work()

__module__ = 'xtd.core.stat.handler'
class xtd.core.stat.handler.DiskHandler(p_directory, p_interval=50, p_fetcher=None)[source]

Bases: xtd.core.stat.handler.BaseHandler

Output counters to filesystem

Given :

This will create a file /var/snmp/a/b/c/counter.name containing the string 55

Parameters:
  • p_directory (str) – target output directory path. If given directory doesn’t exist, the object will attempt to create it (and all necessary parent directories).
  • p_interval (int) – interval between two outputs (in seconds)
Raises:

XtdError – p_directory isn’t writable or could ne be created
__init__(p_directory, p_interval=50, p_fetcher=None)[source]
write(p_counters)[source]

Write all available counters to filesystem

When object fails to write a counte file, an error log is triggered. Raising an exception woudn’t be helpfull since this part of the code runs in a dedicated thread.

Parameters:p_counters (dict) – see BaseHandler.write()
__module__ = 'xtd.core.stat.handler'
class xtd.core.stat.handler.HttpHandler(p_url, p_interval=50, p_fetcher=None)[source]

Bases: xtd.core.stat.handler.BaseHandler

Send counter values to a http server

Counter values are gathered in a single json and sent to the target url as a POST request.

HTTP Request details :

  • Method : POST

  • Content-Type : application/json

  • Body :

    {
  "<namespace>" : {
        "<name-of-counter-1>" : value-of-counter-1,
        "<name-of-counter-2>" : value-of-counter-2,
        ...
  },
  ...
  "<namespace>" : {
    <name-of-counter-N>" : value-of-counter-N
    ...
  }
}

Note

Keep in mind that counter values can be undefined.

In such case, the json value is a string equals to "NaN".

Parameters:
  • p_url – target url for post request
  • p_interval (int) – interval, in second, between two outputs
  • p_fetcher (function) – functor that retrieves data counters
__init__(p_url, p_interval=50, p_fetcher=None)[source]
write(p_counters)[source]

Write all available counters to filesystem

When object fails to send HTTP request, an error log is triggered. Raising an exception woudn’t be helpfull since this part of the code runs in a dedicated thread.

Parameters:p_counters (dict) – see BaseHandler.write()
__module__ = 'xtd.core.stat.handler'
class xtd.core.stat.handler.LoggingHandler(p_name, p_interval=50, p_fetcher=None)[source]

Bases: xtd.core.stat.handler.BaseHandler

Output counter to application logs

Parameters:
  • p_name (str) – logger module name
  • p_interval (int) – interval, in second, between two outputs
  • p_fetcher (function) – functor that retrieves data counters
__init__(p_name, p_interval=50, p_fetcher=None)[source]
__module__ = 'xtd.core.stat.handler'
write(p_counters)[source]

Output all available counters to logging facility :param p_counters: see BaseHandler.write()

xtd.core.stat.manager module
class xtd.core.stat.manager.StatManager[source]

Bases: xtd.core.tools.thread.SafeThreadGroup

__init__()[source]
exists(p_ns, p_name)[source]
register_counter(p_ns, p_counter)[source]

Register a counter in global statistics

Parameters:
  • p_ns (str) – namespace
  • p_counter (BaseCounter) – counter object to add
Raises:
  • XtdError – counter already defined for this namespace
  • XtdError – p_counter is not a valid BaseCounter object
register_handler(p_handler)[source]

Register an counter output handler

Parameters:p_handler (BaseHandler) – new handler to add
Raises:XtdError – given p_handler is not a valid BaseHandler
get(p_ns, p_name)[source]

Get a counter in a particular namespace

Returns:

requests counter
Return type:

BaseCounter
Raises:
  • XtdError – undefined namespace p_ns
  • XtdError – undefined counter p_name for given namespace
write()[source]

Output counter is all registered handlers

get_all()[source]

Get registered counters

Example

{
  "name.space.1" : [ <counter-object-1>, <counter-object-2>, ... ],
  ...
  "name.space.N" : [ <counter-object-N>, <counter-object-N+1>, ... ],
}
Returns:dictionary containing raw counters
Return type:dict
get_json()[source]

Get counter data in a dictionary

Example

{
  "name.space.1" : {
    "counter1" : <value1>,
    "counter2" : <value2>
  },
  ...
  "name.space" : {
    "counterN" : <valueN>
  }
}
Returns:counter’s name and value organized by namespace
Return type:dict
__module__ = 'xtd.core.stat.manager'
xtd.core.tools package
Submodules
xtd.core.tools.daemonize module
xtd.core.tools.daemonize.daemonize(p_pidFilePath)[source]

Detach a process from the controlling terminal and run it in the background as a daemon.

xtd.core.tools.daemonize.get_pid_from_file(p_fileFilePath)[source]
xtd.core.tools.daemonize.is_running(p_pid)[source]
xtd.core.tools.mergedicts module
xtd.core.tools.mergedicts.mergedicts(p_dict1, p_dict2)[source]
xtd.core.tools.thread module
class xtd.core.tools.thread.SafeThread(p_name, p_interval)[source]

Bases: threading.Thread

__init__(p_name, p_interval)[source]
work()[source]
run()[source]
stop()[source]
safe_join()[source]
__module__ = 'xtd.core.tools.thread'
class xtd.core.tools.thread.SafeThreadGroup(p_name)[source]

Bases: object

STATUS_STARTED = 1
STATUS_STOPPED = 2
STATUS_JOINED = 3
__init__(p_name)[source]
add_thread(p_obj)[source]
start()[source]
stop()[source]
join()[source]
__dict__ = mappingproxy({'STATUS_STOPPED': 2, '__dict__': <attribute '__dict__' of 'SafeThreadGroup' objects>, 'add_thread': <function SafeThreadGroup.add_thread at 0x7f24e4228f28>, '__doc__': None, '__init__': <function SafeThreadGroup.__init__ at 0x7f24e4228ea0>, '__weakref__': <attribute '__weakref__' of 'SafeThreadGroup' objects>, '__module__': 'xtd.core.tools.thread', 'join': <function SafeThreadGroup.join at 0x7f24e422a158>, 'start': <function SafeThreadGroup.start at 0x7f24e422a048>, 'STATUS_JOINED': 3, 'STATUS_STARTED': 1, 'stop': <function SafeThreadGroup.stop at 0x7f24e422a0d0>})
__module__ = 'xtd.core.tools.thread'
__weakref__

list of weak references to the object (if defined)

xtd.core.tools.url module
xtd.core.tools.url.parse_unix(p_url)[source]
xtd.core.tools.url.unparse_unix(p_parsed, p_unix)[source]
xtd.core.tools.url.is_pureunix(p_url)[source]
xtd.core.tools.url.parse_pureunix(p_url)[source]
xtd.core.tools.url.pureunix_to_unixhttp(p_url, p_https=False)[source]

Submodules

xtd.core.application module
class xtd.core.application.Application(p_name=None)[source]

Bases: object

XTD main application object

Users must inherit this class :

  • register program’s arguments in their __init__ method
  • and optionally override initialize()
  • override process() and code their program behavior
  • call execute() at top level

p_name parameter is used for various purpose such as:

  • default usage
  • default –config-file value
  • default disk-synced parameters output directory path
  • default statistic disk output directory path
Parameters:p_name (str) – application’s name (optional). Defaults to sys.argv[0]
m_name

str

application’s name

__init__(p_name=None)[source]
config()[source]

Get the ConfigManager instance

Returns:ConfigManager instance
Return type:config.manager.ConfigManager
stat()[source]

Get the StatManager instance

Returns:StatManager instance
Return type:stat.manager.StatManager
process()[source]

Main application body

The child class must override this method. Since default behavior is to log an error, you should not call parent’s method

Returns:program’s exit code and True if object should call stop method before joining
Return type:int, bool
initialize()[source]

Initializes application

Specifically:

Any child class that overrides this method should call super(Application, self).initialize()

start()[source]

Start background modules

Any child class that overrides this method should call super(Application, self).start() or start StatManager by hand

stop()[source]

Stop background modules

Any child class that overrides this method should call super(Application, self).stop() or stop StatManager by hand

join()[source]

Join background modules

Any child class that overrides this method should call super(Application, self).join() or join StatManager by hand

execute(p_argv=None)[source]

Main application entry point

Exits with code returned by process().

Note

During the initializing phase :

  • Any ConfigError leads to the display of the error, followed by the program usage and ends with a sys.exit(1).
  • Any XtdError leads to the display of the error and ends with a sys.exit(1).

During the process phase :

  • Any XtdError leads to the log of the error and ends with a sys.exit(1).
Parameters:p_argv (list) – program’s command-line argument. Defaults to None. If none, arguments are taken from sys.argv
__dict__ = mappingproxy({'stat': <function Application.stat at 0x7f24e4242d90>, '__dict__': <attribute '__dict__' of 'Application' objects>, '__doc__': "XTD main application object\n\n Users must inherit this class :\n\n * register program's arguments in their ``__init__`` method\n * and optionally override :py:meth:`initialize`\n * override :py:meth:`process` and code their program behavior\n * call :py:meth:`execute` at top level\n\n\n ``p_name`` parameter is used for various purpose such as:\n\n * default usage\n * default --config-file value\n * default disk-synced parameters output directory path\n * default statistic disk output directory path\n\n\n Args:\n p_name (str): application's name (optional). Defaults to ``sys.argv[0]``\n\n Attributes:\n m_name (str): application's name\n\n ", '_initialize_stat': <function Application._initialize_stat at 0x7f24e41c8268>, 'join': <function Application.join at 0x7f24e41c80d0>, 'execute': <function Application.execute at 0x7f24e41c8158>, '__module__': 'xtd.core.application', 'initialize': <function Application.initialize at 0x7f24e4242ea0>, '_initialize_param': <function Application._initialize_param at 0x7f24e41c8378>, '__weakref__': <attribute '__weakref__' of 'Application' objects>, '_initialize_log': <function Application._initialize_log at 0x7f24e41c82f0>, '__init__': <function Application.__init__ at 0x7f24e4242268>, 'config': <function Application.config at 0x7f24e4242d08>, '_initialize_config': <function Application._initialize_config at 0x7f24e41c81e0>, 'start': <function Application.start at 0x7f24e4242f28>, 'process': <function Application.process at 0x7f24e4242e18>, 'stop': <function Application.stop at 0x7f24e41c8048>})
__module__ = 'xtd.core.application'
__weakref__

list of weak references to the object (if defined)

xtd.core.error module
exception xtd.core.error.XtdError(p_module, p_message, *p_args, **p_kwds)[source]

Bases: BaseException

__init__(p_module, p_message, *p_args, **p_kwds)[source]
log()[source]

todo

__str__()[source]
__module__ = 'xtd.core.error'
__weakref__

list of weak references to the object (if defined)

exception xtd.core.error.ConfigError(p_message)[source]

Bases: xtd.core.error.XtdError

__init__(p_message)[source]
__module__ = 'xtd.core.error'
exception xtd.core.error.ConfigValueError(p_section, p_option, p_message)[source]

Bases: xtd.core.error.ConfigError

__init__(p_section, p_option, p_message)[source]
__module__ = 'xtd.core.error'
exception xtd.core.error.ConfigValueFileError(p_section, p_option, p_fileName)[source]

Bases: xtd.core.error.ConfigValueError

__init__(p_section, p_option, p_fileName)[source]
__module__ = 'xtd.core.error'
exception xtd.core.error.ConfigValueDirError(p_section, p_option, p_fileName)[source]

Bases: xtd.core.error.ConfigValueError

__init__(p_section, p_option, p_fileName)[source]
__module__ = 'xtd.core.error'
exception xtd.core.error.ConfigValueDirModeError(p_section, p_option, p_fileName, p_read=False, p_write=False, p_execute=False)[source]

Bases: xtd.core.error.ConfigValueError

__init__(p_section, p_option, p_fileName, p_read=False, p_write=False, p_execute=False)[source]
__module__ = 'xtd.core.error'
exception xtd.core.error.ConfigValueFileModeError(p_section, p_option, p_fileName, p_read=False, p_write=False, p_execute=False)[source]

Bases: xtd.core.error.ConfigValueError

__init__(p_section, p_option, p_fileName, p_read=False, p_write=False, p_execute=False)[source]
__module__ = 'xtd.core.error'
exception xtd.core.error.ConfigValueTypeError(p_section, p_option, p_value, p_typeName)[source]

Bases: xtd.core.error.ConfigValueError

INT = 'int'
FLOAT = 'float'
BOOL = 'bool'
__init__(p_section, p_option, p_value, p_typeName)[source]
__module__ = 'xtd.core.error'
exception xtd.core.error.ConfigValueLimitsError(p_section, p_option, p_value, p_minValue=None, p_maxValue=None)[source]

Bases: xtd.core.error.ConfigValueError

__init__(p_section, p_option, p_value, p_minValue=None, p_maxValue=None)[source]
__module__ = 'xtd.core.error'
exception xtd.core.error.ConfigValueEnumError(p_section, p_option, p_value, p_authorizedValues)[source]

Bases: xtd.core.error.ConfigValueError

__init__(p_section, p_option, p_value, p_authorizedValues)[source]
__module__ = 'xtd.core.error'
xtd.core.mixin module
class xtd.core.mixin.Singleton[source]

Bases: type

ms_instances = {}
__call__(*p_args, **p_kwargs)[source]
classmethod reset(mcs, p_item)[source]
__module__ = 'xtd.core.mixin'

xtd.network package

Subpackages

xtd.network.client package
Submodules
xtd.network.client.asynclient module
xtd.network.server package
Submodules
xtd.network.server.application module
class xtd.network.server.application.ServerApplication(p_name='/home/docs/checkouts/readthedocs.org/user_builds/xtd/envs/latest/bin/sphinx-build')[source]

Bases: xtd.core.application.Application

__init__(p_name='/home/docs/checkouts/readthedocs.org/user_builds/xtd/envs/latest/bin/sphinx-build')[source]
default(*p_args, **p_kwds)[source]
initialize()[source]
start()[source]
join()[source]
stop()[source]
process()[source]
__module__ = 'xtd.network.server.application'
xtd.network.server.config module
class xtd.network.server.config.ConfigPage[source]

Bases: object

default(*p_args, **p_kwds)[source]
__dict__ = mappingproxy({'__weakref__': <attribute '__weakref__' of 'ConfigPage' objects>, '__doc__': None, 'default': <function ConfigPage.default at 0x7f24e39d4730>, '__module__': 'xtd.network.server.config', '__dict__': <attribute '__dict__' of 'ConfigPage' objects>})
__module__ = 'xtd.network.server.config'
__weakref__

list of weak references to the object (if defined)

xtd.network.server.counter module
class xtd.network.server.counter.CounterPage[source]

Bases: object

default(*p_args, **p_kwds)[source]
__dict__ = mappingproxy({'__weakref__': <attribute '__weakref__' of 'CounterPage' objects>, '__doc__': None, 'default': <function CounterPage.default at 0x7f24e39d46a8>, '__module__': 'xtd.network.server.counter', '__dict__': <attribute '__dict__' of 'CounterPage' objects>})
__module__ = 'xtd.network.server.counter'
__weakref__

list of weak references to the object (if defined)

xtd.network.server.log module
class xtd.network.server.log.LogPage(p_credentials=None)[source]

Bases: object

__init__(p_credentials=None)[source]
check_password(p_realm, p_username, p_password)[source]
write(*p_args, **p_kwds)[source]
default(*p_args, **p_kwds)[source]
__dict__ = mappingproxy({'_level_to_name': <staticmethod object at 0x7f24e3c7c588>, 'check_password': <function LogPage.check_password at 0x7f24e39d4378>, '__dict__': <attribute '__dict__' of 'LogPage' objects>, '__weakref__': <attribute '__weakref__' of 'LogPage' objects>, 'default': <function LogPage.default at 0x7f24e39d4620>, '__init__': <function LogPage.__init__ at 0x7f24e39d42f0>, 'write': <function LogPage.write at 0x7f24e39d4598>, '__doc__': None, '_name_to_level': <staticmethod object at 0x7f24e3c7c278>, '__module__': 'xtd.network.server.log'})
__module__ = 'xtd.network.server.log'
__weakref__

list of weak references to the object (if defined)

xtd.network.server.manager module
class xtd.network.server.manager.ServerManager[source]

Bases: object

ms_initialized = False
class LoggerFilter(p_name='', p_wrap=False)[source]

Bases: logging.Filter

__init__(p_name='', p_wrap=False)[source]
filter(p_record)[source]
__module__ = 'xtd.network.server.manager'
classmethod ServerManager.listen(p_socket, p_nbThreads=10, p_tls=False, p_cacert=None, p_cert=None, p_key=None)[source]
classmethod ServerManager.initialize(p_logger)[source]
classmethod ServerManager.mount(p_handler, p_path, p_conf=None, p_logger='cherrypy')[source]
classmethod ServerManager.subscribe(p_channel, p_handler, p_prio)[source]
classmethod ServerManager.start()[source]
classmethod ServerManager.join()[source]
classmethod ServerManager.stop()[source]
ServerManager.__dict__ = mappingproxy({'initialize': <classmethod object at 0x7f24e3eacb38>, 'mount': <classmethod object at 0x7f24e3eacb70>, 'listen': <classmethod object at 0x7f24e3eac828>, '__dict__': <attribute '__dict__' of 'ServerManager' objects>, '__weakref__': <attribute '__weakref__' of 'ServerManager' objects>, '__doc__': None, 'subscribe': <classmethod object at 0x7f24e3eac908>, 'LoggerFilter': <class 'xtd.network.server.manager.ServerManager.LoggerFilter'>, 'ms_initialized': False, '__module__': 'xtd.network.server.manager', 'join': <classmethod object at 0x7f24e3e21b38>, 'start': <classmethod object at 0x7f24e3eac5c0>, 'stop': <classmethod object at 0x7f24e3e21dd8>})
ServerManager.__module__ = 'xtd.network.server.manager'
ServerManager.__weakref__

list of weak references to the object (if defined)

xtd.network.server.param module
class xtd.network.server.param.ParamPage(p_credentials=None)[source]

Bases: object

__init__(p_credentials=None)[source]
static get_data()[source]
check_password(p_realm, p_username, p_password)[source]
write(*p_args, **p_kwds)[source]
default(*p_args, **p_kwds)[source]
__dict__ = mappingproxy({'__doc__': None, 'check_password': <function ParamPage.check_password at 0x7f24e39d4840>, '__dict__': <attribute '__dict__' of 'ParamPage' objects>, '__weakref__': <attribute '__weakref__' of 'ParamPage' objects>, 'get_data': <staticmethod object at 0x7f24e3bf2eb8>, 'default': <function ParamPage.default at 0x7f24e39d49d8>, '__init__': <function ParamPage.__init__ at 0x7f24e39d4510>, '__module__': 'xtd.network.server.param', 'write': <function ParamPage.write at 0x7f24e39d4950>})
__module__ = 'xtd.network.server.param'
__weakref__

list of weak references to the object (if defined)

xtd.network.server.tools module
xtd.network.server.tools.log_request_response(p_withResponse)[source]
xtd.network.server.tools.request_logger()[source]
xtd.network.server.tools.response_logger()[source]
xtd.network.server.tools.perf_begin()[source]
xtd.network.server.tools.perf_end()[source]
exception xtd.network.server.tools.JsonHTTPError(p_status, p_message)[source]

Bases: cherrypy._cperror.HTTPError

__init__(p_status, p_message)[source]
set_response()[source]
__module__ = 'xtd.network.server.tools'