The Pod Client

class PodClient

PodClient(url='http://localhost:3030', version='v4', database_key=None, owner_key=None, auth_json=None, register_base_schema=True, verbose=False, default_priority=<Priority.local: 'local'>)

Pymemri communicates with the pod via the PodClient. The PodClient requires you to provide a database key and an owner key. During development, you don't have to worry about these keys, you can just omit the keys when initializing the PodClient, which creates a new user by defining random keys.

If you want to use the same keys for different PodClient instances, you can store a random key pair locally with the store_keys CLI, and create a new client with PodClient.from_local_keys()). When you are using the app, setting the keys in the pod, and passing them when calling a plugin is handled for you by the app itself.

client = PodClient()
client.registered_classes["Photo"]
pymemri.data.photo.Photo

Creating Items and Edges

Now that we have access to the pod, we can create items here and upload them to the pod. All items are defined in the schema of the pod. To create an item in the pod, you have to add the schema first. Schemas can be added as follows

from pymemri.data.schema import EmailMessage, Address, PhoneNumber
succes = client.add_to_schema(EmailMessage, Address, PhoneNumber)

We can now create an item with one of the above item definitions. As a side-effect, our item will be assigned an id.

email_item = EmailMessage.from_data(content="example content field")
client.create(email_item)
print(email_item.id)
0b48d1a5acef4a2085d62222ebb90d33

The types of items in the pod are not limited to definitions to the pymemri schema. We can easily define our own types, or overwrite existing item definitions with the same add_to_schema method.

Note that all keyword arguments need to be added to the properties class variable to let the pod know what the properties of our item are. Additionally, properties in the Pod are statically typed, and have to be inferred from type the annotations of our __init__ method.

class Dog

Dog(name:str=None, age:int=None, bites:bool=False, weight:float=None, friend:EdgeList[Person]=None, **kwargs) :: Item

Item is the baseclass for all of the data classes.
client.add_to_schema(Dog)
dog2 = Dog(name="bob", age=3, weight=33.2)
client.create(dog2);

We can connect items using edges. Let's create another item, a person, and connect the email and the person.

person_item = Person.from_data(firstName="Alice", lastName="X")
succes = client.add_to_schema(person_item)
client.local_db.contains(email_item)
True
client.local_db.nodes
{'0b48d1a5acef4a2085d62222ebb90d33': EmailMessage (#0b48d1a5acef4a2085d62222ebb90d33),
 'ba13433aefc74095a498873fc6cfcc09': Dog (#ba13433aefc74095a498873fc6cfcc09)}
person_item = Account.from_data(displayName="Alice", handle="@alite")
item_succes = client.create(person_item)
edge = Edge(email_item, person_item, "sender")
edge_succes = client.create_edge(edge)
print(client.get_edges(email_item.id))
[{'item': Account (#0edbfb35dee84493b0ea96aecbe21048), 'name': 'sender'}]

If we use the normal client.get (without expanded=False), we also get items directly connected to the Item.

email_from_db = client.get(email_item.id) 
print(email_from_db.sender)
[Account (#0edbfb35dee84493b0ea96aecbe21048)]

Fetching and updating Items

Normal Items

We can use the client to fetch data from the database. This is in particular useful for indexers, which often use data in the database as input for their models. The simplest form of querying the database is by querying items in the pod by their id (unique identifier).

person_item = Person.from_data(firstName="Alice")
client.create(person_item)
# Retrieve person from Pod
person_from_db = client.get(person_item.id, expanded=False)
person_item.to_json(), person_item._in_pod
({'id': '767928a456f6496fb317f12578908015',
  'dateCreated': 1656252612451,
  'dateModified': 1656252612451,
  'dateServerModified': 1656252612451,
  'deleted': False,
  'firstName': 'Alice',
  'type': 'Person'},
 True)
for k, v in person_item.__dict__.items():
    if person_from_db.__dict__[k] != v:
        print(k)
        print(v)
        print(person_from_db.__dict__[k])
        print()

Appart from creating, we might want to update existing items:

person_item.lastName = "Awesome"
client.update_item(person_item)
person_from_db = client.get(person_item.id, expanded=False)
print(person_from_db.lastName)
Awesome

Sync

the PodClient can automatically sync your local items with the Pod with the PodClient.sync method. This method has a priority argument to resolve sync conflicts. The available strategies are:

  • "local": Use the local value if a sync conflict is detected
  • "remote": Use the remote value if a sync conflict is detected
  • "newest": Use the last updated value if a sync conflict is detected. Warning: the pod only stores when a full item is updated. This strategy could lead to unexpected behaviour in cases where sync conflicts happen with a high frequency.
  • "error": Throw a ValueError when a sync conflict is detected.

You can simply add new items to the sync with:

account = Account(handle="alice")
person = Person(firstName="Alice")
account.store(client)
person.store(client)
Person (#6b65ed71b33c4267aefad44cebfb7929)

sync also synchronizes edges if both the source and target item are stored for syncing:

account.add_edge("owner", person)
client.sync()
BULK: Writing 9/9 items/edges
Completed Bulk action, written 9 items/edges
True
client.reset_local_db()
account = client.get(account.id)
assert account.owner[0].id == person.id
all_items = setup_sync_test(client)
for id, node in client.local_db.nodes.items():
    print(id, node._in_pod)

for item in all_items: print(item.id, item._in_pod)

BULK: Writing 3/3 items/edges
Completed Bulk action, written 3 items/edges
8d37bb4ec26442eaab65f503a797969b True
6b65ed71b33c4267aefad44cebfb7929 True
9f1486b274b44dca816b3224d83873ca True
33cdc841744e41a88346c881755919e8 True
1d2edc76f9534d689b574f2f5830aaf1 True
02752f907eae4a819891068fd0844d86 False
1695a8fec3cf43918c8e34b463a67c26 False
aba4d415ec9b478094549b7a9b6844d6 False
9f1486b274b44dca816b3224d83873ca True
33cdc841744e41a88346c881755919e8 True
1d2edc76f9534d689b574f2f5830aaf1 True
02752f907eae4a819891068fd0844d86 False
1695a8fec3cf43918c8e34b463a67c26 False
aba4d415ec9b478094549b7a9b6844d6 False

the PodClient can search through the pod with the search or search_paginate methods, which return the results of a search as a list or generator respectively. Search uses the same arguments as the Pod search API, which can be found here.

To display how search works, we first add a few new items

person_item2 = Person.from_data(firstName="Bob")
person_account = Account(service="testService")
client.create(person_item2)
client.create(person_account)
person_item2.add_edge("account", person_account)
client.create_edges(person_item2.get_edges("account"));
BULK: Writing 1/1 items/edges
Completed Bulk action, written 1 items/edges
all_people = client.search({"type": "Person"}, include_edges=True)
print("Number of results:", len(all_people))
Number of results: 4
ids = [person.id for person in all_people]
result = client.search({"ids": ids})
assert [item.id for item in result] == ids

To hande large volumes of Items, the PodClient.search_paginate method can search through the pod and return a generator which yields batches of items. This method uses the same search arguments as the search method:

client.bulk_action(
    create_items=[
        Account(identifier=str(i), service="paginate_test") for i in range(100)
    ]
)
generator = client.search_paginate({"type": "Account", "service": "paginate_test"}, limit=10)
for page in generator:
    # process accounts
    pass
BULK: Writing 100/100 items/edges
Completed Bulk action, written 100 items/edges

Search last added items

person_item2 = Person.from_data(firstName="Last Person")
client.create(person_item2)
last_added = client.search_last_added(type="Person")

In the near future, Pod will support searching by user defined properties as well. This will allow for the following. warning, this is currently not supported

client.search_last_added(type="Person", with_prop="ImportedBy", with_val="EmailImporter")

Uploading & downloading files

File API

To work with files like Photos or Videos, the PodClient has a separate file api. This api works by posting a blob to the upload_file endpoint, and creating an Item with a property with the same sha256 as the sha used in the endpoint.

For example, we can upload a photo with the file API as follows:

x = np.random.randint(0, 255+1, size=(640, 640), dtype=np.uint8)
photo = Photo.from_np(x)
file = photo.file[0]
succes = client.create(file)
succes2 = client._upload_image(photo.data)
cb = lambda res: print(res)
succes3 = client._upload_image(photo.data, asyncFlag=True, callback=cb)

Photo API

The PodClient implements an easier API for photos separately, which uses the same file API under the hood

print(client.registered_classes["Photo"])
# client.add_to_schema(Photo)
x = np.random.randint(0, 255+1, size=(640, 640), dtype=np.uint8)
photo = Photo.from_np(x)
client.create_photo(photo);
photo.file
<class 'pymemri.data.photo.Photo'>
BULK: Writing 3/3 items/edges
Completed Bulk action, written 3 items/edges
[File (#ab8fb6809b07403599c3abd579ac85cf)]

Some photos come as bytes, for example when downloading them from a third party service. We can use photo.from_bytes to initialize these photos:

byte_photo = b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\xe1\x00\x00\x00\xe1\x08\x03\x00\x00\x00\tm"H\x00\x00\x003PLTE\x04\x02\x04\x00\x00\x00\xa0\xa0\xa0\xa3\xa3\xa3\xaa\xaa\xaa\xb4\xb4\xb4\xbd\xbe\xbd\xbb\xbc\xbb\xde\xde\xde\x9b\x9a\x9b\xfe\xfe\xfe\xf2\xf3\xf2\xe5\xe6\xe5\xd8\xd9\xd8\xd1\xd1\xd1\xc9\xca\xc9\xae\xae\xae\x80k\x98\xfc\x00\x00\x01TIDATx\x9c\xed\xdd;r\xc2P\x00\x04A!\x90\x84\xfc\x01\xee\x7fZ\x138\xb1\x13S\xceF\xaf\xfb\x06\x93o\xd5No\xef\x1f\x9f\xb7\xfb}]\xd7my\xba|;\xff4\xff\xdf\xf9O\x97W<\x96W\xac\xbfm\xd7i9\x1d\xdb\xfe,\x9c\x8e\xec4+\xac{\x16^\x14\xb6)\xecS\xd8\xa7\xb0Oa\x9f\xc2\xbe!\n\xcf\n\xdb\x14\xf6)\xecS\xd8\xa7\xb0Oa\x9f\xc2>\x85}\n\xfb\x14\xf6)\xecS\xd8\xa7\xb0Oa\x9f\xc2>\x85}C\x14\xce\n\xdb\x14\xf6)\xecS\xd8\xa7\xb0Oa\x9f\xc2>\x85}\n\xfb\x14\xf6)\xecS\xd8\xa7\xb0Oa\x9f\xc2>\x85}\n\xfb\x14\xf6)\xecS\xd87\xc4bHa\x9c\xc2>\x85}\n\xfb\x14\xf6)\xecS\xd8\xa7\xb0Oa\x9f\xc2>\x85}\n\xfb\x14\xf6)\xecS\xd8\xa7\xb0Oa\x9f\xc2>\x85}\n\xfb\x86xaQ\x18\xa7\xb0Oa\x9f\xc2>\x85}\n\xfb\x14\xf6)\xecS\xd8\xa7\xb0Oa\x9f\xc2>\x85}\n\xfb\x14\xf6)\xecS\xd87D\xe1\xe3\xf0\x85\x8b\xc26\x85}\n\xfb\x14\xf6)\xecS\xd8\xa7\xb0Oa\x9f\xc2>\x85}\n\xfb\x14\xf6)\xecS\xd8\xa7\xb0Oa\x9f\xc2>\x85}C\x14\xae\n\xdb\x14\xf6)\xecS\xd8\xa7\xb0Oa\x9f\xc2>\x85}C\x14n\xa7c\xdb\xa7\xeb>\x1f\xd9~\xfb\x02\xee\x7f\r\xe5\xe1h\x04"\x00\x00\x00\x00IEND\xaeB`\x82'
photo = Photo.from_bytes(byte_photo)
client.create_photo(photo);
BULK: Writing 3/3 items/edges
Completed Bulk action, written 3 items/edges

Bulk API

Adding each item separately to the pod with the create method can take a lot of time. For this reason, using the bulk API is faster and more convenient in most cases. Here we show creating items and edges, updating and deleting is also possible.

dogs = [Dog(name=f"dog number {i}") for i in range(100)]
person = Person(firstName="Alice")
edge1 = Edge(dogs[0], person, "friend")
edge2 = Edge(dogs[1], person, "friend")
# Simultaneously add the dogs, person, and edges with the bulk API
success = client.bulk_action(create_items=dogs + [person], create_edges=[edge1, edge2])
BULK: Writing 103/103 items/edges
Completed Bulk action, written 103 items/edges

Create items that do not exist in the Pod

The PodClient can deduplicate items with the same externalId with the create_if_external_id_not_exists method.

person_item = Person(firstName="Eve", externalId="gmail_1")
person_item2 = Person(firstName="Eve2", externalId="gmail_1")
client.create_if_external_id_not_exists(person_item)
client.create_if_external_id_not_exists(person_item2)
existing = client.search({"externalId": "gmail_1"})