Skip to content

Instantly share code, notes, and snippets.

@Carreau

Carreau/minirepl

Created May 7, 2019
Embed
What would you like to do?
import asyncio
import trio
import sys
from ast import PyCF_ALLOW_TOP_LEVEL_AWAIT
from inspect import CO_COROUTINE
from textwrap import dedent, indent
g = {}
l = {}
def is_async(co):
return co.co_flags & CO_COROUTINE == CO_COROUTINE
@asyncio.coroutine
def async_exec(co, *args):
"""
An async-version of eval, that need to be awaited
And will be able to run top-level async code obejcts.
"""
if is_async(co):
print('ASYNC')
yield from eval(co, *args)
else:
exec(co, *args)
test_cases = """
import asyncio
from asyncio import sleep
from contextlib import asynccontextmanager
delay = 0.1
count = 0
---
count += await asyncio.sleep(delay, result=1)
---
@asynccontextmanager
async def acm():
print('before')
try:
await sleep(delay)
yield 1
finally:
print('finally')
---
async with acm() as delta:
count += delta
---
async def ticker(to):
for i in range(to):
yield i
await asyncio.sleep(delay)
---
async for i in ticker(3):
count+=i
---
async with asyncio.Lock() as l:
count +=1
---
assert count==6
---
import asyncio, itertools
---
async def ticker():
for i in itertools.count():
print(i)
await asyncio.sleep(1)
---
asyncio.ensure_future(ticker())
---
await sleep(2, result='a')
---
await sleep(2, result='b')
---
await sleep(2, result='c')
---
await sleep(2, result='d')
"""
test_cases = test_cases.split('---')
if sys.stdin.isatty():
def inputs():
while True:
yield input('>>> ')
echo = lambda x: None
else:
inputs = lambda :test_cases
echo = print
async def f():
for inp in inputs():
inp = dedent(inp)
echo(indent(inp, '>>> '))
try:
co = compile(inp, '', 'single', flags=PyCF_ALLOW_TOP_LEVEL_AWAIT)
except:
co = compile(inp, '', 'exec', flags=PyCF_ALLOW_TOP_LEVEL_AWAIT)
await async_exec(co,g)
asyncio.run(f())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.