python sqlite3表无故消失

我有一个python sqlite3 db，其中有一个表没有明显的原因“消失”-我猜它被删除了。数据库文件仍然存在，并且似乎没有“损坏”：我可以通过CLI连接到它并在上面执行基本操作：

sqlite> SELECT name FROM sqlite_schema WHERE type='table' ORDER BY name;

到目前为止，它只发生在dev中，并且只在重启后几天发生。dev服务器上的流量很少，主要是我测试应用程序。

我欢迎任何关于如何调试此问题的想法。在日志中没有指向表删除的 * 原因 * 的错误。我已经添加了一些详细的日志记录，但我看不到任何导致表删除的情况。只是，在某个时候，它消失了，我开始收到no such table错误：

# Everything looking fine....
2023-03-07 17:14:43,039 139982591223616 - application INFO - Updating hash value for PID: 12933
2023-03-07 17:14:43,039 139982591223616 - application INFO - Connecting to: /some_path/mydatabase.db
2023-03-07 17:14:43,047 139982591223616 - application INFO - Connecting to: /some_path/mydatabase.db
2023-03-07 17:14:43,063 139982591223616 - application INFO - Got 7 cache items. This is 7 more than the previous set.
2023-03-07 17:14:43,064 139982591223616 - application INFO - Connecting to: /some_path/mydatabase.db
2023-03-07 17:14:43,072 139982591223616 - application INFO -
MONOMERS STATUS
(pid=12932)  cache_data_hash=d49e11e052832ed7de03f38fa61c09cabdae66473991ae3e9d02041f019983ae
(pid=12933)  cache_data_hash=d49e11e052832ed7de03f38fa61c09cabdae66473991ae3e9d02041f019983ae *****

2023-03-07 17:14:43,072 139982591223616 - uvicorn.error INFO - Application startup complete.
2023-03-07 17:14:43,072 139982591223616 - uvicorn.error INFO - Uvicorn running on socket ('127.0.0.1', 8200) (Press CTRL+C to quit)

# Then, the next day, suddenly...
2023-03-08 15:46:10,190 140122358679360 - application INFO - Connecting to: /some_path/mydatabase.db
2023-03-08 15:46:10,733 140122358679360 - application ERROR - Traceback (most recent call last):
  File "/some_path/my_app/web_service.py", line 74, in whitelist_authentication_and_log_all_exceptions
    if should_update_cache_data():
  File "/some_path/my_app/operations/update_cache_data.py", line 80, in should_update_cache_data
    return _should_update_cache_data(os.getpid())
  File "/some_path/my_app/operations/update_cache_data.py", line 84, in _should_update_cache_data
    cache_data_hash: Union[str, None] = _get_hash_for_pid(pid)
  File "/some_path/my_app/operations/update_cache_data.py", line 94, in _get_hash_for_pid
    cursor.execute(f"select * from {TABLE_NAME} where pid=?", (pid,))
sqlite3.OperationalError: no such table: MY_CACHE_TABLE

我最好的猜测是，我的数据库访问代码中的某些东西，可能与并发问题（多进程）结合在一起，不知何故破坏了某些东西-未关闭的游标/连接，竞争条件......我已经尽我所能仔细检查了代码。
数据库用于跨多个uvicorn进程的内存缓存同步。PID和缓存数据的散列。当我收到缓存更新通知时，我更新当前进程的缓存，然后通过将PID的散列设置为空字符串来使所有其他进程的缓存无效。每个进程轮询表，如果它发现其PID为空字符串，则更新缓存并设置新的散列。

PID     HASH
2385    d49e11e052832ed7de03f38fa61c09cabdae66473991ae3e9d02041f019983ae
9823    d49e11e052832ed7de03f38fa61c09cabdae66473991ae3e9d02041f019983ae

我在部署期间每次都重新创建数据库（我从不在运行时调用create_db()或create_table()），并在启动时更新该高速缓存和表--不需要在部署/重新启动期间保存任何数据。

def create_db():
    create_table()

def get_database_connection() -> sqlite3.Connection:
    try:
        connection = sqlite3.connect(settings.DB_FILE_NAME)
    except sqlite3.Error as exc:
        raise RuntimeError(f"Cannot connect to {os.path.abspath(settings.DB_FILE_NAME)}") from exc

    return connection

