API filtering does not work

[thielem]

I am attempting to use the API with Python. Unfortunately, the filter function does not work. I thought this was an issue with lacking JSON support, but it still persists in version 3.5

Code: Select all

def get_user_info(user_id:int):
    
    field_keys = {
        "8": "user",
        "271": "f1",
        "867": "f2",
        "956": "f3",
        "282": "f4",
        "333": "f5",
        "334": "f6",
        "1039": "f7",
        "574": "f8",
        "1057": "f9",
        "1058": "f10"
    }
    
    params = {
        'action': 'select',
        'entity_id': 1,
        'select_fields': ", ".join(field_keys.keys()),
        'filters': {
            'id': user_id
        }
    }
    
    params = {**API_CREDENTIALS, **params}
    r = requests.post(API_ENDPOINT, data=params, headers=cf_credentials)
    
    res_json = json.loads(r.text)
    if res_json["status"]=="success":
        return res_json['data'][0]
    return None
    

This returns all users in res_json["data"] and the filter has no effect.

[support]

Checked in PHP and filters by ID works ok.

[support]

See my request

Code: Select all

key=bWz3m64fePZSQMjfV2IzzlFPgGXJkRlrWZLb91AK&username=admin&password=admin&action=select&entity_id=21&select_fields=158%2C157%2C255%2C232%2C156%2C+258%2C+158%2C344%2C162&limit=0&filters%5Bid%5D=8

[support]

Also in 3.5 you can send json

Code: Select all

{
    "key": "bWz3m64fePZSQMjfV2IzzlFPgGXJkRlrWZLb91AK",
    "username": "admin",
    "password": "admin",
    "action": "select",
    "entity_id": 21,
    "select_fields": "158,157,255,232,156, 258, 158,344,162",
    "limit": 0,
    "filters": {
        "id": 8
    }
}

[thielem]

Replacing
requests.post(API_ENDPOINT, data=params, headers=cf_credentials)
with
requests.post(API_ENDPOINT, json=params, headers=cf_credentials)

fixed it. Thanks for the JSON update!

[as100]

Will I find somewhere a description of how to use API from JSON? I am interested in overlapping the data in both directions.

[thielem]

Here are a few examples if you want to use it via Python.
The syntax can be inferred from the existing documentation.

Code: Select all

class ApiConnector():
    def __init__(self):
        """
        Initialize the Rukovoditel API connector.
        """
        self.url = cred.rk_api_url
        self.credentials = cred.rk_api_credentials
        if cred.api_external:
            self.headers = cred.cf_credentials
        else:
            self.headers = None
    
    def post(self, params):
        """
        Send a POST request to the Rukovoditel API.
        Args:
            params (dict): The parameters to send with the request.
        Returns:
            requests.Response: The unprocessed response from the Rukovoditel API.
        """
        post_params = {**self.credentials, **params}
        r = requests.post(self.url, json=post_params, headers=self.headers, verify=False)
        return r
    
    def fetchone(self, entity_id: int, items_id: int, select_fields=""):
        """
        Fetch one item from the Rukovoditel API.
        Args:
            entity_id (int): The entity ID.
            items_id (int): The item ID.
            select_fields (str): The fields to select.
        Returns:
            dict: The item. Or None if no item found.
        """
        params = {
            "action": "select",
            "entity_id": entity_id,
            "select_fields": select_fields,
            "filters": {
                "id": items_id
            }
        }
        post_params = {**self.credentials, **params}
        r = requests.post(self.url, json=post_params, headers=self.headers, verify=False)
        
        if r.status_code != 200: raise Exception(f"Request failed with status code {r.status_code}\n{r.text}")

        try:
            res_json = json.loads(r.text)
        except json.JSONDecodeError:
            raise Exception(f"Failed to decode JSON from response\n{r.text}")
        if len(res_json["data"])==0:
            return None
        else:
            return res_json["data"][0]
    
    def fetch_by_filter(self, entity_id: int, filters: dict, select_fields: str, limit_1 = True):
        """
        Fetch items from the Rukovoditel API by filter.
        Args:
            entity_id (int): The entity ID.
            filters (dict): The filters to use.
            select_fields (str): The fields to select.
            limit_1 (bool): Whether to limit the result to 1 item.
        Returns:
            dict: The items. Or None if no items found.
        """
        params = {
            "action": "select",
            "entity_id": entity_id,
            "select_fields": select_fields,
            "filters": filters
        }
        post_params = {**self.credentials, **params}
        r = requests.post(self.url, json=post_params, headers=self.headers, verify=False)
        try:
            res_json = json.loads(r.text)
        except json.JSONDecodeError:
            raise Exception(f"Failed to decode JSON from response\n{r.text}")
        if limit_1:
            assert len(res_json["data"]) == 1, f"Expected 1 item, but got {len(res_json['data'])}"
            return res_json["data"][0]
        else:
            return res_json["data"]