-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathdatabase.py
More file actions
158 lines (130 loc) · 5.74 KB
/
Copy pathdatabase.py
File metadata and controls
158 lines (130 loc) · 5.74 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
import arango
import configparser
''' Database Module '''
class SystemMapDatabase:
''' SystemMapDatabase '''
_default_datbase_options = {'protocol': 'http',
'host': 'localhost',
'port': 8529,
'enable_logging': True}
def _read_conf(self, conffile=None):
''' Read the Database configuration.
:param conffile: The database configuration file.
:type conffile: str | unicode
:returns: The configuration.
:rtype: configpargser.ConfigParser
'''
parser = configparser.ConfigParser(self._default_datbase_options)
parser.read(conffile)
return parser
def __init__(self, dbname=None, configfile=None):
self._conf = self._read_conf(configfile)
client_options = dict(self._conf['Arangodb'])
self.client = arango.ArangoClient(**client_options)
self.database_name = self._conf.get('Database', 'name', fallback=dbname)
if self.database_name not in self.client.databases():
self.database = self.client.create_database(self.database_name)
else:
self.database = self.client.db(self.database_name)
@property
def current_database(self):
''' Currently used database name.
:returns: database name
:rtype: str
'''
return self.database_name
def databases(self):
''' Returns available Databases.
:returns: All available database names.
:rtype: List[str]
'''
return self.client.databases()
def get_graph(self, graphname):
''' Getting a graph object from the Database.
If the requested graph object does not exists, a new one is created.
:param graphname: Name of the Graph.
:type graphname: str
:returns: the requested graph object
:rtype: arango.graph.Graph
'''
if graphname not in [g['name'] for g in self.database.graphs()]:
return self.database.create_graph(graphname)
return self.database.graph(graphname)
def get_collection(self, collectionname):
''' Getting a collection object from the Database.
If the requested collection does not exists, a new one is created.
:param collectionname: Name of the collection
:type collectionname: str
:returns: the requested collection
:rtype: arango.collections.Collection
'''
if collectionname not in [c['name'] for c in self.database.collections()]:
return self.database.create_collection(collectionname)
return self.database.collection(collectionname)
def get_vertex_collection(self, collectionname, graph):
''' Getting a vertex collection from the Database.
If the requested vertex collection does not exists, a new one is
created.
:param collectionname: Name of the collection
:type collectionname: str
:param graph: Graph object where the vertex collection should consists to.
:type graph: graph object
:returns: vertex collection object
:rtype: arango.collections.VertexCollection
'''
try:
v_coll = graph.create_vertex_collection(collectionname)
except arango.exceptions.VertexCollectionCreateError:
v_coll = graph.vertex_collection(collectionname)
return v_coll
def insert_vertex(self, graphname, v_collection, data):
''' Insert vertex data into graph.
:param graphname: Name of the Graph insert into.
:type graphname: str
:param v_collection: Name of the vertex collection of the Graph.
:type v_collection: str
:param data: Vertex data, which should be inserted.
:type data: dict
:returns: List of all inserted document key/revisions.
:rtype: list
'''
graph = self.get_graph(graphname)
v_coll = self.get_vertex_collection(v_collection, graph)
return v_coll.insert(data)
def insert_edges(self, graphname, e_collection, from_coll, to_coll, data):
''' Insert edge collection into graph.
:param graphname: Name of the graph insert into.
:type graphname: str
:param e_collection: Name of the edge collection of the Graph.
:type e_collection: str
:param from_coll: Definition where the edge connection starts.
:type from_coll: string | list of strings
:param to_coll: Definition wher the edge connection ends.
:type to_coll: string | list of strings
:param data: Edge data, which should be inserted.
:type data: dict
'''
graph = self.get_graph(graphname)
try:
e_coll = graph.create_edge_definition(e_collection,
from_collections=from_coll,
to_collections=to_coll)
except arango.exceptions.EdgeDefinitionCreateError:
e_coll = graph.edge_collection(e_collection)
return [e_coll.insert(e) for e in data]
def insert_collection(self, collectionname, data):
''' Returns the result of the insert (e.g. document key, revision)
:param collectionname: Name of collection insert into.
:type collectionname: str
:param data: collection data to insert.
:type data: dict
'''
collection = self.get_collection(collectionname)
return collection.insert(data)
def execute_query(self, query_str, **kwargs):
''' Execute an AQL-query.
:param query_str: The AQL-query that should be executed.
:type query_str: str
'''
self.database.aql.validate(query_str)
return self.database.aql.execute(query_str, **kwargs)