Transactions

After opening a driver with the correct configuration and credentials, you can open transactions for encapsulated querying in your application.

Transaction fundamentals

Transaction types

TypeDB provides three distinct transaction types, each with specific capabilities:

Type Capabilities

READ

  • Execute read queries only

  • Cannot modify data or schema

  • Lightweight

WRITE

  • Execute read queries

  • Execute data write queries (insert, delete, update)

  • Cannot modify schema

  • May conflict and fail on commit

  • Locks out schema transactions

SCHEMA

  • Execute read queries

  • Execute data write queries

  • Execute schema modification queries

  • Exclusive access

  • Locks out write transactions

ACID guarantees

TypeDB transactions provide ACID guarantees up to snapshot isolation. For more details, see the server Transaction documentation.

Transaction lifecycle

Basic transaction usage

Every transaction follows the same lifecycle pattern:

  1. Open: Create transaction with specific type.

  2. Execute: Send queries and process results.

  3. Commit/Close: Persist changes or discard transaction.

Read transactions

Read transactions are the most efficient for queries that don’t modify data.

For example, in Python:

#!test[reset-after]
#{{
from typedb.driver import *

DB_NAME = "my_database"
address = "localhost:1729"
credentials = Credentials("admin", "password")
options = DriverOptions(is_tls_enabled=True, tls_root_ca_path=None)

driver = TypeDB.driver(address, credentials, options)
try:
    driver.databases.create(DB_NAME) # raises already exists exception
finally:
    pass

with driver.transaction(DB_NAME, TransactionType.SCHEMA) as tx:
    tx.query("""
        define
        attribute company-name, value string;
		attribute email, value string;
		attribute name, value string;
		attribute age, value integer;
        entity company, owns company-name, owns email, plays employment:employer;
		entity user, owns name, owns email, owns age, plays employment:employee;
        relation employment, relates employer, relates employee;
    """).resolve()
    tx.commit()

with driver.transaction(DB_NAME, TransactionType.WRITE) as tx:
    tx.query("insert $u isa user, has name 'alice', has email 'alice@example.com', has age 25;").resolve()
    tx.commit()
#}}

# READ transactions for data retrieval
with driver.transaction(DB_NAME, TransactionType.READ) as tx:
    users = tx.query("match $u isa user; fetch { 'name': $u.name, 'email': $u.email };").resolve()
    for user in users.as_concept_documents():
        print(f"User: {user['name']} ({user['email']})")

    # No commit needed - READ transactions are automatically closed
#{{
driver.close()
#}}

Write transactions

Write transactions handle data modifications and require explicit commits.

For example, in Python:

#!test[reset-after]
#{{
from typedb.driver import *

DB_NAME = "my_database"
address = "localhost:1729"
credentials = Credentials("admin", "password")
options = DriverOptions(is_tls_enabled=True, tls_root_ca_path=None)

driver = TypeDB.driver(address, credentials, options)
try:
    driver.databases.create(DB_NAME) # raises already exists exception
finally:
    pass

with driver.transaction(DB_NAME, TransactionType.SCHEMA) as tx:
    tx.query("""
        define
        attribute company-name, value string;
		attribute email, value string;
		attribute name, value string;
		attribute age, value integer;
        entity company, owns company-name, owns email, plays employment:employer;
		entity user, owns name, owns email, owns age, plays employment:employee;
        relation employment, relates employer, relates employee;
    """).resolve()
    tx.commit()
#}}

# WRITE transactions for data modifications
with driver.transaction(DB_NAME, TransactionType.WRITE) as tx:
    # Insert new data
    tx.query("insert $u isa user, has email 'bob@example.com', has age 30;").resolve()

    # Update existing data and ignore the response
    tx.query("""
        match $user isa user, has age $age;
        delete has $age of $user;
        insert $user has age 31;
    """).resolve()

    # All changes are buffered until commit
    tx.commit()
#{{
driver.close()
#}}

Schema transactions

