Hide keyboard shortcuts

Hot-keys on this page

r m x p   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

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

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

""" 

@file 

@brief Manages a sqlite3 database. 

""" 

import sqlite3 

import datetime 

import decimal 

import numpy 

 

 

class DBException(Exception): 

""" 

Exception raised by class @see cl Database. 

""" 

pass 

 

 

class Database: 

""" 

Common functions about sqlite3. 

""" 

 

_field_option = ["PRIMARYKEY", "AUTOINCREMENT", "AUTOFILL"] 

 

def __init__(self, dbfile): 

""" 

@param dbfile filename or ``:memory:`` 

""" 

self._sql_file = dbfile 

self._connection = None 

 

def get_file(self): 

""" 

Returns the file name. 

""" 

return self._sql_file 

 

def _check_connection(self): 

""" 

Check the SQL connection. 

""" 

if self._connection is None: 

message = "Use connect method before doing operation on this database." 

raise Exception(message) 

 

def _is_memory(self): 

""" 

Tells if the database takes place in memory (``:memory:``). 

""" 

return self._sql_file == ":memory:" 

 

def connect(self): 

""" 

Opens a connection to the database. 

""" 

if self._is_memory(): 

if self._connection is None: 

self._connection = sqlite3.connect(self._sql_file) 

elif self._connection is not None: 

raise Exception("A previous connection was not closed.") 

else: 

self._connection = sqlite3.connect(self._sql_file) 

 

def close(self): 

""" 

Close the database. 

""" 

self._check_connection() 

if self._is_memory(): 

# We should not close, otherwise, we lose the data. 

# self._connection = None 

pass 

else: 

self._connection.close() 

self._connection = None 

 

def commit(self): 

""" 

Call this function after any insert request. 

""" 

self._check_connection() 

self._connection.commit() 

 

def execute(self, request): 

""" 

Open a cursor with a query and return it to the user. 

 

@param request SQL request 

@return cursor 

""" 

# classic ways 

self._check_connection() 

cur = self._connection.cursor() 

try: 

cur.execute(request) 

except Exception as e: 

raise DBException( 

"Unable to execute a SQL request (1) (file '%s')" % 

self.get_file(), e, request) from e 

return cur 

 

def get_table_list(self): 

""" 

Returns the list of tables. 

 

@return the table list 

""" 

self._check_connection() 

request = """ SELECT name 

FROM (SELECT * FROM sqlite_master UNION ALL SELECT * FROM sqlite_temp_master) AS temptbl 

WHERE type in('table','temp') AND name != 'sqlite_sequence' ORDER BY name;""" 

 

select = self._connection.execute(request) 

res = [] 

for el in select: 

res.append(el[0]) 

return res 

 

def create_table(self, table, columns, temporary=False): 

""" 

Creates a table. 

 

@param table table name 

@param columns columns definition, dictionary ``{ key:(column_name,python_type) }`` 

if ``PRIMARYKEY`` is added, the key is considered as the primary key. 

Example:: 

 

columns = { -1:("key", int, "PRIMARYKEY", "AUTOINCREMENT"), 

0:("name",str), 1:("number", float) } 

 

@param temporary if True the table is temporary 

@return cursor 

""" 

if table == "sqlite_sequence": 

raise DBException("Unable to create a table named 'sql_sequence'.") 

 

tables = self.get_table_list() 

if table in tables: 

raise DBException("Tables '{0}' is already present.".format(table)) 

 

if isinstance(columns, list): 

columns = {i: v for i, v in enumerate(columns)} 

 

if temporary: 

sql = "CREATE TEMPORARY TABLE " + table + "(" 

else: 

sql = "CREATE TABLE " + table + "(" 

col = [] 

for c, val in columns.items(): 

if isinstance(val[1], tuple): 

v = val[1][0] 

else: 

v = val[1] 

 

if v is str: 

col.append(val[0] + " TEXT") 

elif v is int: 

col.append(val[0] + " INTEGER") 

elif v is float: 

col.append(val[0] + " FLOAT") 

elif v is numpy.int64: 

col.append(val[0] + " INTEGER") 

elif v is numpy.float64: 

col.append(val[0] + " FLOAT") 

elif v is decimal.Decimal: 

col.append(val[0] + " Decimal") 

elif v is datetime.datetime: 

col.append(val[0] + " DATETIME") 

else: 

raise DBException( 

"Unable to add column '{0}' ... {1} v={2}".format(c, val, v)) 

 

if "PRIMARYKEY" in val: 

if val[1] != int: 

raise DBException( 

"unable to create a primary key on something differont from an integer (%s)" % 

str(val)) 

col[-1] += " PRIMARY KEY" 

if "AUTOINCREMENT" in val: 

if self.isMSSQL(): 

col[-1] += " IDENTITY(0,1)" 

else: 

col[-1] += " AUTOINCREMENT" 

 

sql += ",\n ".join(col) 

sql += ");" 

return self.execute(sql) 

 

def has_rows(self, table): 

""" 

Tells if a table has rows. 

 

@param table table name 

@return boolean 

""" 

res = list(self.execute("SELECT * FROM {0} LIMIT 1".format(table))) 

return len(res) > 0