Python etc

Sometimes you want to check the syntax of a py-file without running it. Such naive check may be useful as a commit-hook or a fast continuous integration check. There is no direct way to do this. You can run the file as python -m module.py, that prevents the traditional if __name__ == '__main__' block from running. Still, all imports will be executed, and this may fail if you want to check syntax in the environment where the module can't be and shouldn't be run. However, the python standard library contains the py_compile module that generates byte-code from Python source file without running it. That's exactly what we need:

$ python -m py_compile test.c
  File "test.c", line 1
    int main() {
           ^
SyntaxError: invalid syntax

.

Sometimes you want to clear a collection in Python. You probably something like d = {} (for dictionaries) but it's not exactly clearing, it's creating a new collection and throwing the old one away. It may work for you, but other owners of the same object will still have a reference to the original one. The proper way to clear dictionary, set, deque and other collections is to call x.clear().

A slightly more than half a year left until Python 2 is no longer maintained. You should consider moving your projects to Python 3 until then. There are several things that you have to change to be Python 3 compatible, but the one I consider the more severe yet barely detectable is changing how the / operator works for integers. 1/2 is no more 0, it's 0.5. You should use // instead to save the old behavior.

Python set supports comparison operators, a < b means a is a subset of b:

>>> {1} < {1, 2}
True
>>> {1} < {2, 3}
False

That means that sets are partially ordered, which means there are such a and b that both a < b and b < a are false:

>>> {1} < {2, 3}
False
>>> {1} > {2, 3}
False

Some functions like min, max and sorted rely on total ordering, so you can get confusing results applying them to a list of sets:

>>> min([{1}, {2}])
{1}
>>> min([{2}, {1}])
{2}

Since Python 3.0, raising an exception in an except block will automatically add the caught exception in the __context__ attribute of the new one. That will cause both exceptions to be printed:

try:
    1 / 0
except ZeroDivisionError:
    raise ValueError('Zero!')

Traceback (most recent call last):
  File "test.py", line 2, in <module>
    1 / 0
ZeroDivisionError: division by zero

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "test.py", line 4, in <module>
    raise ValueError('Zero!')
ValueError: Zero!

You also can add __cause__ to any exception with the raise ... from expression:

division_error = None

try:
    1 / 0
except ZeroDivisionError as e:
    division_error = e

raise ValueError('Zero!') from division_error

Traceback (most recent call last):
  File "test.py", line 4, in <module>
    1 / 0
ZeroDivisionError: division by zero

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "test.py", line 8, in <module>
    raise ValueError('Zero!') from division_error
ValueError: Zero!

Letting direct access to an object attributes may be not the best idea. If clients communicate with the object via methods, you can always modify how every request is processed while with direct attribute access it may be not possible. Different languages deal with that problem in different ways. In Ruby, it's syntactically impossible to access an attribute directly, obj.x is a call of the x method. In Java, it's recommended to make all attributes private and write trivial getters instead: public int getX() { return this.x }. Python offers a solution that is somehow similar to that that Ruby has. You can define property so obj.x invokes a method instead of returning the x attribute directly.

class Example:
    def __init__(self, x):
        self._x = x

    @property
    def x(self):
        return self._x

Consider the following class hierarchy:

class GrandParent:
    pass

class Parent1(GrandParent):
    pass

class Parent2(GrandParent):
    pass

class Child(Parent1, Parent2):
    pass

Which order will be used to look up the Child.x() method? The naive approach is to recursively search through all parent classes which gives us Child, Parent1, GrandParent, Parent2. While many programming languages follow this method indeed, it doesn't quite make sense, because Parent2 is more specific than GrandParent and should be looked up first. In order to fix that problem, Python uses C3 superclass linearization, the algorithm that always searches for a method in all child classes before looking up the parent one:

In : Child.__mro__
Out:
(__main__.Child,
 __main__.Parent1,
 __main__.Parent2,
 __main__.GrandParent,
 object)

Both for and with can be asynchronous. async with uses __aenter__ and __aexit__ magic methods, async for uses __aiter__ and __anext__. All of them are async and you can await within them:

import asyncio

class Sleep:
    def __init__(self, t):
        self._t = t

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

    async def __aexit__(self, *args):
        await asyncio.sleep(self._t / 2)

async def main():
    async with Sleep(2):
        print('*')

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

When you implement __iter__ you often don't write an iterator with __next__ method, you just use yield that makes __iter__ a generator:

class Bracketed:                  
    def __init__(self, data):     
        self._data = data         
                                  
    def __iter__(self):           
        for x in self._data:      
            yield '({})'.format(x)
                                  
print(list(Bracketed([1, 2, 3]))) 
# ['(1)', '(2)', '(3)']

PEP 525 allows you do the same with __aiter__. Both yield and await in the function body make it asynchronous generator. While await is used to communicate with the loop, yield deals with for:

import asyncio                          
                                        
class Slow:                             
    def __init__(self, data, t=1):      
        self._data = data               
        self._t = t                     
                                        
    async def __aiter__(self):          
        for x in self._data:            
            await asyncio.sleep(self._t)
            yield x                     
                                        
async def main():                       
    async for x in Slow([1, 2, 3]):     
        print(x)                        
                                        
loop = asyncio.get_event_loop()         
loop.run_until_complete(main())

If you want to pass some information down the call chain, you usually use the most straightforward way possible: you pass it as functions arguments. However, in some cases, it may be highly inconvenient to modify all functions in the chain to propagate some new piece of data. Instead, you may want to set up some kind of context to be used by all functions down the chain. How can this context be technically done? The simplest solution is a global variable. In Python, use also may use modules and classes as context holders since they, strictly speaking, are global variables too. You probably do it on a daily basis for things like loggers. If your application is multi-threaded, a bare global variable won't work for you since they are not thread-safe. You may have more than one call chain running at the same time, and each of them needs its own context. The threading module gets you covered, it provides the threading.local() object that is thread-safe. Store there any data by simply accessing attributes: threading.local().symbol = '@'. Still, both of that approaches are concurrency-unsafe meaning they won't work for coroutine call-chain where functions are not only called but can be awaited too. Once a coroutine does await, an event loop may run a completely different coroutine from a completely different chain. That won't work:

import asyncio
import sys

global_symbol = '.'

async def indication(timeout):
    while True:
        print(global_symbol, end='')
        sys.stdout.flush()
        await asyncio.sleep(timeout)

async def sleep(t, indication_t, symbol='.'):
    loop = asyncio.get_event_loop()

    global global_symbol
    global_symbol = symbol
    loop.create_task(indication(indication_t))
    await asyncio.sleep(t)

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(
    sleep(1, 0.1, '0'),
    sleep(1, 0.1, 'a'),
    sleep(1, 0.1, 'b'),
    sleep(1, 0.1, 'c'),
))

