simon987/hexlib
1
0
Fork 0
mirror of https://github.com/simon987/hexlib.git synced 2025-04-10 14:06:43 +00:00
Code Issues Projects Releases Wiki Activity

Handle raw bytes

Browse Source
This commit is contained in:
simon987 2019-12-29 17:13:35 -05:00
parent 37257f11eb
commit 86e07900ea
2 changed files with 24 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
hexlib/db.py
View File

@ -1,11 +1,15 @@
import base64
import sqlite3
class PersistentState:
"""Quick and dirty persistent dict-like SQLite wrapper"""
def __init__(self, dbfile="state.db"):
def __init__(self, dbfile="state.db", **dbargs):
self.dbfile = dbfile
if dbargs is None:
dbargs = {"timeout": 30000}
self.dbargs = dbargs
def __getitem__(self, table):
return Table(self, table)
@ -17,7 +21,7 @@ class Table:
self._table = table
def __iter__(self):
with sqlite3.connect(self._state.dbfile) as conn:
with sqlite3.connect(self._state.dbfile, **self._state.dbargs) as conn:
conn.row_factory = sqlite3.Row
try:
cur = conn.execute("SELECT * FROM %s" % (self._table,))
@ -27,38 +31,42 @@ class Table:
return None
def __getitem__(self, item):
with sqlite3.connect(self._state.dbfile) as conn:
with sqlite3.connect(self._state.dbfile, **self._state.dbargs) as conn:
conn.row_factory = sqlite3.Row
try:
col_types = conn.execute("PRAGMA table_info(%s)" % self._table).fetchall()
cur = conn.execute("SELECT * FROM %s WHERE id=?" % (self._table,), (item,))
row = cur.fetchone()
if row:
return dict(row)
return dict(
(col[0], _deserialize(row[col[0]], col_types[i]["type"]))
for i, col in enumerate(cur.description)
)
except:
return None
def __setitem__(self, key, value):
with sqlite3.connect(self._state.dbfile) as conn:
with sqlite3.connect(self._state.dbfile, **self._state.dbargs) as conn:
conn.row_factory = sqlite3.Row
sql = "INSERT INTO %s (id,%s) VALUES ('%s',%s)" % \
(self._table, ",".join(value.keys()), key, ",".join("?" for _ in value.values()))
try:
conn.execute(sql, list(_sqlite_value(v) for v in value.values()))
conn.execute(sql, list(_serialize(v) for v in value.values()))
except sqlite3.OperationalError:
conn.execute(
"create table if not exists %s (id text primary key,%s)" %
(self._table, ",".join("%s %s" % (k, _sqlite_type(v)) for k, v in value.items()))
)
conn.execute(sql, list(_sqlite_value(v) for v in value.values()))
conn.execute(sql, list(_serialize(v) for v in value.values()))
except sqlite3.IntegrityError:
sql = "UPDATE %s SET (%s) = (%s) WHERE id=?" \
% (self._table, ",".join(value.keys()), ",".join("?" for _ in value.values()))
args = [key]
args.extend(_sqlite_value(v) for v in value.values())
args.extend(_serialize(v) for v in value.values())
conn.execute(
sql,
args
@ -75,8 +83,13 @@ def _sqlite_type(value):
return "text"
def _sqlite_value(value):
def _serialize(value):
if isinstance(value, bytes):
return _sqlite_value(value.decode("unicode_escape"))
return base64.b64encode(value)
return str(value)
def _deserialize(value, col_type):
if col_type == "blob":
return base64.b64decode(value)
return value

2
setup.py
View File

@ -2,7 +2,7 @@ from setuptools import setup
setup(
name="hexlib",
version="1.5",
version="1.6",
description="Misc utility methods",
author="simon987",
author_email="me@simon987.net",