ahasura 1.4.2
Async and sync Hasura client
License: MIT
Requires Python: >=3.8,<4.0
Keywords: async,sync,ahasura,hasura,client

# ahasura

Async and sync [Hasura](https://hasura.io/) client.

## Install

ahasura is available on PyPi:

```
pip install ahasura
# Or
poetry add ahasura
```

## Quick example

```
from ahasura import ADMIN, Hasura
hasura = Hasura("http://localhost:8080", admin_secret="fake secret")

data = hasura(
    """
    query($id: uuid!) {
        item_by_pk(id: $id) {
            name
        }
    }
    """,
    auth=ADMIN,
    id="00000000-0000-0000-0000-000000000001",
)

item = data["item_by_pk"]
assert item["name"] == "Some name"
```

## Create client

* `hasura = Hasura(...)`
* Args:
    * `endpoint: str` - `HASURA_GRAPHQL_ENDPOINT`, without trailing `/` or `/v1/graphql`
    * `admin_secret: Optional[str]` - `HASURA_GRAPHQL_ADMIN_SECRET`, required for `auth=ADMIN` only
    * `timeout: Optional[float] = 10` - Seconds of network inactivity allowed. `None` disables the timeout
* `hasura` client just keeps the configuration above, so you can reuse global client(s)
* Shortcuts:
    * `hasura(...)` is a shortcut for sync GraphQL client: `hasura.gql(...)`
    * You can define a shortcut for Async GraphQL client: `ahasura = hasura.agql`

## Execute GraphQL query

* With shortcuts:
    * Sync: `data = hasura(...)`
    * Async: `data = await ahasura(...)`
* Without shortcuts:
    * Sync: `data = hasura.gql(...)`
    * Async: `data = await hasura.agql(...)`
* Args:
    * `query: str` - GraphQL query, e.g. `query { item { id }}`
    * `auth: str` - Either `ADMIN` or value of `Authorization` header, e.g. `Bearer {JWT}`
    * `headers: Optional[Dict[str, str]]` - Custom headers, if any
    * `**variables` - Variables used in `query`, if any
* Returns: GraphQL response data, e.g. `{"item": [{"id": "..."}]}`
* Raises: `HasuraError` - If JSON response from Hasura contains `errors` key

## Execute SQL query

* Sync: `rows = hasura.sql(...)`
* Async: `rows = await hasura.asql(...)`
* Args:
    * `query: str` - SQL query, e.g. `SELECT "id" FROM "item"`
    * `headers: Optional[Dict[str, str]]` - Custom headers, if any
* Returns:
    * Rows selected by `SELECT` query, e.g. `[{"id": "..."}]`
    * Or `[{"ok": True}]` for non-`SELECT` query
* Raises: `HasuraError` - If JSON response from Hasura contains `error` key

## Auth

* SQL queries are [admin-only](https://hasura.io/docs/latest/graphql/core/api-reference/schema-api/run-sql.html#run-sql)
* GraphQL queries can use both admin and non-admin `auth`
* `auth=ADMIN` is not default, because:
    * `sudo` is not default
    * It's easy to forget to propagate `Authorization` header of the user to Hasura
    * Declarative Hasura permissions are better than checking permissions in Python
    * When we set Hasura permissions, we should test them for each role supported

## How to test

### `test_item.py`

```
from typing import Any, Dict

from ahasura import Hasura, HasuraError
import pytest


def test_reader_reads_item_ok(
    hasura: Hasura,
    reader_auth: str,
    ok_item: Dict[str, Any],
) -> None:
    data = hasura(
        """
        query($id: uuid!) {
            item_by_pk(id: $id) {
                name
            }
        }
        """,
        auth=reader_auth,
        id=ok_item["id"],
    )

    item = data["item_by_pk"]
    assert item["name"] == "Some name"


def test_error(hasura: Hasura, reader_auth: str) -> None:
    with pytest.raises(HasuraError) as error:
        hasura("bad query", auth=reader_auth)

    assert error.value.response == {"errors": [...]}
```

### `conftest.py`

```
from typing import Any, Dict, Generator, List

from ahasura import ADMIN, Hasura
import jwt
import pytest

_TABLE_NAMES = ["item"]


@pytest.fixture(scope="session")
def hasura() -> Hasura:
    return Hasura("http://localhost:8080", admin_secret="fake secret")


@pytest.fixture(scope="session")
def reader_auth() -> str:
    decoded_token = ...
    encoded_token = jwt.encode(decoded_token, "")
    return f"Bearer {encoded_token}"


@pytest.fixture(scope="session")
def test_row_ids() -> List[str]:
    """
    When a test function creates a row in any table,
    it should append this `row["id"]` to `test_row_ids`

    UUIDs are unique across all tables with enough probability
    """
    return []


@pytest.fixture(scope="function")
def ok_item(hasura: Hasura, test_row_ids: List[str]) -> Dict[str, Any]:
    data = hasura(
        """
        mutation($item: item_insert_input!) {
            insert_item_one(object: $item) {
                id
                name
            }
        }
        """,
        auth=ADMIN,
        name="Some name",
    )
    item: Dict[str, Any] = data["insert_item_one"]
    test_row_ids.append(item["id"])
    return item


@pytest.fixture(scope="function", autouse=True)
def cleanup(hasura: Hasura, test_row_ids: List[str]) -> Generator[None, None, None]:
    """
    When the test function ends,
    this autouse fixture deletes all test rows from all tables
    """
    yield

    if test_row_ids:
        for table_name in _TABLE_NAMES:
            hasura(
                """
                mutation($ids: [uuid!]!) {
                    delete_{table_name}(where: {id: {_in: $ids}}) {
                        affected_rows
                    }
                }
                """.replace(
                    "{table_name}", table_name
                ),
                auth=ADMIN,
                ids=test_row_ids,
            )
        test_row_ids.clear()
```