Python etc

Welcome to weekend task section. Below is the task the you can solve in Python. My solution will be publish in 36 hours.

A priority queue is a data structure that supports two operations: add element and extract the minimum of all elements among previously added. One of the most common implementations of a priority queue is a binary heap. It's a complete binary tree with the following property: the key stored in each node is equal to or less than (≤) the keys in the node's children. The minimum of all elements is a root of such tree.

             1

      3               7

  5       4       9       8

15 16   17 18   19

In a binary heap, both inserting and extraction operations' complexity is O(log n). The common way of storing a complete binary tree in memory is an array, where children of x[i] are x[2*i+1] and x[2*i+2]:

[1, 3, 7, 5, 4, 9, 8, 15, 16, 17, 18, 19]

Python doesn't provide a binary heap as a class, but it does provide a number of functions that treat list like a binary heap. They are placed in the heapq module.

In [1]: from heapq import *
In [2]: heap = [3,2,1]
In [3]: heapify(heap)
In [4]: heap
Out[4]: [1, 2, 3]
In [5]: heappush(heap, 0)
In [6]: heap
Out[6]: [0, 1, 3, 2]
In [7]: heappop(heap)
Out[7]: 0
In [8]: heap
Out[8]: [1, 2, 3]

All objects in Python are created via the call to the __new__ method. Even if you provide custom __new__ for your class, you have to call super().__new__(...). You might think that object.__new__ is a root implementation that is responsible for the creation of all objects. That is not entirely true. There are several such implementations, and they are incompatible. For example, dict has its own low-level __new__ and objects of types derived from dict can't be created with object.__new__:

In : class D(dict):
...:     pass
...:

In : class A:
...:     pass
...:

In : object.__new__(A)
Out: <__main__.A at 0x7f200c8902e8>

In : object.__new__(D)
...
TypeError: object.__new__(D) is not safe,
use D.__new__()

Python lets you overload many different operators and the shift operator is one of them. Here is an example of how to create a function composition using this operator. Here, arrow-like signs show the data-flow direction:

from collections import deque
from math import sqrt


class Compose:
    def __init__(self):
        self._functions = deque()

    def __call__(self, *args, **kwargs):
        result = None
        for f in self._functions:
            result = f(*args, **kwargs)
            args = [result]
            kwargs = dict()
        return result

    def __rshift__(self, f):
        self._functions.append(f)
        return self

    def __lshift__(self, f):
        self._functions.appendleft(f)
        return self


compose = Compose


sqrt_abs = (compose() << sqrt << abs)
sqrt_abs2 = (compose() >> abs >> sqrt)

print(sqrt_abs(-4))  # 2.0
print(sqrt_abs2(-4))  # 2.0

Your task is to create a rate limiting object. Users calls rl.query(id, ts). id is the user unique identifier. ts is a current time which must be non-decreasing across all calls. The method returns False if the user has already made N or more queries in the last T seconds. N and T are specified during rl creation.

Welcome to weekend task section. Below is the task the you can solve in Python. My solution will be publish in 36 hours.

If you override the __setattr__ method but still wants to assign some instance attributes, you should call the base class method via super():

class DictConstructor:
    def __init__(self, _d=None, _path=None):
        if _d is None:
            _d = {}
        if _path is None:
            _path = ()
        super().__setattr__('_d', _d)
        super().__setattr__('_path', _path)

    def to_dict(self):
        return self._d

    def __getattr__(self, name):
        return type(self)(
            self._d,
            self._path + (name,)
        )

    def __setattr__(self, name, value):
        d = self._d
        for p in self._path:
            d.setdefault(p, {})
            d = d[p]

        d[name] = value


d = DictConstructor()
d.a = 1
d.x.y.z = 2

print(d.to_dict())
# {'a': 1, 'x': {'y': {'z': 2}}}

The itertools.chain function is a way to iterate over many iterables as though they are glued together:

In : list(chain(['a', 'b'], range(3), set('xyz')))
Out: ['a', 'b', 0, 1, 2, 'x', 'z', 'y']

Sometimes you want to know whether a generator is empty (rather say, exhausted). To do this, you have to try getting the next element from the generator. If it works, you would like to put element back in the generator, which of course is not possible. You can glue it back with chain instead:

def sum_of_odd(gen):
    try:
        first = next(gen)
    except StopIteration:
        raise ValueError('Empty generator')

    return sum(
        x for x in chain([first], gen)
        if x % 2 == 1
    )

Usage example:

In : sum_of_odd(x for x in range(1, 6))
Out: 9
In : sum_of_odd(x for x in range(2, 3))
Out: 0
In : sum_of_odd(x for x in range(2, 2))
...
ValueError: Empty generator

The class of an object is available through the __class__ attribute:

>>> [1, 2].__class__
<class 'list'>

The more conventional way to get the class, however, is to use the type function. It's also the only way that works for old-style classes.

>>> type([1, 2])
<class 'list'>

Also, if you want to check whether some object is an instance of the given class, you should use isinstance instead of comparison:

>>> class A:
...   pass
...
>>> class B(A):
...   pass
...
>>> type(B())
<class '__main__.B'>
>>> isinstance(B(), A)
True

Your task is to create a function that lets you iterate over parts (batches) of some iterable. Every part is an iterable itself that only iterates over the original iterable and doesn’t store any data. The new part is started once the key function returns result that is not equal to the previous one.

for batch in batches(range(9), lambda y: y > 4):
    print('[{}]'.format(','.join(
        str(x) for x in batch
    )))

Output:

[0,1,2,3,4]
[5,6,7,8]

Iteration should be strictly sequential. The user must not ask for the next batch before the previous one is exhausted.

# RuntimeError
list(batches(range(9), lambda y: y > 4))

Welcome to weekend task section. Below is the task the you can solve in Python. My solution will be publish in 36 hours.

Before arguments unpacking was introduced, you could use the apply built-in function, which is now deprecated (and completely removed from Python 3). The exact equivalent of f(*a, **k) is apply(f, a, k). If you still want to use apply (for backward compatibility or due to semantic reasons) you could do something like this:

def apply(func, args, kwargs):
    return func(*args, **kwargs)

asyncio loop doesn’t have to be run to have tasks. You can create and stop tasks even though the loop is stopped right now. If loop is stopped, some tasks may stay incompleted for good.

import asyncio


async def printer():
    try:
        try:
            while True:
               print('⚙️')
               await asyncio.sleep(1)
        except asyncio.CancelledError:
            print('❌')
    finally:
        await asyncio.sleep(2)
        print('🛑')  # never happens


loop = asyncio.get_event_loop()
run = loop.run_until_complete
task = loop.create_task(printer())

run(asyncio.sleep(1))  # printer works here
print('⏸️')

run(asyncio.sleep(1))  # printer works here
task.cancel()  # nothing happens
run(asyncio.sleep(1))  # ❌ printed

Output:

⚙️
⚙️
⏸️
⚙️
❌

You have to be sure to await all tasks before stopping the loop. In case you don’t you may have some finally blocks being skipped and some context managers not being exited.

You have some iterable that yields integers. You task is to save all these integers and let others to iterate over them. The catch is you have to save all non-decreasing subsequences as ranges so you don’t consume too much memory. For example, [1, 2, 3, 10, 11, 12, 13, 14, 15] should be internally stored like range(1, 4) and range(10, 16).

Welcome to weekend task section. Below is the task the you can solve in Python. My solution will be publish in 36 hours.

Annotating a factory method is not as simple as it may seem. The immediate urge is to use something like this:

class A:
    @classmethod
    def create(cls) -> 'A':
        return cls()

However, that is not a right thing to do. The catch is, create doesn’t return A, it returns an instance of cls that is A or any of its descendants. Look at this code:

class A:
    @classmethod
    def create(cls) -> 'A':
        return cls()


class B(A):
    @classmethod
    def create(cls) -> 'B':
        return super().create()

The mypy check result is error: Incompatible return value type (got "A", expected "B"). Again, the problem is super().create() is annotated to return A while it clearly returns B in this case. You can fix that by annotating cls with TypeVar:

AType = TypeVar('AType')
BType = TypeVar('BType')


class A:
    @classmethod
    def create(cls: Type[AType]) -> AType:
        return cls()


class B(A):
    @classmethod
    def create(cls: Type[BType]) -> BType:
        return super().create()

Now create returns the instance of the cls class. However, this annotations are too loose, we lost the information that cls is a subtype of A:

AType = TypeVar('AType')


class A:
    DATA = 42
    @classmethod
    def create(cls: Type[AType]) -> AType:
        print(cls.DATA)
        return cls()

The error is "Type[AType]" has no attribute "DATA". To fix that you have to explicitly define AType as a subtype of A with the bound argument of TypeVar:

AType = TypeVar('AType', bound='A')
BType = TypeVar('BType', bound='B')


class A:
    DATA = 42
    @classmethod
    def create(cls: Type[AType]) -> AType:
        print(cls.DATA)
        return cls()


class B(A):
    @classmethod
    def create(cls: Type[BType]) -> BType:
        return super().create()

Some modules may contain such cryptic constructions:

try:
    cache
except NameError:
    cache = {}

Looks like there is no point to do something like this. cache definitely causes NameError at the beginning of the module since it wasn't assigned before. However, it's not the case if the module is reloaded. When this happens, the dictionary containing all module attributes is reused, giving the module opportunity to reuse attributes of its previous incarnation. If the module is designed to be reloaded, it can rely on this feature. For example, the above code helps it to keep some cache intact upon reloading.