Python etc

There is no ++ operator in Python, x += 1 is used instead. However, even ++x is still a valid syntax (but x++ is not). The catch is Python has unary plus operator, and ++x is actually x.__pos__().__pos__(). We can abuse this fact and make ++ work as increment (not recommended though):

class Number:
    def __init__(self, value):
        self._value = value

    def __pos__(self):
        return self._Incrementer(self)

    def inc(self):
        self._value += 1

    def __str__(self):
        return str(self._value)

    class _Incrementer:
        def __init__(self, number):
            self._number = number

        def __pos__(self):
            self._number.inc()


x = Number(4)
print(x)  # 4
++x
print(x)  # 5

Concise arguments Thesis: arguments have to cater to the function’s needs, not another way around. Motivation If a function’s arguments are conveniently structured, do not contain any unused data, and are organized straightforwardly, then it greatly simplifies testing. By a convenient structure I mean the data is presented on the top level, without the need to dig for what is actually used to produce the output. In tests, this lets us fake less and setup less because now it is not necessary to create more data than the function actually uses. Otherwise, we most likely would need to provide comparably complex data structures as test input. The bottom line is that it is helpful to separate data lookup from the business logic, instead of managing them as a single entity. As a beneficial side effect, when you have it extracted, you can immediately see how much work this data lookup takes. If there is a lot, and it includes complex calculations, then your object graph and data structures may be unnecessarily complicated or poorly designed. However, now you can informedly gauge whether the current implementation is justified. Approach The params need to contain only the relevant data and have a simple structure that supports allows to access it easily. A refactoring to achieve this is very affordable and does not require any significant changes, because it only affects the function itself and its direct clients. To perform it, you extract the logic which searches the relevant data in the arguments from the function and then replace it with the already retrieved data. It has to be structured as simple as possible, being directly assessable without unnecessary nesting and do not contain anything that function does not use. The required data may differ for the same function, but optimally, you aim to achieve 100% usage of the input data by tailoring it for each call. The extracted logic now is handled by the clients. Even though this is already enough to reap the benefits of this pattern, it may be useful to reconsider the initially chosen data structures of the arguments. If they aren’t optimal for the refactored function, they might be a not fit that well in the other places as well. Example Before refactoring:

# Method that does some work
# The interesting data is a part of "param", but it's not directly accessible
def do_work(param):
    # Navigating object graph to get to the array
    data_array = param.related_object.some_dict['data_array']

    # This array contains more data, we remove it
    relevant_array = [x for x in data_array if not useless(x)]

    # Only now we can start processing business logic
    result = [process(x) for x in relevant_array]

    return result
 
# In tests
def test():
    # Since the "param" is some complex structure,
    # that is probably used for something else,
    # we need to do a lot of effort to mock it
    
    # Produces more data, which is mostly "useless()", hence it's slower
    data_array = Factory.get_generic_data_array()  
    fake_param = MagicMock()
    fake_param.related_object.some_dict['data_array'] = data_array

    # Only now we can finally execute it
    assert do_work(fake_param) == Factory.get_do_work_result()

After refactoring:

