TypeDBInterface

class ros_typedb.typedb_interface.TypeDBInterface(address, database_name, schema_path=None, data_path=None, force_database=False, force_data=False, infer=False)

Class used to interact with typeDB databases.

Connect to a typeDB server and interacts with it.

Connects TypeDBInterface to typeDB server, creating a database, loading a schema and a data file.

Parameters:

address (str) – TypeDB server address.
database_name (str) – database name.
schema_path (Union[list[str], str, None]) – list with paths to schema files (.tql).
data_path (Union[list[str], str, None]) – list with paths to data files (.tql).
force_database (Optional[bool]) – if database should override an existing database
force_data (Optional[bool]) – if the database data should be overriden.
infer (Optional[bool]) – if inference engine should be used.

Methods

`__init__`(address, database_name[, ...])	Connect to a typeDB server and interacts with it.
`attribute_dict_to_query`(attribute_dict)	Convert python dict with typedb attributes to a query.
`connect_driver`(address)	Connect to typedb server.
`create_database`(database_name[, force])	Create database.
`create_match_query`(things_list[, prefix])	Create match query from a list of tuples specifying a Thing individual.
`create_relationship_query`(relationship, ...)	Create a query for relationships.
`create_session`(database_name, session_type)	Create session with the database.
`database_query`(session_type, ...[, options])	Query database.
`delete_all_data`()	Delete all data from database.
`delete_attribute_from_thing`(thing, key, ...)	Delete attribute value from a instance of a thing.
`delete_attributes_from_thing`(match_dict[, ...])	Delete attributes from thing individual.
`delete_data_event`()	Delete data event.
`delete_data_event_`()	Generate delete data event.
`delete_database`(database_name)	Delete database.
`delete_from_database`(query)	Perform delete query.
`delete_thing`(thing, key, key_value)	Delete thing individual in the database.
`dict_to_query`(things_dict[, attribute_str, ...])	Convert python dict to query.
`fetch_attribute_from_thing`(thing, ...)	Get attribute value from a instance of a thing.
`fetch_attribute_from_thing_raw`(thing, ...)	Get raw attribute values from a instance of a thing.
`fetch_database`(query)	Perform match query.
`get_aggregate_database`(query)	Perform get aggregate query.
`get_database`(query)	Perform get query.
`insert_attribute_in_thing`(thing, key, ...)	Insert attribute value in a instance of a thing.
`insert_attributes_in_thing`(match_dict[, ...])	Insert attributes to thing individual.
`insert_data_event`()	Insert data event.
`insert_data_event_`()	Generate insert data event.
`insert_database`(query)	Perform insert query.
`insert_entity`(entity[, attribute_list])	Insert entity individual in the database.
`insert_relationship`(relationship, related_dict)	Insert relationship individual in the database.
`load_data`(data_path[, force])	Load .tql data file to database.
`load_schema`(schema_path)	Load .tql schema file to database.
`update_attribute_in_thing`(thing, key, ...)	Update attribute value in a instance of a thing.
`update_attributes_in_thing`(match_dict)	Update attributes of a thing individual.
`write_database_file`(session_type, ...)	Write .tql schema or data file content to database.

__init__(address, database_name, schema_path=None, data_path=None, force_database=False, force_data=False, infer=False)

Connect to a typeDB server and interacts with it.

Connects TypeDBInterface to typeDB server, creating a database, loading a schema and a data file.

Parameters:

address (str) – TypeDB server address.
database_name (str) – database name.
schema_path (Union[list[str], str, None]) – list with paths to schema files (.tql).
data_path (Union[list[str], str, None]) – list with paths to data files (.tql).
force_database (Optional[bool]) – if database should override an existing database
force_data (Optional[bool]) – if the database data should be overriden.
infer (Optional[bool]) – if inference engine should be used.

attribute_dict_to_query(attribute_dict)

Convert python dict with typedb attributes to a query.

Parameters:: attribute_dict (dict[str, str | int | float | bool | datetime]) – Dictionary with attributes to be converted.
Return type:: str
Returns:: Converted query.
Example:

The following dict:

{
    'email': 'test@test.test',
    'height': 1.8,
    'age': 18,
    'alive': True,
    'birth-date': datetime.datetime(2024, 1, 9, 15, 20, 0, 997315)
}

Converts to the string:

has email 'test@test.test',
has height 1.8,
has age 18,
has alive true,
has birth-date 2024-01-09T15:27:10.385

connect_driver(address)

Connect to typedb server.

Parameters:: address (str) – typedb server address.
Return type:: None

create_database(database_name, force=False)

Create database.

Parameters:

database_name (str) – database name.
force (Optional[bool]) – if database should override an existing database

Return type:

None

create_match_query(things_list, prefix='t')

