Coverage for src/pyensae/sql/sql_interface.py : 82%
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# -*- coding: utf-8 -*-
2"""
3@file
4@brief Abstract class to connect to a SQL server using various way.
5It will be used to implement magic functions
6"""
8import re
9import sqlite3
12class InterfaceSQLException(BaseException):
14 """
15 a specific exception
16 """
18 def __init__(self, message):
19 """
20 @param message exception message
21 """
22 self.message = message
24 def __str__(self):
25 """
26 usual
27 """
28 return self.message
31class AutoCompletionSQLObject:
33 """
34 a simple class which allows auto completion
35 for tables, columns...
36 """
38 def __init__(self, name):
39 """
40 creates an instance with a given name
41 """
42 self._true_name = name
43 self._filt_name = AutoCompletionSQLObject._filter_name(name)
45 @staticmethod
46 def _filter_name(name):
47 """
48 removes unavailable characters
49 """
50 return name.replace(".", "_").replace(" ", "_")
52 @property
53 def _(self):
54 """
55 returns the true name of the object
56 """
57 return self._true_name
59 @property
60 def _f(self):
61 """
62 returns the filtered name
63 """
64 return self._filt_name
66 def _add(self, name):
67 """
68 add a subname to the class
70 @param name string
71 @return an AutoCompletionSQLObject
73 the filtered name (``_f``) of the new object will
74 be added to ``self.__dict__``, if an object
75 already exists with the same name, it will raise an exception
76 """
77 au = AutoCompletionSQLObject(name)
78 af = au._f
79 if af in self.__dict__:
80 raise KeyError(
81 "the object %s was already added to %s" %
82 (af, self._f))
83 self.__dict__[af] = au
84 return au
87class InterfaceSQL:
89 """
90 Abstract class to connect to a SQL server using various way.
91 It will be used to implement magic functions
92 """
94 @staticmethod
95 def create(obj):
96 """
98 @param obj a filename, a connection string, ...
100 ``obj`` can be a:
102 * file --> the class :class:`Database <pyensae.sql.database_main.Database>` will be used, we assume this file
103 is sqlite database, the file does not have to exist, in that case, it will created
104 * sqlite3.Connection --> the object will be wrapped into a :class:`Database <pyensae.sql.database_main.Database>`
105 * InterfaceSQL --> returns the object itself
107 .. versionchanged:: 1.1
108 Parameter *dbfile* can be of type `sqlite3.Connection <https://docs.python.org/3/library/sqlite3.html#sqlite3.Connection>`_.
109 """
110 if isinstance(obj, str):
111 from .sql_interface_database import InterfaceSQLDatabase
112 return InterfaceSQLDatabase(obj)
113 elif isinstance(obj, sqlite3.Connection):
114 from .sql_interface_database import InterfaceSQLDatabase
115 return InterfaceSQLDatabase(obj)
116 elif isinstance(obj, InterfaceSQL):
117 return obj
118 else:
119 raise NotImplementedError(
120 "nothing is implemented for type: %s" % str(
121 type(obj)))
123 def populate_completion(self):
124 """
125 the method create an object which contains a class
126 the user could use to speed the typing SQL queries,
127 functions in a notebook
129 This object will added with the name ``CC``,
130 it is returned by the function.
132 @return @see cl AutoCompletionSQLObject
134 The method should be called when the method @see me connect
135 is called.
136 """
137 self.CC = AutoCompletionSQLObject("TBL")
138 tbls = self.get_table_list()
139 for tb in tbls:
140 compl = self.CC._add(tb)
141 cols = self.get_table_columns(tb)
142 for k, v in cols.items():
143 compl._add(v[0])
144 return self.CC
146 def __init__(self, obj):
147 """
148 Initializes the object.
150 @param obj anything, see below
152 ``obj`` can be a:
153 * file --> the class :class:`Database <pyensae.sql.database_main.Database>` will be used, we assume this file
154 is sqlite database, the file does not have to exist, in that case, it will created
155 """
156 raise NotImplementedError()
158 def connect(self):
159 """
160 connection to the database
161 """
162 raise NotImplementedError()
164 def close(self):
165 """
166 close the connection to the database
167 """
168 raise NotImplementedError()
170 def get_table_list(self):
171 """
172 returns the list of tables in the database
174 @return list of strings
175 """
176 raise NotImplementedError()
178 def get_table_columns(self, table_name, as_dict=True):
179 """
180 returns the list of columns in a table
182 @param table_name table name
183 @param as_dict True, as dictionary, as a list otherwise
184 @return dictionary { "column": (position, type) } or a list
185 """
186 raise NotImplementedError()
188 def execute(self, sql_query):
189 """
190 execute a SQL query
192 @param sql_query query to execute
193 @return pandas DataFrame
195 The function takes care of the unexpected syntax introduction
196 by the autocompletion object: it just replaces
197 ``DB.CC.<name>`` by the ``true_name``.
198 """
199 sql_query = self.process_query(sql_query)
200 return self.execute_clean_query(sql_query)
202 def execute_clean_query(self, sql_query):
203 """
204 The function does the same thing as @see me execute
205 but it does not replace autocompletion object.
206 It is this function which should be overloaded by
207 classes inheriting from this one.
209 @param sql_query query to execute
210 @return pandas DataFrame
211 """
212 raise NotImplementedError()
214 _exp = re.compile("(DB[.]CC[.][a-zA-Z0-9_]+([.][a-zA-Z0-9_]+)*)")
216 def process_query(self, sql_query):
217 """
218 replaces autocompletion object by their real names
220 @param sql_query SQL query
221 @return clean sql_query
222 """
223 # i don't remember the syntax but it should be replaced using regular expression, not
224 # string replace
225 fi = InterfaceSQL._exp.findall(sql_query)
226 if len(fi) > 0:
227 only = [_[0] for _ in fi]
228 only.sort(reverse=True)
229 for o in only:
230 co = "self." + o[3:]
231 ev = eval(co)
232 sql_query = sql_query.replace(o, ev._)
233 return sql_query
235 def import_flat_file(self, filename, table_name):
236 """
237 import a flat file as a table, we assume the columns
238 separator is ``\\t`` and the file name contains a header
240 @param filename filename
241 @param table table name
242 @return the number of added rows
243 """
244 raise NotImplementedError()
246 def drop_table(self, table_name):
247 """
248 drops a table
250 @param table table name
251 """
252 raise NotImplementedError()
254 def refresh_completion(self):
255 """
256 refresh the auto completion
258 @return completion object
259 """
260 return self.populate_completion()
262 def add_function(self, code_function):
263 """
264 add a function to the database which can be called in a SELECT statement
266 @param code_function pointer to the function
267 """
268 raise NotImplementedError()
270 def import_dataframe(self, tablename, df):
271 """
272 import a dataframe into the database
274 @param tablename name of the table
275 @param df dataframe
276 @return the number of added rows
277 """
278 raise NotImplementedError()