trio-universal-queue.py

class UniversalQueue:
    def __init__(self, *args, **kwargs):
        self._queue = trio.Queue(*args, **kwargs)
        self._portal = trio.BlockingTrioPortal()
        
    async def trio_get(self):
        return await self._queue.trio_get()
        
    def thread_get(self):
        return self._portal.run(self._queue.trio_get)
        
    ...
        
# OR...

# Use anything, from anywhere
# (Well, or at least... any trio object, from either trio or a thread)
class UniversalSquared:
    def __init__(self, obj):
        self._obj = obj
        self._portal = trio.BlockingTrioPortal()
        
    def __getattr__(self, name):
        value = getattr(self._obj, name)
        if callable(value) and sniffio.current_async_library() is None:
            @functools.wraps(value)
            def wrapper(*args, **kwargs):
                async def inner_wrapper():
                    result = value(*args, **kwargs)
                    if hasattr(result, "__await__"):
                        return await result
                    else:
                        return result
                return self._portal.run(inner_wrapper)
            return wrapper
        return value