decisionengine.framework.engine package

Subpackages

• decisionengine.framework.engine.tests package
  • Submodules
  • decisionengine.framework.engine.tests.conftest module
    • _redis_server_running()
    • pytest_runtest_call()
  • decisionengine.framework.engine.tests.fixtures module
    • DEServer()
    • PG_DE_DB_WITHOUT_SCHEMA()
    • PG_PROG()
    • SQLALCHEMY_PG_WITH_SCHEMA()
    • SQLALCHEMY_TEMPFILE_SQLITE()
  • decisionengine.framework.engine.tests.test_ChannelWorkers module
    • TaskManager
      • TaskManager.name
      • TaskManager.set_loglevel_value()
    • global_config()
    • test_worker_logger_sized_rotation()
    • test_worker_logger_timed_rotation()
    • test_worker_name()
  • decisionengine.framework.engine.tests.test_SourceWorkers module
    • global_config()
    • source_config()
    • source_worker_for()
    • test_worker_logger_sized_rotation()
    • test_worker_logger_timed_rotation()
    • test_worker_logger_wrong_rotation_method()
    • test_worker_name()
  • decisionengine.framework.engine.tests.test_client_only module
    • test_client_err_returned_as_rc()
    • test_client_err_returned_verbose_as_rc()
    • test_client_help()
    • test_client_with_no_command_says_use_help()
    • test_client_with_no_server()
    • test_client_with_no_server_verbose()
    • test_exclusive_options()
  • decisionengine.framework.engine.tests.test_query_tool_only module
    • test_client_err_returned_as_rc()
    • test_client_err_returned_verbose_as_rc()
    • test_query_tool_help()
    • test_query_tool_with_no_server()
    • test_query_tool_with_no_server_verbose()
  • decisionengine.framework.engine.tests.test_startup module
    • _check_override()
    • metrics_env_setup()
    • test_change_port()
    • test_check_metrics_env_no_webserver()
    • test_check_metrics_env_var_set()
    • test_check_metrics_env_var_unset()
    • test_default_config()
  • decisionengine.framework.engine.tests.test_verify_redis_server module
    • fake_redis_server()
    • test_verify_bad_broker()
    • test_verify_bad_redis_server()
    • test_verify_bad_url()
    • test_verify_redis_server()
    • test_verify_redis_server_int()
    • test_verify_redis_url()
  • Module contents

Submodules

decisionengine.framework.engine.ChannelWorkers module

class decisionengine.framework.engine.ChannelWorkers.ChannelWorker(task_manager, logger_config)

Bases: Process

Class that encapsulates a channel’s task manager as a separate process.

This class’ run function is called whenever the process is started. If the process is abruptly terminated–e.g. the run method is preempted by a signal or an os._exit(n) call–the ChannelWorker object will still exist even if the operating-system process no longer does.

To determine the exit code of this process, use the ChannelWorker.exitcode value, provided by the multiprocessing.Process base class.

get_consumes()

get_produces()

get_state_name()

run()

Method to be run in sub-process; can be overridden in sub-class

setup_logger()

wait_while(state, timeout=None)

class decisionengine.framework.engine.ChannelWorkers.ChannelWorkers

Bases: object

This class manages and provides access to the task-manager workers.

The intention is that the decision engine never directly interacts with the workers but refers to them via a context manager:

with workers.access() as ws:

# Access to ws now protected ws[‘new_channel’] = ChannelWorker(…)

In cases where the decision engine’s block_while method must be called (e.g. during tests), one should use unguarded access:

ws = workers.get_unguarded() # Access to ws is unprotected ws[‘new_channel’].wait_while(…)

Calling a blocking method while using the protected context manager (i.e. workers.access()) will likely result in a deadlock.

class Access(workers, lock)

Bases: object

access()

accessed_by_another_thread()

get_unguarded()

decisionengine.framework.engine.ClientMessageReceiver module

class decisionengine.framework.engine.ClientMessageReceiver.ClientMessageReceiver(exchange_name, exchange_type, broker_url, routing_key_suffix, logger_name)

