使用Python跨SFTP拷贝活跃SQLite数据库时频繁出现数据库镜像畸形问题

我编写了Python脚本，通过SFTP协议跨设备拷贝SQLite数据库文件，但由于源端数据库处于高频写入的活跃状态，拷贝出来的库经常出现镜像畸形（malformed DB）错误。作为Python开发新手，我梳理了几个可能的解决方向，但不清楚具体落地的实现方式：

• 是否存在其他可替代的SQLite数据库拷贝方案？
• 能否跳过文件拷贝步骤，直接在远端设备上查询这份SQLite文件？由于SQLite是典型的嵌入式本地数据库，我不清楚它能否像MySQL这类C/S架构数据库那样实现远程查询。
• 能否增加循环重试逻辑？比如在捕获异常时重新调用拉取函数，但我不确定如何实现后续流程的可靠重试。
• 考虑到数据库镜像畸形问题可能在其他数据读取环节也会触发，是否需要提前运行PRAGMA quick_check对库文件做完整性校验？

以上是我日常运行时遇到的典型场景。我一直困惑的是，为什么这类问题的发生频率这么高？毕竟我在本地设备上手动加载这份SQLite文件执行查询时，文件完全能正常运行。

操作截图参考：

• 第一张为执行PRAGMA quick_check完整性校验的界面截图
• 第二张为普通数据库查询执行的结果截图

运行报错输出如下：

(venv) dulanic@mediaserver:/opt/python_scripts/rpi$  cd /opt/python_scripts/rpi ; /usr/bin/env /opt/python_scripts/rpi/venv/bin/python /home/dulanic/.vscode-server/extensions/ms-python.python-2021.2.636928669/pythonFiles/lib/python/debugpy/launcher 37599 -- /opt/python_scripts/rpi/rpdb.py 
An error occurred: database disk image is malformed

以下是我当前使用的完整脚本代码：

#!/usr/bin/env python3
import psycopg2, sqlite3, sys, paramiko, sys, os, socket, time 

scpuser=os.getenv('scpuser')
scppw = os.getenv('scppw')
sqdb = os.getenv('sqdb')
sqlike = os.getenv('sqlike')
pgdb = os.getenv('pgdb')
pguser = os.getenv('pguser')
pgpswd = os.getenv('pgpswd')
pghost = os.getenv('pghost')
pgport = os.getenv('pgport')
pgschema = os.getenv('pgschema')
database = r"./pihole.db"
pihole = socket.gethostbyname('pi.hole')
tabnames=[]
tabgrab = ''

def pullsqlite(): 
    sftp.get('/etc/pihole/pihole-FTL.db','pihole.db')
    sftp.close()

# SFTP拉取配置
ssh_client=paramiko.SSHClient()
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh_client.connect(hostname=pihole,username=scpuser,password=scppw)
sftp=ssh_client.open_sftp()

# 拉取SQLite文件
pullsqlite()

# 读取SQLite表名列表
consq=sqlite3.connect(sqdb)
cursq=consq.cursor()    
cursq.execute(f"SELECT name FROM sqlite_master WHERE type='table' AND name in ({sqlike})" )
tabgrab = cursq.fetchall()

# 连接Postgres数据库
conpg = psycopg2.connect(database=pgdb, user=pguser, password=pgpswd,
                        host=pghost, port=pgport)

# 将SQLite数据同步到Postgres
for item in tabgrab:
    tabnames.append(item[0])
    start = time.perf_counter() 

for table in tabnames:
    curpg = conpg.cursor()
    if table=='queries':
        curpg.execute(f"SELECT max(id)  FROM {table};")
        max_id = curpg.fetchone()[0]
        cursq.execute(f"SELECT * FROM {table} where id > {max_id};")
    else:
        cursq.execute(f"SELECT * FROM {table};")
    try:     
        rows=cursq.fetchall()
    except sqlite3.Error as e:
        print("An error occurred:", e.args[0])

    colcount=len(rows[0])
    pholder=('%s,'*colcount)[:-1]
 
    try:
        curpg.execute(f"SET search_path TO {pgschema};" )
        curpg.executemany(f"INSERT INTO {table} VALUES ({pholder}) ON CONFLICT DO NOTHING;" ,rows)
        conpg.commit()
        print(f'Inserted {len(rows)} rows into {table}')
 
    except psycopg2.DatabaseError as e:
        print (f'Error {e}') 
        sys.exit(1)

if 'start' in locals():
    elapsed = time.perf_counter() - start
    print(f'Time {elapsed:0.4}')
consq.close()

内容的提问来源于stack exchange，提问作者Dulanic