File: app_operation.py

package info (click to toggle)
pypy3 7.3.19%2Bdfsg-2
  • links: PTS, VCS
  • area: main
  • in suites: trixie
  • size: 212,236 kB
  • sloc: python: 2,098,316; ansic: 540,565; sh: 21,462; asm: 14,419; cpp: 4,451; makefile: 4,209; objc: 761; xml: 530; exp: 499; javascript: 314; pascal: 244; lisp: 45; csh: 12; awk: 4
file content (83 lines) | stat: -rw-r--r-- 2,312 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
import _operator

def bin(x):
    """Return the binary representation of an integer.

    >>> bin(2796202)
    '0b1010101010101010101010'

    """
    value = _operator.index(x)
    return int.__format__(value, "#b")

def oct(x):
    """Return the octal representation of an integer.

    >>> oct(342391)
    '0o1234567'

    """
    x = _operator.index(x)
    return int.__format__(x, "#o")

def hex(x):
    """Return the hexadecimal representation of an integer.

    >>> hex(12648430)
    '0xc0ffee'

    """
    x = _operator.index(x)
    return int.__format__(x, "#x")


# anext and aiter adapted from: https://github.com/python/cpython/pull/8895

_NOT_PROVIDED = object()  # sentinel object to detect when a kwarg was not given


def aiter(obj):
    """aiter(async_iterable) -> async_iterator
    aiter(async_callable, sentinel) -> async_iterator
    Like the iter() builtin but for async iterables and callables.
    """
    typ = type(obj)
    try:
        meth = typ.__aiter__
    except AttributeError:
        raise TypeError(f"'{type(obj).__name__}' object is not an async iterable")
    ait = meth(obj)
    if not hasattr(ait, '__anext__'):
        raise TypeError(f"aiter() returned not an async iterator of type '{type(ait).__name__}'")
    return ait

def anext(iterator, default=_NOT_PROVIDED):
    """anext(async_iterator[, default])
    Return the next item from the async iterator.
    If default is given and the iterator is exhausted,
    it is returned instead of raising StopAsyncIteration.
    """
    typ = type(iterator)

    try:
        __anext__ = typ.__anext__
    except AttributeError:
        raise TypeError(f"'{type(iterator).__name__}' object is not an async iterator")

    if default is _NOT_PROVIDED:
        return __anext__(iterator)

    async def anext_impl():
        try:
            # The C code is way more low-level than this, as it implements
            # all methods of the iterator protocol. In this implementation
            # we're relying on higher-level coroutine concepts, but that's
            # exactly what we want -- crosstest pure-Python high-level
            # implementation and low-level C anext() iterators.
            return await __anext__(iterator)
        except StopAsyncIteration:
            return default

    return anext_impl()