Snippets

Xavier C. SQLight 1.0

Created by Xavier C. last modified
sqlight.py
Raw 
  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)

  1. Xavier C.

    Download as sqlight.py

    Import:

    import sqlight

    Usage:

    # https://artfromcode.wordpress.com/2016/10/20/sqlite-python-module-sqlight-1-0/
test = SQLight("/tmp/test.db")
print 'Connecting...'
test.connect()

print 'Selecting all rows...'
for r in test.select('username', table = 'users', where = "username='javi'"):
    print "Username %s " % (r[0])


print 'Selecting one row...'
rows = test.select('username', 'password', table = 'users', where = "username='javi'")
if (len(rows) > 0):
    row = rows[0]
    print "Username is %s" % row[0]
    print "Password is %s" % row[1]

print 'Inserting a new row...'
test.insert_into("users", username="javi", password="mypassword")

print 'Updating a row with conditions...'
test.update("users", username = 'javi_updated', password = 'MyPassword', where = "username='javi'")

print 'Massive update'
test.update("users", username = 'javi_massive_updated', password = 'MyPassword')

print 'Deleting set of rows...'
test.delete("users", where="username = 1")

print 'Massive Deleting...'
test.delete("users")
HTTPS SSH

You can clone a snippet to your computer for local editing. Learn more.