class PodClient[source]

PodClient(url='http://localhost:3030', version='v4', database_key=None, owner_key=None, auth_json=None, verbose=False, register_base_schema=True)

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. Note that this will create a new database for your every time you create a PodClient, if you want to access the same database with multiple PodClients, you have to set the same keys When you are using the app, setting the keys in the pod, and passing them when calling an integrator is handled for you by the app itself.

client = PodClient()
success = client.test_connection()
assert success
Succesfully connected to pod

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)
assert succes

We can now create our item. As a side-effect, our item will be assigned an id.

email_item = EmailMessage.from_data(content="example content field")

We can easily define our own types, and use them in the pod.

class Dog[source]

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.

dog = Dog("", 0, True, 0.)
dog2 = Dog(name="bob", age=3, weight=33.2)
dog_from_db = client.get(
assert == "bob"
assert dog_from_db.age == 3
assert dog_from_db.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)
assert succes
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)
assert item_succes
assert edge_succes
[{'item': Person (#15bc9019cf5c6727a71d0cf789156b4d), '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( 
assert isinstance(email_from_db.sender[0], Person)

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")
assert client.create(person_item)
person_from_db = client.get(, expanded=False)
assert person_from_db is not None
assert person_from_db == person_item
assert is not None

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

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

When we don't know the ids of the items we want to fetch, we can also search by property. We can use this for instance when we want to query all items from a particular type to perform some indexing on. We can get all Person Items from the db by:

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
# To test search with and without include_edges, make a fresh client (no local_db)
new_client = PodClient(owner_key=client.owner_key, database_key=client.database_key)

all_people ={"type": "Person"}, include_edges=False)
assert all([isinstance(p, Person) for p in all_people]) and len(all_people) > 0
assert all([len(p.account)==0 for p in all_people])
all_people ={"type": "Person"}, include_edges=True)
assert all([isinstance(p, Person) for p in all_people]) and len(all_people) > 0
assert any([len(p.account) for p in all_people])

Search last added items

person_item2 = Person.from_data(firstName="Last Person")
assert client.search_last_added(type="Person").firstName == "Last 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, the PodClient has a file api. The file 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.

from import *
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(x)
assert succes
assert succes2
data = client.get_file(file.sha256)
arr = np.frombuffer(data, dtype=np.uint8)
assert (arr.reshape(640,640) == x).all()

Photo API

For photos we do this automatically using PodClient.create on a Photo and PodClient.get_photo:

x = np.random.randint(0, 255+1, size=(640, 640), dtype=np.uint8)
photo = Photo.from_np(x)
succes = client.add_to_schema(Photo.from_np(x))
BULK: Writing 2/2 items/edges
Completed Bulk action, written 2 items/edges
res = client.get_photo(, size=640)
assert ( == x).all()

Some photos come as bytes, we can use Iphoto.from_bytes to initialize those

_bytes = 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(_bytes)
assert client.create_photo(photo)
BULK: Writing 2/2 items/edges
Completed Bulk action, written 2 items/edges

We create a new client to prevent caching

new_client = PodClient(database_key=client.database_key, owner_key=client.owner_key)
res = new_client.get_photo(, size=225) ==
[File (#f97f96d594a36ba33bbece3dba8a62ac)]

Bulk API

Currently, an api call takes a lot of time (~0.1 second). If you want to write many data items to the pod, you can use the bulk api for efficiency reasons. Currently, only creating Items and Edges is supported, support for updating and deletion will follow.

dogs = [Dog(name=f"dog number {i} " +"a" * 10000) for i in range(500)]
person = Person(firstName="Alice")
edge1 = Edge(dogs[0], person, "label")
edge2 = Edge(dogs[1], person, "label")

client.bulk_action(create_items=dogs + [person], create_edges=[edge1,edge2])

dogs_with_edge = [item for item in{"type": "Dog"}) if"dog number 0")
                  or"dog number 1 ")]

assert len(dogs_with_edge) == 2
for d in dogs_with_edge:
    assert len(d.label) > 0
BULK: Writing 495/503 items/edges
BULK: Writing 503/503 items/edges
Completed Bulk action, written 503 items/edges
# Change person name, delete first dog :(
person.firstName = "Bob"
to_delete = [dogs[0]]
to_update = [person]

client.bulk_action(delete_items=to_delete, update_items=to_update)
dogs_with_edge = [
    item for item in{"type": "Dog"}) if"dog number 0") or"dog number 1 ")

assert len(dogs_with_edge) == 1

dog = dogs_with_edge[0]
assert dog.label[0].firstName == "Bob"
BULK: Writing 2/2 items/edges
Completed Bulk action, written 2 items/edges
if False:
    to = ""
    client.send_email(to=to, subject="test", body="test2")