The simplest native coroutine demo

(that I could imagine)



In [1]:

    
import types

@types.coroutine
def gen():
    yield 42



In [2]:

    
async def delegating():
    await gen()

The driving code starts here:



In [3]:

    
coro = delegating()
coro









    Out[3]:





<coroutine object delegating at 0x10496fd00>



In [4]:

    
coro.send(None)









    Out[4]:





42



In [5]:

    
# coro.send(None)  # --> StopIteration

A slightly more interesting demo

Now the generator-coroutine yields 3 times.



In [6]:

    
@types.coroutine
def gen123():
    return (i for i in range(1, 4))



In [7]:

    
async def delegating():
    await gen123()

Driving code:



In [8]:

    
coro = delegating()
coro.send(None)









    Out[8]:





1



In [9]:

    
coro.send(None)









    Out[9]:





2



In [10]:

    
coro.send(None)









    Out[10]:





3



In [11]:

    
# coro.send(None)  # --> StopIteration



In [12]:

    
# coro.send(None)  # --> RuntimeError

A generator-coroutine that receives values

The driving code can send values other than None.



In [13]:

    
import types

@types.coroutine
def times10(terms):
    n = yield 'Ready to begin!'
    for _ in range(terms):
        n = yield n * 10
    return n * 10



In [14]:

    
async def delegating(terms):
    res = await times10(terms)
    return res

Driving code must prime the coroutine by sending None initially:



In [15]:

    
coro = delegating(3)
coro.send(None)









    Out[15]:





'Ready to begin!'



In [16]:

    
coro.send(5)









    Out[16]:





50



In [17]:

    
coro.send(6)









    Out[17]:





60



In [18]:

    
coro.send(7)









    Out[18]:





70

To retrieve the last result, we must catch StopIteration and get its value attribute:



In [19]:

    
try:
    coro.send(8)
except StopIteration as e:
    res = e.value
res









    Out[19]:





80