Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_db(self):
cogdb = Cog('/tmp/cogtestdb2')
# create a namespace
cogdb.create_namespace("my_namespace")
# create new table
cogdb.create_or_load_table("new_db", "my_namespace")
# put some data
cogdb.put(('A', 'val'))
cogdb.put(('B', 'val'))
cogdb.put(('key3', 'val'))
cogdb.put(('key3', 'val_updated'))
self.assertEqual(cogdb.get('key3')[1][1], 'val_updated')
def __init__(self, graph_name, cog_dir):
'''
:param graph_name:
:param cog_dir:
'''
self.predicates = self.list_predicate_tables(cog_dir, graph_name)
self.graph_name = graph_name
self.cog_dir = cog_dir
self.cogs = {}
for predicate in self.predicates:
cog = Cog(db_path=cog_dir)
cog.use_table(predicate, graph_name)
self.cogs[predicate] = cog
def __init__(self, db_path=None, config=cfg):
self.cog = Cog(db_path=db_path, config=config)
def put(self, vertex1, predicate, vertex2):
cog = Cog(db_path=self.cog_dir)
cog.create_namespace(self.graph_name)
cog.use_table(predicate, self.graph_name) # it wont create if it exists.
put_node(cog, vertex1, predicate, vertex2)
self.cogs[predicate] = cog
return self