You can fix that by having the loop set and restore the context every time it resumes some coroutine. The aiotask_context module does exactly this by changing the way how tasks are created with loop.set_task_factory. This works:

import asyncio                                
import sys                                    
import aiotask_context as context             
                                              
async def indication(timeout):                
    while True:                               
        print(context.get('symbol'), end='')  
        sys.stdout.flush()                    
        await asyncio.sleep(timeout)          
                                              
async def sleep(t, indication_t, symbol='.'): 
    loop = asyncio.get_event_loop()           
                                              
    context.set(key='symbol', value=symbol)   
    loop.create_task(indication(indication_t))
    await asyncio.sleep(t)                    
                                              
loop = asyncio.get_event_loop()               
loop.set_task_factory(context.task_factory)   
loop.run_until_complete(asyncio.gather(       
    sleep(1, 0.1, '0'),                       
    sleep(1, 0.1, 'a'),                       
    sleep(1, 0.1, 'b'),                       
    sleep(1, 0.1, 'c'),                       
))

Once you write a coroutine, you need three things to get it running: create it, obtain a loop and schedule coroutine's execution. The first two steps are trivial:

import asyncio
import sys

async def indication(timeout):
    while True:
        print('.', end='')
        sys.stdout.flush()
        await asyncio.sleep(timeout)

async def sleep(t, indication_t):    # steps
    coro = indication(indication_t)  # 1
    loop = asyncio.get_event_loop()  # 2

    # ...
    await asyncio.sleep(t)

loop = asyncio.get_event_loop()
loop.run_until_complete(sleep(5, 0.5))

The slightly trickier part is to schedule execution of a coroutine within an already running loop. There are two different ways to achieve that: loop.create_task(coro) and asyncio.ensure_future(coro, loop=loop). Though they both work, they have pretty different semantic. create_task does precisely what we need: it schedules execution of a coroutine and returns a future that allows you to track the coroutine execution (likely by awaiting it). That future is strictly speaking a task (asyncio.Task), the particular type of futures that wrap a coroutine. ensure_future(x) just ensures that x is a future or wrap it in one. If x is a coroutine, is uses create_task for such wrapping, obviously scheduling x. The previous name for ensure_future is async; it was changed to respect the fact that async is a keyword now.

