Snippets
Created by
Xavier C.
last modified
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 | #!/usr/bin/python
import sqlite3 as lite
import sys
class SQLight:
""" Usage: see comments """
cur = None # Cursor de SQL
con = None # Conexion SQL
def __init__(self, path):
self.path = path
def connect(self):
try:
self.con = lite.connect(self.path)
self.cur = self.con.cursor()
except lite.Error, e:
print "Error %s:" % e.args[0]
if self.con:
self.con.close()
sys.exit(1)
def disconnect(self):
if self.con:
self.con.close()
def select(self, *fields, **kwargs):
""" sintax: test.select('username', table = 'table', where = "username='javi'") """
f = ", ".join(fields)
table = " "
where = " "
for k in kwargs:
if k == "table":
table = kwargs[k]
if k == "where":
where = kwargs[k]
sql_query = "SELECT %s FROM %s WHERE %s" % (f, table, where)
return self._fetch_rows(sql_query)
def _fetch_rows(self, sqlState):
if (self.con):
self.cur.execute(sqlState)
return self.cur.fetchall()
else:
print "[-] Error: There is not an active connection"
sys.exit(1)
def get_native_cursor(self):
return self.cur
def insert_into(self, table_name, **kwargs):
if (self.con == None):
sys.exit("Error: No active connection found")
insert_query = "INSERT INTO %s" % table_name
tables = []
values = []
for field in kwargs:
# concat tables
tables.append(field)
# quote if not int
if (type(kwargs[field]) == str):
string_value = kwargs[field].center(len(kwargs[field]) + 2, "'")
values.append(string_value)
else:
values.append(str(kwargs[field]))
# then we join them with comma
table_string = "%s%s%s" % ("(", ",".join(tables), ")")
values_string = "%s%s%s" % ("(", ",".join(values), ")")
final_insert = "%s %s VALUES %s" % (insert_query, table_string, values_string)
self.cur.execute(final_insert)
self.con.commit()
def delete(self, table_name, **kwargs):
if (self.con == None):
sys.exit("Error: No active connection found")
delete_query = "DELETE FROM %s " % table_name
where = ""
for field in kwargs:
if field == "where":
where = "WHERE %s" % kwargs[field]
delete_query += where
self.cur.execute(delete_query)
self.con.commit()
def update(self, table_name, **kwargs):
if (self.con == None):
sys.exit("Error: No active connection found")
update_query = "UPDATE %s SET " % table_name
setters = []
values = []
where = ""
for field in kwargs:
if (field != "where"):
if type(kwargs[field]) == str:
setters.append("%s = '%s'" % (field, kwargs[field]))
else:
setters.append("%s = %d" % (field, kwargs[field]))
else:
where = " WHERE %s" % kwargs[field]
update_query += ", ".join(setters)
update_query += where
self.cur.execute(update_query)
self.con.commit()
|
Comments (1)
You can clone a snippet to your computer for local editing. Learn more.
Download as sqlight.py
Import:
Usage: