adb_python/pyadb/internal/android_db.py

import shlex
import re
import csv
import sys
csv.field_size_limit(2**31-1)
dict_params = re.compile(r':([^\ \,\.\\\(\)\=]+)')
list_params = re.compile(r'\?')
class AndroidSQLConn:
    def __init__(self,device,filepath,use_root=True):
        self.device = device
        self.filepath = filepath
        self.root = use_root

    def _quote_param_(param):
        if isinstance(param,str):
            return "'{}'".format(param.replace("'","''"))
        elif isinstance(param,bool):
            return '1' if param else '0'
        elif param is None:
            return 'NULL'
        else:
            return str(param)
    def _sub_params_(SQL,params):
        params = iter(params)
        try:
            return list_params.sub(lambda match: AndroidSQLConn._quote_param_(next(params)),SQL)
        except StopIteration:
            raise TypeError("Not enough parameters")

    def _sub_params_dict_(SQL,params):
        try:
            return dict_params.sub(lambda match: AndroidSQLConn._quote_param_(params[match.group(1)]),SQL)
        except KeyError:
            raise TypeError("Parameter specified but not in mapping")

    def execute(self,SQL,params = None):
        if params:
            if isinstance(params,dict):
                SQL = AndroidSQLConn._sub_params_dict_(SQL,params)
            else:
                SQL = AndroidSQLConn._sub_params_(SQL,params)
        SQL = shlex.quote(SQL)
        if self.root:
            shell = self.device.sudo
        else:
            shell = self.device.shell
        out = shell('sqlite3','-header','-csv',shlex.quote(self.filepath),SQL,output="out")
        if out:
            return csv.DictReader(out.splitlines())

if __name__ == "__main__":
    print(AndroidSQLConn._sub_params_("SELECT * FROM table WHERE name=?,age=?;",["boby;DROP TABLE table",1]))