# Now the clients use this method to prepare data for do_work()
# But they also can improve their data structures
# to get rid if this lookup for good.
def extract_relevant_array(param):
    data_array = param.related_object.some_hash['data_array']
    relevant_array = [x for x in data_array if not useless(x)

def do_work(relevant_array):
    return [process(x) for x in relevant_array]
 
# In tests
def test():
    # Setup is incredibly less simpler
    # Now input contains only objects which return false on "useless()"  
    relevant_array = Factory.get_tailored_data_array()
    
    assert do_work(relevant_array) == Factory.get_do_work_result()

This post is the seventh installment of the series of 16 software design patterns dedicated to improving code testability. Published every Tuesday. Provided by @NikolayRys.

An object instantiation includes two significant steps. First, the __new__ method of a class is called. It creates and returns a brand new object. Second, Python calls the __init__ method of that object. Its work is to set up the initial state of the object. However, __init__ isn't called if __new__ returns an object that is not an instance of the original class. The reason for this is that it was probably created by another class, hence __init__ was already called for that object:

class Foo:
    def __new__(cls, x):
        return dict(x=x)

    def __init__(self, x):
        print(x)  # Never called

print(Foo(0))

That also means that you should not ever create instances of the same class in __new__ with a regular constructor (Foo(...)). It could lead to the double __init__ execution or even infinite recursion. Infinite recursion:

class Foo:
    def __new__(cls, x):
        return Foo(-x)  # Recursion

Double __init__:

class Foo:
    def __new__(cls, x):
        if x < 0:
            return Foo(-x)
        return super().__new__(cls)

    def __init__(self, x):
        print(x)
        self._x = x

The proper way:

class Foo:
    def __new__(cls, x):
        if x < 0:
            return cls.__new__(cls, -x)
        return super().__new__(cls)

    def __init__(self, x):
        print(x)
        self._x = x

Since Python 3.7, contextlib provides the asynccontextmanager decorator which allow you to define asynchronous context manager in the exact same manner as contextmanager does:

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def slow(delay):
    half = delay / 2
    await asyncio.sleep(half)
    yield
    await asyncio.sleep(half)

async def main():
    async with slow(1):
        print('slow')

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

For older versions, you could use @asyncio_extras.async_contextmanager.

If you want a context manager to suspend coroutine on entering or exiting context, you should use asynchronous context managers. Instead of calling m.__enter__() and m.__exit__() Python does await m.__aenter__() and await m.__aexit__() respectively. Asynchronous context managers should be used with async with syntax:

import asyncio

class Slow:
    def __init__(self, delay):
        self._delay = delay

    async def __aenter__(self):
        await asyncio.sleep(self._delay / 2)

    async def __aexit__(self, *exception):
        await asyncio.sleep(self._delay / 2)

async def main():
    async with Slow(1):
        print('slow')

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Example

# We refactor the do_job method
class Worker:
    def do_job(self, params):
        # Original implementation
        return calculated_value

    def do_job_new(self, params):
        # Some different implementation
        return calculated_value

In tests:

# Tests for the old method
def test_do_job():
    # We need to construct output ourselves, which may be hard
    for input, output in input_output_10_pairs:
        worker = Worker()
        assert worker.do_job(input) == output

# Tests for the new method
def test_do_job_new():
    # Now we need just input data, so we can prepare more of it
    for input in input_100_pieces:
        worker = Worker()
        assert worker.do_job_new(input) == worker.do_job(input)

When we are happy with the result, we can remove the old method and let the new one to take its place.

Approach If we have an already implemented function that we intend to change, then instead of modifying it in place, create a copy of it with a different name, and change it as required, while keeping the existing method intact until the new one is ready to take its place. We use the old function to help produce output test data from the input values, against which we validate the changed one. For this technique, both functions must retain a certain similarity in their observable behavior, so it’s going to be less applicable if the differences between them are significant. That is so because you need to write additional code to convert the old data sets into the new ones, so the refactoring is going to be the most favorable case. Besides that, you need to be able to execute both functions sustainably in the same script, which requires them to be isolatable and adhere to the patterns presented here. To maximize the efficiency of the approach, you always need to look out for opportunities to increase the share of refactoring in development. For example, if instead of a single code change, you can refactor it first to reduce the required change, then you get support with the generation of test data during the whole refactoring step. When the development and validation of the new function is done, you need to record the output of the old method before removing it. Most likely you can just the serialize (marshal) the produced data, for example, using a tool like Google Protocol Buffers, but this depends on your language and available libraries.

Hot spare Motivation The name of this pattern refers to the concept of a “hot spare" in civil engineering. It implies the simultaneous existence of two similar operational components in a single system. It may be for the sake of reliability or as a temporary measure to facilitate the transition from one component to another. The latter is a common situation in the software development, because every code change may be viewed as a swap of a function with a different one. Thus, we can use this idea by keeping the existing function along with the modified version to validate the latter more easily. This approach works tremendously well with refactoring, but it stays applicable as long as the new behavior retains at least some similarity with the original. If it has changed, then some additional code may be necessary to overcome the difference. However, it almost always pays off by greatly simplifying creation of test examples, because for every set of input data you now have its output counterpart virtually free of charge. You may be asking, why do we need to create new test cases if the coverage for a function already exists and we can simply reuse it. This is so because a different implementation is likely to require a different set of examples to satisfactorily validated its correctness. Consider this: if the same functionality may be implemented with either 1 or 100 lines of code, these versions are expected to have very different demand for testing. Of course, it is higher for the latter, but it needs to be provided nevertheless. The same stays true if there are any differences in the execution paths - it might affect the equivalence classes, and therefore rearrange the boundary values that you need to verify. As a general rule, the adequacy of the present test cases is always jeopardized by the refactoring. So even though it’s possible to reuse the set-up, exercise and clean-up steps of the existing tests, you still need to be prepared to come up with additional cases for the new implementation. This approach synergizes well with the concepts presented here in the Pure Functions chapter. If both the new and the old methods are wrapped together with their sandbox environments that resemble pure functions, then the data produced by the old wrapper can be easily fed to the new one, in order to check the output for the same input data. It also creates an opportunity to generate test examples automatically, which I will elaborate in my next article, along with a tool for accomplishing it in the context of xUnit frameworks, stay tuned.

This post is the sixth installment of the series of 16 software design patterns dedicated to improving code testability. Published every Tuesday.

asyncio.shield allows you to protect an awaitable object from being cancelled. However, the caller (code that does await) is still cancelled as though there is no shield.

import asyncio
from asyncio import shield

async def main():
    print('BEGIN')
    loop = asyncio.get_event_loop()
    task = loop.create_task(job())
    await asyncio.sleep(1)
    task.cancel()
    await asyncio.sleep(10)
    print('END')

async def job():
    print('start job')
    await shield(action())  # cancelled here
    print('stop job')  # never executed


async def action():
    await asyncio.sleep(2)
    print('action')  # executed nevertheless

asyncio.get_event_loop().run_until_complete(main())

Mind, that it's not the same as suppressing CancelledError, it's quite the opposite: catching the error make the caller continue working, but the callee is still cancelled.

import asyncio
from asyncio import shield
from contextlib import suppress

async def main():
    print('BEGIN')
    loop = asyncio.get_event_loop()
    task = loop.create_task(job())
    await asyncio.sleep(1)
    task.cancel()
    await asyncio.sleep(5)
    await task  # no exception here
    print('END')

async def job():
    print('start job')
    with suppress(asyncio.CancelledError):
        await action()
    print('stop job')  # still executed


async def action():
    await asyncio.sleep(2)  # cancelled here
    print('action')  # never executed

asyncio.get_event_loop().run_until_complete(main())

Ignoring cancellation is usually discouraged though.

To create a class method, you should use the @classmethod decorator. This method can be called from the class directly, not from its instances, and accepts the class as a first argument (usually called cls, not self). However, there are two implicit class methods in Python data model: __new__ and __init_subclass__. They work exactly as though they are decorated with @classmethod except they aren't. (__new__ creates new instances of a class, __init_subclass__ is a hook that is called when a derived class is created.)

class Foo:
    def __new__(cls, *args, **kwargs):
        print(cls)
        return super().__new__(
            cls, *args, **kwargs
        )

Foo()  # <class '__main__.Foo'>

Instead of the classic ternary operator (?:) Python has the conditional expression: if_true if condition else if_false. Mind, that it has slightly unexpected order of operands: if_true, condition, if_false instead of classic condition, if_true, if_false. You should also be aware that this is not any kind of statement expression, the if and else keywords are merely reused. Only expressions can be used as operands, so you have no way to do something like print(x) if x else break.

Every method can be treated as a plain function and called with a custom self:

In : class A:
   ...:     def foo(self):
   ...:         return self
   ...:

In : A().foo
Out: <bound method A.foo of <...>>

In : A.foo
Out: <function __main__.A.foo>

In : A.foo(A())
Out: <__main__.A at 0x7f55ddd32898>

You can even convert a function back to the bound method. Any function is a descriptor, so it can be abused by calling __get__:

In [8]: b = A()

In [9]: A.foo.__get__(b, A)
Out[9]: <bound method A.foo of <...>>

Since Python 3.5, it's actually possible to use unpacking with dictionary and list literals.

In : {**{'a': 1}, 'b': 2, **{'c': 3}}
Out: {'a': 1, 'b': 2, 'c': 3}

In : [1, 2, *[3, 4]]
Out: [1, 2, 3, 4]

For dictionaries, this form is even more powerful than the dict function, since it allows values to be overridden:

In : {**{'a': 1, 'b': 1}, 'a': 2, **{'b': 3}}
Out: {'a': 2, 'b': 3}

The holiday season has started in Russia. See you next week :).

