aito.client.aito_client.AitoClient

class aito.client.aito_client.AitoClient(instance_url: str, api_key: str, check_credentials: bool = True, raise_for_status: bool = True)

Bases: object

A versatile client that connects to the Aito Database Instance

Parameters:

instance_url (str) – Aito instance url
api_key (str) – Aito instance API key
check_credentials (bool) – check the given credentials by requesting the Aito instance version, defaults to True
raise_for_status (bool) – automatically raise RequestError for each failed response, defaults to True

Raises:

BaseError – an error occurred during the creation of AitoClient

>>> aito_client = AitoClient(your_instance_url, your_api_key)
>>> # Change the API key to READ-WRITE or READ-ONLY
>>> aito_client.api_key = new_api_key

Methods

`async_request`(session, *[, method, ...])	execute a request asynchronously using aiohttp ClientSession
`async_requests`(methods, endpoints, queries)
`batch_requests`(requests[, ...])	execute a batch of requests asynchronously
`bounded_async_request`(semaphore, args, *kwargs)	bounded concurrent requests with asyncio semaphore
`request`(*[, method, endpoint, query, ...])	make a request to an Aito API endpoint The client returns a JSON response if the request succeed and a `RequestError` if the request fails

Attributes

`headers`	the headers that will be used to send a request to the Aito instance
`is_multitenant`	Check if the client is connected to a multitenant instance.

execute a request asynchronously using aiohttp ClientSession

Parameters:

session (ClientSession) – aiohttp ClientSession for making request
method (str) – method for the new AitoRequest object
endpoint (str) – endpoint for the new AitoRequest object
query (Optional[Union[Dict, List]]) – an Aito query if applicable, defaults to None
request_obj (AitoRequest) – use an AitoRequest object
raise_for_status (Optional[bool]) – raise RequestError if the request fails instead of returning the error If set to None, value from Client will be used. Defaults to True

Raises:

RequestError – an error occurred during the execution of the request and raise_for_status

async_requests(methods: List[str], endpoints: List[str], queries: List[List | Dict], batch_size: int = 10) → List[BaseResponse]: Deprecated since version 0.4.0.

Use batch_requests() instead

batch_requests(requests: List[AitoRequest], max_concurrent_requests: int = 10) → List[BaseResponse | RequestError]

execute a batch of requests asynchronously

This method is useful when sending a batch of requests, for example, when sending a batch of predict requests.

Parameters:

requests (List[AitoRequest]) – list of request objects
max_concurrent_requests (int) – the number of queries to be sent per batch

Returns:

list of request response or exception if a request did not succeed

Return type:

List[Union[BaseResponse, RequestError]]

Find products that multiple users would most likely buy

>>> from aito.client import MatchRequest
>>> users = ['veronica', 'larry', 'alice']
>>> responses = client.batch_requests([
...     MatchRequest(
...         query = {
...             'from': 'impressions',
...             'where': { 'context.user': usr },
...             'match': 'product'
...         }
...     )
...     for usr in users
... ])
>>> # Print top product for each customer
>>> for idx, usr in enumerate(users):
...     print(f"{usr}: {responses[idx].top_match}")
veronica: {"$p": ..., "category": ..., "id": ..., "name": ..., "price": ..., "tags": ...}
larry: {"$p": ..., "category": ..., "id": ..., "name": ..., "price": ..., "tags": ...}
alice: {"$p": ..., "category": ..., "id": ..., "name": ..., "price": ..., "tags": ...}

async bounded_async_request(semaphore: Semaphore, *args, **kwargs) → BaseResponse | RequestError

bounded concurrent requests with asyncio semaphore

Parameters:

semaphore (asyncio.Semaphore) – asyncio Semaphore
args – async_request() positional arguments
kwargs – async_request() keyword arguments

Returns:

tuple of request status code and request json content

Return type:

Tuple[int, Union[Dict, List]]

property headers

the headers that will be used to send a request to the Aito instance

Return type:: Dict

property is_multitenant: bool

Check if the client is connected to a multitenant instance.

Multitenant URLs have the format: https://shared.aito.ai/db/{database_name}

Returns:: True if connected to a multitenant instance
Return type:: bool

make a request to an Aito API endpoint The client returns a JSON response if the request succeed and a RequestError if the request fails

Parameters:

method (str) – method for the new AitoRequest object
endpoint (str) – endpoint for the new AitoRequest object
query (Optional[Union[Dict, List]]) – an Aito query if applicable, defaults to None
request_obj (AitoRequest) – use an AitoRequest object
raise_for_status (Optional[bool]) – raise RequestError if the request fails instead of returning the error If set to None, value from Client will be used. Defaults to True

Raises:

RequestError – an error occurred during the execution of the request and raise_for_status

Returns:

a response object or RequestError if an error occurred and not raise_for_status

Return type:

Union[BaseResponse, RequestError]

Simple request to get the schema of a table:

>>> res = client.request(method="GET", endpoint="/api/v1/schema/impressions")
>>> print(res.to_json_string(indent=2, sort_keys=True))
{
  "columns": {
    "context": {
      "link": "contexts.id",
      "nullable": false,
      "type": "String"
    },
    "product": {
      "link": "products.id",
      "nullable": false,
      "type": "String"
    },
    "purchase": {
      "nullable": false,
      "type": "Boolean"
    }
  },
  "type": "table"
}

Sends a PREDICT query:

>>> from aito.client import PredictRequest
>>> res = client.request(request_obj=PredictRequest(
...    query={
...        "from": "impressions",
...        "where": { "context": "veronica" },
...        "predict": "product.name"
...    }
... ))
>>> print(res.top_prediction)
{"$p": ..., "$value": ...}

Returns an error when make a request to an incorrect path:

>>> client.request(method="GET", endpoint="api/v1/incorrect-path")
Traceback (most recent call last):
   ...
ValueError: invalid endpoint 'api/v1/incorrect-path' for BaseRequest