Create match query from a list of tuples specifying a Thing individual.

Create match query from list of tuples specifying a Thing individual. A tuple has the following form: (THING_NAME, ATTR_NAME, ATTR_VALUE).

Parameters:

things_list (list[Tuple[str, str, str]]) – list of tuples specifying a Thing individual. E.g., [(‘person’, ‘email’, ‘test@email.test’), (‘person’, ‘email’, ‘test2@email.test’)].
prefix (Optional[str]) – prefix for the variable of each individual in the match query. E.g., employee results in the typedb variables $employee_0, $employee_1 etc.

Return type:

Tuple[str, list[str]]

Returns:

Converted query.

Example:

The following list:

my_list = [
    ('person', 'email', 'test@email.test'),
    ('person', 'email', 'test2@email.test')
]

query, prefix_list = self.create_match_query(my_list, 'employee')

Results in:

# query
$employee_0 isa person, has email 'test@email.test';
$employee_1 isa person, has email 'test2@email.test';

# prefix_list
['employee_0', 'employee_1']

create_relationship_query(relationship, related_dict, attribute_list=[], prefix='r')

Create a query for relationships.

Parameters:

relationship (str) – relationship name.
related_dict (dict[str, list[Tuple[str, str, str]]]) – dictionary with related things.
attribute_list (Optional[list[Tuple[str, str | int | float | bool | datetime]]]) – list with the relationships attributes.
prefix (Optional[str]) – prefix for query variables.

Example:

Return type:

str

related_dict = {
    'employee': [
        ('person', 'email', 'test@email.test'),
        ('person', 'email', 'test2@email.test')],
    'employer': [
        ('person', 'email', 'test3@email.test'),
        ('person', 'email', 'test4@email.test')
    ]
}

attribute_list = [
    ('salary', 2333), ('role-name', 'boss')]

query = self.create_relationship_query(
    'employment', related_dict, attribute_list, 'employment')

# query
$employment (employee:$employee_0, employee:$employee_1,
    employer:$employer_0, employer:$employer_1) isa employment,
    has salary 2333 , has role-name 'boss';

create_session(database_name, session_type, options=<typedb.api.connection.options.TypeDBOptions object>)

Create session with the database.

Parameters:

database_name (str) – database name.
session_type (SessionType) – session type, e.g., schema or data.
options (Optional[TypeDBOptions]) – typedb options.

Return type:

TypeDBSession

database_query(session_type, transaction_type, query_type, query, options=<typedb.api.connection.options.TypeDBOptions object>)

Query database.

Helper method to query the database, it handles creating a session and managaging a transaction. It can perform queries of the type define, insert, fetch, get_aggregate, or delete.

Parameters:

session_type (SessionType) – TypeDB session type.
transaction_type (TransactionType) – TypeDB transaction type.
query_type (Literal['define', 'insert', 'delete', 'fetch', 'get', 'get_aggregate']) – TypeDB query type.
query (str) – Query to be performed.
options (Optional[TypeDBOptions]) – TypeDB options.

Return type:

Union[Literal[True], Iterator[ConceptMap], list[dict[str, MatchResultDict]], None, int, float]

Returns:

Query result, type depends on the query_type.

delete_all_data()

Delete all data from database.

Return type:: None

delete_attribute_from_thing(thing, key, key_value, attr)

Delete attribute value from a instance of a thing.

Parameters:

thing (str) – thing name
key (str) – attribute name to identify the instance
key_value (str | int | float | bool | datetime) – attribute value to identify the instance
attr (str) – attribute name to be deleted

Return type:

Optional[Literal[True]]

Returns:

True.

delete_attributes_from_thing(match_dict, attribute_str='delete_attributes')

Delete attributes from thing individual.

Parameters:: match_dict (dict[str, list[ThingPrefixAttrDict]]) – Dictionary describing the delete operation.
Return type:: Optional[Literal[True]]
Returns:: Delete query result.
Example:

The following code:

delete_dict = {
    'person': [
        {
            'prefix': 'p1',
            'attributes': {
                'email': 'test@test.test',
            },
            'delete_attributes': ['height', 'age']
        },
    ],
},

self.delete_attributes_from_thing(
    delete_dict, delete_attribute_str='delete_attributes')

Performs the query:

# query
match $p1 isa person, has email 'test@test.test';
    $p1 has height $p1_height, has age $p1_age;
delete $p1 has $p1_height, has $p1_age;

delete_data_event(): Delete data event.

delete_data_event_(): Generate delete data event.

delete_database(database_name)

Delete database.

Parameters:: database_name (str) – database name.
Return type:: None

delete_from_database(query)

Perform delete query.

Parameters:: query (str) – Query to be performed.
Return type:: Optional[Literal[True]]
Returns:: Query result, if query fails return None.

delete_thing(thing, key, key_value)

Delete thing individual in the database.

Parameters:

thing (str) – thing name.
key (str) – attribute name to identify the individual.
key_value (str | int | float | bool | datetime) – attribute value to identify the individual.

Return type:

Optional[Literal[True]]

Returns:

True.

dict_to_query(things_dict, attribute_str='attributes', delete_attribute_str='delete-attributes')

Convert python dict to query.

Convert python dict that describes how to insert a thing, or how to update a thing’s attributes, or how to delete a thing’s attributes.

Parameters:: things_dict (dict[str, list[ThingPrefixAttrDict]]) – Dictionary describing the thing operation.
Return type:: str
Returns:: Converted query.
Example:

Adding thing

insert_dict = {
    'person': [
        {
            'prefix': 'p1',
            'attributes': {
                'email': 'test@email.test',
                'nickname': 't',
            }
        },
        {
            'prefix': 'p2',
            'attributes': {
                'email': 'test2@email.test',
                'nickname': 't2',
                'height': 1.33,
            }
        },
    ],
    'employment': [
        {
            'prefix': 'e',
            'attributes': {
                'salary': 2333,
                'role-name': ['boss', 'super boss'],
            },
            'relationship': {
                'employee': 'p1',
                'employer': 'p2'
            }
        },
    ]
}

query = typedb_interface.dict_to_query(insert_dict)
insert_result = typedb_interface.insert_database("insert " + query)

The output of dict_to_query is:

$p1  isa person,  has email 'test@email.test';
$p2  isa person,  has email 'test2@email.test', has height 1.33;
$e (employee:$p1,employer:$p2) isa employment,  has salary 2333,
    has role-name 'boss', has role-name 'super boss';

Insert attribute to thing:

insert_dict = {
    'person': [
        {
            'prefix': 'p1',
            'attributes': {
                'email': 'test_person@test.test',
            },
            'insert_attributes': {
                'height': 1.80,
                'alive': True,
            }
        },
    ],
},

match_query = self.dict_to_query(insert_dict)
insert_query = self.dict_to_query(insert_dict, 'insert_attributes')

# match_query
$p1  isa person,  has email 'test_person@test.test';

# insert_query
$p1  isa person,  has height 1.8, has alive true;

Delete attribute from thing:

delete_dict = {
    'person': [
        {
            'prefix': 'p1',
            'attributes': {
                'email': 'test@test.test',
            },
            'delete_attributes': ['height', 'age']
        },
    ],
},

query = self.dict_to_query(
    delete_dict, delete_attribute_str='delete_attributes')

# query
$p1 isa person, has email 'test@test.test';
$p1 has height $p1_height, has age $p1_age;
delete $p1 has $p1_height, has $p1_age;

Update attribute from thing:

update_dict = {
    'person': [
        {
            'prefix': 'p1',
            'attributes': {
                'email': 'test@test.test',
            },
            'update_attributes': {
                'height': 1.50,
                'age': 17,
            }
        },
    ],
},

typedb_interface.update_attributes_in_thing(update_dict)

fetch_attribute_from_thing(thing, key_attr_list, attr)

Get attribute value from a instance of a thing.

Parameters:

thing (str) – thing name
key_attr_list (list[Tuple[str, str | int | float | bool | datetime]]) – list with attribute tuple (name, value)
attr (str) – attribute name to be fetched, e.g., ‘person-name’

Return type:

list[str | int | float | bool | datetime]

Returns:

List with the attribute values of type attr.

fetch_attribute_from_thing_raw(thing, key_attr_list, attr)

Get raw attribute values from a instance of a thing.

Parameters:

thing (str) – thing name
key_attr_list (list[Tuple[str, str | int | float | bool | datetime]]) – list with attribute tuple (name, value)
attr (str) – attribute name to be fetched, e.g., ‘person-name’

Return type:

list[dict[str, MatchResultDict]]

Returns:

List of dictionary with the query result.

fetch_database(query)

Perform match query.

Parameters:: query (str) – Query to be performed.
Return type:: list[dict[str, MatchResultDict]]
Returns:: Query result, if query fails return None.

get_aggregate_database(query)

Perform get aggregate query.

Parameters:: query (str) – Query to be performed.
Return type:: int | float | None
Returns:: Query result.

get_database(query)

Perform get query.

Parameters:: query (str) – Query to be performed.
Return type:: int | float | None
Returns:: Query result.

insert_attribute_in_thing(thing, key, key_value, attr, attr_value)

Insert attribute value in a instance of a thing.

Parameters:

thing (str) – thing name
key (str) – attribute name to identify the instance
key_value (str | int | float | bool | datetime) – attribute value to identify the instance
attr (str) – attribute name to be inserted
attr_value (str | int | float | bool | datetime) – attribute value to be inserted

