lib/portage/util/futures/_sync_decorator.py

# Copyright 2018 Gentoo Foundation
# Distributed under the terms of the GNU General Public License v2

import functools

import portage
portage.proxy.lazyimport.lazyimport(globals(),
	'portage.util.futures:asyncio',
)


def _sync_decorator(func, loop=None):
	"""
	Decorate an asynchronous function (either a corouting function or a
	function that returns a Future) with a wrapper that runs the function
	synchronously.
	"""
	loop = asyncio._wrap_loop(loop)
	@functools.wraps(func)
	def wrapper(*args, **kwargs):
		return loop.run_until_complete(func(*args, **kwargs))
	return wrapper


def _sync_methods(obj, loop=None):
	"""
	For use with synchronous code that needs to interact with an object
	that has coroutine methods, this function generates a proxy which
	conveniently converts coroutine methods into synchronous methods.
	This allows coroutines to smoothly blend with synchronous
	code, eliminating clutter that might otherwise discourage the
	proliferation of coroutine usage for I/O bound tasks.
	"""
	loop = asyncio._wrap_loop(loop)
	return _ObjectAttrWrapper(obj,
		lambda attr: _sync_decorator(attr, loop=loop)
		if asyncio.iscoroutinefunction(attr) else attr)


class _ObjectAttrWrapper(portage.proxy.objectproxy.ObjectProxy):

	__slots__ = ('_obj', '_attr_wrapper')

	def __init__(self, obj, attr_wrapper):
		object.__setattr__(self, '_obj', obj)
		object.__setattr__(self, '_attr_wrapper', attr_wrapper)

	def __getattribute__(self, attr):
		obj = object.__getattribute__(self, '_obj')
		attr_wrapper = object.__getattribute__(self, '_attr_wrapper')
		return attr_wrapper(getattr(obj, attr))

	def _get_target(self):
		return object.__getattribute__(self, '_obj')