单进程交互环境下SQLite数据库锁定：如何不终止进程解锁？

一、不终止进程释放数据库锁的步骤 #

1. 排查并关闭所有未释放的SQLAlchemy会话 #

即使调用了close_all_sessions()，仍可能存在未被追踪的会话实例，可通过GC模块遍历并强制关闭：

import gc
from sqlalchemy.orm import Session

for obj in gc.get_objects():
    if isinstance(obj, Session):
        if not obj.is_closed():
            obj.close()
            print(f"已关闭未释放会话: {obj}")

如果使用了scoped_session，额外执行：

from sqlalchemy.orm import scoped_session
# 替换为你的scoped_session实例
scoped_session_instance.remove()

2. 彻底释放引擎及连接池资源 #

除了engine.dispose()，需清空连接池确保所有底层连接被释放：

engine.dispose()
# 强制清空连接池
if hasattr(engine, 'pool'):
    engine.pool.clear()

3. 回滚残留事务 #

加索引失败可能导致事务未回滚，即使会话关闭，底层连接仍持有锁，需手动执行回滚：

with engine.connect() as conn:
    conn.execute("ROLLBACK")
    conn.commit()

4. 排查底层sqlite3连接/游标 #

如果直接使用过原生sqlite3模块，需检查并关闭未释放的连接或游标：

import gc
import sqlite3

for obj in gc.get_objects():
    if isinstance(obj, sqlite3.Connection):
        if not obj.closed:
            obj.close()
            print(f"已关闭未释放sqlite3连接: {obj}")
    elif isinstance(obj, sqlite3.Cursor):
        obj.close()

二、列出进程中持有文件锁的Python对象 #

借助psutil模块查看进程打开的文件句柄，结合GC模块关联到Python对象：

1. 先安装psutil（Jupyter环境下执行）：

!pip install psutil

2. 执行排查代码：

import psutil
import gc
from sqlalchemy.engine.base import Connection
import sqlite3

# 替换为你的数据库文件路径
DB_FILE = "C:/path/to/your/database.db"

# 获取当前进程的所有打开文件
current_proc = psutil.Process()
open_files = current_proc.open_files()
db_handles = [f for f in open_files if DB_FILE in f.path]

# 遍历Python对象，匹配持有锁的实例
for obj in gc.get_objects():
    # 检查SQLAlchemy连接
    if isinstance(obj, Connection):
        try:
            fd = obj.connection.fileno()
            for handle in db_handles:
                if handle.fd == fd:
                    print(f"持有锁的SQLAlchemy连接: {obj} (文件描述符: {fd})")
        except Exception:
            pass
    # 检查原生sqlite3连接
    elif isinstance(obj, sqlite3.Connection):
        if not obj.closed:
            try:
                fd = obj.fileno()
                for handle in db_handles:
                    if handle.fd == fd:
                        print(f"持有锁的sqlite3连接: {obj} (文件描述符: {fd})")
            except Exception:
                pass
    # 检查sqlite3游标（关联到连接）
    elif isinstance(obj, sqlite3.Cursor):
        try:
            conn = obj.connection
            if not conn.closed:
                fd = conn.fileno()
                for handle in db_handles:
                    if handle.fd == fd:
                        print(f"持有锁的游标（关联连接: {conn}）: {obj}")
        except Exception:
            pass

内容的提问来源于stack exchange，提问作者Ark-kun