如何使用 Python 脚本通过 ADB 运行 sqlite 语句以从 JSON 文件填充表？

Question

虽然标题可能不太好，但我认为这个疑问是合法的。我正在为 Android 开发并在我的持久层上使用 SQLite。为了进行测试（不是单元或仪器测试，只是开发测试，例如打开屏幕并检查一些数据以查看布局）我需要将数据填充到我的数据库中 - 这个这是一件非常烦人的事情，因为我需要插入大量数据并且我不想用它来污染我的代码。

我想以一种非常简单的方式来实现这一点，以便我可以在以后开发的任何应用程序、库或模块中重复使用，因此我决定使用 Android 调试桥 (ADB) 来访问模拟器 shell并从那里运行 sql 脚本以插入我想要的数据 - 从 json 文件中读取。为此，我想出了以下脚本：

import sys
import json
import subprocess

# ADDS ROOT DIR TO PATH
sys.path.insert(0, '..')

COMMANDS = "abd root ; adb remount ; adb shell; sqlite3 {} ; .headers on ; insert into {} ({}) values ({}); exit ;"

ERROR_INVALID_INPUT = "You must specify the TableName and database full path"


def get_columns_and_values(object):
    columns = ""
    values = ""
    for key, value in object.items():
        columns += str(key) + ","
        values += str(value) + ","
    return [columns, values]


def get_table_name():
    return sys.argv[1]


def get_database_path():
    return sys.argv[2]


def validate_input():
    return len(sys.argv) == 3


def run_commands(columns, values):
    formatted_commands = COMMANDS.format(
        get_database_path(),
        get_table_name(),
        columns,
        values
    )

    process = subprocess.Popen(formatted_commands.split(), stdout=subprocess.PIPE)
    output, error = process.communicate()


if validate_input():
    with open(ARG_SAMPLE_JSON_FILE) as json_file:
        data = json.load(json_file)
        for object in data:
            normalizedRow = get_columns_and_values(object)
            run_commands(normalizedRow[0], normalizedRow[1])
else:
    print ERROR_INVALID_INPUT

问题是：脚本从访问 bash 开始，然后从那里进入模拟器的 shell，只是为了进入 sqlite3 命令行。发生这种情况时，我不知道如何使用 Python 在那里运行命令 - 也许问题的标题可能是如何将命令从 Python 运行到 Android Emulator 的 shell 中？ :p

非常感谢任何帮助。

---

PS：我知道有更优雅的方法可以实现我想要的，比如切换 Dagger2 依赖项以提供模拟数据，但这需要大量配置和调试而且，有时，您只是想要一种快速开始在屏幕上查看内容的方法。

Answer 1

好的，我最终发现您可以在一行中运行 sqlite 命令。下面的脚本正是我想要的：

import sys
import json
import subprocess

# ADDS ROOT DIR TO PATH
sys.path.insert(0, '..')

ARG_SAMPLE_JSON_FILE = "resources/WorkingDays.json"

CMD_SQL_INSERT = "adb shell sqlite3 {} \"INSERT INTO {} ({}) VALUES ({})\""
CMD_SQL_SELECT = "adb shell sqlite3 {} \"SELECT {} FROM {}\""
CMD_SQL_DELETE = "adb shell sqlite3 {} \"DELETE FROM {}\""

ACTION_INSERT = 'INSERT'
ACTION_DELETE = 'DELETE'
ACTION_SELECT = 'SELECT'

ERROR_INVALID_INPUT = "You must specify the table name, database full path and SQL action to be executed"
ERROR_INVALID_ACTION = "Unrecognized actions. Valid actions are: SELECT, DELETE and INSERT"


def get_columns_and_values(object):
    columns = ""
    values = ""

    for key, value in object.items():
        columns += str(key) + ","
        if is_number(value):
            values += str(value) + ","
        else:
            values += "'" + str(value) + "'" + ","

    return [columns[:-1], values[:-1].replace("False", '0').replace("True", '1')]


def is_number(s):
    try:
        int(s)
        return True
    except ValueError:
        return False


def get_json_file():
    return sys.argv[1]


def get_table_name():
    return sys.argv[2]


def get_database_path():
    return sys.argv[3]


def get_action():
    return sys.argv[4]


def validate_input():
    return len(sys.argv) == 5


def run_commands(columns, values):
    action = get_action()

    if action == ACTION_SELECT:
        formatted_command = CMD_SQL_SELECT.format(get_database_path(), columns, get_table_name())
    elif action == ACTION_DELETE:
        formatted_command = CMD_SQL_DELETE.format(get_database_path(), get_table_name())
    elif action == ACTION_INSERT:
        formatted_command = CMD_SQL_INSERT.format(get_database_path(), get_table_name(), columns, values)
    else:
        print ERROR_INVALID_ACTION
        return

    print "Executing command: {}".format(formatted_command)

    process = subprocess.Popen(
        formatted_command.split(),
        stdout=subprocess.PIPE
    )

    output, error = process.communicate()

    formatted_output = "OUTPUT: {}".format(output)
    formatted_error = "ERROR: {}".format(error)

    print formatted_output, formatted_error


if validate_input():
    with open(get_json_file()) as json_file:
        data = json.load(json_file)
        for object in data:
            normalizedRow = get_columns_and_values(object)
            run_commands(normalizedRow[0], normalizedRow[1])
else:
    print ERROR_INVALID_INPUT