Source code for ai.backend.client.base

import functools

__all__ = (

def _wrap_method(cls, orig_name, meth):

    def _method(*args, **kwargs):
        assert cls.session is not None, \
               'You must use API wrapper functions via a Session object.'
        # We need to keep the original attributes so that they could be correctly
        # bound to the class/instance at runtime.
        func = getattr(cls, orig_name)
        coro = func(*args, **kwargs)
        if hasattr(cls.session, 'worker_thread'):
            return cls.session.worker_thread.execute(coro)
            return coro

    return _method

[docs]def api_function(meth): ''' Mark the wrapped method as the API function method. ''' setattr(meth, '_backend_api', True) return meth
[docs]class APIFunctionMeta(type): ''' Converts all methods marked with :func:`api_function` into session-aware methods that are either plain Python functions or coroutines. ''' _async = True def __init__(cls, name, bases, attrs, **kwargs): assert 'session' in attrs, \ 'An API function class must define the session attribute.' super().__init__(name, bases, attrs) for attr_name, attr_value in attrs.items(): if hasattr(attr_value, '_backend_api'): orig_name = '_orig_' + attr_name setattr(cls, orig_name, attr_value) wrapped = _wrap_method(cls, orig_name, attr_value) setattr(cls, attr_name, wrapped)
[docs]class BaseFunction(metaclass=APIFunctionMeta): ''' The class used to build API functions proxies bound to specific session instances. ''' session = None