adb_python/android_db.py

import shlex
import re
import csv
import sys
csv.field_size_limit(2**31-1)
param_str = re.compile(r'(?<=\()[ ,\?]+(?=\))')
validator = re.compile(r'^\?(?: *\, *\?)*$')
param_sep = 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):
        p_string = param_str.search(SQL).group(0)
        if not validator.match(p_string):
            raise ValueError('Invalid substitution')

        n_params = len(param_sep.split(p_string))
        if len(params) < n_params:
            raise ValueError('Not enough parameters supplied')
        new_str = ','.join(map(
            AndroidSQLConn._quote_param_,params[:n_params]
            ))
        print(*map(repr,(new_str,SQL)))
        return param_str.sub(re.escape(new_str),SQL,1)
    def execute(self,SQL,params = None):
        if params:
            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())