Return type:

Optional[Iterator[ConceptMap]]

Returns:

Insert query result.

insert_attributes_in_thing(match_dict, attribute_str='insert_attributes')

Insert attributes to thing individual.

Parameters:: match_dict (dict[str, list[ThingPrefixAttrDict]]) – Dictionary describing the insert operation.
Return type:: Optional[Iterator[ConceptMap]]
Returns:: Insert query result.
Example:

The following code:

insert_dict = {
    'person': [
        {
            'prefix': 'p1',
            'attributes': {
                'email': 'test_person@test.test',
            },
            'insert_attributes': {
                'height': 1.80,
                'alive': True,
            }
        },
    ],
},

self.insert_attributes_in_thing(insert_dict, 'insert_attributes')

Performs the following query:

match $p1 isa person, has email 'test_person@test.test';
insert $p1 isa person, has height 1.8, has alive true;

insert_data_event(): Insert data event.

insert_data_event_(): Generate insert data event.

insert_database(query)

Perform insert query.

Parameters:: query (str) – Query to be performed.
Return type:: Optional[Iterator[ConceptMap]]
Returns:: Query result, if query fails return None.

insert_entity(entity, attribute_list=[])

Insert entity individual in the database.

Parameters:

entity (str) – entity name.
attribute_list (Optional[list[Tuple[str, str | int | float | bool | datetime]]]) – list with attribute tuple (name, value).

Return type:

Optional[Iterator[ConceptMap]]

Returns:

query result.

insert_relationship(relationship, related_dict, attribute_list=[])

Insert relationship individual in the database.

Parameters:

relationship (str) – relationship name.
related_dict (dict[str, list[Tuple[str, str, str]]]) – dictionary with related things.
attribute_list (Optional[list[Tuple[str, str | int | float | bool | datetime]]]) – list with the relationships attributes.

Return type:

Optional[Iterator[ConceptMap]]

Returns:

query result

Example:

The following code:

related_dict = {
    'employee': [
        ('person', 'email', 'test@email.test'),
        ('person', 'email', 'test2@email.test')],
    'employer': [
        ('person', 'email', 'test3@email.test'),
        ('person', 'email', 'test4@email.test')
    ]
}

attribute_list = [
    ('salary', 2333), ('role-name', 'boss')]

self.insert_relationship(
    'employment', related_dict, attribute_list)

Performs the following query:

# query
match $employee_0 isa person, has email 'test@email.test';
      $employee_1 isa person, has email 'test2@email.test';
      $employer_0 isa person, has email 'test3@email.test';
      $employer_1 isa person, has email 'test4@email.test';
insert  $employment (employee:$employee_0, employee:$employee_1,
        employer:$employer_0,employer:$employer_1) isa employment,
        has salary 2333 , has role-name 'boss';

load_data(data_path, force=False)

Load .tql data file to database.

Parameters:

data_path (str) – .tql file path.
force (bool) – if database should be overwritten.

Return type:

None

load_schema(schema_path)

Load .tql schema file to database.

Parameters:: schema_path (str) – .tql file path.
Return type:: None

update_attribute_in_thing(thing, key, key_value, attr, attr_value)

Update attribute value in a instance of a thing.

Parameters:

thing (str) – thing name
key (str) – attribute name to identify the instance
key_value (str | int | float | bool | datetime) – attribute value to identify the instance
attr (str) – attribute name to be updated
attr_value (str | int | float | bool | datetime) – attribute value to be inserted

Return type:

Optional[Iterator[ConceptMap]]

Returns:

Insert query result.

update_attributes_in_thing(match_dict)

Update attributes of a thing individual.

Parameters:: match_dict (dict[str, list[ThingPrefixAttrDict]]) – Dictionary describing the update operation.
Return type:: Optional[Iterator[ConceptMap]]
Returns:: Insert query result.
Example:

The following code:

update_dict = {
    'person': [
        {
            'prefix': 'p1',
            'attributes': {
                'email': 'test@test.test',
            },
            'update_attributes': {
                'height': 1.50,
                'age': 17,
            }
        },
    ],
},

typedb_interface.update_attributes_in_thing(update_dict)

Performs the following two query:

# delete query
match $p1 isa person, has email 'test@test.test';
    $p1 has height $p1_height, has age $p1_age;
delete $p1 has $p1_height, has $p1_age;

# insert query
match $p1 isa person, has email 'test@test.test';
insert $p1 isa person, has height 1.50, has age 17;

write_database_file(session_type, query_type, file_path)

Write .tql schema or data file content to database.

Parameters:

session_type (SessionType) – session type, e.g., schema or data.
query_type (Literal['define', 'insert']) – query type, e.g., ‘define’ or ‘insert’.
file_path (str) – .tql file path.

Return type:

None