12
\$\begingroup\$

I asked this question a while ago, and this was one of the comments:

There (mostly) seems to be a one-to-one mapping of a lot of your functions to IRC command verbs. For example, leave_server -> QUIT, join_channel -> JOIN, send_private_message -> PRIVMSG, etc. Part of me just feels that (as you say) there is just a tad too much repetition in that architecture. (Perhaps some more abstract encapsulation of a server command and arguments?

This made sense to me; a lot of the general work about making sure a command looks correct can probably be abstracted out such that I can define what valid parameters to a command would look like. I started by defining a set of parameters (note, I'm using the numpydoc package style for Sphinx documentation generation throughout):

class CommandParameterSet(object):
    """A set of parameters to a command."""

    def __init__(self, *parameters):
        """Create a new CommandParameterSet.

        Parameters
        ----------
        parameters: iterable[CommandParameter]
            Collection of parameters that we expect to see. Must be an
            empty list if no parameters expected.
        """

        self.parameters = list(parameters) or []

    def insert_parameter(self, parameter, index=None):
        """Add another parameter to the parameter list in the given
        location, or append if not specified.

        We can't guarantee the order in which the decorators will 
        execute, so this is a little flexible. If you insert at a
        greater index than is currently supported, we'll fill it out
        with `None` until we can fit it in. If you insert at a location
        that is already populated we just replace it.

        Parameters
        ----------
        parameter: CommandParameter
            A parameter to insert into this set.
        index: indexer, default=None
            Where to insert the parameter; if None then just append it.
        """

        if index is None:
            index = len(self.parameters)
        if index < 0:
            raise IndexError(
                "CommandParameterSet doesn't support negative indices")

        diff = index - len(self.parameters)
        if diff == 0:
            self.parameters.append(parameter)
        elif diff > 0:
            self.parameters.extend([None]*diff + [parameter])
        else:
            self.parameters[index] = parameter

    def validate(self, values):
        """Check if the given values pass parameter validation.

        We have to have filled in all of the parameters; if any are
        `None` then this will fail as well.

        Parameters
        ----------
        values: iterable[Object]
            Collection of the values that we expect to see. Must be an
            empty list if no values to check.

        Returns
        -------
        result: generator[string]
            Yields all of the errors found; if none, then an empty generator
            is returned.
        """

        if any(param is None for param in self.parameters):
            yield "Not all parameters are populated."
        elif len(values) != len(self.parameters):
            yield "Expected {} parameters, got {}".format(
                len(self.parameters), len(values))
        else:
            for i, (val, param) in enumerate(zip(values, self.parameters)):
                problem = param.validate(val)
                if problem:
                    yield "{i}-{name} was {val} but {problem}".format(
                        name=param.name, **locals())

    def __str__(self):
        """Returns a string representation of the set."""

        return "CommandParameterSet[{}]".format(len(self.parameters))

    def __unicode__(self):
        """Returns the unicode string representation of the set."""

        return unicode(str(self))

    def __repr__(self):
        """Returns a round-trippable representation of the set."""

        return "CommandParameterSet({})".format(
                    ", ".join(repr(param) for param in self.parameters))

    def __eq__(self, other):
        """Check whether or not two CommandParameterSets are equal."""

        if len(self) != len(other):
            return False

        return all(me == them for me, them in izip_longest(self, other, fillvalue=None))

    def __ne__(self, other):
        """Check whether or not two CommandPArameterSets are not equal."""

        return not (self == other)

    def __len__(self):
        """Returns the number of parameters in the set."""

        return len(self.parameters)

    def __iter__(self):
        """Return an iterator for this command parameter set."""

        for param in self.parameters:
            yield param

All this class really does is provide a nice abstraction over a list of parameters. These parameters are defined by the following class:

import enum

@enum.unique
class CountType(enum.Enum):
    """Possible types of "counts" we could have.

    Attributes
    ----------
    MAX
        The given "count" is the most allowed.
    MIN
        The given "count" is the least allowed.
    EXACT
        The given "count" is exactly the number expected.
    """

    MAX = 1
    MIN = 2
    EXACT = 3


class CommandParameter(object):
    """Parameter to a command.

    Can specify validation code and options for validation.
    """

    def __init__(self, name, validator, optional=False, count_type=None, count=1):
        """Create a new CommandParameter.

        Parameters
        ----------
        name: string
            Name of the parameter; used to describe the parameter.
        validator: function(value) -> string | None
            Function that takes in a value and returns an error message,
            or None if it was okay. If there are multiple values allowed
            (i.e. `count_type != None`) then this is called for each 
            item in the collection, _instead_ of on the entire 
            collection.
        optional: boolean, default=False
            Whether or not the parameter is optional. If it is optional,
            then `None` should be passed for validation.
        count_type: CountType, default=None
            Required if a `count` is given; describes how to interpret 
            the count (as a max, min, or exact requirement).
        count: integer, default=1
            Describes how many instances of this value are allowed. If 
            greater than 1, then `count_type` is required. If 0 or more
            are allowed, then pass `count=0` and
            `count_type=CountTypes.MIN`
        """

        self.name = name
        self.validator = validator
        self.optional = optional
        self.count = count
        self.count_type = count_type

        if not callable(validator):
            raise TypeError("Validator must be callable")
        if count < 0:
            raise AttributeError("Count must be positive; was {}".format(count))
        if int(count) != count:
            raise TypeError("Count must be an integer; was {}".format(count))
        if count > 1 and count_type is None:
            raise AttributeError(
                "Count type is required if multiple values are allowed.")
        if count == 0 and count_type is not CountType.MIN:
            raise AttributeError(
                "Count and count type would require 0 or fewer values.")

    def validate(self, value):
        """Check if the given value satisfies this parameter.

        Parameters
        ----------
        value: Object
            The value to check for this parameter.

        Returns
        -------
        result: string|None
            None if the result is okay, otherwise an error message.
        """

        if self.optional and value is None:
            return None

        if not self.count_type:
            return self.validator(value)
        elif self.count_type == CountType.MAX and len(value) > self.count:
            return "Expected no more than {} parameters, but got {}".format(
                self.count, len(value))
        elif self.count_type == CountType.MIN and len(value) < self.count:
            return "Expected no less than {} parameters, but got {}".format(
                self.count, len(value))
        elif len(value) != self.count:
            return "Expected exactly {} parameters, but got {}".format(
                self.count, len(value))

        return None

    def __str__(self):
        """Returns a string representation of the parameter."""

        return "CommandParameter"

    def __unicode__(self):
        """Returns the unicode string representation of the set."""

        return unicode(str(self))

    def __repr__(self):
        """Returns a round-trippable representation of the set."""

        return \
            "CommandParameter('{}', {}, optional={}, count_type={}, count={})"\
            .format(
                self.name, self.validator.__name__, self.optional, 
                self.count_type, self.count)

    def __eq__(self, other):
        """Check whether or not two CommandParameters are equal."""

        try:
            return (self.name == other.name 
                and self.validator is other.validator
                and self.optional is other.optional
                and self.count_type is other.count_type
                and self.count == other.count)
        except AttributeError:
            raise TypeError(
                "CommandParameter and type <{}> can't be compared.".format(
                    type(other)))

    def __ne__(self, other):
        """Check whether or not two CommandParameters are not equal."""

        return not (self == other)

There were a few things I wanted to encapsulate here:

  • A parameter may be optional; in this situation, it must be explicitly passed as None for the framework to pick up on that
  • A parameter may be repeated in any number of ways; at least n repetitions, at most n repetitions, or exactly n repetitions
  • A parameter must be in some detectably valid state

You may have noticed a somewhat questionable design choice here: I return strings instead of raising an error when something fails validation. I did this for a few reasons:

  • I want to be able to get the reason validation failed back, so just returning a boolean doesn't work
  • Similarly, once you're returning a string, you don't really need the boolean so returning a tuple was out
  • I want to be able to handle a whole bunch of these parameters at once (i.e. in a CommandParameterSet) without having to enclose each one in a try...except

That being said, I'm not convinced that what I've done is actually better (it would be trivial to make an equivalent implementation using exceptions), and it definitely looks un-Pythonic. Looking for feedback here for sure.

Now we get to the fun part. I wanted it to be easy to clearly define what commands are available, how they are parameterized, and what happens when you execute them. For this, I used enums and decorators. The basic idea is like so:

class CommandEnum(ExecutableCommandMixin, enum.Enum):
   COMMAND1 = 1
   COMMAND2 = 2
   ...

@CommandEnum.COMMAND1.register_parameter(name, index, **kwargs)
def validate_name_param(value):
   # Do some validation
   return None # no error

@CommandEnum.COMMAND1.register_execution
def execute_some_command(all, these, parameters):
   # execute something
   return super_awesome_value

To accomplish this, I wrote the following mixin class:

class InvalidCommandParametersException(Exception):
    """Raised when the parameters to a command are invalid."""

    def __init__(self, command, param_problems):
        """Raise an error that a command's parameters are invalid.
        Parameters
        ----------
        command: Command
            The command type that failed validation.
        param_problems: iterable[string]
            Collection of problems with the parameters
        """

        message = '\n'.join(["Command: {}".format(command.name)] + param_problems)
        super(InvalidCommandParametersException, self).__init__(message)

class ExecutableCommandMixin(object):
    """Mixin to make an enum of commands executable.

    Enables parameterizing a given command and specifying how to 
    validate each parameter, as well as specifying what "executing" the
    command means. Does so by providing decorator functions that will
    register functions as parameter validation or command execution.

    Notes
    -----
    Does not subclass from `enum.Enum` because working around the
    non-extensibility of enums with defined members is pretty inelegant.

    Properties
    ----------
    parameters: CommandParameterSet
        Set of parameters for this command.
    execution: function(*values) -> Object
        Function that takes in the parameters and returns something.
    """

    @property
    def parameters(self):
        return self.__class__._parameters[self]

    _executions = {}
    @property
    def execution(self):
        return self.__class__._executions[self]

    def execute_command(self, *values):
        problems = self._validate_arguments(values)
        if problems:
            raise InvalidCommandParametersException(self, problems)

        return self.execution(*values)

    def register_execution(self, func):
        """Register a function as this command's action.

        Parameters
        ----------
        func: function
            Function to execute for this parameter.

        Returns
        -------
        func: function
            The original function, unchanged.
        """

        self.__class__._executions[self] = func
        return func

    _parameters = defaultdict(CommandParameterSet)
    def register_parameter(self, name, n_th, optional=False, count=1, count_type=None):
        """Decorator to register a function to validate a given parameter.

        Parameters
        ----------
        name: string
            Name of the parameter; used to describe the parameter.
        n_th: indexer
            Which parameter this should be in the parameter list.
        optional: boolean, default=False
            Whether or not the parameter is optional. If it is optional,
            then `None` should be passed for validation.
        count: integer, default=1
            Describes how many instances of this value are allowed. If 
            greater than 1, then `count_type` is required. If 0 or more
            are allowed, then pass `count=0` and
            `count_type=CountTypes.MIN`
        count_type: CountType, default=None
            Required if a `count` is given; describes how to interpret 
            the count (as a max, min, or exact requirement).

        Returns
        -------
        decorator: function -> function
            Wrapper function that wraps its argument function and adds
            it as validation for this parameter
        """

        def decorator(validator):
            """Add the actual validator function for this parameter.

            Parameters
            ----------
            validator: function(value) -> string | None
                Function that takes in a value and returns an error 
                message, or None if it was okay. If there are multiple 
                values allowed (i.e. `count_type != None`) then this is 
                called for each item in the collection, _instead_ of on 
                the entire collection.

            Returns
            -------
            validator: function(value) -> string | None
                The original function, unchanged.
            """

            self.__class__._parameters[self].insert_parameter(
                CommandParameter(name, validator, optional=optional, 
                    count=count, count_type=count_type),
                n_th)

            return validator

        return decorator

    def _validate_arguments(self, arguments):
        """Validate the command's arguments.

        Parameters
        ----------
        arguments: collection[Object]
            List of the arguments being passed.

        Returns
        -------
        list[string]
            List of all of the problems with the arguments. Will be an
            empty list if no problems are present.
        """

        errors = list(self.__class__._parameters[self].validate(arguments))

        if not errors or all(err is None for err in errors):
            return []
        return errors

I wish I could make this subclass enum.Enum itself, but unfortunately you can't subclass an enum that has members defined. If there is a clean way to handle that, then please let me know.

I'm also not positive that operating on self.__class__ to get the attributes is necessary, or even makes sense. It definitely doesn't look pretty, so suggestions here are welcome as well.

Lastly, although this was a lot of fun to make, it seems pretty complicated. Am I overdoing it here? Is there any real value in what I've done, or does it make more sense just to stick to the one-to-one function-to-command paradigm?


Here is a (simple) motivating example, with the IRC commands referenced in the original post:

@enum.unique
class IrcCommand(ExecutableCommandMixin, enum.Enum):
    LIST = 1


@IrcCommand.LIST.register_parameter(
    "channels", 0, optional=True, count=1, count_type=CountType.MIN)
def validate_channel_name(name):
    """Validate that a given name is a valid channel name.

    Uses RFC 1459 validation.

    Parameters
    ----------
    name: string
        Name of the channel.

    Returns
    -------
    err: string | None
        Returns an error string describing what is wrong, otherwise 
        returns None if it is okay.
    """

    ## TODO: implement this
    return None

@IrcCommand.LIST.register_parameter("server", 1, optional=True)
def validate_server_name(hostname):
    """Validate that a given name is a valid server name.

    Uses RFC 952, 1123, and 2810.

    Parameters
    ----------
    hostname: string
        Name of the server.

    Returns
    -------
    err: string | None
        Returns an error string describing what is wrong, otherwise 
        returns None if it is okay.
    """

    ## TODO: implement this
    return None

@IrcCommand.LIST.register_execution
def list_channels(channels=None, server=None):

    ## TODO: implement this
    return None

I've intentionally left out their implementations, as they don't really matter in this context.


Last, but not least, here are my tests. I run them using nose, but I assume it should work with the builtin unittest runner as well. I've made a test class for each class above; test method names are structured like so: test_<method_name>_case_description.

from PyIRC.irc.executable_command import (
    ExecutableCommandMixin, InvalidCommandParametersException, CommandParameter,
    CommandParameterSet, CountType)
from PyIRC.irc import IrcCommand

import unittest

try:
    range = xrange
except NameError:
    pass

try:
    import enum34 as enum
except ImportError:
    import enum


def something_bad(x):
    return "Something Bad"

def something_good(x):
    return None


class TestCommandParameterSet(unittest.TestCase):

    def test___init___no_parameters(self):
        """
        Test that constructing a CommandParameterSet without any
        parameters doesn't error.
        """

        cps = CommandParameterSet()
        self.assertEquals(cps.parameters, [])

    def test___init___some_parameters(self):
        """
        Test that constructing a CommandParameterSet with some
        parameters doesn't error.
        """

        params = ['i', 'am', 'a', 'parameter', 'list']
        cps = CommandParameterSet(*params)
        self.assertEquals(cps.parameters, params)

    def test_insert_parameter_no_index(self):
        """Test that inserting without an index just appends."""

        param = object()
        cps = CommandParameterSet()
        cps.insert_parameter(param)
        self.assertEquals(len(cps.parameters), 1)
        self.assertIs(cps.parameters[0], param)

    def test_insert_parameter_negative_index(self):
        """Test that a negative index raises."""

        cps = CommandParameterSet()
        params = (object(), -1)
        self.assertRaises(IndexError, cps.insert_parameter, *params)

    def test_insert_parameter_replace(self):
        """Test that inserting at an existing location replaces."""

        original_param = object()
        cps = CommandParameterSet(original_param)
        new_param = object()
        cps.insert_parameter(new_param, 0)
        self.assertIsNot(cps.parameters[0], original_param)
        self.assertIs(cps.parameters[0], new_param)
        self.assertEquals(len(cps.parameters), 1)

    def test_insert_parameter_append(self):
        """Test that inserting at length just appends."""

        cps = CommandParameterSet('test', 'params')
        param = object()
        cps.insert_parameter(param, 2)
        self.assertEquals(len(cps.parameters), 3)
        self.assertIs(cps.parameters[2], param)

    def test_insert_parameter_extend(self):
        """Test that inserting beyond length backfills with None."""

        cps = CommandParameterSet()
        param = object()
        cps.insert_parameter(param, 2)
        self.assertEquals(len(cps.parameters), 3)
        self.assertIs(cps.parameters[2], param)
        for i in range(2):
            self.assertIs(cps.parameters[i], None)

    def test_validate_still_empty(self):
        """Test that if we still have null parameters, it errors."""

        cps = CommandParameterSet()
        cps.insert_parameter("Test", 2)
        errors = list(cps.validate([]))

        self.assertEquals(len(errors), 1)
        self.assertNotEquals(errors[0], "")


    def test_validate_mismatch_length(self):
        """
        Test that if the number of values and parameters doesn't match
        it errors.
        """

        cps = CommandParameterSet()
        cps.insert_parameter("Test")
        errors = list(cps.validate([]))

        self.assertEquals(len(errors), 1)
        self.assertNotEquals(errors[0], "")

    def test_validate_values_good(self):
        """Test that all parameters validate correctly."""

        cps = CommandParameterSet()
        cps.insert_parameter(CommandParameter("test", something_good))
        errors = list(cps.validate([1]))

        self.assertEquals(len(errors), 0)

    def test_validate_values_bad(self):
        """Test that all parameters validate correctly."""

        cps = CommandParameterSet()
        cps.insert_parameter(CommandParameter("test", something_bad))
        errors = list(cps.validate([1]))

        self.assertEquals(len(errors), 1)
        self.assertNotEquals(errors[0], "")

    def test_validate_values_some_bad(self):
        """Test that all parameters validate correctly."""

        cps = CommandParameterSet()
        cps.insert_parameter(CommandParameter("test", something_bad))
        cps.insert_parameter(CommandParameter("test2", something_good))
        errors = list(cps.validate([1, 2]))

        self.assertEquals(len(errors), 1)
        self.assertNotEquals(errors[0], "")

    def test___repr___empty_round_trippable(self):
        """Test that __repr__ can be round-tripped for empty parameters."""

        cps = CommandParameterSet()
        cps2 = eval(repr(cps))

        self.assertEquals(cps, cps2)

    def test___repr___nonempty_round_trippable(self):
        """Test that __repr__ can be round-tripped non-empty params."""

        cps = CommandParameterSet(CommandParameter("test", something_bad))
        cps2 = eval(repr(cps))

        self.assertTrue(cps == cps2)

    def test___eq___empty(self):
        """Test that two empty CommandParameterSets are equal."""

        cps = CommandParameterSet()
        cps2 = CommandParameterSet()

        self.assertEquals(cps, cps2)

    def test___eq___same_length(self):
        """Test that two non-empty CommandParameterSets are equal."""

        cps = CommandParameterSet(CommandParameter("test", something_bad))
        cps2 = CommandParameterSet(CommandParameter("test", something_bad))

        self.assertTrue(cps, cps2)

    def test___eq___diff_length(self):
        """Test that two non-empty CommandParameterSets are not equal."""

        cps = CommandParameterSet(CommandParameter("test", something_bad))
        cps2 = CommandParameterSet(
            CommandParameter("test", something_bad),
            CommandParameter("test", something_bad))

        self.assertFalse(cps == cps2)

    def test___ne___empty(self):
        """Test that two empty CommandParameterSets are equal."""

        cps = CommandParameterSet()
        cps2 = CommandParameterSet()

        self.assertFalse(cps != cps2)

    def test___ne___same_length(self):
        """Test that two non-empty CommandParameterSets are equal."""

        cps = CommandParameterSet(CommandParameter("test", something_bad))
        cps2 = CommandParameterSet(CommandParameter("test", something_bad))

        self.assertFalse(cps != cps2)

    def test___ne___diff_length(self):
        """Test that two non-empty CommandParameterSets are not equal."""

        cps = CommandParameterSet(CommandParameter("test", something_bad))
        cps2 = CommandParameterSet(
            CommandParameter("test", something_bad),
            CommandParameter("test", something_bad))

        self.assertTrue(cps != cps2)

    def test___len__(self):
        """
        Test that the length of a CommandParameterSet is the length of 
        its parameters.
        """

        cps = CommandParameterSet(
            CommandParameter("test", something_bad),
            CommandParameter("test", something_bad))

        self.assertEquals(len(cps), len(cps.parameters))


class TestCommandParameter(unittest.TestCase):

    def test___init___no_error(self):
        """Test that constructing a normal CommandParameter works."""

        cp = CommandParameter("test", something_bad)

        self.assertEquals(cp.name, "test")
        self.assertEquals(cp.count, 1)
        self.assertIs(cp.validator, something_bad)
        self.assertIs(cp.optional, False)
        self.assertIs(cp.count_type, None)

    def test___init___not_callable(self):
        """Test that a non-callable validator is an error."""

        args = ["test", None]

        self.assertRaises(TypeError, CommandParameter, *args)

    def test___init___neg_count(self):
        """Test that a negative count is an error."""

        args = ["test", something_bad]
        kwargs = {"count": -1}

        self.assertRaises(AttributeError, CommandParameter, *args, **kwargs)

    def test___init___non_int_count(self):
        """Test that a non-integer count is an error."""

        args = ["test", something_bad]
        kwargs = {"count": 3.14159}

        self.assertRaises(TypeError, CommandParameter, *args, **kwargs)

    def test___init___missing_count_type(self):
        """Test that for count > 1, the count type is required."""

        args = ["test", something_bad]
        kwargs = {"count": 3}

        self.assertRaises(AttributeError, CommandParameter, *args, **kwargs)

    def test___init___nonsense_count_type(self):
        """Test that for count == 0, the count type must be min."""

        args = ["test", something_bad]
        kwargs = {"count": 0}

        self.assertRaises(AttributeError, CommandParameter, *args, **kwargs)

    def test_validate_optional(self):
        """Test that an optional parameter can be missing."""

        cp = CommandParameter("test", something_bad, optional=True)

        self.assertIs(cp.validate(None), None)

    def test_validate_no_count_type(self):
        """Test that we can validate a single value when no count type."""

        cp = CommandParameter("test", something_good)

        self.assertIs(cp.validate("anything"), None)

    def test_validate_count_exceeded(self):
        """Test that we don't exceed the count on CountType.MAX."""

        cp = CommandParameter(
            "test", something_good, count=2, count_type=CountType.MAX)

        error = cp.validate([1, 2, 3])

        self.assertIsNot(error, None)

    def test_validate_count_below(self):
        """Test that we don't come below the count on CountType.MIN."""

        cp = CommandParameter(
            "test", something_good, count=2, count_type=CountType.MIN)

        error = cp.validate([1])

        self.assertIsNot(error, None)

    def test_validate_count_exact(self):
        """Test that we get the exact count on CountType.EXACT."""

        cp = CommandParameter(
            "test", something_good, count=2, count_type=CountType.EXACT)

        error = cp.validate([1])

        self.assertIsNot(error, None)

    def test___repr___round_trippable(self):
        """Test that __repr__ can be round-tripped."""

        cp = CommandParameter(
            "test", something_good, count=2, count_type=CountType.EXACT)
        cp2 = eval(repr(cp))

        self.assertEquals(cp, cp2)

    def test___eq___equal(self):
        """Test that two equal CommandParameters can test for equality."""

        cp = CommandParameter(
            "test", something_good, count=2, count_type=CountType.EXACT)
        cp2 = eval(repr(cp))

        self.assertTrue(cp == cp2)

    def test___eq___inequal(self):
        """Test that two inequal CommandParameters can test for equality."""

        cp = CommandParameter(
            "test", something_good, count=2, count_type=CountType.EXACT)
        cp2 = eval(repr(cp))
        cp2.name = "not test"

        self.assertFalse(cp == cp2)

    def test___ne___equal(self):
        """Test that two equal CommandParameters can test for inequality."""

        cp = CommandParameter(
            "test", something_good, count=2, count_type=CountType.EXACT)
        cp2 = eval(repr(cp))

        self.assertFalse(cp != cp2)

    def test___ne___inequal(self):
        """Test that two inequal CommandParameters can test for inequality."""

        cp = CommandParameter(
            "test", something_good, count=2, count_type=CountType.EXACT)
        cp2 = eval(repr(cp))
        cp2.name = "not test"

        self.assertTrue(cp != cp2)

class TestExecutableCommandMixin(unittest.TestCase):

    def test_register_execution(self):
        """Test that an execution can be registered."""

        class ExecutableCommandMixinTester(ExecutableCommandMixin, enum.Enum):
            TESTPROPERTY = 1

        @ExecutableCommandMixinTester.TESTPROPERTY.register_execution
        def test_func():
            return None

        self.assertIs(ExecutableCommandMixinTester.TESTPROPERTY.execution, test_func)

    def test_register_execution_overwrite(self):
        """Test that an execution can be overwritten."""

        class ExecutableCommandMixinTester(ExecutableCommandMixin, enum.Enum):
            TESTPROPERTY = 1

        @ExecutableCommandMixinTester.TESTPROPERTY.register_execution
        def test_func():
            return None

        self.assertIs(ExecutableCommandMixinTester.TESTPROPERTY.execution, test_func)

        @ExecutableCommandMixinTester.TESTPROPERTY.register_execution
        def test_func_2():
            return "something bad"

        self.assertIs(ExecutableCommandMixinTester.TESTPROPERTY.execution, test_func_2)

    def test_register_parameter(self):
        """Test that a parameter can be registered."""

        class ExecutableCommandMixinTester(ExecutableCommandMixin, enum.Enum):
            TESTPROPERTY = 1

        @ExecutableCommandMixinTester.TESTPROPERTY.register_parameter(
            "test", 0, optional=True)
        def validate_func(param):
            return None

        cps = CommandParameterSet(
            CommandParameter("test", validate_func, optional=True))

        self.assertEquals(
            ExecutableCommandMixinTester.TESTPROPERTY.parameters, cps)

    def test_register_parameter_overwrite(self):
        """Test that a parameter can be overwritten."""

        class ExecutableCommandMixinTester(ExecutableCommandMixin, enum.Enum):
            TESTPROPERTY = 1

        @ExecutableCommandMixinTester.TESTPROPERTY.register_parameter(
            "test", 0, optional=True)
        def validate_func(param):
            return None
        @ExecutableCommandMixinTester.TESTPROPERTY.register_parameter(
            "test", 0, count=4, count_type=CountType.MAX)
        def validate_func_2(param):
            return "something good"

        cps = CommandParameterSet(
            CommandParameter(
                "test", validate_func_2, count=4, count_type=CountType.MAX))

        self.assertEquals(
            ExecutableCommandMixinTester.TESTPROPERTY.parameters, cps)

    def test_execute_command_invalid_args(self):
        """Test that there is an error if the parameters are invalid."""

        class ExecutableCommandMixinTester(ExecutableCommandMixin, enum.Enum):
            TESTPROPERTY = 1

        @ExecutableCommandMixinTester.TESTPROPERTY.register_parameter("test", 0)
        def validate_func(param):
            return "something bad"
        @ExecutableCommandMixinTester.TESTPROPERTY.register_execution
        def execute_func(value):
            return "something good"

        self.assertRaises(
            InvalidCommandParametersException, 
            ExecutableCommandMixinTester.TESTPROPERTY.execute_command)

    def test_execute_command_valid_args(self):
        """Test that there is an error if the parameters are invalid."""

        class ExecutableCommandMixinTester(ExecutableCommandMixin, enum.Enum):
            TESTPROPERTY = 1

        @ExecutableCommandMixinTester.TESTPROPERTY.register_parameter(
            "test", 0, optional=True)
        def validate_func(param):
            return None
        @ExecutableCommandMixinTester.TESTPROPERTY.register_execution
        def execute_func(value):
            return "something good"

        self.assertEquals(
            ExecutableCommandMixinTester.TESTPROPERTY.execute_command(None), 
            "something good")

If you need it, my folder structure looks like this (also available on GitHub)GitHub PyIRC Repository URL:

PyIRC/
    irc/
        __init.py
        executable_command.py
    __init__.py
test
    test_executable_command.py

executable_command.py is all of the code besides the sample (IrcCommand) and the tests, plus a handful of imports.

I haven't included the __init__.py files as they do about what you'd expect. Welcoming any and all reviews!

\$\endgroup\$
3
  • \$\begingroup\$ Something that I don't understand is why in the docstring you document parameters as Parameters ---------- name: string Name of the channel. Since Python has a convention for documenting parameters, and also return values \$\endgroup\$
    – A. Romeu
    Feb 27, 2018 at 17:21
  • \$\begingroup\$ @ARomeu I'm using the numpydoc package style for Sphinx documentation generation \$\endgroup\$ Feb 27, 2018 at 17:53
  • \$\begingroup\$ I imagined there was a good reason about it :) \$\endgroup\$
    – A. Romeu
    Feb 27, 2018 at 18:34

1 Answer 1

3
\$\begingroup\$

Might as well post what I did to improve this.

Don't use __class__ if you don't have to

I was already taking advantage of enum values being instances of the class, so I can just access things via self.attribute instead of self.__class__.attribute_dict[self].

def execute_command(self, *values):
    problems = self._validate_arguments(values)
    if problems:
        raise InvalidCommandParametersException(self, problems)

    return self.execution(*values)

_execution = None
@property
def execution(self):
    return self._execution

def register_execution(self, func):
    """Register a function as this command's action.

    Parameters
    ----------
    func: function
        Function to execute for this parameter.

    Returns
    -------
    func: function
        The original function, unchanged.
    """

    self._execution = func
    return func

_parameters = None
@property
def parameters(self):
    if self._parameters is None:
        self._parameters = CommandParameterSet()
    return self._parameters

def register_parameter(self, name, n_th, optional=False, count=1, count_type=None):
    """Register a function to validate a given parameter.

    Parameters
    ----------
    name: string
        Name of the parameter; used to describe the parameter.
    n_th: indexer
        Which parameter this should be in the parameter list.
    optional: boolean, default=False
        Whether or not the parameter is optional. If it is optional,
        then `None` should be passed for validation.
    count: integer, default=1
        Describes how many instances of this value are allowed. If 
        greater than 1, then `count_type` is required. If 0 or more
        are allowed, then pass `count=0` and
        `count_type=CountTypes.MIN`
    count_type: CountType, default=None
        Required if a `count` is given; describes how to interpret 
        the count (as a max, min, or exact requirement).

    Returns
    -------
    decorator: function -> function
        Wrapper function that wraps its argument function and adds
        it as validation for this parameter
    """

    def decorator(validator):
        """Add the actual validator function for this parameter.

        Parameters
        ----------
        validator: function(value) -> string | None
            Function that takes in a value and returns an error 
            message, or None if it was okay. If there are multiple 
            values allowed (i.e. `count_type != None`) then this is 
            called for each item in the collection, _instead_ of on 
            the entire collection.

        Returns
        -------
        validator: function(value) -> string | None
            The original function, unchanged.
        """

        self.parameters.insert_parameter(
            CommandParameter(name, validator, optional=optional, 
                count=count, count_type=count_type),
            n_th)

        return validator

    return decorator

def _validate_arguments(self, arguments):
    """Validate the command's arguments.

    Parameters
    ----------
    arguments: collection[Object]
        List of the arguments being passed.

    Returns
    -------
    list[string]
        List of all of the problems with the arguments. Will be an
        empty list if no problems are present.
    """

    errors = list(self.parameters.validate(arguments))

    if not errors or all(err is None for err in errors):
        return []
    return errors

Use iterators better

In ExecutableCommandMixin._validate_arguments I consume a generator by turning it into a list and then filtering it. Instead, it should look like this:

return [error for error in self.parameters.validate(arguments) 
                if error is not None

Whole object validation

Sometimes you might have dependencies between parameters. For example, the IRC JOIN command has the following syntax:

JOIN <channel>{,<channel>} [<key>{,<key>}]

In this case, if you do provide any keys then there definitely shouldn't be more than there are channels, and maybe should even be exactly as many (I'm not sure yet if you can do something like chan1,chan2,chan3 key1,,key3 or chan1,chan2,chan3 key1. Add a way to validate the whole set of parameters.

Handle errors better

It's imaginable that a given executable command might end up with specific errors we know can happen, and we might want to handle them a specific way. We can add this into the framework as well. I think we would generally want one of two things: raise a specific exception from the error, or compute something and return it. The below functionality enables decorating a function or exception class to take a command and an error as the error handler, which you would then use as MyCommandEnum.MEMBER.handle_error(SOME_ERROR_CODE).

class NoHandlerExcepetion(Exception):
    """
    Exception to indicate that there is no handler for a given error.
    """

    def __init__(self, command, error):
        super(NoHandlerExcepetion, self).__init__(
            "Command {} - Error {}".format(command, error))

class ExecutableCommandMixin(object):

    # All of the other code

    _error_handlers = None
    @property
    def error_handlers(self):
        """Error handler for this command."""

        if self._error_handlers is None:
            self._error_handlers = {}
        return self._error_handlers

    def handle_error(self, error):
        """Handle the error for this command.

        Parameters
        ----------
        error: hashable
            The error to handle.
        """

        handler = self.error_handlers.get(error, NoHandlerExcepetion)

        result = handler(self, error)

        try:
            raise result
        except TypeError:
            return result

    def register_error_handler(self, error):
        """Register an error handler for a given command/error pair.

        Parameters
        ----------
        error: hashable
            The error to handle.

        Returns
        -------
        decorator: function
            Decorator function that registers error handlers for a 
            command.
        """

        def decorator(f):
            """Do the actual registration."""

            if not callable(f):
                raise TypeError(
                    "Error handler must be a function or Exception type")

            self.error_handlers[error] = f
            return f

        return decorator

And of course, you should write tests for that

def test_register_error_handler_callable(self):
    """Test that the handler can be a function."""

    class ExecutableCommandMixinTester(ExecutableCommandMixin, enum.Enum):
        TESTPROPERTY = 1

    @ExecutableCommandMixinTester.TESTPROPERTY.register_error_handler(
        "I AM A COOL ERROR CODE")
    def handle_error(command, error):
        return command, error

    # No error, yay!

def test_register_error_handler_exception(self):
    """Test that the handler can be an exception."""

    class ExecutableCommandMixinTester(ExecutableCommandMixin, enum.Enum):
        TESTPROPERTY = 1

    @ExecutableCommandMixinTester.TESTPROPERTY.register_error_handler(
        "I AM A COOL ERROR CODE")
    class CustomException(Exception): pass

    # No error, yay!

def test_register_error_handler_invalid(self):
    """Test that the handler can't be non-callable."""

    class ExecutableCommandMixinTester(ExecutableCommandMixin, enum.Enum):
        TESTPROPERTY = 1

    dec = ExecutableCommandMixinTester.TESTPROPERTY.register_error_handler(
        "I AM A COOL ERROR CODE")

    self.assertRaises(TypeError, dec, "I AM NOT CALLABLE")

def test_handle_error_function(self):
    """Test that an error handler function is called."""

    class ExecutableCommandMixinTester(ExecutableCommandMixin, enum.Enum):
        TESTPROPERTY = 1

    @ExecutableCommandMixinTester.TESTPROPERTY.register_error_handler(
        "I AM A COOL ERROR CODE")
    def handle_error(command, error):
        return command, error

    self.assertEquals(
        handle_error(ExecutableCommandMixinTester.TESTPROPERTY, 
                     "I AM A COOL ERROR CODE"),
        ExecutableCommandMixinTester.TESTPROPERTY.handle_error(
            "I AM A COOL ERROR CODE"))

def test_handle_error_exception(self):
    """Test that an error handler exception is raised."""

    class ExecutableCommandMixinTester(ExecutableCommandMixin, enum.Enum):
        TESTPROPERTY = 1

    @ExecutableCommandMixinTester.TESTPROPERTY.register_error_handler(
        "I AM A COOL ERROR CODE")
    class CustomException(Exception): pass

    self.assertRaises(
        CustomException, 
        ExecutableCommandMixinTester.TESTPROPERTY.handle_error,
        "I AM A COOL ERROR CODE")

def test_handle_error_none(self):
    """Test that the handler must be present."""

    class ExecutableCommandMixinTester(ExecutableCommandMixin, enum.Enum):
        TESTPROPERTY = 1

    self.assertRaises(
        NoHandlerExcepetion,
        ExecutableCommandMixinTester.TESTPROPERTY.handle_error,
        "I AM ANY UNREGISTERED ERROR CODE")

Actually use exceptions

I actually haven't made this change yet, but every time I come back to seeing the validation functions returning strings instead of raising exceptions it feels really weird. I think I'll change this eventually, probably like so:

  1. Individual validation functions (the ones that get the CommandEnum.Member.register_parameter decorator) should return a tuple of form (is_valid, problem_description), however, if these functions are allowed to raise instead (if they want) to signal some/all erroneous conditions. This let's me write those functions flexibly, in case they should be used somewhere else where the paradigm of (is_valid, problem_description) doesn't make sense.
  2. CommandParameter should raise an exception (probably InvalidCommandParameterException) where the problem description is the error message. If the validation function raises an exception, then re-raise as an InvalidCommandParameterException.
  3. CommandParameterSet should catch all of these exceptions, save their data, and then raise a single aggregate exception with all problems in it. This avoids losing any data.
  4. ExecutableCommandMixin should let the CommandParameterSet handle exceptions, and let any problems percolate up to the consumer.

This provides a more familiar and Pythonic interface from all of the special executable command infrastructure, while still allowing you to write functions that don't raise if you don't want them to.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.