Schema transactions provide exclusive access for schema modifications, and allow you to submit schema modification queries.

For example, in Python:

#!test[reset-after]
#{{
from typedb.driver import *

DB_NAME = "my_database"
address = "localhost:1729"
credentials = Credentials("admin", "password")
options = DriverOptions(is_tls_enabled=True, tls_root_ca_path=None)

driver = TypeDB.driver(address, credentials, options)
try:
    driver.databases.create(DB_NAME) # raises already exists exception
finally:
    pass
#}}

# SCHEMA transactions for schema changes
with driver.transaction(DB_NAME, TransactionType.SCHEMA) as tx:
    # Define new schema elements
    tx.query("""
        define
        attribute company-name, value string;
		attribute email, value string;
		attribute name, value string;
		attribute age, value integer;
        entity company, owns company-name, owns email, plays employment:employer;
		entity user, owns name, owns email, owns age, plays employment:employee;
        relation employment, relates employer, relates employee;
    """).resolve()

    # Can also modify data in same transaction, but it will not be as efficient as using a WRITE transaction
    tx.query("insert $c isa company, has company-name 'TypeDB Inc';").resolve()

    tx.commit()  # Commit schema and data changes atomically

#{{
driver.close()
#}}

Concurrency and locking

Read transactions can run concurrently with all other transaction types at any time.

Write transactions can be opened concurrently to other read or write transactions. They may conflict and fail on commit.

Schema transactions are exclusive against other schema and other write transactions, but do not block read transactions. There is a configurable timeout to obtain the schema lock, as noted below.

Transaction options and timeouts

For the complete documentation, see your driver’s reference page:

Transaction timeouts

Transactions have configurable timeout settings. The default setting is a 5 minute timeout. This helps avoid resource leaks when users forget to close transactions.

You can configure transactions with a specific timeout.

For example, in Python:

#!test[reset-after]
#{{
from typedb.driver import *

DB_NAME = "my_database"
address = "localhost:1729"
credentials = Credentials("admin", "password")
options = DriverOptions(is_tls_enabled=True, tls_root_ca_path=None)

driver = TypeDB.driver(address, credentials, options)
try:
    driver.databases.create(DB_NAME) # raises already exists exception
finally:
    pass
with driver.transaction(DB_NAME, TransactionType.SCHEMA) as tx:
    #tag::tx_define_query[]
    tx.query("""
        define
		attribute email, value string;
		entity user, owns email;
    """).resolve()
    #end::tx_define_query[]
    tx.commit()
#}}

from typedb.driver import TransactionOptions

tx_options = TransactionOptions(transaction_timeout_millis=60_000)

with driver.transaction(DB_NAME, TransactionType.WRITE, options=tx_options) as tx:
    # This transaction will timeout after 60 seconds if not committed
    tx.query("insert $u isa user, has email 'test@example.com';")
    tx.commit()

#{{
driver.close()
#}}

Schema lock timeout

For schema transactions, you can configure how long to wait for exclusive access. The default timeout is 30 seconds.

For example, in Python:

#!test[reset-after]
#{{
from typedb.driver import *

DB_NAME = "my_database"
address = "localhost:1729"
credentials = Credentials("admin", "password")
options = DriverOptions(is_tls_enabled=True, tls_root_ca_path=None)

driver = TypeDB.driver(address, credentials, options)
try:
    driver.databases.create(DB_NAME) # raises already exists exception
finally:
    pass
#}}

from typedb.driver import TransactionOptions

tx_options = TransactionOptions(schema_lock_acquire_timeout_millis=60_000)

with driver.transaction(DB_NAME, TransactionType.SCHEMA, options=tx_options) as tx:
    # Will wait up to 60 seconds to acquire exclusive schema lock, else error
    tx.query("define entity another-type;")
    tx.commit()
#{{
driver.close()
#}}

Transaction best practices

Keep transactions short

Minimize transaction duration to reduce conflicts and resource usage. This is particularly important for write transactions.

Batching operations

Group related operations but avoid transactions that are too large. Optimal throughput is achieved with commits of 10-100 queries per transaction.

#!test[reset-after]
#{{
from typedb.driver import *

DB_NAME = "my_database"
address = "localhost:1729"
credentials = Credentials("admin", "password")
options = DriverOptions(is_tls_enabled=True, tls_root_ca_path=None)

driver = TypeDB.driver(address, credentials, options)
try:
    driver.databases.create(DB_NAME) # raises already exists exception
finally:
    pass

with driver.transaction(DB_NAME, TransactionType.SCHEMA) as tx:
    tx.query("""
        define
        attribute company-name, value string;
		attribute email, value string;
		attribute name, value string;
		attribute age, value integer;
        entity company, owns company-name, owns email, plays employment:employer;
		entity user, owns name, owns email, owns age, plays employment:employee;
        relation employment, relates employer, relates employee;
    """).resolve()
    tx.commit()
#}}

# Batch related operations
def create_users_batch(driver, user_data_list):
    with driver.transaction(DB_NAME, TransactionType.WRITE) as tx:
        for user_data in user_data_list:
            tx.query(f"insert $u isa user, has email '{user_data['email']}', has name '{user_data['name']}';")
        tx.commit()

# For large datasets, process in chunks
def create_users_chunked(driver, user_data_list, chunk_size=100):
    for i in range(0, len(user_data_list), chunk_size):
        chunk = user_data_list[i:i + chunk_size]
        create_users_batch(driver, chunk)

create_users_chunked(
    driver,
    [
        {'email': 'carrol@c.com', 'name': 'carrol'},
        {'email': 'danny@d.com', 'name': 'danny'},
        {'email': 'errol@e.com', 'name': 'errol'},
        {'email': 'fares@f.com', 'name': 'fares'},
    ],
    chunk_size=2
)
#{{
driver.close()
#}}

Handle commit failures gracefully

Implement retry logic for write transaction conflicts:

#!test
#{{
from typedb.driver import *

DB_NAME = "my_database"
address = "localhost:1729"
credentials = Credentials("admin", "password")
options = DriverOptions(is_tls_enabled=True, tls_root_ca_path=None)

driver = TypeDB.driver(address, credentials, options)
try:
    driver.databases.create(DB_NAME) # raises already exists exception
finally:
    pass

with driver.transaction(DB_NAME, TransactionType.SCHEMA) as tx:
    tx.query("""
        define
        attribute company-name, value string;
		attribute email, value string;
		attribute name, value string;
		attribute age, value integer;
        entity company, owns company-name, owns email, plays employment:employer;
		entity user, owns name, owns email, owns age, plays employment:employee;
        relation employment, relates employer, relates employee;
    """).resolve()
    tx.commit()

with driver.transaction(DB_NAME, TransactionType.WRITE) as tx:
    tx.query("insert $u isa user, has name 'alice', has email 'alice@example.com', has age 25;").resolve()
    tx.commit()
#}}
import time
import random

def update_user_with_retry(driver, email, age, max_retries=3):
    for attempt in range(max_retries):
        try:
            with driver.transaction(DB_NAME, TransactionType.WRITE) as tx:
                tx.query(f"""
                    match $u isa user, has age $old_age;
                    delete $u has $old_age;
                    insert $u has age {age};
                """)
                tx.commit()
                return  # Success!
        except TypeDBDriverException as e:
            if "conflict" in str(e).lower() and attempt < max_retries - 1:
                # Transaction conflict - wait and retry
                wait_time = 0.1 * (2 ** attempt) + random.uniform(0, 0.1)
                time.sleep(wait_time)
                continue
            else:
                raise  # Re-raise if not retryable or max retries exceeded
    raise Exception(f"Failed to update user after {max_retries} attempts")

update_user_with_retry(driver, "alice@example.com", 30)

Next steps

  • Queries - Learn about query execution and result processing

  • Best Practices - Transaction management and performance optimization patterns