You can define dictionaries in two ways, using literals or the dict function:

>>> dict(a=1, b=2)
{'a': 1, 'b': 2}
>>> {'a': 1, 'b': 2}
{'a': 1, 'b': 2}

Literals work faster than dict, but the function has some advantages. First, you don't need to add additional quotes. However, it only works as long as all keys are valid Python identifiers.

>>> dict(a=1)
{'a': 1}
>>> dict(1='a')
  File "<stdin>", line 1
SyntaxError: keyword can't be an expression

Second, you can't accidentally provide the same key twice:

>>> {'a': 1, 'a': 1}
{'a': 1}
>>> dict(a=1, a=1)
  File "<stdin>", line 1
SyntaxError: keyword argument repeated

Third, you can easily create new the new dictionary based on some already existed one.

>>> d = dict(b=2)
>>> dict(a=1, **d)
{'a': 1, 'b': 2}

Mind, however, that keys can't be redefined with this syntax:

>>> dict(b=3, **d)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: type object got multiple values for keyword argument 'b'

If you want some code running with some global variable modified, you better use context manager instead of directly changing it:

from contextlib import contextmanager

QUIT_MESSAGE = 'Bye'

def print_quit_mesage():
    global QUIT_MESSAGE
    print(QUIT_MESSAGE)

