Python etc

You can slice dicts using dict comprehensions:

In [1]: d = dict(
   ...:     a=1,
   ...:     b=2,
   ...:     c=3,
   ...: )

In [2]: {k: d.get(k) for k in ['a', 'c', 'e']}
Out[2]: {'a': 1, 'c': 3, 'e': None}

In [3]: {k: d.get(k) for k in ['a', 'c', 'e'] if k in d}
Out[3]: {'a': 1, 'c': 3}

In [4]: {k: d.get(k) for k in d.keys() if k in {'a', 'c', 'e'}}
Out[4]: {'a': 1, 'c': 3}

One of the techniques of metaprogramming is to use code generation. That means that you write some program that produces another program as an output. Even if your goal is to create Python code, the program that generates code can be written in any language. Here is the Perl program that generates a line of Python source code, that is executed by Python interpreter afterward:

$ perl -e 'print "print(100)"' | python
100

It's worth noting that not every program that is syntactically correct may be executed, it's possible that expression is simply too long. That usually doesn't happen when you write source code manually, but it's entirely possible to have such problem while generating code:

$ python -c 'print("not " * 1000000 + "False")' | python
s_push: parser stack overflow
MemoryError

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, you also may use modules and classes as context holders since they are, strictly speaking, 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
    task = loop.create_task(
        indication(indication_t)
    )
    await asyncio.sleep(t)
    task.cancel()

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 switches between coroutines. You can do it with the contextvars module since Python 3.7.

import asyncio
import sys
import contextvars

global_symbol = contextvars.ContextVar('symbol')

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

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

    global_symbol.set(symbol)
    task = loop.create_task(indication(indication_t))
    await asyncio.sleep(t)
    task.cancel()

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'),
))

The problem with calling repr of other objects in your own __repr__ method is that you can't guarantee none of the other objects is not equal to self and the call isn't recursive:

In : p = Pair(1, 2)
In : p
Out: Pair(1, 2)
In : p.right = p
In : p
Out: [...]
RecursionError: maximum recursion depth exceeded while calling a Python object

To easily solve this problem you can use the reprlib.recursive_repr decorator:

@reprlib.recursive_repr()
def __repr__(self):
        class_name = type(self).__name__
        return f'{class_name}({self.left!r}, {self.right!r})'

Now it works:

In : p = Pair(1, 2)
In : p.right = p
In : p
Out: Pair(1, ...)

Nested context managers normally don’t know that they are nested. You can make them know by spawning inner context managers by the outer one:

from contextlib import AbstractContextManager
import time


class TimeItContextManager(AbstractContextManager):
    def __init__(self, name, parent=None):
        super().__init__()

        self._name = name
        self._parent = parent
        self._start = None
        self._substracted = 0

    def __enter__(self):
        self._start = time.monotonic()
        return self
        
    def __exit__(self, exc_type, exc_value, traceback):
        delta = time.monotonic() - self._start
        if self._parent is not None:
            self._parent.substract(delta)

        print(self._name, 'total', delta)
        print(self._name, 'outer', delta - self._substracted)

        return False

    def child(self, name):
        return type(self)(name, parent=self)

    def substract(self, n):
        self._substracted += n


timeit = TimeItContextManager


def main():
    with timeit('large') as large_t:
        with large_t.child('medium') as medium_t:
            with medium_t.child('small-1'):
                time.sleep(1)
            with medium_t.child('small-2'):
                time.sleep(1)
            time.sleep(1)
        time.sleep(1)


main()

If you want to measure time between two events you should use time.monotonic() instead of time.time(). time.monotonic() never goes backwards even if system clock is updated:

from contextlib import contextmanager
import time


@contextmanager
def timeit():
    start = time.monotonic()
    yield
    print(time.monotonic() - start)

def main():
    with timeit():
        time.sleep(2)

main()

Feel like you versed in Data Science but can't organize your knowledges properly and don't have enough practice? Online-education center SkillFactory launches Data Science course https://clc.to/I9sEKQ This course will allow you not only to increase your existing skills but learn somethig new. All our teachers are industry professionals, who worked for Yandex and NVIDIA. They will share with you industry insides and intricacies of the job, which are not written in any books. Our course program includes a Python learning block (including Pandas for Data Analysis), mathematics, statistics probability theory with solving problems in NumPy and landing on Data Science, introduction to Machine Learning, course of Data Engineering, Neural Networks and AI and management for a data scientist: the skill of implementing data analysis systems, machine learning and neural networks. Want to become a one of a kind specialist and take part in a variety of competitions on Kaggle (with solutions analyses and various models of Machine Learning and Neural Networks training) - our course starts 11.06, click to register now https://clc.to/I9sEKQ

Ads time :).

Sometimes during iteration you may want to know whether it’s the first or the last element step of the iteration. Simple way to handle this is to use explicit flag:

def sparse_list(iterable, num_of_zeros=1):
    result = []
    zeros = [0 for _ in range(num_of_zeros)]

    first = True
    for x in iterable:
        if not first:
            result += zeros
        result.append(x)

        first = False

    return result

