Last active
December 7, 2017 13:05
-
-
Save AileenLumina/15f720a2b64a272b0ef87aa9c02975f6 to your computer and use it in GitHub Desktop.
Testing a way to allow the execution of coroutine methods as synchronous methods
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
| import types | |
| import asyncio | |
| import inspect | |
| import time | |
| class API: | |
| # def __new__(cls, | |
| def self_reference(function): | |
| function.__defaults__ = (function,) + function.__defaults__[-1:] | |
| return function | |
| @self_reference | |
| def _run_in_loop(self, this_method=None, *args, **kwargs): | |
| self.loop.run_until_complete(this_method.coro(self, *args, **kwargs)) | |
| def __init__(self, is_async=False, name=None): | |
| print("API.__init__ was called") | |
| if not is_async: | |
| print("API is not async") | |
| self.loop = asyncio.get_event_loop() | |
| for method_name, method_value in inspect.getmembers(self): | |
| if inspect.iscoroutinefunction(method_value): | |
| self.setattr('async_' + method_name, method_value) | |
| new_sync_method = _run_in_loop | |
| new_sync_method.__name__ = method_name | |
| new_sync_method.coro = method_value | |
| self.setattr(method_name, new_sync_method) | |
| class ExampleAPI(API): | |
| def __init__(self, is_async=False, name=None): | |
| super() | |
| self.is_async = is_async | |
| self.name = name | |
| print("My name is " + self.name) | |
| async def change_name(self, name): | |
| if self.is_async: | |
| if inspect.iscoroutinefunction(self.change_name): | |
| print("I am a coroutine") | |
| print("Changing name...") | |
| await asyncio.sleep(1) | |
| self.name = name | |
| print("Changed name to " + self.name) | |
| return | |
| else: | |
| print("Changing name...") | |
| time.sleep(1) | |
| self.name = name | |
| print("Changed name to " + self.name) | |
| async def test_async(): | |
| eapi = ExampleAPI(is_async=True, name="Spam") | |
| await eapi.change_name("Eggs") | |
| def test(): | |
| eapi = ExampleAPI(name="Spam") | |
| eapi.change_name("Eggs") | |
| if __name__ == '__main__': | |
| loop = asyncio.get_event_loop() | |
| loop.run_until_complete(test_async()) | |
| loop.close() | |
| test() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment