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()

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 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")

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, **kwargs) :: Item

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

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)
{'29c0b7a3f8754d02a7d0e641523c25bb': EmailMessage (#29c0b7a3f8754d02a7d0e641523c25bb),
 '93d7acb23bdc4902bd8e4c44d49a3953': Dog (#93d7acb23bdc4902bd8e4c44d49a3953)}
person_item = Person.from_data(firstName="Alice", lastName="X")
item_succes = client.create(person_item)
edge = Edge(email_item, person_item, "sender")
edge_succes = client.create_edge(edge)
[{'item': Person (#5594a26fa6494f68a939eb610e34e7fe), '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( 
[Person (#5594a26fa6494f68a939eb610e34e7fe)]

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")
# Retrieve person from Pod
person_from_db = client.get(, expanded=False)
person_item.to_json(), person_item._in_pod
({'id': '1bdefab867c74f22ad156eebac482c1c',
  'dateCreated': 1644321461975,
  'dateModified': 1644321461975,
  'dateServerModified': 1644321461975,
  'deleted': False,
  'firstName': 'Alice',
  'type': 'Person'},
for k, v in person_item.__dict__.items():
    if person_from_db.__dict__[k] != v:

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

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


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")
Person (#e340f889884b4e4c91bc6711b38d9652)

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

account.add_edge("owner", person)
BULK: Writing 9/9 items/edges
Completed Bulk action, written 9 items/edges
account = client.get(
assert account.owner[0].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._in_pod)

BULK: Writing 3/3 items/edges
Completed Bulk action, written 3 items/edges
4c2913f892c8463ebdd2ee2a8197cb8e True
e340f889884b4e4c91bc6711b38d9652 True
5411889908c44b4bbebf469308065b66 True
a7d7e0687c36451bb105097e5c8bb304 True
eb4c84d002ac4ffebfae69a9a4597b5e True
a6f801efb04946b4885b9d62fb445a0f False
c8bae9ccff19410099d7e5ed8ae0e954 False
60e6e59dbf104d4b976f68fbbce8d325 False
5411889908c44b4bbebf469308065b66 True
a7d7e0687c36451bb105097e5c8bb304 True
eb4c84d002ac4ffebfae69a9a4597b5e True
a6f801efb04946b4885b9d62fb445a0f False
c8bae9ccff19410099d7e5ed8ae0e954 False
60e6e59dbf104d4b976f68fbbce8d325 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")
person_item2.add_edge("account", person_account)
BULK: Writing 1/1 items/edges
Completed Bulk action, written 1 items/edges
all_people ={"type": "Person"}, include_edges=True)
print("Number of results:", len(all_people))
Number of results: 5
ids = [ for person in all_people]
result ={"ids": ids})
assert [ 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:

        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
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")
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(
cb = lambda res: print(res)
succes3 = client._upload_image(, asyncFlag=True, callback=cb)

Photo API

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

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

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)
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, "label")
edge2 = Edge(dogs[1], person, "label")
# 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")
existing ={"externalId": "gmail_1"})