import asyncio
import sys

async def indication(timeout):
    while True:
        print('.', end='')
        sys.stdout.flush()
        await asyncio.sleep(timeout)

async def sleep(t, indication_t):
    coro = indication(indication_t)
    loop = asyncio.get_event_loop()

    # # Choose one:
    # loop.create_task(coro)
    # asyncio.ensure_future(coro)

    await asyncio.sleep(t)

loop = asyncio.get_event_loop()
loop.run_until_complete(sleep(5, 0.5))

Guido explicitly recommends using create_task() since it's intent is more clear.

In asyncio, the main thread loop is automatically created for you once you call asyncio.get_event_loop(). That doesn't happen in any other thread:

In : asyncio.get_event_loop()
Out: <_UnixSelectorEventLoop running=False closed=False debug=False>

In : Thread(target=asyncio.get_event_loop).start()
In : Exception in thread Thread-385:
...
RuntimeError: There is no current event loop in thread 'Thread-385'.

The get_event_loop() method returns the loop bound to the current thread. You can use set_event_loop(loop) to bind the loop to the current thread after creating it with loop = asyncio.new_event_loop(). You can run any loop, even if another one is bound to the thread. That's why in Python 3.6 get_event_loop() works differently within a coroutine. It returns not the loop bound to the thread, but the currently running loop. It's important when a coroutine tries to interact with its loop:

import asyncio
import sys

async def sleep(t, indication_t):
    async def indication():
        while True:
            print('.', end='')
            sys.stdout.flush()
            await asyncio.sleep(indication_t)

    loop = asyncio.get_event_loop()  # <-- here
    task = loop.create_task(indication())
    await asyncio.sleep(t)
    task.cancel()

loop = asyncio.get_event_loop()
loop.run_until_complete(sleep(5, 0.5))

Even though you can run any loop you can never run two loops in the single thread. Though it's technically possible, it's explicitly forbidden by asyncio:

In : async def run_another():
...:     loop = asyncio.new_event_loop()
...:     loop.run_forever()
In : loop = asyncio.get_event_loop()
In : loop.run_until_complete(run_another())
RuntimeError: Cannot run the event loop while another loop is running

Objects with the __await__ method defined are called Future-like. __await__ is meant to yield a value to the loop (asyncio.Future does exactly this). However, you may want to await inside __await__. The problem is, __await__ is not async, so await is a syntax error:

class Future:
    def __await__(self):
        await asyncio.sleep(1)

#     await asyncio.sleep(1)
#                 ^
# SyntaxError: invalid syntax

You can use yield from instead:

class Future:
    def __await__(self):
        yield from asyncio.sleep(1)

Another problem is, you can only yield from native coroutine in another coroutine (whether native or generator-based). __await__ is neither though:

async def sleep_one_sec():
    await asyncio.sleep(1)

class Future:
    def __await__(self):
        yield from sleep_one_sec()

loop.run_until_complete(Future())

#      1 class Future:
#      2     def __await__(self):
#----> 3         yield from sleep_one_sec()
#      4
# TypeError: cannot 'yield from' a coroutine object
# in a non-coroutine generator

One of the solutions would be to call __await__ manually:

class Future:
    def __await__(self):
        yield from sleep_one_sec().__await__()

Another one is to use generator-based coroutine as an adapter:

@asyncio.coroutine
def adapter(coroutine):
    yield from coroutine

class Future:
    def __await__(self):
        yield from adapter(sleep_one_sec())

Once an asyncio coroutine wants to stop and communicate with the event loop, it uses await obj (or yield from obj before Python 3.6). An obj should be another coroutine, asyncio.Future or any custom Future-like object (any object with the __await__ method defined).

async def coroutine():
    await another_coroutine()

async def another_coroutine():
    future = asyncio.Future()
    await future

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

Once the coroutine awaits another one, the second starts to run instead of the first. If it awaits the third one, the third one runs. It goes on and on until some coroutine awaits a future. The future actually yields the value, so the loop finally gains control. What value does the future yield? It yields itself. Can you yield a future directly? No, it's an internal detail you shouldn't normally worry about.

class Awaitable:
    def __await__(self):
        future = asyncio.Future()
        yield future
            # RuntimeError: yield was used
            # instead of yield from in task

async def coroutine():
    await Awaitable()

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

Why does this error occur? How does asyncio know it's you who yield the future, not the future itself? There is a simple protection: the future raises the internal flag before yielding.