def create_table() -> None:
    """ Establishing a connection creates the db if it doesn't exist. """
    database_connection: sqlite3.Connection = get_database_connection()
    cursor: sqlite3.Cursor

    try:
        cursor = database_connection.cursor()
        cursor.execute(f"DROP TABLE IF EXISTS {TABLE_NAME}")
        cursor.execute(
            f"CREATE TABLE {TABLE_NAME} (PID INTEGER, CACHE_HASH TEXT)")
    except sqlite3.Error as err:
        raise RuntimeError("Error creating database.") from err
    finally:
        cursor.close()
        database_connection.commit()
        database_connection.close()

if __name__ == "__main__":
    create_db()

这里是所有涉及数据库的其他代码。基本上，我尽可能使用上下文管理器（with），否则尝试确保我正确提交和关闭所有内容：

PID = 0
CACHE_DATA_HASH = 1

async def update_cache_data_and_invalidate_cache():
    result_message: dict[str, str] = await update_cache_data()
    invalidate_cache_for_all_other_processes()
    _log_cache_data_info()
    return result_message

async def update_cache_data() -> dict[str, str]:

    cache_data = await _get_cache_data()
    thin_cache_data: CacheData = CacheData(cache_data)
    cache_data_refresh_diff = len(thin_cache_data.items) - len(settings.cache_data.items)
    if len(thin_cache_data.items) > 0:
        settings.cache_data = thin_cache_data
        set_cache_data_up_to_date()
        return {"message": f"Successfully received {len(thin_cache_data.items)} items."}

    error_message: str = ("ERROR: items refresh request did not error, but we received 0 items.")
    logger.error(error_message)
    _log_cache_data_info()
    raise RuntimeError(error_message)

async def _get_cache_data() -> list[dict]:
    async with httpx.AsyncClient(verify=False) as client:
        response = await client.get(cache_data_url, timeout=15.0)
        cache_data: list[dict] = response.json()
        return cache_data

def should_update_cache_data() -> bool:
    return _should_update_cache_data(os.getpid())

def _should_update_cache_data(pid: int) -> bool:
    cache_data_hash: Union[str, None] = _get_hash_for_pid(pid, include_connection_logging=False)
    if cache_data_hash is None or cache_data_hash == '':
        return True
    return False

def _get_hash_for_pid(pid: int, include_connection_logging=True) -> Union[str, None]:
    cache_data_hash: Union[str, None] = None
    with get_database_connection(include_logging=include_connection_logging) as conn:
        cursor: sqlite3.Cursor = conn.cursor()
        cursor.execute(f"select * from {TABLE_NAME} where pid=?", (pid,))
        result: Union[tuple, None] = cursor.fetchone()
        cursor.close()

        if result is not None:
            cache_data_hash = result[CACHE_DATA_HASH]
    return cache_data_hash

def set_cache_data_up_to_date() -> None:
    current_pid: int = os.getpid()
    _set_cache_data_up_to_date(current_pid)

def _set_cache_data_up_to_date(pid: int) -> None:
    cache_data_hash: Union[str, None] = _get_hash_for_pid(pid)
    with get_database_connection() as conn:
        if cache_data_hash is None:
            conn.execute(f"insert into {TABLE_NAME} values (?, ?)", (pid, settings.cache_data.hash))
        else:
            conn.execute(
                f"update {TABLE_NAME} set cache_data_hash = ? where pid = ?", (settings.cache_data.hash, pid))

def invalidate_cache_for_all_other_processes() -> None:
    with get_database_connection() as conn:
        process_ids = []
        for row in conn.execute(f"select * from {TABLE_NAME}"):
            process_ids.append(row[PID])

        # Invalidate the cache for all other processes by setting the hash to an empty string
        this_process_pid: int = os.getpid()
        for pid in process_ids:
            if pid != this_process_pid:
                conn.execute(f"update {TABLE_NAME} set cache_data_hash = ? where pid = ?", ('', pid))

def _generate_cache_data_info() -> str:
    cache_data_info: str = "\nCACHE STATUS\n"
    got_rows: bool = False
    with get_database_connection() as conn:
        for row in conn.execute(f"select * from {TABLE_NAME}"):
            got_rows = True
            if row[PID] == os.getpid():
                cache_data_info += f"(pid={row[PID]})  cache_data_hash={row[CACHE_DATA_HASH]} *****\n"
            else:
                cache_data_info += f"(pid={row[PID]})  cache_data_hash={row[CACHE_DATA_HASH]}\n"
    if not got_rows:
        cache_data_info += f"{TABLE_NAME} is empty."
    return cache_data_info

def _log_cache_data_info() -> None:
    logger.info(_generate_cache_data_info())

python sqlite3表无故消失

1条答案

相关问题

热门标签

最新问答