@contextmanager
def global_variable_changed(name, value):
    orig_value = globals()[name]
    globals()[name] = value
    yield
    globals()[name] = orig_value

with global_variable_changed(
    'QUIT_MESSAGE',
    'Tschüss'
):
    print_quit_mesage()

If you want objects of a class to have an auto-incremented ID, you can make it happen by tracking current ID in the class attribute:

class Task:
    _task_id = 0

    def __init__(self):
        self._id = self._task_id
        type(self)._task_id += 1

Mind, that you can't do self._task_id += 1. That creates the _task_id attribute within the instance, not the class. You should consider using a factory method instead of __init__ to make it look prettier:

class Task:
    _task_id = 0

    def __init__(self, task_id):
        self._id = task_id

    @classmethod
    def create(cls):
        obj = cls(cls._task_id)
        cls._task_id += 1
        return obj

This version is also easier to test since any custom ID can be easily provided.

After refactoring:

@dataclass
class WorkDoerData:
    a: Any
    b: Any
    c: Any
    d: Any
    # It can also host private methods from WorkDoer

class  WorkDoer:
    # We may make it public, but we don't have to, it's a matter of convenience.
    def position(data: WorkDoerData)
        self._data = data
  
    def locate(self)
        return self._data.copy()
  
    # If we don't give the public access to self._data, we need accept through the constructor.
    # We have done it here, but for the sake of illustration let's assume that we haven't.
    def __init__(self, data):
        self._data = data
  
    # Some methods that affect the state of the object
    def prepare_c(self):
        self._data.c = self.prepare_c_from_ab(self._data.a, self._data.b)

    def prepare_d(self):
        self._data.d = self.prepare_d_from_ab(self._data.a, self._data.b)

    def prepare_ab(self):
        self._data.a = self.make_new_a_from_c(self._data.c)
        self._data.b = self.make_new_b_from_d(self._data.d)

    def do_job(self):
        # The same method that we intend to test, which now uses self._data.xxx object

# Test is much simpler now
def test():
    # data is already in the correct state
    data = Factory.create_correct_data()
    work_doer = WorkDoer.new(data)

    # Exercising it immidiately after creation in the right state
    assert work_doer.do_job() == correct_result
    # Data object should have been affected, so we can validate it as well
    # We already have a reference to it, so we don't need to dig in the private attributes  
    assert data == changed_data