Created
November 23, 2016 09:38
-
-
Save mpontillo/941f662b30f80781a69097dde40d79fa to your computer and use it in GitHub Desktop.
IPython startup script version of: https://github.com/Gr1N/asyncio-ipython-magic/blob/master/asynciomagic.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# -*- 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