assert sparse_list([1, 2, 3], 2) == [
    1,
    0, 0,
    2,
    0, 0,
    3,
]

You also could process the first element outside of the loop, that may seem more clear but leads to code duplication to the certain extent. It is also not a simple thing to do while working with abstract iterables:

def sparse_list(iterable, num_of_zeros=1):
    result = []
    zeros = [0 for _ in range(num_of_zeros)]

    iterator = iter(iterable)
    try:
        result.append(next(iterator))
    except StopIteration:
        return []

    for x in iterator:
        result += zeros
        result.append(x)

    return result

You also could use enumerate and check for the i == 0 (works only for the detection of the first element, not the last one), but the ultimate solution might be a generator that returns first and last flags along with the element of an iterable:

def first_last_iter(iterable):
    iterator = iter(iterable)

    first = True
    last = False
    while not last:
        if first:
            try:
                current = next(iterator)
            except StopIteration:
                return
        else:
            current = next_one

        try:
            next_one = next(iterator)
        except StopIteration:
            last = True

        yield (first, last, current)

        first = False

The initial function now may look like this:

def sparse_list(iterable, num_of_zeros=1):
    result = []
    zeros = [0 for _ in range(num_of_zeros)]

    for first, last, x in first_last_iter(iterable):
        if not first:
            result += zeros
        result.append(x)

    return result

Context managers and function decorators are pretty similar and usually interchangeable. Any context manager can be used as a function decorator as long as it's derived from contextlib.ContextDecorator. (Mind, that the @contextlib.contextmanager decorator inherits from that class automatically.) On the other hand, you can't use any decorator as a context manager due to severe limitation: a context manager always runs a code block exactly once while a decorated function may call original one as many times as it wants (zero included). PEP 377 proposed the change that allows __enter__ to ask Python not to run a code block at all, but it was rejected. As a workaround, you should extract a function and apply a decorator to it:

def retry(attempts, exceptions=(Exception,)):       
    def decorator(func):                            
        def decorated(*args, **kwargs):             
            for _ in range(attempts - 1):           
                with suppress(*exceptions):          
                    return func(*args, **kwargs)    
            return func(*args, **kwargs)  # last try
        return decorated                            
    return decorator                                

@retry(10, HTTPError) 
def get(url):
    requests.get(url).raise_for_status()

A method in Python is a callable object. It’s not, however, the same object as the function it was originally created from. Every time you access a method, a new object is created that stores information not only about the function that will be executed but also about self, that will be used as a first argument:

class User:
    def __init__(self, name):
        self._name = name

    def __repr__(self):
        return '{klass}("{name}")'.format(
            klass=type(self).__name__,
            name=self._name,
        )

    def get_name(self):
        return self._name


vadim = User('vadim')

# That is False:
print(vadim.get_name is vadim.get_name)

Both original function and encapsulated self can be directly accessed though:

>>> vadim.get_name 
<bound method User.get_name of User("vadim")>
>>> m = vadim.get_name
>>> m
<bound method User.get_name of User("vadim")>
>>> m()
'vadim'
>>> m.__func__(m.__self__)
'vadim'

A decorator creates a new object (usually function) using another single function as an argument. However, you may want to provide more than one function. This can't be done straightforwardly due to limitations of Python syntax, but you could use a simple trick to solve the issue. The returned function may contain another decorator that can be reapplied to additional functions to bring in additional behavior. That is kinda what @property does:

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

@x.setter
def x(self, value):
    self._x = value

This is the example of how you can define a function that uses additional functions for special cases:

from functools import wraps

def make_case_decorator(func):
    def case_decorator(*case_decorator_args):
        def decorator(special_case_func):
            @wraps(func)
            def decorated(*args):
                if case_decorator_args == args:
                    return special_case_func(*args)
                return func(*args)

            decorated.case = make_case_decorator(decorated)

            return decorated

        return decorator

    return case_decorator


def special_cases(func):
    @wraps(func)
    def decorated(*args):
        return func(*args)

    decorated.case = make_case_decorator(decorated)

    return decorated

@special_cases
def fact(x):
    return x * fact(x - 1)

@fact.case(0)
def fact(x):
    return 1

@fact.case(10)
def fact(x):
    print(f'(optimization worked for {x})')
    return 3628800

You can’t mutate closure variables by simply assigning them. Python treats assignment as a definition inside a function body and doesn’t make closure at all Works fine, prints 2:

def make_closure(x):
    def closure():
        print(x)

    return closure

make_closure(2)()

Throws UnboundLocalError: local variable 'x' referenced before assignment:

def make_closure(x):
    def closure():
        print(x)
        x *= 2
        print(x)

    return closure

make_closure(2)()

To make it work you should use nonlocal. It explicitly tells the interpreter not to treat assignment as a definition:

def make_closure(x):
    def closure():
        nonlocal x
        print(x)
        x *= 2
        print(x)

    return closure

make_closure(2)()

Any running asyncio coroutine can be cancelled via the cancel() method. CancelledError will be thrown into the coroutine that will lead for it and all wrapping coroutines to be terminated, unless the error is caught and suppressed. CancelledError is a subclass of Exception that means that it can be accidentally caught by try ... except Exception that is meant to catch “any error”. To safely do this within a coroutine, you stuck with something like this:

try:
    await action()
except asyncio.CancelledError:
    raise
except Exception:
    logging.exception('action failed')

Conditional using of context managers is usually a pain; you can’t simply place with inside an if-block without putting the entire with-block there. That usually leads to duplicate code:

def print_whole_file(
    *,
    path: Optional[str] = None,
    file_obj: Optional[TextIO] = None
):
    assert path or file_obj

    if path:
        with open(path) as f:
            print(f.read(), end='')
    else:
        print(file_obj.read(), end='')

The way to fight the problem is to use ExitStack and place enter_context inside if:

def print_whole_file(
    *,
    path: Optional[str] = None,
    file_obj: Optional[TextIO] = None
):
    assert path or file_obj

    with ExitStack() as stack:
        if path:
            file_obj = stack.enter_context(
                open(path)
            )

        print(file_obj.read(), end='')

However, the more obvious way to achieve the same is to use trivial context managers that do nothing when you don’t need them instead of the real ones. Since Python 3.7, you can obtain one with contextlib.nullcontext:

def print_whole_file(
    *,
    path: Optional[str] = None,
    file_obj: Optional[TextIO] = None
):
    assert path or file_obj

    if path:
        context = open(path)
    else:
        context = nullcontext(file_obj)

    with context as f:
        print(f.read(), end='')

Have you ever heard about defer construction from languages like go, swift or kotlin? It allows you to execute some cleanup code upon function exit:

func main() {
    f := createFile("/tmp/defer.txt")
    defer closeFile(f)
    err = writeFile(f)
    if err != nil {
        // closeFile is called here
        return
    }
    // closeFile is called here
}

Typical error happens once you forget to invoke this logic before one of the returns and defer solves this problem. In Pytho, we usually use context managers or finally block for such tasks. However, it may be too wordy to create context manager every time you need defer. contextlib.ExitStack.callback does exactly what defer would do:

from contextlib import ExitStack

def foo(a: int) -> str:
    with ExitStack() as stack:
        stack.callback(lambda: print('Leaving...'))
        if (a > 0):
            return 'Positive'

        if a == 0:
            return 'Zero'

        if a < 0:
            return 'Negative'


if __name__ == '__main__':
    print(foo(1))
    print(foo(0))
    print(foo(-20))

Output:

Leaving...
Positive
Leaving...
Zero
Leaving...
Negative

Another post by @karbachinsky. Enjoy!

itertools.tee() creates multiple iterators from the single one. That may be helpful if numerous consumers need to read the same stream.

In : a, b, c = tee(iter(input, ''), 3)

In : next(a), next(c)
FIRST
Out: ('FIRST', 'FIRST')

In : next(a), next(b)
SECOND
Out: ('SECOND', 'FIRST')

In : next(a), next(b), next(c)
THIRD
Out: ('THIRD', 'SECOND', 'SECOND')

The data that is not yet used by all iterators is stored in memory. If some of the created iterators are not yet started at the time another one is finished, that means that all of the generated elements are saved in memory for future use. In that case, it's simpler and more efficient to use list(iter(input, '')) instead of tee.

In Python, you can chain comparison operators:

>>> 0 < 1 < 2
True
>>> 0 < 1 < 0
False

Such chains don’t have to be mathematically valid, you can mix > and <:

>>> 0 < 1 > 2
False
>>> 0 < 1 < 2 > 1 > 0
True

Other operators such as ==, is and in are also supported:

>>> [] is not 3 in [1, 2, 3]
True

Every operator is applied to the two nearest operands. a OP1 b OP2 c is strictly equal to (a OP1 b) AND (b OP2 c). No comparison between a and c is implied:

class Spy:
    def __init__(self, x):
        self.x = x

    def __eq__(self, other):
        print(f'{self.x} == {other.x}')
        return self.x == other.x

    def __ne__(self, other):
        print(f'{self.x} != {other.x}')
        return self.x != other.x

    def __lt__(self, other):
        print(f'{self.x} < {other.x}')
        return self.x < other.x

    def __le__(self, other):
        print(f'{self.x} <= {other.x}')
        return self.x <= other.x

    def __gt__(self, other):
        print(f'{self.x} > {other.x}')
        return self.x > other.x

    def __ge__(self, other):
        print(f'{self.x} >= {other.x}')
        return self.x >= other.x


s1 = Spy(1)
s2 = Spy(2)
s3 = Spy(3)

print(s1 is s1 < s2 <= s3 == s3)

Output:

1 < 2
2 <= 3
3 == 3
True

There are two built-in functions that let you analyze iterables without writing trivial and redundant for loops. These are all and any. any returns True if some of the values are true; all returns True if all of them are. all returns True for an empty iterable while any returns False in that case. Both functions are usually useful while used together with list comprehensions:

package_broken = any(
    part.is_broken() for part in package.get_parts()
)
package_ok = all(
    part.ok() for part in package.get_parts()
)

any and all are usually interchangeable thanks to De Morgan's laws. Choose one that is easier to understand.