--- title: Title keywords: fastai sidebar: home_sidebar nb_path: "nbs/pod.api.ipynb" ---
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)
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)
version = api.pod_version
assert "cargo" in version
person = api.get_item(person.id)
assert len(person)
assert person[0]["displayName"] == 'Alice'
new_relation = {"label": "friendship", "proximityValue": 5, "type": "Relationship"}
new_id = api.create_item(new_relation)
assert len(new_id) and isinstance(new_id, str)
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"
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"]
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
api.delete_item(new_relation[0]["id"])
item = api.get_item(new_relation[0]["id"])
assert item[0]["deleted"] == True
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])
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
result = api.search({"type": "Account", "service": "wrong_service"})
assert result == []
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
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
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