--- title: Title keywords: fastai sidebar: home_sidebar nb_path: "nbs/pod.api.ipynb" ---
{% raw %}
{% endraw %} {% raw %}
{% endraw %} {% raw %}
{% endraw %} {% raw %}

class PodError[source]

PodError(status=None, message=None, **kwargs) :: Exception

Common base class for all non-exit exceptions.
{% endraw %} {% raw %}
{% endraw %} {% raw %}

class PodAPI[source]

PodAPI(database_key:str, owner_key:str, url:str='http://localhost:3030', version:str='v4', auth_json:dict=None, verbose:bool=True)

{% endraw %} {% raw %}
{% endraw %}

Test setup

{% raw %}
from pymemri.pod.client import PodClient
from pymemri.data.schema import Account, Person, Relationship
from pymemri.data.itembase import Edge

client = PodClient()
api = PodAPI(database_key=client.database_key, owner_key=client.owner_key)
{% endraw %} {% raw %}
person = Person(displayName="Alice")
accounts = [
    Account(identifier="Alice", service="whatsapp"),
    Account(identifier="Alice", service="instagram"),
    Account(identifier="Alice", service="gmail")
]

edges = [
    Edge(account, person, "owner") for account in accounts
]

client.add_to_schema(Account, Person, Relationship)
client.bulk_action(create_items = [person] + accounts, create_edges=edges)

# Create data for search
search_accounts = [Account(identifier=str(i), service="search") for i in range(100)]
client.bulk_action(create_items=search_accounts)
BULK: Writing 7/7 items/edges
Completed Bulk action, written 7 items/edges
BULK: Writing 100/100 items/edges
Completed Bulk action, written 100 items/edges
True
{% endraw %}

Tests

{% raw %}
version = api.pod_version

assert "cargo" in version
{% endraw %}

Get, Create

{% raw %}
person = api.get_item(person.id)

assert len(person)
assert person[0]["displayName"] == 'Alice'
{% endraw %} {% raw %}
new_relation = {"label": "friendship", "proximityValue": 5, "type": "Relationship"}
new_id = api.create_item(new_relation)

assert len(new_id) and isinstance(new_id, str)
{% endraw %} {% raw %}
new_relation_updated = {"proximityValue": 6, "id": new_id, "label": "attachment"}
api.update_item(new_relation_updated)

new_relation = api.get_item(new_id)
assert new_relation[0]["proximityValue"] == 6
assert new_relation[0]["label"] == "attachment"
{% endraw %}

Edges

{% raw %}
edges = api.get_edges(person[0]["id"], direction="Incoming")
assert len(edges) == 3
edge_items = [edge["item"] for edge in edges]
assert set(item["id"] for item in edge_items) == set(account.id for account in accounts)

# Get incoming edges
for account in accounts:
    edges = api.get_edges(account.id, direction="Outgoing")
    assert len(edges) == 1
    item = edges[0]["item"]
    assert item["id"] == person[0]["id"]
{% endraw %} {% raw %}
src_id = person[0]["id"]
tgt_id = new_relation[0]["id"]
edge = {"_source": src_id, "_target": tgt_id, "_name": "relationship"}

edge_id = api.create_edge(edge)

edges = api.get_edges(person[0]["id"], direction="Outgoing")
assert len(edges) == 1
assert edges[0]["name"] == "relationship"
assert edges[0]["item"]["id"] == tgt_id
{% endraw %}

Delete

{% raw %}
api.delete_item(new_relation[0]["id"])

item = api.get_item(new_relation[0]["id"])
assert item[0]["deleted"] == True
{% endraw %}
{% raw %}
results = api.search({"type": "Account", "service": "search"})

assert len(results) == 100
assert set([result["id"] for result in results]) == set([account.id for account in search_accounts])
{% endraw %} {% raw %}
paginator = api.search_paginate({"type": "Account", "service": "search"}, limit=10, even_page_size=False)

results = []
for page in paginator:
    assert len(page)
    results.extend(page)
    
assert len(results) == 100, len(results)
assert set([result["id"] for result in results]) == set([account.id for account in search_accounts])
identifiers = [result["identifier"] for result in results]

# Test paginated search with even page sizes
paginator = api.search_paginate({"type": "Account", "service": "search"}, limit=10, even_page_size=True)

results = []
for page in paginator:
    assert len(page) == 10
    results.extend(page)
    
assert len(results) == 100, len(results)
assert set([result["id"] for result in results]) == set([account.id for account in search_accounts])
identifiers2 = [result["identifier"] for result in results]

assert identifiers == identifiers2
{% endraw %} {% raw %}
result = api.search({"type": "Account", "service": "wrong_service"})
assert result == []
{% endraw %} {% raw %}
result = api.search_paginate({"type": "Account", "service": "wrong_service"}, limit=10)

assert isinstance(result, Generator)
try:
    page = next(result)
    assert False, "StopIteration expected for empty generator"
except StopIteration:
    pass
{% endraw %}

Bulk

{% raw %}
bulk_accounts = [
    {"type": "Account", "identifier": str(i), "service": "bulk"} for i in range(10)
]

result = api.bulk(create_items=bulk_accounts)
assert len(result["createItems"]) == 10
{% endraw %}

Error handling

{% raw %}
new_person = {"displayName": "clara", "type": "Person"}
new_id = api.create_item(new_person)

new_person = {"displayName": "clara", "type": "Person", "id": new_id}
try:
    api.create_item(new_person)
    assert False, "PodError 500 expected"
except PodError as e:
    print("Expected error raised:")
    print(e)
    pass
Expected error raised:
500 Failure: Database rusqlite error UNIQUE constraint failed: items.id, Failed to execute insert_item with parameters
{% endraw %}