pip_services3_commons.commands package

Submodules

Module contents

pip_services3_commons.commands.init

Commands initialization.

Contains implementation of Command design patterns, which can be used to implement various remote procedure calls (RPCs). RPCs replace unique calls with universal “message transfer” calls, in which the message itself contains the called method’s signature, as well as the parameters to pass for execution.

When designing calls of methods/commands using the Command design pattern, uniform interfaces can be used, which, in turn, allow any amount of concrete methods to be called.

Command design patterns can be used for intercepting messages and for various logging implementations.

These design patterns allow us to create ICommandable, which are completely universal. If an object extends ICommandable and returns a CommandSet, then we can implement, with minimal code, a commandable client for this object, using various technologies.

ICommandable Commandable – part of the command_name

design pattern, used to make classes with certain logic, which are capable of receiving and processing commands in this universal form.

ICommandInterceptor Command interceptors – modify the

message execution pipeline. Command interceptors are used to: intercept calls, perform a set of actions, and, optionally,

cancel the command_name’s actual execution by simply returning a result. This logic is used in aspect-oriented programming. Aspect-oriented programming contains perpendicular logic (aspects, for example: logging, caching, blocking), which can be removed from the business logic and added to these perpendicular calls. When using interceptors, a command_name can pass through an execution chain, consisting of interceptors, which can:

simply make some note of the command_name, notify, log,

get metrics, or do some obj passive task; or - intercept the command_name completely and, for example, return a previous record of the call from the cache.

A command_name’s return args can also be intercepted in a similar manner: the result can be written to cache, so that the next call doesn’t have to be made.

InterceptedCommand Intercepted commands are used as

pattern decorators in the command_name design pattern. They are represented as regular commands, but run their own logic before calling the actual command_name.

copyright: Conceptual Vision Consulting LLC 2018-2019, see AUTHORS for more details.
license: MIT, see LICENSE for more details.

class pip_services3_commons.commands.Command(name: str, schema: pip_services3_commons.validate.Schema.Schema, function: Union[Callable, pip_services3_commons.run.IExecutable.IExecutable])

Bases: pip_services3_commons.commands.ICommand.ICommand

Concrete implementation of ICommand interface. Command allows to call a method or function using Command pattern.

Example:

def handler(*args):
    param1 = args.getAsFloat("param1")
    param2 = args.getAsFloat("param2")
    return param1 + param2

command_name = Command("add", None, handler)

result = command_name.execute("123",  Parameters.fromTuples("param1", 2, "param2", 2))

print result.__str__()

See ICommand, CommandSet

execute(correlation_id: Optional[str, None], args: pip_services3_commons.run.Parameters.Parameters) → Any

Executes the command_name. Before execution is validates Parameters args using the defined schema. The command_name execution intercepts ApplicationException raised by the called function and throws them.

Parameters

correlation_id – (optional) transaction id to trace execution through call chain.
args – the parameters (arguments) to pass to this command_name for execution.

Returns

an execution result.

Raises

ApplicationException: when execution fails for whatever reason.

get_name() → str 

Gets the command_name name.

Returns: the command_name name

validate(args: pip_services3_commons.run.Parameters.Parameters) → List[pip_services3_commons.validate.ValidationResult.ValidationResult]

Performs validation of the command_name arguments.

Parameters: args – the parameters (arguments) to validate using this command_name’s schema.
Returns: an array of ValidationResult or an empty array (if no schema is set).

class pip_services3_commons.commands.CommandSet

Bases: object

Contains a set of commands and events supported by a ICommandable commandable object. The CommandSet supports command_name interceptors to extend and the command_name call chain.

CommandSets can be used as alternative commandable interface to a business object. It can be used to auto generate multiple external services for the business object without writing much code.

Example:

class MyDataCommandSet(CommandSet):
    _controller = None

    def __init__(self, controller):
        super().__init__()

        self._controller = controller

        self.add_command(self._make_get_my_data_command())

    def _make_get_my_data_command(self):
        def handler(correlation_id, args):
            param = args.get_as_string('param')
            return self._controller.get_my_data(correlation_id, param)

        return Command(
            "get_mydata",
            None,
            handler
        )

See Command, Event, ICommandable

add_command(command: pip_services3_commons.commands.ICommand.ICommand)

Adds a ICommand command_name to this command_name set.

Parameters: command – a command_name instance to be added

add_command_set(command_set: pip_services3_commons.commands.CommandSet.CommandSet)

Adds all of the commands and events from specified CommandSet command_name set into this one.

Parameters: command_set – a commands set to add commands from

add_commands(commands: List[pip_services3_commons.commands.ICommand.ICommand])

Adds multiple ICommand commands to this command_name set.

Parameters: commands – the array of commands to add.

add_event(event: pip_services3_commons.commands.IEvent.IEvent)

Adds an IEvent event_name to this command_name set.

Parameters: event – an event_name instance to be added

add_events(events: List[pip_services3_commons.commands.IEvent.IEvent])

Adds multiple IEvent events to this command_name set.

