File: async_with2.py

package info (click to toggle)
giac 1.6.0.41%2Bdfsg1-1
  • links: PTS, VCS
  • area: main
  • in suites: bullseye
  • size: 64,540 kB
  • sloc: cpp: 351,842; ansic: 105,138; python: 30,545; javascript: 8,675; yacc: 2,690; lex: 2,449; makefile: 1,243; sh: 579; perl: 314; lisp: 216; asm: 62; java: 41; sed: 16; csh: 7; pascal: 6
file content (37 lines) | stat: -rw-r--r-- 854 bytes parent folder | download | duplicates (5)
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
# test waiting within async with enter/exit functions

import sys
if sys.implementation.name == 'micropython':
    # uPy allows normal generators to be awaitables
    coroutine = lambda f: f
else:
    import types
    coroutine = types.coroutine

@coroutine
def f(x):
    print('f start:', x)
    yield x + 1
    yield x + 2
    return x + 3

class AContext:
    async def __aenter__(self):
        print('enter')
        print('f returned:', await f(10))
    async def __aexit__(self, exc_type, exc, tb):
        print('exit', exc_type, exc)
        print('f returned:', await f(20))

async def coro():
    async with AContext():
        print('body start')
        print('body f returned:', await f(30))
        print('body end')

o = coro()
try:
    while True:
        print('coro yielded:', o.send(None))
except StopIteration:
    print('finished')