Bases: object

_receive(body, message)

execute(func, *args)

decisionengine.framework.engine.DecisionEngine module

Main loop for Decision Engine. The following environment variable points to decision engine configuration file: DECISION_ENGINE_CONFIG_FILE if this environment variable is not defined the DE-Config.py file from the ``../tests/etc/` directory will be used.

class decisionengine.framework.engine.DecisionEngine.DecisionEngine(global_config, channel_config_loader, server_address)

Bases: ThreadingMixIn, SimpleXMLRPCServer

_dataframe_to_column_names(df)

_dataframe_to_csv(df)

_dataframe_to_json(df)

_dataframe_to_table(df)

_dataframe_to_vertical_tables(df)

_dispatch(method, params)

Dispatches the XML-RPC method.

XML-RPC calls are forwarded to a registered function that matches the called XML-RPC method name. If no such function exists then the call is forwarded to the registered instance, if available.

If the registered instance has a _dispatch method then that method will be called with the name of the XML-RPC method and its parameters as a tuple e.g. instance._dispatch(‘add’,(2,3))

If the registered instance does not have a _dispatch method then the instance will be searched to find a matching method and, if found, will be called.

Methods beginning with an ‘_’ are considered private and will not be called.

block_while(state, timeout=None)

create_channel(channel_name, channel_config)

get_logger()

metrics()

reaper_start(delay)

reaper_status()

reaper_stop()

rm_channel(channel, maybe_timeout)

rpc_block_while(client_queue, state_str, timeout=None)

rpc_get_channel_log_level(client_queue, channel)

rpc_get_log_level(client_queue)

rpc_get_source_log_level(client_queue, source)

rpc_kill_channel(client_queue, channel, timeout=None)

rpc_metrics(client_queue)

Display collected metrics

rpc_ping(client_queue)

rpc_print_product(client_queue, product, columns=None, query=None, types=False, format=None)

rpc_print_products(client_queue)

rpc_product_dependencies(client_queue)

rpc_query_tool(client_queue, product, format=None, start_time=None)

rpc_queue_status(client_queue)

rpc_reaper_start(client_queue, delay=0)

Start the reaper process after ‘delay’ seconds. Default 0 seconds delay. :type delay: int

rpc_reaper_status(client_queue)

rpc_reaper_stop(client_queue)

rpc_rm_channel(client_queue, channel, maybe_timeout)

rpc_set_channel_log_level(client_queue, channel, log_level)

Assumes log_level is a string corresponding to the supported logging-module levels.

rpc_set_source_log_level(client_queue, source, log_level)

Assumes log_level is a string corresponding to the supported logging-module levels.

rpc_show_config(client_queue, channel)

Show the configuration for a channel.

rpc_show_de_config(client_queue)

rpc_start_channel(client_queue, channel_name)

rpc_start_channels(client_queue)

rpc_status(client_queue)

rpc_stop(client_queue=None)

rpc_stop_channel(client_queue, channel)

rpc_stop_channels(client_queue)

service_actions()

Called by the serve_forever() loop.

May be overridden by a subclass / Mixin to implement any code that needs to be run during the loop.

start_channel(channel_name, src_workers)

start_channels()

start_webserver()

Start CherryPy webserver using configured port. If port is not configured use default webserver port.

stop_channels()

stop_worker(worker, timeout)

class decisionengine.framework.engine.DecisionEngine.RequestHandler(request, client_address, server)

Bases: SimpleXMLRPCRequestHandler

rpc_paths = ('/RPC2',)

class decisionengine.framework.engine.DecisionEngine.StopState(value)

Bases: Enum

An enumeration.

Clean = 2

NotFound = 1

Terminated = 3

decisionengine.framework.engine.DecisionEngine._channel_preamble(name)

decisionengine.framework.engine.DecisionEngine._check_metrics_env(options)

decisionengine.framework.engine.DecisionEngine._create_de_server(global_config, channel_config_loader)

Create the DE server with the passed global configuration and config manager

decisionengine.framework.engine.DecisionEngine._get_de_conf_manager(global_config_dir, channel_config_dir, options)

decisionengine.framework.engine.DecisionEngine._get_global_config(config_file, options)

decisionengine.framework.engine.DecisionEngine._initial_start_channels(server)

decisionengine.framework.engine.DecisionEngine._queue_name_and_type(redis_obj, redis_member)

decisionengine.framework.engine.DecisionEngine._requests_in_flight(redis_obj, exchange_name)

decisionengine.framework.engine.DecisionEngine._start_de_server(server)

Start the DE server and listen forever

decisionengine.framework.engine.DecisionEngine._verify_redis_server(broker_url)

decisionengine.framework.engine.DecisionEngine._verify_redis_url(broker_url)

decisionengine.framework.engine.DecisionEngine.main(args=None)

If args is None, sys.argv will be used instead If args is a list, it will be used instead of sys.argv (for unit testing)

decisionengine.framework.engine.DecisionEngine.parse_program_options(args=None)

If args is a list, it will be used instead of sys.argv

decisionengine.framework.engine.SourceWorkers module

class decisionengine.framework.engine.SourceWorkers.SourceWorker(key, config, logger_config, channel_name, exchange, broker_url)

Bases: Process

Provides interface to loadable modules an events to synchronize execution

get_loglevel()

run()

Get the data from source

set_loglevel_value(log_level)

Assumes log_level is a string corresponding to the supported logging-module levels.

setup_logger()

take_offline()

class decisionengine.framework.engine.SourceWorkers.SourceWorkers(exchange, broker_url, logger=<BoundLoggerLazyProxy(logger=None, wrapper_class=None, processors=None, context_class=None, initial_values={}, logger_factory_args=('decisionengine', ))>)

Bases: object

This class manages and provides access to the Source workers.

The intention is that the decision engine never directly interacts with the workers but refers to them via a context manager:

with workers.access() as ws:

# Access to ws now protected ws[‘new_source’] = SourceWorker(…)

In cases where the decision engine’s block_while method must be called (e.g. during tests), one should use unguarded access:

ws = workers.get_unguarded() # Access to ws is unprotected ws[‘new_source’].wait_while(…)

Calling a blocking method while using the protected context manager (i.e. workers.access()) will likely result in a deadlock.

class Access(workers, lock)

Bases: object

access()

detach(channel_name, source_names)

get_unguarded()

prune(channel_name, source_names)

remove_all(timeout)

update(channel_name, source_configs, logger_config)

decisionengine.framework.engine.de_client module

decisionengine.framework.engine.de_client.command_for_args(argsparsed, de_socket)

argsparsed should be from create_parser in this file

decisionengine.framework.engine.de_client.console_scripts_main(args_to_parse=None)

This is the entry point for the setuptools auto generated scripts. Setuptools thinks a return from this function is an error message.

decisionengine.framework.engine.de_client.create_parser()

decisionengine.framework.engine.de_client.main(args_to_parse=None, logger_name='de_client')

If you pass a list of args, they will be used instead of sys.argv

decisionengine.framework.engine.de_query_tool module

decisionengine.framework.engine.de_query_tool.command_for_args(argsparsed, de_socket)

Calls the proper function for the arguments passed to de_query_tool.

Parameters:

• argsparsed (Namespace) – Should be from create_parser in this file.

• de_socket (ServerProxy) – RPC Server Proxy.

Returns:

Output of the command.

Return type:

str

decisionengine.framework.engine.de_query_tool.console_scripts_main(args_to_parse=None)

This is the entry point for the setuptools auto generated scripts. Setuptools thinks a return from this function is an error message.

decisionengine.framework.engine.de_query_tool.create_parser()

decisionengine.framework.engine.de_query_tool.main(args_to_parse=None, logger_name='de_query_tool')

Main function for de_query_tool

Parameters:

args_to_parse (list, optional) – If you pass a list of args, they will be used instead of sys.argv. Defaults to None.

Returns:

Query result

Return type:

str

Module contents