Python Directory Client

You can initialize a directory client as follows:

from aserto.client.directory.v3 import Directory

ds = Directory(api_key="my_api_key", tenant_id="1234", address="localhost:9292")

• address (optional): hostname:port of directory service.hostname:port.
• api_key (optional): API key for directory service (required if using hosted directory).
• tenant_id (optional): Aserto tenant ID (required if using hosted directory).
• ca_cert_path (optional): Path to the grpc service certificate when connecting to local topaz instance.
• reader_address (optional): hostname:port of directory reader service (required if using a diferent address for directory reader service than the address).
• writer_address (optional): hostname:port of directory writer service (required if using a diferent address for directory writer service than the address).
• importer_address (optional): hostname:port of directory importer service (required if using a diferent address for directory importer service than the address).
• exporter_address (optional): hostname:port of directory exporter service (required if using a diferent address for directory exporter service than the address).
• model_address (optional): hostname:port of directory model service (required if using a diferent address for directory model service than the address).

get_object

Get a directory object instance with the type and the id, optionally with the object's relations.

# without relations:
user = ds.get_object(object_type="user", object_id="euang@acmecorp.com")

# with relations:
page = PaginationRequest(size=10)
while True:
    resp = ds.get_object(object_type="user", object_id="euang@acmecorp.com", with_relations=True, page=page)
    user = resp.result               # The returned object.
    relations_page = resp.relations  # A page of relations.

    if not resp.page.next_token:
        # we've reached the last page.
        break

    # request the next page.
    page.token = resp.page.next_token

get_object_many

Similar to get_object but can retrieve multiple object instances in a single request.

objects = ds.get_object_many(
    [
        ObjectIdentifier(type="user", id="euan@acmecorp.com"),
        ObjectIdentifier(type="group", id="marketing"),
    ]
)

get_objects

Get object instances with an object type type pagination info (page size and pagination token).

from aserto.client.directory.v3 import PaginationRequest

users = ds.get_objects(object_type="user", page=PaginationRequest(size=10))

set_object

Create an object instance with the specified properties. If an etag is specified and is different from the current object's etag, the call raises an ETagMismatchError.

# pass object fields as arguments:
user = ds.set_object(
    object_type="user",
    object_id="new-user@acmecorp.com",
    display_name="John Doe",
    "properties": {"active": True, "department": "Engineering"},
}

# set_object can also take an Object parameter:
user.display_name = "Jane Doe"
user.properties["title"] = "Senior Engineer"
updated_user = ds.set_object(object=user)

delete_object

Delete an object instance and optionally its relations, using its type and id:

# delete an object
ds.delete_object(object_type="user", object_id="test-object")

# delete an object and all its relations
ds.delete_object(object_type="user", object_id="test-object", with_relations=True)

get_relation

Retrieve a single relation from the directory or raise a NotFoundError if no matching relation exists.

# get the manager of euang@acmecorp.com:
relation = ds.get_relation(
    object_type="user",
    relation="manager",
    subject_type="user",
    subject_id="euang@acmecorp.com",
)

assert relation.object_id

# include the relation's object and subject in the response:
response = ds.get_relation(
    object_type="user",
    relation="manager",
    subject_type="user",
    subject_id="euang@acmecorp.com",
    with_relations=True,
)

assert response.relation.object_id
assert response.subject.display_name == "Euan Garden"
assert response.object.properties["department"] == "Sales"
#

get_relations

Searches the directory for relations matching the specified criteria, optionally including the object and subject of each returned relation.

# find all groups a user is a member of:
page = PaginationRequest(size=10)

while True:
    response = ds.get_relations(
        object_type="group",
        "relation"="member",
        "subject_type": "user",
        "subject_id": "euang@acmecorp.com",
        with_objects=True,
        page=page,
    )

    if not response.page.next_token:
        break

    page.token = response.page.next_token

set_relation

Create a new relation.

ds.set_relation(
    object_type="group",
    object_id="admin",
    relation="member",
    subject_type="user",
    subject_id="euang@acmecorp.com",
)

delete_relation

Delete a relation.

ds.delete_relation(
    object_type="group",
    object_id="admin",
    relation="member",
    subject_type="user",
    subject_id="euang@acmecorp.com",
)

check

Check if a subject has a given relation or permission on an object.

allowed = ds.check(
    object_type="folder",
    object_id="/path/to/folder",
    relation="can_delete",
    subject_type="user",
    subject_id="euang@acmecorp.com",
)

get_manifest

Download the directory manifest.

manifest = ds.get_manifest()

print(manifest.body)    # yaml manifest

# conditionally get the manifest if its etag has changed
new_manifest = ds.get_manifest(etag=manifest.etag)

assert new_manifest is None   # the manifest hasn't changed

set_manifest

Upload a new directory manifest.

with open("manifest.yaml", "rb") as f:
    manifest = f.read()

ds.set_manifest(manifest)

import_data

Bulk-insert objects and/or relations to the directory. Returns a summary of the number of objects/relations affected.

# import an object and a relation.
data = [
    Object(type="user", id="test@acmecorp.com"),
    Relation(
        object_type="user",
        object_id="euang@acmecorp.com",
        relation="manager",
        subject_type="user",
        subject_id="test@acmecorp.com",
    ),
]

response = ds.import_data(data)

assert response.objects.set == 1
assert response.object.error == 0
assert response.relations.set == 1
assert response.relations.error == 0

export_data

Bulk-retrieve objects and/or relations from the directory.

from aserto.client.directory.v3 import ExportOption, Object, Relation

# export all objects and relations
for item in ds.export(ExportOption.OPTION_DATA):
    if isinstance(item, Object):
        print("object:", item)
    elif isinstance(item, Relation):
        print("relation:", item)

Async Directory Client

You can initialize an asynchronous directory client as follows:

from aserto.client.directory.v3.aio import Directory

ds = Directory(api_key="my_api_key", tenant_id="1234", address="localhost:9292")

The methods on the async directory have the same signatures as their synchronous counterparts.

Directory v2 client

To interact with older instances of the directory service, a v2 client is available with limited functionality. The v2 client doesn't support get_manifest/set_manifest, and import_data/export_data.

from aserto.client.directory.v2 import Directory
ds = Directory(api_key="my_api_key", tenant_id="1234", address="localhost:9292")