Skip to content

BQL

ptnad.api.bql.BQLAPI 

BQLAPI(client)

Methods:

Name Description
execute

Execute a BQL query on a specific source.
execute_raw

Execute a BQL query on a specific source and return the full response.
Source code in src/ptnad/api/bql.py 
24
25
def __init__(self, client) -> None:
    self.client = client

execute 

execute(query: str, source: str = '2') -> Any

Execute a BQL query on a specific source.

Args: query (str): The BQL query to execute. source (str): The identifier of the storage to query. Defaults to "2" (live).

Returns: Any: The result of the query execution.

Raises: PTNADAPIError: If there's an error executing the query.

Source code in src/ptnad/api/bql.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def execute(self, query: str, source: str = "2") -> Any:
    """
    Execute a BQL query on a specific source.

    Args:
        query (str): The BQL query to execute.
        source (str): The identifier of the storage to query. Defaults to "2" (live).

    Returns:
        Any: The result of the query execution.

    Raises:
        PTNADAPIError: If there's an error executing the query.

    """
    try:
        response = self._send_query(query, source)
        return response["result"]
    except PTNADAPIError as e:
        e.operation = "execute BQL query"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to execute BQL query: {str(e)}")

execute_raw 

execute_raw(query: str, source: str = '2') -> BQLResponse

Execute a BQL query on a specific source and return the full response.

Args: query (str): The BQL query to execute. source (str): The identifier of the storage to query. Defaults to "2" (live).

Returns: BQLResponse: An object containing the query results, execution time, total hits, and debug info.

Raises: PTNADAPIError: If there's an error executing the query.

Source code in src/ptnad/api/bql.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def execute_raw(self, query: str, source: str = "2") -> BQLResponse:
    """
    Execute a BQL query on a specific source and return the full response.

    Args:
        query (str): The BQL query to execute.
        source (str): The identifier of the storage to query. Defaults to "2" (live).

    Returns:
        BQLResponse: An object containing the query results, execution time, total hits, and debug info.

    Raises:
        PTNADAPIError: If there's an error executing the query.

    """
    try:
        response = self._send_query(query, source)
        return BQLResponse(
            result=response["result"],
            took=response["took"],
            total=response["total"],
            debug=response.get("debug")
        )
    except PTNADAPIError as e:
        e.operation = "execute BQL query"
        raise
    except Exception as e:
        raise PTNADAPIError(f"Failed to execute BQL query: {str(e)}")

ptnad.api.bql.BQLResponse 

BQLResponse(result: Any, took: int, total: int, debug: Dict[str, Any] | None = None)
Source code in src/ptnad/api/bql.py 
 7
 8
 9
10
11
def __init__(self, result: Any, took: int, total: int, debug: Dict[str, Any] | None = None) -> None:
    self.result = result
    self.took = took
    self.total = total
    self.debug = debug