Skip to content

Instantly share code, notes, and snippets.

@mpontillo
Created November 23, 2016 09:38
Show Gist options
  • Save mpontillo/941f662b30f80781a69097dde40d79fa to your computer and use it in GitHub Desktop.
Save mpontillo/941f662b30f80781a69097dde40d79fa to your computer and use it in GitHub Desktop.
# -*- coding: utf-8 -*-
# IPython startup script version of Gr1N's asyncio-ipython-magic.
# To use, place this file in: ~/.ipython/profile_default/startup
import ast as _ast
import asyncio as _asyncio
from IPython.core.magic import Magics, magics_class, cell_magic
class RewriteAwait(_ast.NodeTransformer):
def visit_Await(self, node):
self.generic_visit(node)
from ast import Call, Attribute, Name, Load
new = Call(
func=Attribute(
value=Call(
func=Attribute(
value=Name(
id='_asyncio',
ctx=Load()
),
attr='get_event_loop',
ctx=Load()),
args=[],
keywords=[]
),
attr='run_until_complete',
ctx=Load()
),
args=[node.value],
keywords=[]
)
return _ast.copy_location(new, node)
def visit_AsyncFunctionDef(self, node):
# Don't transform awaits inside an 'async def' function
return node
def visit_Return(self, node):
raise SyntaxError('Return outside function definition')
@magics_class
class AsyncIOMagics(Magics):
@cell_magic
def async(self, line, cell):
from IPython.utils.text import indent
coro_wrapper = 'async def __f():\n{cell}'.format(cell=indent(cell))
coro_wrapper = _ast.parse(coro_wrapper)
coro_wrapper = coro_wrapper.body[0].body
nodes = [RewriteAwait().visit(node) for node in coro_wrapper]
module = _ast.Module(nodes)
_ast.fix_missing_locations(module)
coro = compile(module, filename='<asynciomagic>', mode='exec')
self.shell.ex(coro)
get_ipython().user_global_ns['_asyncio'] = _asyncio
get_ipython().register_magics(AsyncIOMagics)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment