grai_client
endpoints
client

ClientOptions Objects

class ClientOptions(BaseModel)

validate_connection_arguments

def validate_connection_arguments(
    url: Optional[str] = None,
    host: Optional[str] = None,
    port: Optional[str] = None,
    protocol: Optional[ProtocolType] = None,
    insecure: Optional[bool] = None
) -> Tuple[str, str, str, ProtocolType, bool]

Arguments:

  • url Optional[str], optional - (Default value = None)
  • host Optional[str], optional - (Default value = None)
  • port Optional[str], optional - (Default value = None)
  • protocol Optional[ProtocolType], optional - (Default value = None)
  • insecure Optional[bool], optional - (Default value = None)

Returns:

AuthValues Objects

class AuthValues(BaseModel)

is_valid

def is_valid() -> bool

Arguments:

Returns:

get_auth

def get_auth() -> Auth

Arguments:

Returns:

async_requires_auth

def async_requires_auth(func)

Arguments:

func:

Returns:

requires_auth

def requires_auth(func)

Arguments:

func:

Returns:

BaseClient Objects

class BaseClient(abc.ABC)

get_session_args

def get_session_args() -> Dict

Arguments:

Returns:

default_options

@property
def default_options() -> ClientOptions

Arguments:

Returns:

server_health_status

def server_health_status() -> Response

Arguments:

Returns:

auth

@property
def auth() -> Auth

Arguments:

Returns:

auth

@auth.setter
def auth(auth: Auth) -> None

Arguments:

auth (Auth):

Returns:

authenticate

def authenticate(username: Optional[str] = None,
                 password: Optional[str] = None,
                 api_key: Optional[str] = None) -> None

Arguments:

  • username Optional[str], optional - (Default value = None)
  • password Optional[str], optional - (Default value = None)
  • api_key Optional[str], optional - (Default value = None)

Returns:

check_authentication

@abc.abstractmethod
def check_authentication() -> Response

Arguments:

Returns:

get_url

@multimethod
def get_url(grai_type: Any) -> str

Arguments:

grai_type (Any):

Returns:

session_manager

def session_manager(func: Callable,
                    *args,
                    options: Optional[OptionType] = None,
                    **kwargs)

Arguments:

func (Callable): *args:

  • options Optional[OptionType], optional - (Default value = None) **kwargs:

Returns:

get

@requires_auth
def get(*args, options: Optional[OptionType] = None, **kwargs)

Arguments:

*args:

  • options Optional[OptionType], optional - (Default value = None) **kwargs:

Returns:

post

@requires_auth
def post(*args, options: Optional[OptionType] = None, **kwargs)

Arguments:

*args:

  • options Optional[OptionType], optional - (Default value = None) **kwargs:

Returns:

patch

@requires_auth
def patch(*args, options: Optional[OptionType] = None, **kwargs)

Arguments:

*args:

  • options Optional[OptionType], optional - (Default value = None) **kwargs:

Returns:

delete

@requires_auth
def delete(*args, options: Optional[OptionType] = None, **kwargs)

Arguments:

*args:

  • options Optional[OptionType], optional - (Default value = None) **kwargs:

Returns:

type_segmentation

def type_segmentation(
    objs: Sequence[T], priority_order: Optional[Tuple[Type[T]]]
) -> List[Tuple[List[int], Union[Generator[T, None, None], Iterable[T]], str]]

Arguments:

objs (Sequence): priority_order (Optional[Tuple[Type[T]]]):

Returns:

segmented_caller

def segmented_caller(
    func: Callable[[BaseClient, Sequence[T], ClientOptions], R],
    priority_order: Optional[Tuple[Type[T]]] = None
) -> Callable[[BaseClient, Sequence[T], ClientOptions], list[R]]

Arguments:

func (Callable[[BaseClient, Sequence[T]): ClientOptions]: R]:

  • priority_order Optional[Tuple], optional - (Default value = None)

Returns:

get_sequence

@get.register
def get_sequence(
    client: BaseClient,
    objs: Sequence,
    options: ClientOptions = ClientOptions()) -> Sequence[T]

Arguments:

client (BaseClient): objs (Sequence):

  • options ClientOptions, optional - (Default value = ClientOptions())

Returns:

delete_sequence

@delete.register
def delete_sequence(
    client: BaseClient,
    objs: Sequence,
    options: ClientOptions = ClientOptions()) -> None

Arguments:

client (BaseClient): objs (Sequence):

  • options ClientOptions, optional - (Default value = ClientOptions())

Returns:

post_sequence

@post.register
def post_sequence(
    client: BaseClient,
    objs: Sequence,
    options: ClientOptions = ClientOptions()) -> List[T]

Arguments:

client (BaseClient): objs (Sequence):

  • options ClientOptions, optional - (Default value = ClientOptions())

Returns:

patch_sequence

@patch.register
def patch_sequence(
    client: BaseClient,
    objs: Sequence,
    options: ClientOptions = ClientOptions()) -> List[T]

Arguments:

client (BaseClient): objs (Sequence):

  • options ClientOptions, optional - (Default value = ClientOptions())

Returns: