Officially out now: The TypeDB 3.0 Roadmap

Python driver tutorial

In this tutorial, we’ll build a sample application with the Python driver capable of basic interaction with TypeDB:

  • Connect to a TypeDB server (Core or Cloud),

  • Manage databases, sessions, and transactions,

  • Send different types of queries.

Follow the steps below or see the full source code.

Environment setup

To run this sample application, you’ll need:

  1. TypeDB: either a TypeDB Cloud deployment or a TypeDB Core server. For TypeDB Core installation instructions, see the TypeDB Core installation guide page.

  2. Python and TypeDB Python driver. For the driver installation instructions, see the Python driver page.

Use pip for the Python driver installation:

$ pip install typedb-driver

Imported modules

To be able to use the TypeDB Python driver API in the Sample application, use the following import statements:

from typedb.driver import TypeDB, SessionType, TransactionType, TypeDBOptions, TypeDBCredential
from enum import Enum

Default values

We store default values as constants in the source code:

DB_NAME = "sample_app_db"
SERVER_ADDR = "127.0.0.1:1729"
TYPEDB_EDITION = Edition.Core
CLOUD_USERNAME = "admin"
CLOUD_PASSWORD = "password"

where DB_NAME — the name of the database to use; SERVER_ADDR — address of the TypeDB server to connect to; TYPEDB_EDITION — TypeDB Core or Cloud edition selector; CLOUD_USERNAME/CLOUD_PASSWORD — credentials to connect to TypeDB Cloud.

Program structure

The main workflow of this sample application includes establishing a connection to TypeDB, performing a new database setup, and querying.

def main():
    with connect_to_TypeDB(TYPEDB_EDITION, SERVER_ADDR) as driver:
        if db_setup(driver, DB_NAME, db_reset=False):
            queries(driver, DB_NAME)
        else:
            print("Terminating...")
            exit()

The entire main() function code is executed in the context of the network connection, represented by the driver object that is returned by the function.

TypeDB connection

The connect_to_TypeDB() function takes edition and addr as mandatory parameters.

def connect_to_TypeDB(edition, addr, username=CLOUD_USERNAME, password=CLOUD_PASSWORD):
    if edition is Edition.Core:
        return TypeDB.core_driver(addr)
    if edition is Edition.Cloud:
        credentials = TypeDBCredential(username, password, tls_enabled=True)
        return TypeDB.cloud_driver(addr, credentials)

The edition is expected to be an Enum for selecting a TypeDB edition. Depending on the TypeDB edition selected, this function initializes either a TypeDB Core or TypeDB Cloud connection.

TypeDB Cloud connection requires an object of the TypeDBCredential class that is initialized with a username and password. For our sample application, we have the default credentials for the admin account provided as default values for the function’s optional parameters.

TypeDB Cloud requires the default password for the default admin account to be changed before any other request can be accepted.

Database setup

To set up a TypeDB database, we need to make sure that it exists and has the correct schema and data. First, we check whether a database with the provided name already exists on the server.

If such a database doesn’t exist, we create a new database, define its schema, and load initial data.

To prevent data loss, avoid deleting an existing database without confirmation from a user.

If a database with the specified name already exists, we check whether we need to replace it. To do so, we check the db_reset parameter, and, if it’s False, ask for an input from a user. If any of the two suggesting replacement of the database is acceptable, we replace the database by deleting the existing database and then creating a new one.

As the final step of the database setup, we test it.

def db_setup(driver, db_name, db_reset=False) -> bool:
    print(f"Setting up the database: {db_name}")
    if driver.databases.contains(db_name):
        if db_reset or (input("Found a pre-existing database. Do you want to replace it? (Y/N) ").lower() == "y"):
            if not replace_database(driver, db_name):
                return False
        else:
            print("Reusing an existing database.")
    else:  # No such database found on the server
        if not create_database(driver, db_name):
            print("Failed to create a new database. Terminating...")
            return False
    if driver.databases.contains(db_name):
        with driver.session(db_name, SessionType.DATA) as session:
            return db_check(session)
    else:
        print("Database not found. Terminating...")
        return False

Creating a new database

We create a new database with the specified name (sample_app_db by default) and call functions to define its schema and load initial data.

def create_database(driver, db_name) -> bool:
    print("Creating a new database", end="...")
    driver.databases.create(db_name)
    print("OK")
    with driver.session(db_name, SessionType.SCHEMA) as session:
        db_schema_setup(session)
    with driver.session(db_name, SessionType.DATA) as session:
        db_dataset_setup(session)
    return True

Replacing a database

We delete a database with the specified name (sample_app_db by default) and call a function to create a new one instead:

def replace_database(driver, db_name) -> bool:
    print("Deleting an existing database", end="...")
    driver.databases.get(db_name).delete()  # Delete the database if it exists already
    print("OK")
    if create_database(driver, db_name):
        return True
    else:
        print("Failed to create a new database. Terminating...")
        return False

Defining a schema

We use a Define query to define a schema for the newly created database:

def db_schema_setup(schema_session, schema_file='iam-schema.tql'):
    with open(schema_file, 'r') as data:
        define_query = data.read()
    with schema_session.transaction(TransactionType.WRITE) as tx:
        print("Defining schema", end="...")
        tx.query.define(define_query)
        tx.commit()
        print("OK")

The schema for the sample application is stored in the iam-schema.tql file.

See the full schema
define

credential sub attribute, value string;
full-name sub attribute, value string;
id sub attribute, abstract, value string;
email sub id, value string;
name sub id, value string;
number sub id, value string;
path sub id, value string;
object-type sub attribute, value string;
ownership-type sub attribute, value string;
review-date sub attribute, value datetime;
size-kb sub attribute, value long;
validity sub attribute, value boolean;

access sub relation,
    relates action,
    relates object,
    plays change-request:change,
    plays permission:access;

change-request sub relation,
    relates change,
    relates requestee,
    relates requester;

membership sub relation,
    relates member,
    relates parent;

collection-membership sub membership,
    relates collection as parent;

group-membership sub membership,
    relates group as parent;

set-membership sub membership,
    relates set as parent;

ownership sub relation,
    relates owned,
    relates owner;

group-ownership sub ownership,
    owns ownership-type,
    relates group as owned;

object-ownership sub ownership,
    owns ownership-type,
    relates object as owned;

permission sub relation,
    owns review-date,
    owns validity,
    relates access,
    relates subject;

segregation-policy sub relation,
    owns name,
    relates action,
    plays segregation-violation:policy;

violation sub relation,
    abstract;

segregation-violation sub violation,
    relates object,
    relates policy,
    relates subject;

action sub entity,
    abstract,
    owns name,
    owns object-type,
    plays access:action,
    plays membership:member,
    plays segregation-policy:action;

operation sub action;

operation-set sub action,
    plays set-membership:set;

object sub entity,
    abstract,
    owns object-type,
    plays access:object,
    plays membership:member,
    plays object-ownership:object,
    plays segregation-violation:object;

resource sub object,
    abstract;

file sub resource,
    owns path,
    owns size-kb;

record sub resource,
    owns number;

resource-collection sub object,
    abstract,
    plays collection-membership:collection;

database sub resource-collection,
    owns name;

directory sub resource-collection,
    owns path,
    owns size-kb;

subject sub entity,
    abstract,
    owns credential,
    plays change-request:requestee,
    plays change-request:requester,
    plays membership:member,
    plays ownership:owner,
    plays permission:subject,
    plays segregation-violation:subject;

user sub subject,
    abstract;

person sub user,
    owns email,
    owns full-name;

user-group sub subject,
    abstract,
    plays group-membership:group,
    plays group-ownership:group;

business-unit sub user-group,
    owns name;

user-account sub user-group,
    owns email;

user-role sub user-group,
    owns name;

rule add-view-permission: when {
    $modify isa action, has name "modify_file";
    $view isa action, has name "view_file";
    $ac_modify (object: $obj, action: $modify) isa access;
    $ac_view (object: $obj, action: $view) isa access;
    (subject: $subj, access: $ac_modify) isa permission;
} then {
    (subject: $subj, access: $ac_view) isa permission;
};

We use a session object passed as a parameter to open a transaction. Then we send the contents of the file as a TypeQL Define query and commit the changes made by the transaction.

Loading initial data

With the schema defined, we can load initial data into our database with the Insert query:

def db_dataset_setup(data_session, data_file='iam-data-single-query.tql'):
    with open(data_file, 'r') as data:
        insert_query = data.read()
    with data_session.transaction(TransactionType.WRITE) as tx:
        print("Loading data", end="...")
        tx.query.insert(insert_query)
        tx.commit()
        print("OK")

We read the iam-data-single-query.tql file, send its contents as a single query, and then commit the changes.

See the full Insert query
insert
$p1 isa person,
    has full-name "Masako Holley",
    has email "masako.holley@typedb.com";
$p2 isa person,
    has full-name "Pearle Goodman",
    has email "pearle.goodman@typedb.com";
$p3 isa person,
    has full-name "Kevin Morrison",
    has email "kevin.morrison@typedb.com";
$f1 isa file,
    has path "iopvu.java",
    has size-kb 55;

$modify isa operation, has name "modify_file";
$view isa operation, has name "view_file";

$a1 (object: $f1, action: $modify) isa access;
$a11 (object: $f1, action: $view) isa access;
$permission1 (subject: $p3, access: $a1) isa permission;
$f2 isa file,
    has path "zlckt.ts",
    has size-kb 143;
$a2 (object: $f2, action: $modify) isa access;
$a22 (object: $f2, action: $view) isa access;
$permission2 (subject: $p3, access: $a2) isa permission;
$f3 isa file,
    has path "psukg.java",
    has size-kb 171;
$a3 (object: $f3, action: $modify) isa access;
$a33 (object: $f3, action: $view) isa access;
$permission3 (subject: $p3, access: $a3) isa permission;
$f4 isa file,
    has path "axidw.java",
    has size-kb 212;
$a4 (object: $f4, action: $modify) isa access;
$a44 (object: $f4, action: $view) isa access;
$permission4 (subject: $p3, access: $a4) isa permission;
$f5 isa file,
    has path "lzfkn.java",
    has size-kb 70;
$a5 (object: $f5, action: $modify) isa access;
$a55 (object: $f5, action: $view) isa access;
$permission5 (subject: $p3, access: $a5) isa permission;
$f6 isa file,
    has path "budget_2022-05-01.xlsx",
    has size-kb 758;
$a6 (object: $f6, action: $modify) isa access;
$a66 (object: $f6, action: $view) isa access;
$permission6 (subject: $p3, access: $a6) isa permission;
$permission66 (subject: $p2, access: $a66) isa permission;
$f7 isa file,
    has path "zewhb.java";
$a7 (object: $f7, action: $modify) isa access;
$a77 (object: $f7, action: $view) isa access;
$permission7 (subject: $p3, access: $a7) isa permission;
$permission77 (subject: $p2, access: $a77) isa permission;
$f8 isa file,
    has path "budget_2021-08-01.xlsx",
    has size-kb 1705;
$a8 (object: $f8, action: $modify) isa access;
$a88 (object: $f8, action: $view) isa access;
$permission8 (subject: $p3, access: $a8) isa permission;
$permission88 (subject: $p2, access: $a88) isa permission;
$f9 isa file,
    has path "LICENSE";
$a9 (object: $f9, action: $modify) isa access;
$a99 (object: $f9, action: $view) isa access;
$permission9 (subject: $p3, access: $a9) isa permission;
$permission99 (subject: $p2, access: $a99) isa permission;
$f10 isa file,
    has path "README.md";
$a10 (object: $f10, action: $modify) isa access;
$a100 (object: $f10, action: $view) isa access;
$permission10 (subject: $p3, access: $a10) isa permission;
$permission100 (subject: $p2, access: $a100) isa permission;

Testing a database

With the schema defined and data loaded, we test our database to make sure it’s ready. To test the database, we send a query to count the number of users in the database:

def db_check(data_session) -> bool:
    with data_session.transaction(TransactionType.READ) as tx:
        test_query = "match $u isa user; get $u; count;"
        print("Testing the database", end="...")
        response = tx.query.get_aggregate(test_query)
        result = response.resolve().as_value().as_long()
        if result == 3:
            print("Passed")
            return True
        else:
            print("Failed the test with the result:", result, "\n Expected result: 3.")
            return False

Query examples

After database setup is complete, we proceed with querying our database with different types of queries in the queries() function:

def queries(driver, db_name):
    print("\nRequest 1 of 6: Fetch all users as JSON objects with full names and emails")
    users = fetch_all_users(driver, DB_NAME)
    assert len(users) == 3

    new_name = "Jack Keeper"
    new_email = "jk@typedb.com"
    print(f"\nRequest 2 of 6: Add a new user with the full-name {new_name} and email {new_email}")
    insert_new_user(driver, DB_NAME, new_name, new_email)

    name = "Kevin Morrison"
    print(f"\nRequest 3 of 6: Find all files that the user {name} has access to view (no inference)")
    files = get_files_by_user(driver, DB_NAME, name)
    assert files is not None
    assert len(files) == 0

    print(f"\nRequest 4 of 6: Find all files that the user {name} has access to view (with inference)")
    files = get_files_by_user(driver, DB_NAME, name, inference=True)
    assert files is not None
    assert len(files) == 10

    old_path = 'lzfkn.java'
    new_path = 'lzfkn2.java'
    print(f"\nRequest 5 of 6: Update the path of a file from {old_path} to {new_path}")
    updated_files = update_filepath(driver, DB_NAME, old_path, new_path)
    assert updated_files is not None
    assert len(updated_files) == 1

    path = 'lzfkn2.java'
    print(f"\nRequest 6 of 6: Delete the file with path {path}")
    deleted = delete_file(driver, DB_NAME, path)
    assert deleted

The queries are as follows:

  1. Fetch query — to retrieve information in a JSON format

  2. Insert query — to insert new data into the database

  3. Get query — to retrieve data from the database as stateful objects

  4. Get query with inference — to retrieve data from the database as stateful objects using inference

  5. Update query — to replace data in the database

  6. Delete query — to delete data from the database

Every query is implemented as a function that includes some output of the query response and returns some meaningful data.

Fetch query

The main way to retrieve data from a TypeDB database is to use fetching to get values of attributes, matched by a pattern.

Let’s use a Fetch query to fetch names and emails for all users in the database:

def fetch_all_users(driver, db_name) -> list:
    with driver.session(db_name, SessionType.DATA) as data_session:
        with data_session.transaction(TransactionType.READ) as tx:
            users = list(
                tx.query.fetch("match $u isa user; fetch $u: full-name, email;")
            )
            for i, JSON in enumerate(users, start=0):
                print(f"User #{i + 1} — Full-name:", JSON['u']['full-name'][0]['value'],
                      "E-mail:", JSON['u']['email'][0]['value'])
            return users

We collect response in a list and store it in the users variable that is returned by the function. We iterate through the list and print the results from every JSON in the list.

Since we know that there is only one name and one email every time, we can print only the first element every time. In general case, there can be multiple attributes of the same type owned, so we need to consider iterating through all returned values returned for every attribute in every JSON. That can be done as in the following example:

Printing JSON for any number of attributes fetched
for i, JSON in enumerate(users, start=0):
    print(f"User #{i + 1} —", end="")
    for result in JSON['u']['full-name']:
        print(f" Full-name:", result['value'], end="")
    for result in JSON['u']['email']:
        print(f" E-mail:", result['value'], end="")
    print()

Insert query

Let’s insert a new user with full-name and email attributes to the database.

def insert_new_user(driver, db_name, name, email) -> list:
    with driver.session(db_name, SessionType.DATA) as data_session:
        with data_session.transaction(TransactionType.WRITE) as tx:
            response = list(
                tx.query.insert(
                    f"insert $p isa person, has full-name $fn, has email $e; $fn == '{name}'; $e == '{email}';")
            )
            tx.commit()
            for i, concept_map in enumerate(response, start=1):
                name = concept_map.get("fn").as_attribute().get_value()
                email = concept_map.get("e").as_attribute().get_value()
                print("Added new user. Name: " + name + ", E-mail:" + email)
            return response

We collect the response of the Insert query as a list and use it to print the inserted data after we commit the transaction.

We iterate through the response, retrieve attribute values and print them for every return result. Since the Insert query has no match clause, the insert clause is executed exactly once. But the Insert query always returns a list of ConceptMap objects, where every ConceptMap represents an inserted result.

Get query

Let’s retrieve all files available for a user with a get_files_by_user() function. It can be used with or without inference enabled.

def get_files_by_user(driver, db_name, name, inference=False):
    options = TypeDBOptions(infer=inference)
    with driver.session(db_name, SessionType.DATA) as data_session:
        with data_session.transaction(TransactionType.READ, options) as tx:
            users = list(
                tx.query.get(f"match $u isa user, has full-name '{name}'; get;")
            )
            if len(users) > 1:
                print("Error: Found more than one user with that name.")
                return None
            elif len(users) == 1:
                response = list(
                    tx.query.get(f"""
                                    match
                                    $fn == '{name}';
                                    $u isa user, has full-name $fn;
                                    $p($u, $pa) isa permission;
                                    $o isa object, has path $fp;
                                    $pa($o, $va) isa access;
                                    $va isa action, has name 'view_file';
                                    get $fp; sort $fp asc;
                                    """)
                )
                for i, file in enumerate(response, start=1):
                    print(f"File #{i}:", file.get("fp").as_attribute().get_value())
                if len(response) == 0:
                    print("No files found. Try enabling inference.")
                return response
            else:
                print("Error: No users found with that name.")
                return None

We call the function without enabling the inference and expect it to return no results, as the query pattern matches only files available for view_file action, and there are no such files initially in the database.

The get_files_by_user() function checks that there is only one user matched with the name provided by an input parameter. It then executes the query and iterates through the list of results to print a value of every matched path attribute.

Get query with inference

To get query results with inferred data, let’s enable the infer parameter of the TypeDB transaction options. We use the same get_files_by_user() function, but set the inference parameter to True when we call it again. The add-view-permission provides us with some inferred results this time.

Update query

Let’s replace a path for one of the files with a new path. We can do that by deleting ownership of the old path attribute from the file entity and assigning it with ownership of the new path attribute with the Update query:

def update_filepath(driver, db_name, old, new):
    with driver.session(db_name, SessionType.DATA) as data_session:
        with data_session.transaction(TransactionType.WRITE) as tx:
            response = list(
                tx.query.update(f"""
                                match
                                $f isa file, has path $old_path;
                                $old_path = '{old}';
                                delete
                                $f has $old_path;
                                insert
                                $f has path $new_path;
                                $new_path = '{new}';
                                """)
            )
            if len(response) > 0:
                tx.commit()
                print(f"Total number of paths updated: {len(response)}.")
                return response
            else:
                print("No matched paths: nothing to update.")
                return None

We collect the response of the Update query in a list and check the length of the list to determine the number of times the delete and insert clauses are executed. We then commit the changes only if the number meets our expectation.

Delete query

Finally, let’s delete the same file we updated the path for. First, we match the file in a Get (or Fetch) query to check how many files get matched to prevent unplanned deletes. If the number (and any other relevant parameters) of deletes is as expected, we proceed with a Delete query with the same match clause.

By using the same write transaction we employ snapshot isolation to prevent any other transactions from changing the expected results. If any other transaction makes a conflicting change before we commit this transaction, then our transaction fails upon a commit.

def delete_file(driver, db_name, path):
    with driver.session(db_name, SessionType.DATA) as data_session:
        with data_session.transaction(TransactionType.WRITE) as tx:
            response = list(
                tx.query.get(f"""
                                match
                                $f isa file, has path '{path}';
                                get;
                                """)
            )
            if len(response) == 1:
                tx.query.delete(f"""
                                match
                                $f isa file, has path '{path}';
                                delete
                                $f isa file;
                                """).resolve()
                tx.commit()
                print("The file has been deleted.")
                return True
            elif len(response) > 1:
                print("Matched more than one file with the same path.")
                print("No files were deleted.")
                return False
            else:
                print("No files matched in the database.")
                print("No files were deleted.")
                return False

Learn more

The full source code of this sample application.

The full API reference for the TypeDB Python driver.

Provide Feedback