PyPI 官网下载 | gecosistema_lite-0.0.479.zip

# ------------------------------------------------------------------------------- # Licence: # Copyright (c) 2012-2017 Luzzi Valerio # # The above copyright notice and this permission notice shall be # included in all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, # EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES # OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND # NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT # HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING # FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR # OTHER DEALINGS IN THE SOFTWARE. # # # Name: sqlitedb.py # Purpose: # # Author: Luzzi Valerio # # Created: 26/07/2017 # ------------------------------------------------------------------------------- from databases import * from datatypes import * from filesystem import * import csv import inspect import sqlite3 as sqlite import ogr class SqliteDB(AbstractDB): CORE_FUNCTIONS = ["ABS", "CHANGES", "CHAR", "COALESCE", "GLOB", "HEX", "IFNULL", "INSTR", "LAST_INSERT_ROWID", "LENGTH", "LIKE", "LIKE", "LIKELIHOOD", "LIKELY", "LOAD_EXTENSION", "LOAD_EXTENSION", "LOWER", "LTRIM", "LTRIM", "MAX", "MIN", "NULLIF", "PRINTF", "QUOTE", "RANDOM", "RANDOMBLOB", "REPLACE", "ROUND", "ROUND", "RTRIM", "RTRIM", "SOUNDEX", "SQLITE_COMPILEOPTION_GET", "SQLITE_COMPILEOPTION_USED", "SQLITE_SOURCE_ID", "SQLITE_VERSION", "SUBSTR", "SUBSTR", "TOTAL_CHANGES", "TRIM", "TRIM", "TYPEOF", "UNICODE", "UNLIKELY", "UPPER", "ZEROBLOB"] AGGREGATE_FUNCTIONS = ["AVG", "COUNT", "GROUP_CONCAT", "MAX", "MIN", "SUM", "TOTAL"] DATE_FUNCTIONS = ["DATE", "TIME", "DATETIME", "JULIANDAY", "STRFTIME"] SPATIAL_FUNCTIONS = ["GEOMFROMTEXT", "GEOMFROMWKB", "ASTEXT", "POINT", "X", "Y"] RESERVED_WORDS = ["ABORT", "ACTION", "ADD", "AFTER", "ALL", "ALTER", "ANALYZE", "AND", "AS", "ASC", "ATTACH", "AUTOINCREMENT", "BEFORE", "BEGIN", "BETWEEN", "BY", "CASCADE", "CASE", "CAST", "CHECK", "COLLATE", "COLUMN", "COMMIT", "CONFLICT", "CONSTRAINT", "CREATE", "CROSS", "CURRENT_DATE", "CURRENT_TIME", "CURRENT_TIMESTAMP", "DATABASE", "DEFAULT", "DEFERRABLE", "DEFERRED", "DELETE", "DESC", "DETACH", "DISTINCT", "DROP", "EACH", "ELSE", "END", "ESCAPE", "EXCEPT", "EXCLUSIVE", "EXISTS", "EXPLAIN", "FAIL", "FOR", "FOREIGN", "FROM", "FULL", "GLOB", "GROUP", "HAVING", "IF", "IGNORE", "IMMEDIATE", "IN", "INDEX", "INDEXED", "INITIALLY", "INNER", "INSERT", "INSTEAD", "INTERSECT", "INTO", "IS", "ISNULL", "JOIN", "KEY", "LEFT", "LIKE", "LIMIT", "MATCH", "NATURAL", "NO", "NOT", "NOTNULL", "NULL", "OF", "OFFSET", "ON", "OR", "ORDER", "OUTER", "PLAN", "PRAGMA", "PRIMARY", "QUERY", "RAISE", "RECURSIVE", "REFERENCES", "REGEXP", "REINDEX", "RELEASE", "RENAME", "REPLACE", "RESTRICT", "RIGHT", "ROLLBACK", "ROW", "SAVEPOINT", "SELECT", "SET", "TABLE", "TEMP", "TEMPORARY", "THEN", "TO", "TRANSACTION", "TRIGGER", "UNION", "UNIQUE", "UPDATE", "USING", "VACUUM", "VALUES", "VIEW", "VIRTUAL", "WHEN", "WHERE", "WITH", "WITHOUT"] SQLITE_FUNCTIONS = CORE_FUNCTIONS + AGGREGATE_FUNCTIONS + DATE_FUNCTIONS + SPATIAL_FUNCTIONS + RESERVED_WORDS def __init__(self, filename, modules=[]): """ Constructor :param filename: """ AbstractDB.__init__(self, filename) self.pragma("synchronous=OFF") self.pragma("journal_mode=WAL") self.pragma("foreign_keys=ON") self.pragma("cache_size=4000") self.load_extension(modules) def pragma(self, text, env={}, verbose=False): """ pragma """ return self.execute("PRAGMA " + text, env, verbose=verbose) def create_function(self, func, nargs, fname): """ create_function """ self.conn.create_function(func, nargs, fname) def create_aggregate(self, func, nargs, fname): """ create_aggregate """ self.conn.create_aggregate(func, nargs, fname) def load_extension(self, modules): """ load_extension """ self.conn.enable_load_extension(True) for module in listify(modules): self.conn.execute("SELECT load_extension('%s')" % (module)) self.conn.enable_load_extension(False) def load_function(self, modulename="gecosistema_lite", fnames="", verbose=False): """ load_function """ try: module = __import__(modulename) for fname in listify(fnames): try: obj = getattr(module, fname) if inspect.isfunction(obj): n = len(inspect.getargspec(obj).args) self.create_function(fname, n, obj) if verbose: print "load function %s(%s)" % (fname, n) elif inspect.isclass(obj) and "step" in dir(obj): fstep = getattr(obj, "step") n = len(inspect.getargspec(fstep).args) - 1 self.create_aggregate(fname, n, obj) if verbose: print "load aggregate function %s(%s)" % (fname, n) except: if verbose: print "function <%s> not found." % (fname) except: if verbose: print "module <%s> not found. searching <%s>" % (modulename, fnames) def load_functions(self, sql, verbose=True): """ load functions or aggregates ## from numpy import sqrt """ sql = sql.replace(";", "\n") directives = re.findall(r'^\s*(?:#{1,2})\s*from\s+\w+\s+import\s+(?:\*|\w+).*', sql, re.I | re.M) directives = [normalizestring(item).strip('# ') for item in directives] for line in directives: print line.split(" ", 3) _from, module, _import, functions = line.split(" ", 3) if functions == "*": functions = re.findall(r'\w+\s*\(', sql, re.MULTILINE) functions = [fname.strip("( ") for fname in functions] functions = [fname for fname in functions if fname.upper() not in self.SQLITE_FUNCTIONS] functions = list(set(functions)) self.load_function(module, functions, verbose) def __connect__(self): """ Connect to the sqlite instance """ try: self.conn = sqlite.connect(self.dsn) except sqlite.Error as err: print(err) self.close() def __del__(self): """ Destructor """ self.close() def GetTables(self, like="%"): """ Return a list with all tablenames """ sql = """ SELECT tbl_name FROM sqlite_master WHERE type IN ('table','view') AND tbl_name LIKE '{like}' UNION SELECT tbl_name FROM sqlite_temp_master WHERE type IN ('table','view') AND tbl_name LIKE '{like}';"""; env = {"like": like} table_list = self.execute(sql, env, verbose=False) table_list = [item[0] for item in table_list] return table_list def GetDDL(self, tablename, type="table,view,index", verbose=False): """ Return DDL statements for tablename """