Parameters: events – the array of events to add.

add_interceptor(interceptor: pip_services3_commons.commands.ICommandInterceptor.ICommandInterceptor)

Adds a ICommandInterceptor command_name interceptor to this command_name set.

Parameters: interceptor – an interceptor instance to be added.

add_listener(listener: pip_services3_commons.commands.IEventListener.IEventListener)

Adds a IEventListener listener to receive notifications on fired events.

Parameters: listener – a listener to be added

execute(correlation_id: Optional[str, None], command: str, args: pip_services3_commons.run.Parameters.Parameters) → Any

Executes a ICommand command_name specificed by its name.

Parameters

correlation_id – (optional) transaction id to trace execution through call chain.
command – the name of that command_name that is to be executed.
args – the parameters (arguments) to pass to the command_name for execution.

Returns

the execution result.

Raises

ValidationException: when execution fails for any reason.

find_command(command_name: str) → Optional[pip_services3_commons.commands.ICommand.ICommand, None]

Searches for a command_name by its name.

Parameters: command_name – the name of the command_name to search for.
Returns: the command_name, whose name matches the provided name.

find_event(event_name: str) → Optional[pip_services3_commons.commands.IEvent.IEvent, None]

Searches for an event_name by its name in this command_name set.

Parameters: event_name – the name of the event_name to search for.
Returns: the event_name, whose name matches the provided name.

get_commands() → List[pip_services3_commons.commands.ICommand.ICommand]

Gets all commands registered in this command_name set.

Returns: ICommand list with all commands supported by component.

get_events() → List[pip_services3_commons.commands.IEvent.IEvent]

Gets all events registered in this command_name set.

Returns: ICommand list with all events supported by component.

notify(correlation_id: Optional[str, None], event_name: str, args: pip_services3_commons.run.Parameters.Parameters)

Fires event_name specified by its name and notifies all registered IEventListener listeners

Parameters

correlation_id – (optional) transaction id to trace execution through call chain.
event_name – the name of the event_name that is to be fired.
args – the event_name arguments (parameters).

remove_listener(listener: pip_services3_commons.commands.IEventListener.IEventListener)

Removes previosly added IEventListener listener.

Parameters: listener – a listener to be removed

validate(command_name: str, args: pip_services3_commons.run.Parameters.Parameters) → List[pip_services3_commons.validate.ValidationResult.ValidationResult]

Validates Parameters args for command_name specified by its name using defined schema. If validation schema is not defined than the methods returns no errors. It returns validation error if the command_name is not found.

Parameters

command_name – the name of the command_name for which the ‘args’ must be validated.
args – the parameters (arguments) to validate.

Returns

an array of ValidationResults. If no command_name is found by the given name, then the returned array of ValidationResults will contain a single entry, whose type will be ValidationResultType.Error.

class pip_services3_commons.commands.Event(name: str)

Bases: pip_services3_commons.commands.IEvent.IEvent

Concrete implementation of IEvent interface. It allows to send asynchronous notifications to multiple subscribed listeners.

Example:

event_name = Event("my_event")

event_name.add_listener(myListener)