The upcoming week will be fully dedicated to one topic. Vote 🐧 for it to be ctypes, or vote 🐴 for asyncio.

Tail recursion is a special case of recursion where the recursive call is the last expression in the function:

def fact(x, result=1):
    if x == 0:
        return result
    else:
        return fact(x - 1, result * x)

The cool thing about it is you don't have to return to the caller once callee returns the result since the caller has nothing more to do. That means that you don't have to save the stack frame of the caller. That technique is called TRE, tail recursion elimination. And Python doesn't support it. It was considered and declined by Guido, mostly because removing stack frames makes stack trace looks cryptic.

Sometimes you want to run a code block with multiple context managers:

with open('f') as f:
    with open('g') as g:
        with open('h') as h:
            pass

Since Python 2.7 and 3.1, you can do it with a single with expression:

o = open
with o('f') as f, o('g') as g, o('h') as h:
    pass

Before that, you could you use the contextlib.nested function:

with nested(o('f'), o('g'), o('h')) as (f, g, h):
    pass

It still can be used while working with an unknown number of context managers (nested(*managers)), but it throws the warning in the modern Python interpreter. Instead, the more advanced tool is provided. contextlib.ExitStack allows you to enter any number of contexts at the arbitrary time but guarantees to exit them at the end:

with ExitStack() as stack:
    f = stack.enter_context(o('f'))
    g = stack.enter_context(o('g'))
    other = [
        stack.enter_context(o(filename))
        for filename in filenames
    ]

While using asyncio, you rarely need to mess with futures directly, they are usually concealed and are dealt by the loop itself. However, custom futures might be a mighty tool. This example demonstrates how a coroutine can be stopped for batch processing. The sqr coroutine creates and awaits future that can be processed not by the loop, but by the custom Queue, that sends HTTP requests once every second, not immediately on demand.

UTF-8 is a variable-width encoding. One character can be encoded by one, two, three or four bytes. That means that you can't start reading a utf8-encoded string from any bite; that can accidentally break a character:

In : lion = 'Löwe'
In : lion.encode('utf-8')[2:]
Out: b'\xb6we'
In : lion.encode('utf-8')[2:].decode('utf-8')
...
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xb6 in position 0: invalid start byte

This also means that to skip the first N characters of a string you can read and decode them, calculating offset upfront is not possible. You can, however, skip some fixed number of bytes with some precautions. Let's look how a symbol can be decoded:

0xxxxxxx
110xxxxx 10xxxxxx
1110xxxx 10xxxxxx 10xxxxxx
11110xxx 10xxxxxx 10xxxxxx 10xxxxxx

As you can see, the byte is the leading byte of a character unless it's 10xxxxxx. Such non-leading bytes are called continuation bytes. Let's skip them:

def cut_bytes(s, n):
    result = s.encode('utf-8')[n:]
    mask = int('11000000', 2)
    conbyte = int('10000000', 2)
    while result[0] and result[0] & mask == conbyte:
        result = result[1:]

    return result.decode('utf-8')

In : cut_bytes(lion, 2)
Out: 'we'
In : cut_bytes(lion, 1)
Out: 'öwe'

Concatenating parts of file path can be done with os.path.join:

In : dir_path = '/home/vadim/'
In : file_name = 'test.py'
In : os.path.join(dir_path, file_name)
Out: '/home/vadim/test.py'

It's usually better than using string concating like this:

In : dir_path + '/' + file_name
Out: '/home/vadim//test.py'

os.path.join uses the correct delimiter for the current platform (e. g. \ for Windows). It also never produces a double delimiter (//). Since Python 3.4, you also can use the Path class from the pathlib module. (It also can be used as an os.path.join argument since Python 3.6.) Path supports concatenation via / operator:

In : Path('/home/vadim/') / Path('test.py')
Out: PosixPath('/home/vadim/test.py')

Native Python float values use your computer hardware directly, so any value is represented internally as a binary fraction. That means that you usually work with approximations, not exact values:

In : format(0.1, '.17f')
Out: '0.10000000000000001'

The decimal module lets you use decimal floating point arithmetic with arbitrary precision:

In : Decimal(1) / Decimal(3)
Out: Decimal('0.3333333333333333333333333333')

That's still can be not enough:

In [61]: Decimal(1) / Decimal(3) * Decimal(3) == Decimal(1)
Out[61]: False

For perfect computations, you can use fractions, that stores any number as a rational one:

In : Fraction(1) / Fraction(3) * Fraction(3) == Fraction(1)
Out: True

The obvious limitation is you still have to use approximations to irrational numbers (such as π).