-
Notifications
You must be signed in to change notification settings - Fork 0
/
sqlite_server.py
38 lines (32 loc) · 1.29 KB
/
sqlite_server.py
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
import os
from subprocess import Popen, PIPE, STDOUT
g_attach_query = ""
def _run_query(query_str):
global g_attach_query
query_str = g_attach_query + query_str + ";"
script_dir = os.path.dirname(os.path.realpath(__file__))
open(os.path.join(script_dir, "sqlite_server_IN"), "wa").write(query_str);
return open(os.path.join(script_dir, "sqlite_server_OUT"), "r").read()
def load(filename, dbname):
global g_attach_query
if not os.path.isfile(filename):
raise Exception("Database %s does not exists." % repr(filename))
g_attach_query = "attach '%s' as '%s';" % (filename, dbname)
def _sqlite_escape(s):
return s.replace("'", "''")
def query(query_str, *rest):
# replace ? with escaped version of arguments
last = 0
for obj in rest:
i = query_str.find("?", last)
if i == -1:
raise Exception("Too many arguments, not enough ?")
inserted = "'" + _sqlite_escape(str(obj)) + "'"
query_str = query_str[:i] + inserted + query_str[i+1:]
last = i + len(inserted) + 1
result = _run_query(query_str)
# parse the results
if result[:6] == "Error:":
raise Exception("Invalid query: " +
repr(query_str) + "," + result[6:])
return [x.split("|") for x in result.splitlines()]