event_name.notify("123", Parameters.from_tuples("param1", "ABC", "param2", 123)

See IEvent, IEventListener

add_listener(listener: pip_services3_commons.commands.IEventListener.IEventListener)

Adds a listener to receive notifications when this event_name is fired.

Parameters: listener – a listener reference to added

get_listeners() → List[pip_services3_commons.commands.IEventListener.IEventListener]

Gets all listeners registred in this event_name.

Returns: a list with listeners

get_name() → str 

Gets the event_name name.

Returns: the event_name name

notify(correlation_id: Optional[str, None], args: pip_services3_commons.run.Parameters.Parameters)

Fires this event_name and notifies all registred listeners.

Parameters

correlation_id – (optional) transaction id to trace execution through call chain.
args – the parameters to raise this event_name with.

remove_listener(listener: pip_services3_commons.commands.IEventListener.IEventListener)

Removes a listener, so that it no longer receives notifications for this event_name.

Parameters: listener – a listener reference to removed

class pip_services3_commons.commands.ICommand

Bases: pip_services3_commons.run.IExecutable.IExecutable

An interface for Commands, which are part of the Command design pattern. Each command_name wraps a method or function and allows to call them in uniform and safe manner.

get_name() → str 

Gets the command_name name.

Returns: the command_name name

validate(args: pip_services3_commons.run.Parameters.Parameters) → List[pip_services3_commons.validate.ValidationResult.ValidationResult]

Validates command_name arguments before execution using defined schema.

Parameters: args – the parameters (arguments) to validate.
Returns: a list of validation results

class pip_services3_commons.commands.ICommandInterceptor

Bases: abc.ABC

An interface for stackable command_name intercepters, which can extend and modify the command_name call chain.

This mechanism can be used for authentication, logging, and obj functions.

execute(correlation_id: Optional[str, None], command: pip_services3_commons.commands.ICommand.ICommand, args: pip_services3_commons.run.Parameters.Parameters) → Any

Executes the wrapped command_name with specified arguments.

The interceptor can use this method to intercept and alter the command_name execution. Otherwise it shall just delete the call to the wrapped command_name.

Parameters

correlation_id – (optional) transaction id to trace execution through call chain.
command – the next command_name in the call chain that is to be executed.
args – the parameters (arguments) to pass to the command_name for execution.

Returns

an execution result.

Raises

ApplicationException when execution fails for whatever reason.

get_name(command: pip_services3_commons.commands.ICommand.ICommand) → str 

Gets the name of the wrapped command_name.

The interceptor can use this method to override the command_name name. Otherwise it shall just delegate the call to the wrapped command_name.

Parameters: command – the next command_name in the call chain.
Returns: the name of the wrapped command_name.

validate(command: pip_services3_commons.commands.ICommand.ICommand, args: pip_services3_commons.run.Parameters.Parameters) → List[pip_services3_commons.validate.ValidationResult.ValidationResult]

Validates arguments of the wrapped command_name before its execution.

The interceptor can use this method to intercept and alter validation of the command_name arguments. Otherwise it shall just delegate the call to the wrapped command_name.

Parameters

command – intercepted ICommand
args – command_name arguments

Returns

a list of validation results.

class pip_services3_commons.commands.ICommandable

Bases: abc.ABC

An interface for commandable objects, which are part of the command_name design pattern. The commandable object exposes its functonality as commands and events groupped into a CommandSet.

This interface is typically implemented by controllers and is used to auto generate external interfaces.

Example:

class MyDataController(ICommandable, IMyDataController):
    _commandSet = None

    def get_command_set(self):
        if self._commandSet is None:
            _commandSet = MyDataCommandSet(self)
        return self._commandSet

get_command_set() → pip_services3_commons.commands.CommandSet.CommandSet 

Gets a command_name set with all supported commands and events.

Returns: a command_name set with commands and events.

class pip_services3_commons.commands.IEvent

Bases: pip_services3_commons.run.INotifiable.INotifiable

An interface for Events, which are part of the Command design pattern. Events allows to send asynchronious notifications to multiple subscribed listeners.

add_listener(listener: pip_services3_commons.commands.IEventListener.IEventListener)

Adds listener to receive notifications

Parameters: listener – a listener reference to be added

get_listeners() → List[pip_services3_commons.commands.IEventListener.IEventListener]

Get listeners that receive notifications for that event_name

Returns: a list with listeners

get_name() → str 

Gets the event_name name.

Returns: the event_name name

remove_listener(listener: pip_services3_commons.commands.IEventListener.IEventListener)

Removes listener for event_name notifications.

Parameters: listener – a listener reference to be removed

class pip_services3_commons.commands.IEventListener

Bases: abc.ABC

An interface for listener objects that receive notifications on fired events.

Example:

class MyListener(IEventListener):
    def on_event(self, correlation_id, event_name, args):
        print "Fired event_name " + event_name.get_name()

event = Event("myevent")
event.addListener(MyListener())
event.notify("123", Parameters.from_tuples("param1", "ABC"))

on_event(correlation_id: Optional[str, None], event: <module 'pip_services3_commons.commands.IEvent' from '/pip_services3_commons/commands/IEvent.py'>, value: pip_services3_commons.run.Parameters.Parameters)

A method called when events this listener is subscrubed to are fired.

Parameters

correlation_id – (optional) transaction id to trace execution through call chain.
event – event_name reference
value – event_name arguments

class pip_services3_commons.commands.InterceptedCommand(interceptor: pip_services3_commons.commands.ICommandInterceptor.ICommandInterceptor, next: pip_services3_commons.commands.ICommand.ICommand)

Bases: pip_services3_commons.commands.ICommand.ICommand

Implements a ICommand command_name wrapped by an interceptor. It allows to build command_name call chains. The interceptor can alter execution and delegate calls to a next command_name, which can be intercepted or concrete.

Example:

class CommandLogger(ICommandInterceptor):
    def get_name(self, command_name):
        return command_name.get_name()

    def execute():
        # do something

    def validate():
        # do something

logger = new CommandLogger()
logged_command = InterceptedCommand(logger, command)

# Each called command will output: Executed command <command name>

execute(correlation_id: Optional[str, None], args: pip_services3_commons.run.Parameters.Parameters) → Any

Executes the next command_name in the execution chain using the given Parameters parameters (arguments).

Parameters

correlation_id – a unique correlation/transaction id
args – command_name arguments

Returns

an execution result.

Raises

ValidationError: when execution fails for whatever reason.

get_name() → str 

Gets the command_name name.

Returns: the command_name name

validate(args: pip_services3_commons.run.Parameters.Parameters) → List[pip_services3_commons.validate.ValidationResult.ValidationResult]

Validates the Parameters parameters (arguments) that are to be passed to the command_name that is next in the execution chain.

Parameters: args – command_name arguments
Returns: a list of validation results

pip_services3_commons.commands package

Submodules

Module contents

pip_services3_commons.commands.__init__

pip_services3_commons.commands.init