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]))