From e6291fe3a49ae0564267ad83bf15bfce10a00922 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Sat, 27 Sep 2025 21:18:06 +0800 Subject: [PATCH] =?UTF-8?q?modify=20=E7=89=88=E6=9C=AC=E6=9B=B4=E6=96=B0?= =?UTF-8?q?=200.2.0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .python-version | 2 +- pyproject.toml | 14 +- scripts/code_content.py | 28 +- scripts/to_json_utils.py | 26 +- src/core/src/lib.rs | 13 - src/core/src/queue_operation.rs | 530 ------------ src/queue_sqlite/__init__.py | 46 + src/queue_sqlite/constant/__init__.py | 11 + src/queue_sqlite/constant/message_priority.py | 11 + src/queue_sqlite/constant/message_status.py | 11 + src/queue_sqlite/constant/message_type.py | 11 + src/queue_sqlite/core/__init__.py | 0 src/queue_sqlite/core/__init__.pyi | 167 ---- src/queue_sqlite/cycle/__init__.py | 9 + src/queue_sqlite/cycle/async_task_cycle.py | 11 + src/queue_sqlite/cycle/task_cycle.py | 11 + src/queue_sqlite/model/__init__.py | 11 + src/queue_sqlite/model/message_item.py | 73 +- src/queue_sqlite/mounter/__init__.py | 11 + src/queue_sqlite/mounter/listen_mounter.py | 11 + src/queue_sqlite/mounter/task_mounter.py | 11 + src/queue_sqlite/queue_operation/__init__.py | 12 +- .../queue_operation/listen_operation.py | 11 + .../queue_operation/queue_operation.py | 127 --- src/queue_sqlite/scheduler/__init__.py | 11 + src/queue_sqlite/scheduler/_async/__init__.py | 19 +- .../_async/async_listen_data_scheduler.py | 11 + .../_async/async_receive_scheduler.py | 24 +- .../scheduler/_async/async_task_scheduler.py | 24 +- .../scheduler/cleanup_scheduler.py | 35 +- .../scheduler/standard/__init__.py | 19 +- .../standard/listen_data_scheduler.py | 11 + .../scheduler/standard/receive_scheduler.py | 27 +- .../scheduler/standard/task_scheduler.py | 23 +- src/{core => queue_sqlite_core}/.gitignore | 0 src/{core => queue_sqlite_core}/Cargo.lock | 304 ++++++- src/{core => queue_sqlite_core}/Cargo.toml | 16 +- .../pyproject.toml | 26 +- src/queue_sqlite_core/queue_sqlite_core.pyi | 38 + src/queue_sqlite_core/src/lib.rs | 15 + src/queue_sqlite_core/src/queue_operation.rs | 814 ++++++++++++++++++ .../src/task_mounter.rs | 0 tests/listen/listen_data.py | 7 +- tests/test_async_scheduler.py | 4 +- tests/test_core.py | 4 +- tests/test_stress.py | 3 + 46 files changed, 1650 insertions(+), 953 deletions(-) delete mode 100644 src/core/src/lib.rs delete mode 100644 src/core/src/queue_operation.rs delete mode 100644 src/queue_sqlite/core/__init__.py delete mode 100644 src/queue_sqlite/core/__init__.pyi delete mode 100644 src/queue_sqlite/queue_operation/queue_operation.py rename src/{core => queue_sqlite_core}/.gitignore (100%) rename src/{core => queue_sqlite_core}/Cargo.lock (73%) rename src/{core => queue_sqlite_core}/Cargo.toml (50%) rename src/{core => queue_sqlite_core}/pyproject.toml (52%) create mode 100644 src/queue_sqlite_core/queue_sqlite_core.pyi create mode 100644 src/queue_sqlite_core/src/lib.rs create mode 100644 src/queue_sqlite_core/src/queue_operation.rs rename src/{core => queue_sqlite_core}/src/task_mounter.rs (100%) diff --git a/.python-version b/.python-version index 9b1d4ba..c8cfe39 100644 --- a/.python-version +++ b/.python-version @@ -1 +1 @@ -3.13t +3.10 diff --git a/pyproject.toml b/pyproject.toml index e46b01f..f72e061 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,18 +6,20 @@ classifiers = [ "Development Status :: 4 - Beta", "Intended Audience :: Developers", "License :: OSI Approved :: MIT License", - "Programming Language :: Python :: 3.13", + "Programming Language :: Python :: 3", "Topic :: System :: Distributed Computing", "Typing :: Typed", "Operating System :: Microsoft :: Windows", "Operating System :: POSIX :: Linux", ] -dependencies = [] +dependencies = [ + "queue_sqlite_core>=0.2.0", +] description = "A high-performance, SQLite-based distributed task queue system with Rust-backed core operations. Supports task mounting, message listening, priority handling, retry mechanisms, and automatic cleanup of expired messages. Ideal for building reliable, scalable background task processing systems." name = "queue-sqlite" readme = "README.md" requires-python = ">=3.7" -version = "0.1.0" +version = "0.2.0" [project.optional-dependencies] pyside6 = ["PySide6"] @@ -41,9 +43,3 @@ dev = [ [tool.hatch.build.targets.wheel] packages = ["src/queue_sqlite"] -platforms = ["windows_x86_64", "linux_x86_64"] - -[tool.hatch.build] -artifacts = [ - "src/queue_sqlite/core/*.pyd", -] diff --git a/scripts/code_content.py b/scripts/code_content.py index ae621af..b8b54a4 100644 --- a/scripts/code_content.py +++ b/scripts/code_content.py @@ -1,27 +1,29 @@ import os import sys + def generate_markdown_from_py_files(directory, output_file): - with open(output_file, 'w', encoding='utf-8') as md_file: + with open(output_file, "w", encoding="utf-8") as md_file: for root, dirs, files in os.walk(directory): # 排除 venv 目录 - dirs[:] = [d for d in dirs if d != '.venv'] - dirs[:] = [d for d in dirs if d != '.vscode'] - dirs[:] = [d for d in dirs if d != 'scripts'] - dirs[:] = [d for d in dirs if d != 'build'] + dirs[:] = [d for d in dirs if d != ".venv"] + dirs[:] = [d for d in dirs if d != ".vscode"] + dirs[:] = [d for d in dirs if d != "scripts"] + dirs[:] = [d for d in dirs if d != "build"] for file in files: - if file.endswith('.py') or file.endswith('.rs'): + if file.endswith(".py") or file.endswith(".rs"): file_path = os.path.join(root, file) - md_file.write(f'`{file_path}`\n') - md_file.write('```python\n') - with open(file_path, 'r', encoding='utf-8') as py_file: + md_file.write(f"`{file_path}`\n") + md_file.write("```python\n") + with open(file_path, "r", encoding="utf-8") as py_file: md_file.write(py_file.read()) - md_file.write('\n```\n\n') + md_file.write("\n```\n\n") + if __name__ == "__main__": # 指定目录和输出文件名 - target_directory = sys.argv[1] # 替换为你的目标目录 - output_markdown_file = 'output.md' # 输出的 Markdown 文件名 + target_directory = sys.argv[1] # 替换为你的目标目录 + output_markdown_file = "output.md" # 输出的 Markdown 文件名 generate_markdown_from_py_files(target_directory, output_markdown_file) - print(f'Markdown 文件已生成：{output_markdown_file}') + print(f"Markdown 文件已生成：{output_markdown_file}") diff --git a/scripts/to_json_utils.py b/scripts/to_json_utils.py index 6dd5c62..dd66c48 100644 --- a/scripts/to_json_utils.py +++ b/scripts/to_json_utils.py @@ -1,3 +1,13 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : to_json_utils.py +@Time : 2025-09-27 16:36:18 +@Author : chakcy +@Email : 947105045@qq.com +@description : xxxxxxxxx +""" + import sqlite3 import json import tkinter as tk @@ -21,11 +31,11 @@ class SQLiteToJSONConverter: def create_widgets(self): # 主框架 main_frame = ttk.Frame(self.root, padding="10") - main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) + main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) # type: ignore # 数据库选择部分 db_frame = ttk.LabelFrame(main_frame, text="数据库选择", padding="5") - db_frame.grid(row=0, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=5) + db_frame.grid(row=0, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=5) # type: ignore ttk.Entry(db_frame, textvariable=self.db_path, width=70).grid( row=0, column=0, padx=5 @@ -39,13 +49,13 @@ class SQLiteToJSONConverter: # 表选择部分 table_frame = ttk.LabelFrame(main_frame, text="表选择", padding="5") - table_frame.grid(row=1, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=5) + table_frame.grid(row=1, column=0, columnspan=2, sticky=(tk.W, tk.E), pady=5) # type: ignore ttk.Label(table_frame, text="选择表:").grid(row=0, column=0, sticky=tk.W) self.table_combo = ttk.Combobox( table_frame, textvariable=self.selected_table, state="readonly" ) - self.table_combo.grid(row=0, column=1, sticky=(tk.W, tk.E), padx=5) + self.table_combo.grid(row=0, column=1, sticky=(tk.W, tk.E), padx=5) # type: ignore ttk.Button(table_frame, text="加载数据", command=self.load_table_data).grid( row=0, column=2, padx=5 ) @@ -53,7 +63,7 @@ class SQLiteToJSONConverter: # 数据显示部分 data_frame = ttk.LabelFrame(main_frame, text="表数据", padding="5") data_frame.grid( - row=2, column=0, columnspan=2, sticky=(tk.W, tk.E, tk.N, tk.S), pady=5 + row=2, column=0, columnspan=2, sticky=(tk.W, tk.E, tk.N, tk.S), pady=5 # type: ignore ) # 创建Treeview显示数据 @@ -63,9 +73,9 @@ class SQLiteToJSONConverter: hsb = ttk.Scrollbar(data_frame, orient="horizontal", command=self.tree.xview) self.tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set) - self.tree.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) - vsb.grid(row=0, column=1, sticky=(tk.N, tk.S)) - hsb.grid(row=1, column=0, sticky=(tk.W, tk.E)) + self.tree.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) # type: ignore + vsb.grid(row=0, column=1, sticky=(tk.N, tk.S)) # type: ignore + hsb.grid(row=1, column=0, sticky=(tk.W, tk.E)) # type: ignore # 按钮部分 button_frame = ttk.Frame(main_frame) diff --git a/src/core/src/lib.rs b/src/core/src/lib.rs deleted file mode 100644 index 4f6c2d4..0000000 --- a/src/core/src/lib.rs +++ /dev/null @@ -1,13 +0,0 @@ -mod queue_operation; -mod task_mounter; - -use pyo3::prelude::*; -use queue_operation::QueueOperation; -use task_mounter::TaskMounter; // 导入结构体 // 导入结构体 - -#[pymodule(gil_used = false)] -fn core(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> { - m.add_class::()?; - m.add_class::()?; - Ok(()) -} diff --git a/src/core/src/queue_operation.rs b/src/core/src/queue_operation.rs deleted file mode 100644 index 701f6d2..0000000 --- a/src/core/src/queue_operation.rs +++ /dev/null @@ -1,530 +0,0 @@ -use chrono::{DateTime, Duration, Utc}; -use pyo3::prelude::*; -use pyo3::types::*; -use r2d2::Pool; -use r2d2_sqlite::SqliteConnectionManager; -use rusqlite::params; - -#[pyclass] -pub struct QueueOperation { - pool: Pool, -} - -#[pymethods] -impl QueueOperation { - #[new] - pub fn new(queue_path: String) -> Self { - let manager = SqliteConnectionManager::file(queue_path.as_str()); - let pool = Pool::new(manager).unwrap(); - QueueOperation { pool } - } - - pub fn init_db(&self) -> PyResult<()> { - let conn = self - .pool - .get() - .map_err(|e| PyErr::new::(e.to_string()))?; - conn.execute( - " - CREATE TABLE IF NOT EXISTS messages ( - id TEXT PRIMARY KEY, - type TEXT NOT NULL, - status INTEGER NOT NULL, - content TEXT NOT NULL, - createtime DATETIME NOT NULL, - updatetime DATETIME NOT NULL, - result TEXT DEFAULT NULL, - priority INTEGER NOT NULL, - source TEXT NOT NULL, - destination TEXT NOT NULL, - retry_count INTEGER NOT NULL, - expire_time DATETIME DEFAULT NULL, - tags TEXT DEFAULT NULL, - metadata TEXT DEFAULT NULL, - is_deleted INTEGER NOT NULL DEFAULT 0 - ); - CREATE INDEX IF NOT EXISTS idx_status ON messages(status); - CREATE INDEX IF NOT EXISTS idx_priority ON messages(priority); - CREATE INDEX IF NOT EXISTS idx_dequeue ON messages( - status, - priority DESC, - createtime ASC - ) WHERE is_deleted = 0 AND status = 0; - ", - params![], - ) - .map_err(|e| PyErr::new::(e.to_string()))?; - let pragmas = [ - "PRAGMA journal_mode=WAL", - "PRAGMA synchronous=NORMAL", - "PRAGMA cache_size=-20000", - "PRAGMA mmap_size=1073741824", - "PRAGMA temp_store=MEMORY", - "PRAGMA busy_timeout=5000", - ]; - for pragma in pragmas.iter() { - let mut stmt = conn - .prepare(pragma) - .map_err(|e| PyErr::new::(e.to_string()))?; - - let mut rows = stmt - .query([]) - .map_err(|e| PyErr::new::(e.to_string()))?; - - while let Some(_) = rows - .next() - .map_err(|e| PyErr::new::(e.to_string()))? - { - } - } - Ok(()) - } - - pub fn enqueue<'py>( - &self, - py: Python<'py>, - message: &Bound<'py, PyDict>, - ) -> PyResult> { - let mut conn = self - .pool - .get() - .map_err(|e| PyErr::new::(e.to_string()))?; - - let get_field = |key: &str| -> PyResult<_> { - message.get_item(key)?.ok_or_else(|| { - PyErr::new::(format!("{} not found", key)) - }) - }; - - let id = get_field("id")?.extract::()?; - let type_ = get_field("type")?.extract::()?; - let status = get_field("status")?.extract::()?; - let content = get_field("content")?.extract::()?; - let createtime = get_field("createtime")?.extract::()?; - let updatetime = get_field("updatetime")?.extract::()?; - let result = get_field("result")?.extract::()?; - let priority = get_field("priority")?.extract::()?; - let source = get_field("source")?.extract::()?; - let destination = get_field("destination")?.extract::()?; - let retry_count = get_field("retry_count")?.extract::()?; - let expire_time = get_field("expire_time")?.extract::()?; - let tags = get_field("tags")?.extract::()?; - let metadata = get_field("metadata")?.extract::()?; - - let tx = conn - .transaction() - .map_err(|e| PyErr::new::(e.to_string()))?; - tx.execute( - " - INSERT INTO messages ( - id, type, status, content, createtime, - updatetime, result, priority, source, - destination, retry_count, expire_time, tags, metadata - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14) - ", - params![ - id, - type_, - status, - content, - createtime, - updatetime, - result, - priority, - source, - destination, - retry_count, - expire_time, - tags, - metadata - ], - ) - .map_err(|e| PyErr::new::(e.to_string()))?; - tx.commit() - .map_err(|e| PyErr::new::(e.to_string()))?; - Ok(Bound::from(PyString::new(py, &id))) - } - - pub fn dequeue<'py>(&self, py: Python<'py>, size: i32) -> PyResult> { - let conn = self - .pool - .get() - .map_err(|e| PyErr::new::(e.to_string()))?; - let now: DateTime = Utc::now(); - let iso_format = now.to_rfc3339(); - - // Fixed SQL query with proper subquery alias - let mut stmt = conn - .prepare( - "UPDATE messages - SET status = 1 - WHERE id IN ( - SELECT id - FROM ( - SELECT id, ROW_NUMBER() OVER (ORDER BY priority DESC, createtime ASC) AS rn - FROM messages - WHERE is_deleted = 0 - AND status = 0 - AND (expire_time IS NULL OR expire_time > ?1) - ) AS sub - WHERE rn <= ?2 - ) - RETURNING *;", - ) - .map_err(|e| PyErr::new::(e.to_string()))?; - - let rows = PyList::empty(py); - let mapped_rows = stmt - .query_map(params![iso_format, size], |row| { - let dict = PyDict::new(py); - - // Helper to convert PyErr to rusqlite::Error - let to_rusqlite_error = - |e: PyErr| rusqlite::Error::ToSqlConversionFailure(Box::new(e)); - - // Extract and set each field with proper error handling - dict.set_item("id", row.get::<_, String>("id")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("type", row.get::<_, String>("type")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("status", row.get::<_, i32>("status")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("content", row.get::<_, String>("content")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("createtime", row.get::<_, String>("createtime")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("updatetime", row.get::<_, String>("updatetime")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("result", row.get::<_, Option>("result")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("priority", row.get::<_, i32>("priority")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("source", row.get::<_, Option>("source")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("destination", row.get::<_, Option>("destination")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("retry_count", row.get::<_, i32>("retry_count")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("expire_time", row.get::<_, Option>("expire_time")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("tags", row.get::<_, Option>("tags")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("metadata", row.get::<_, Option>("metadata")?) - .map_err(to_rusqlite_error)?; - - // Return the dictionary as Py - Ok(dict) - }) - .map_err(|e| PyErr::new::(e.to_string()))?; - - for row in mapped_rows { - let dict = - row.map_err(|e| PyErr::new::(e.to_string()))?; - rows.append(dict)?; - } - - Ok(rows.into()) - } - - pub fn get_queue_length(&self) -> PyResult { - let conn = self - .pool - .get() - .map_err(|e| PyErr::new::(e.to_string()))?; - let now = Utc::now().to_rfc3339(); - - // Using query_row for single result handling - let count: i32 = conn - .query_row( - "SELECT COUNT(*) FROM messages - WHERE is_deleted = 0 - AND status = ?1 - AND (expire_time IS NULL OR expire_time > ?2)", - params![0, now], - |row| row.get(0), - ) - .map_err(|e| PyErr::new::(e.to_string()))?; - - Ok(count) - } - pub fn get_completed_messages<'py>(&self, py: Python<'py>) -> PyResult> { - let conn = self - .pool - .get() - .map_err(|e| PyErr::new::(e.to_string()))?; - let mut stmt = conn - .prepare( - "SELECT - * - FROM - messages - WHERE - is_deleted = 0 - AND (status = ?1 OR status = ?2)", - ) - .map_err(|e| PyErr::new::(e.to_string()))?; - let rows = PyList::empty(py); - let mapped_rows = stmt - .query_map(params![2, 3], |row| { - let dict = PyDict::new(py); - - // Helper to convert PyErr to rusqlite::Error - let to_rusqlite_error = - |e: PyErr| rusqlite::Error::ToSqlConversionFailure(Box::new(e)); - - // Extract and set each field with proper error handling - dict.set_item("id", row.get::<_, String>("id")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("type", row.get::<_, String>("type")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("status", row.get::<_, i32>("status")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("content", row.get::<_, String>("content")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("createtime", row.get::<_, String>("createtime")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("updatetime", row.get::<_, String>("updatetime")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("result", row.get::<_, Option>("result")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("priority", row.get::<_, i32>("priority")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("source", row.get::<_, Option>("source")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("destination", row.get::<_, Option>("destination")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("retry_count", row.get::<_, i32>("retry_count")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("expire_time", row.get::<_, Option>("expire_time")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("tags", row.get::<_, Option>("tags")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("metadata", row.get::<_, Option>("metadata")?) - .map_err(to_rusqlite_error)?; - - // Return the dictionary as Py - Ok(dict) - }) - .map_err(|e| PyErr::new::(e.to_string()))?; - - for row in mapped_rows { - let dict = - row.map_err(|e| PyErr::new::(e.to_string()))?; - rows.append(dict)?; - } - - Ok(rows.into()) - } - - pub fn get_result<'py>(&self, py: Python<'py>, id: String) -> PyResult> { - let conn = self - .pool - .get() - .map_err(|e| PyErr::new::(e.to_string()))?; - let mut stmt = conn - .prepare("SELECT * FROM messages WHERE id = ?1 AND (status = ?2 OR status = ?3)") - .map_err(|e| PyErr::new::(e.to_string()))?; - let mut mapped_rows = stmt - .query_map(params![&id, 2, 3], |row| { - let dict = PyDict::new(py); - - // Helper to convert PyErr to rusqlite::Error - let to_rusqlite_error = - |e: PyErr| rusqlite::Error::ToSqlConversionFailure(Box::new(e)); - - // Extract and set each field with proper error handling - dict.set_item("id", row.get::<_, String>("id")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("type", row.get::<_, String>("type")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("status", row.get::<_, i32>("status")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("content", row.get::<_, String>("content")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("createtime", row.get::<_, String>("createtime")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("updatetime", row.get::<_, String>("updatetime")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("result", row.get::<_, Option>("result")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("priority", row.get::<_, i32>("priority")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("source", row.get::<_, Option>("source")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("destination", row.get::<_, Option>("destination")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("retry_count", row.get::<_, i32>("retry_count")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("expire_time", row.get::<_, Option>("expire_time")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("tags", row.get::<_, Option>("tags")?) - .map_err(to_rusqlite_error)?; - - dict.set_item("metadata", row.get::<_, Option>("metadata")?) - .map_err(to_rusqlite_error)?; - - // Return the dictionary as Py - Ok(dict) - }) - .map_err(|e| PyErr::new::(e.to_string()))?; - - let dict = mapped_rows - .next() - .transpose() - .map_err(|e| PyErr::new::(e.to_string()))?; - - Ok(dict.unwrap().into()) - } - - pub fn update_status(&self, id: String, status: i32) -> PyResult<()> { - let conn = self - .pool - .get() - .map_err(|e| PyErr::new::(e.to_string()))?; - conn.execute( - " - UPDATE messages SET status = ?1 WHERE id = ?2 - ", - params![status, id], - ) - .map_err(|e| PyErr::new::(e.to_string()))?; - Ok(()) - } - - pub fn update_result(&self, id: String, result: String) -> PyResult<()> { - let conn = self - .pool - .get() - .map_err(|e| PyErr::new::(e.to_string()))?; - conn.execute( - " - UPDATE messages SET result = ?1 WHERE id = ?2 - ", - params![result, id], - ) - .map_err(|e| PyErr::new::(e.to_string()))?; - Ok(()) - } - - pub fn delete_message(&self, id: String) -> PyResult<()> { - let conn = self - .pool - .get() - .map_err(|e| PyErr::new::(e.to_string()))?; - conn.execute( - " - UPDATE messages SET is_deleted = 1 WHERE id = ?1 - ", - params![id], - ) - .map_err(|e| PyErr::new::(e.to_string()))?; - Ok(()) - } - - pub fn remove_expired_messages(&self, days: i64) -> PyResult<()> { - let now: DateTime = Utc::now(); - let expire_time = now - Duration::days(days); - let iso_format = expire_time.to_rfc3339(); - - let conn = self - .pool - .get() - .map_err(|e| PyErr::new::(e.to_string()))?; - - conn.execute( - " - UPDATE messages - SET is_deleted = 1 - WHERE is_deleted = 0 - AND expire_time IS NOT NULL - AND expire_time < ?1 - ", - params![iso_format], - ) - .map_err(|e| PyErr::new::(e.to_string()))?; - - Ok(()) - } - - pub fn clean_expired_messages(&self) -> PyResult<()> { - let now: DateTime = Utc::now(); - let iso_format = now.to_rfc3339(); - let conn = self - .pool - .get() - .map_err(|e| PyErr::new::(e.to_string()))?; - conn.execute( - " - UPDATE messages - SET is_deleted = 1 - WHERE is_deleted = 0 - AND expire_time IS NOT NULL - AND expire_time < ?1 - ", - params![iso_format], - ) - .map_err(|e| PyErr::new::(e.to_string()))?; - Ok(()) - } - - pub fn clean_old_messages(&self, days: i64) -> PyResult<()> { - // 获取七天前的时间 - let now: DateTime = Utc::now(); - let mut old_time = now - Duration::days(days); - let iso_format = old_time.to_rfc3339(); - let conn = self - .pool - .get() - .map_err(|e| PyErr::new::(e.to_string()))?; - conn.execute( - " - UPDATE messages - SET is_deleted = 1 - WHERE is_deleted = 0 - AND status IN (?1, ?2) - AND updatetime < ?3 - ", - params![2, 3, iso_format], - ) - .map_err(|e| PyErr::new::(e.to_string()))?; - Ok(()) - } -} diff --git a/src/queue_sqlite/__init__.py b/src/queue_sqlite/__init__.py index e69de29..0aeb255 100644 --- a/src/queue_sqlite/__init__.py +++ b/src/queue_sqlite/__init__.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : __init__.py +@Time : 2025-09-27 17:05:40 +@Author : chakcy +@Email : 947105045@qq.com +@License : (C)Copyright 2020-2025, chakcy +@package : queue_sqlite +@description : A high-performance, SQLite-based distributed task queue system with Rust-backed core operations. Supports task mounting, message listening, priority handling, retry mechanisms, and automatic cleanup of expired messages. Ideal for building reliable, scalable background task processing systems. +""" + +from . import constant +from . import cycle +from . import model +from . import mounter +from . import queue_operation +from . import scheduler +import logging + +logging.basicConfig( + level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" +) +logger = logging.getLogger(__name__).addHandler(logging.NullHandler()) + +__title__ = "queue_sqlite" +__version__ = "0.1.0" +__author__ = "chakcy" +__email__ = "947105045@qq.com" +__license__ = "MIT" +__copyright__ = "Copyright 2020-2025, chakcy" +__all__ = ["constant", "cycle", "model", "mounter", "queue_operation", "scheduler"] +__version_info__ = tuple(int(i) for i in __version__.split(".")) +__description__ = "A high-performance, SQLite-based distributed task queue system with Rust-backed core operations. Supports task mounting, message listening, priority handling, retry mechanisms, and automatic cleanup of expired messages. Ideal for building reliable, scalable background task processing systems." + + +def get_info(): + return { + "name": __title__, + "version": __version__, + "author": __author__, + "email": __email__, + "license": __license__, + "copyright": __copyright__, + "description": __description__, + } diff --git a/src/queue_sqlite/constant/__init__.py b/src/queue_sqlite/constant/__init__.py index 4e27c92..a711b42 100644 --- a/src/queue_sqlite/constant/__init__.py +++ b/src/queue_sqlite/constant/__init__.py @@ -1,3 +1,14 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : __init__.py +@Time : 2025-09-27 16:56:34 +@Author : chakcy +@Email : 947105045@qq.com +@description : 常量模块 +""" + + from .message_priority import MessagePriority from .message_status import MessageStatus from .message_type import MessageType diff --git a/src/queue_sqlite/constant/message_priority.py b/src/queue_sqlite/constant/message_priority.py index 1ca4c1f..d1e143f 100644 --- a/src/queue_sqlite/constant/message_priority.py +++ b/src/queue_sqlite/constant/message_priority.py @@ -1,3 +1,14 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : message_priority.py +@Time : 2025-09-27 16:56:58 +@Author : chakcy +@Email : 947105045@qq.com +@description : 消息优先级枚举类 +""" + + from enum import IntEnum diff --git a/src/queue_sqlite/constant/message_status.py b/src/queue_sqlite/constant/message_status.py index e4a2ec5..0fee50a 100644 --- a/src/queue_sqlite/constant/message_status.py +++ b/src/queue_sqlite/constant/message_status.py @@ -1,3 +1,14 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : message_status.py +@Time : 2025-09-27 16:57:25 +@Author : chakcy +@Email : 947105045@qq.com +@description : 消息状态枚举类 +""" + + from enum import IntEnum diff --git a/src/queue_sqlite/constant/message_type.py b/src/queue_sqlite/constant/message_type.py index 6e2abfb..f27aad2 100644 --- a/src/queue_sqlite/constant/message_type.py +++ b/src/queue_sqlite/constant/message_type.py @@ -1,3 +1,14 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : message_type.py +@Time : 2025-09-27 16:57:58 +@Author : chakcy +@Email : 947105045@qq.com +@description : 消息类型枚举类 +""" + + from enum import Enum diff --git a/src/queue_sqlite/core/__init__.py b/src/queue_sqlite/core/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/queue_sqlite/core/__init__.pyi b/src/queue_sqlite/core/__init__.pyi deleted file mode 100644 index 16b489c..0000000 --- a/src/queue_sqlite/core/__init__.pyi +++ /dev/null @@ -1,167 +0,0 @@ -from ..mounter import TaskMounter -from typing import Callable, List -from ..constant import MessageStatus - -class core: - class QueueOperation: - def __init__(self, db_path: str): - """ - Initialize the queue operation. - - Args: - db_path (str): The path to the database file. - - """ - ... - - def init_db(self): - """ - Initialize the database. - """ - ... - - def enqueue(self, message_dict: dict) -> str: - """ - Enqueue a message. - - Args: - message_dict (dict): The message to be enqueued. - - Returns: - str: The message id. - - Raises: - ValueError: If the message is not valid. - """ - ... - - def dequeue(self, size: int = 1) -> List[dict]: - """ - Dequeue messages. - - Args: - size (int, optional): The number of messages to dequeue. Defaults to 1. - - Returns: - List[dict]: The dequeued messages. - """ - ... - - def get_queue_length(self) -> int: - """ - Get the length of the queue. - - Returns: - int: The length of the queue. - """ - ... - - def get_completed_messages(self) -> List[dict]: - """ - Get the completed messages. - - Returns: - List[dict]: The completed messages. - """ - ... - - def get_result(self, message_id: str): - """ - Get the result of a message. - - Args: - message_id (str): The message id. - - Returns: - str: The result of the message. - """ - ... - - def update_status(self, message_id: str, status: MessageStatus): - """ - Update the status of a message. - Args: - message_id (str): The message id. - status (MessageStatus): The status of the message. - Raises: - ValueError: If the status is not valid. - """ - ... - - def update_result(self, message_id: str, result: str): - """ - Update the result of a message. - - Args: - message_id (str): The message id. - result (str): The result of the message. - """ - ... - - def delete_message(self, message_id: str): - """ - Delete a message. - Args: - message_id (str): The message id. - Raises: - ValueError: If the message id is not valid. - """ - ... - - def clean_old_messages(self, days: int): - """ - Clean old messages. - - Args: - days (int): The number of days to keep messages. - """ - ... - - def clean_expired_messages(self): - """ - Clean expired messages. - Raises: - Exception: If an unknown error occurs. - """ - ... - - def remove_expired_messages(self, days: int): - """ - Remove expired messages. - Args: - days (int): The number of days to keep messages. - """ - ... - - class TaskMounter: - def __init__(self, task_class: type[TaskMounter]): - """ - Initialize the task mounter. - - Args: - task_class (type[TaskMounter]): The task class. - """ - ... - - def get_task_list(self) -> List[str]: - """ - Get the task list. - - Returns: - List[str]: The task list. - """ - ... - - def get_task_function(self, name: str) -> Callable: - """ - Get the task function. - - Args: - name (str): The name of the task. - - Returns: - Callable: The task function. - Raises: - ValueError: If the task name is not valid. - """ - ... diff --git a/src/queue_sqlite/cycle/__init__.py b/src/queue_sqlite/cycle/__init__.py index e69de29..4a5a0fe 100644 --- a/src/queue_sqlite/cycle/__init__.py +++ b/src/queue_sqlite/cycle/__init__.py @@ -0,0 +1,9 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : __init__.py +@Time : 2025-09-27 16:58:34 +@Author : chakcy +@Email : 947105045@qq.com +@description : 任务周期模块 +""" diff --git a/src/queue_sqlite/cycle/async_task_cycle.py b/src/queue_sqlite/cycle/async_task_cycle.py index b26a77b..18199d5 100644 --- a/src/queue_sqlite/cycle/async_task_cycle.py +++ b/src/queue_sqlite/cycle/async_task_cycle.py @@ -1,3 +1,14 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : async_task_cycle.py +@Time : 2025-09-27 16:59:17 +@Author : chakcy +@Email : 947105045@qq.com +@description : 异步任务周期类 +""" + + from ..model import MessageItem from ..constant import MessageStatus import json diff --git a/src/queue_sqlite/cycle/task_cycle.py b/src/queue_sqlite/cycle/task_cycle.py index c837394..14243ac 100644 --- a/src/queue_sqlite/cycle/task_cycle.py +++ b/src/queue_sqlite/cycle/task_cycle.py @@ -1,3 +1,14 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : task_cycle.py +@Time : 2025-09-27 16:59:31 +@Author : chakcy +@Email : 947105045@qq.com +@description : 任务周期模块 +""" + + from ..model import MessageItem from typing import Callable, Optional from ..constant import MessageStatus diff --git a/src/queue_sqlite/model/__init__.py b/src/queue_sqlite/model/__init__.py index 5e7bb6a..0a18a01 100644 --- a/src/queue_sqlite/model/__init__.py +++ b/src/queue_sqlite/model/__init__.py @@ -1,3 +1,14 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : __init__.py +@Time : 2025-09-27 16:59:58 +@Author : chakcy +@Email : 947105045@qq.com +@description : 模型模块 +""" + + from .message_item import MessageItem diff --git a/src/queue_sqlite/model/message_item.py b/src/queue_sqlite/model/message_item.py index 4f44dcd..00f61af 100644 --- a/src/queue_sqlite/model/message_item.py +++ b/src/queue_sqlite/model/message_item.py @@ -1,3 +1,14 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : message_item.py +@Time : 2025-09-27 17:00:27 +@Author : chakcy +@Email : 947105045@qq.com +@description : 消息数据模型 +""" + + from dataclasses import dataclass, field from typing import Optional, Dict, Any from datetime import datetime @@ -5,6 +16,7 @@ from ..constant import MessageStatus, MessagePriority, MessageType import json from enum import Enum import uuid +import logging @dataclass @@ -115,7 +127,13 @@ class MessageItem: # 处理枚举字段 if "type" in data: if isinstance(data["type"], str): - data["type"] = MessageType(data["type"]) # 使用枚举名称 + # 根据字符串值查找对应的枚举成员 + for member in MessageType: + if member.value == data["type"]: + data["type"] = member + break + else: + raise ValueError(f"Invalid message type: {data['type']}") elif isinstance(data["type"], MessageType): pass # 已经是枚举类型 else: @@ -123,7 +141,13 @@ class MessageItem: if "status" in data: if isinstance(data["status"], str): - data["status"] = MessageStatus(data["status"]) # 使用枚举名称 + # 尝试将字符串转换为整数，如果失败则使用字符串作为枚举名称 + try: + data["status"] = MessageStatus( + int(data["status"]) + ) # 尝试作为枚举值 + except ValueError: + data["status"] = MessageStatus(data["status"]) # 使用枚举名称 elif isinstance(data["status"], int): data["status"] = MessageStatus(data["status"]) # 使用枚举值 elif isinstance(data["status"], MessageStatus): @@ -132,13 +156,38 @@ class MessageItem: raise ValueError(f"Invalid message status: {data['status']}") if "priority" in data: - if isinstance(data["priority"], (int, str)): - data["priority"] = MessagePriority(data["priority"]) # type: ignore + if isinstance(data["priority"], str): + # 尝试将字符串转换为整数，如果失败则使用字符串作为枚举名称 + try: + data["priority"] = MessagePriority( + int(data["priority"]) + ) # 尝试作为枚举值 + except ValueError: + # 如果转换失败，尝试通过枚举名称查找 + for member in MessagePriority: + if member.name == data["priority"]: + data["priority"] = member + break + else: + raise ValueError( + f"Invalid message priority: {data['priority']}" + ) + elif isinstance(data["priority"], int): + data["priority"] = MessagePriority(data["priority"]) # 使用枚举值 elif isinstance(data["priority"], MessagePriority): - pass + pass # 已经是枚举类型 else: raise ValueError("请使用 'int' 或 'str' 类型设置 priority 字段") + if "retry_count" in data: + if isinstance(data["retry_count"], str): + if data["retry_count"].isdigit(): + data["retry_count"] = int(data["retry_count"]) + else: + raise ValueError("请使用 'int' 类型设置 retry_count 字段") + elif isinstance(data["retry_count"], int): + pass # 已经是整数类型 + return cls(**data) def to_dict(self) -> Dict[str, Any]: @@ -235,13 +284,13 @@ class MessageItem: """打印所有字段的信息，包括类型、默认值和描述""" from dataclasses import fields - print(f"=== {cls.__name__} 字段信息 ===") + logging.info(f"=== {cls.__name__} 字段信息 ===") for field_info in fields(cls): - print(f"字段名: {field_info.name}") - print(f"类型: {field_info.type}") - print(f"默认值: {field_info.default}") + logging.info(f"字段名: {field_info.name}") + logging.info(f"类型: {field_info.type}") + logging.info(f"默认值: {field_info.default}") if field_info.default_factory is not None: - print(f"默认工厂: {field_info.default_factory}") + logging.info(f"默认工厂: {field_info.default_factory}") if "description" in field_info.metadata: - print(f"描述: {field_info.metadata['description']}") - print("-" * 50) + logging.info(f"描述: {field_info.metadata['description']}") + logging.info("-" * 50) diff --git a/src/queue_sqlite/mounter/__init__.py b/src/queue_sqlite/mounter/__init__.py index 5df2047..5632403 100644 --- a/src/queue_sqlite/mounter/__init__.py +++ b/src/queue_sqlite/mounter/__init__.py @@ -1,3 +1,14 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : __init__.py +@Time : 2025-09-27 17:00:47 +@Author : chakcy +@Email : 947105045@qq.com +@description : 挂载器模块 +""" + + from .task_mounter import task from .listen_mounter import listener diff --git a/src/queue_sqlite/mounter/listen_mounter.py b/src/queue_sqlite/mounter/listen_mounter.py index eb9ac1d..bbc5bd5 100644 --- a/src/queue_sqlite/mounter/listen_mounter.py +++ b/src/queue_sqlite/mounter/listen_mounter.py @@ -1,3 +1,14 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : listen_mounter.py +@Time : 2025-09-27 17:01:23 +@Author : chakcy +@Email : 947105045@qq.com +@description : 监听装饰器 +""" + + from typing import Callable, List diff --git a/src/queue_sqlite/mounter/task_mounter.py b/src/queue_sqlite/mounter/task_mounter.py index ec523c5..1507d15 100644 --- a/src/queue_sqlite/mounter/task_mounter.py +++ b/src/queue_sqlite/mounter/task_mounter.py @@ -1,3 +1,14 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : task_mounter.py +@Time : 2025-09-27 17:01:38 +@Author : chakcy +@Email : 947105045@qq.com +@description : 任务挂载器 +""" + + from typing import Callable, List diff --git a/src/queue_sqlite/queue_operation/__init__.py b/src/queue_sqlite/queue_operation/__init__.py index b356f3b..0ead17e 100644 --- a/src/queue_sqlite/queue_operation/__init__.py +++ b/src/queue_sqlite/queue_operation/__init__.py @@ -1,3 +1,9 @@ -from .queue_operation import QueueOperation - -__all__ = ["QueueOperation"] \ No newline at end of file +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : __init__.py +@Time : 2025-09-27 17:01:59 +@Author : chakcy +@Email : 947105045@qq.com +@description : 队列操作模块 +""" diff --git a/src/queue_sqlite/queue_operation/listen_operation.py b/src/queue_sqlite/queue_operation/listen_operation.py index 12e5222..19583b4 100644 --- a/src/queue_sqlite/queue_operation/listen_operation.py +++ b/src/queue_sqlite/queue_operation/listen_operation.py @@ -1,3 +1,14 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : listen_operation.py +@Time : 2025-09-27 17:02:18 +@Author : chakcy +@Email : 947105045@qq.com +@description : 监听操作模块 +""" + + from ..mounter.listen_mounter import ListenMounter import sqlite3 from typing import List, Tuple, Union diff --git a/src/queue_sqlite/queue_operation/queue_operation.py b/src/queue_sqlite/queue_operation/queue_operation.py deleted file mode 100644 index 31dd029..0000000 --- a/src/queue_sqlite/queue_operation/queue_operation.py +++ /dev/null @@ -1,127 +0,0 @@ -import os -import hashlib -from typing import Tuple, Union, List -import threading -import random -from ..core import core -from ..constant import MessageStatus - - -class QueueOperation: - def __init__(self, shard_num: int = 4, queue_name: str = "default"): - self.shard_num = shard_num - self.db_dir = os.path.join("cache", queue_name) - self.shard_connections = threading.local() - if not os.path.exists(self.db_dir): - os.makedirs(self.db_dir) - self.init_shards() - - def _get_shard_index(self, message_id: str) -> int: - """计算消息的分片索引""" - hash_obj = hashlib.sha256(message_id.encode()) - return int(hash_obj.hexdigest(), 16) % self.shard_num - - def _get_shard_path(self, shard_index: int) -> str: - """获取分片数据库路径""" - return os.path.join(self.db_dir, f"queue_shard_{shard_index}.db") - - def _get_shard_conn(self, shard_index: int) -> core.QueueOperation: - """获取分片数据库连接（线程安全）""" - if not hasattr(self.shard_connections, "shards"): - self.shard_connections.shards = {} - - if shard_index not in self.shard_connections.shards: - db_path = self._get_shard_path(shard_index) - conn = core.QueueOperation(db_path) - self.shard_connections.shards[shard_index] = conn - - return self.shard_connections.shards[shard_index] - - def init_shards(self): - """初始化分片数据库""" - for i in range(self.shard_num): - conn = self._get_shard_conn(i) - conn.init_db() - - # 入队 - def enqueue(self, message: dict) -> str: - shard_index = self._get_shard_index(message["id"]) - conn = self._get_shard_conn(shard_index) - conn.enqueue(message) - return message["id"] - - def dequeue(self, size: int = 1) -> List[dict]: - messages = [] - collected = 0 - - # 随机轮询分片顺序 - shard_order = list(range(self.shard_num)) - random.shuffle(shard_order) - - for shard_index in shard_order: - if collected >= size: - break - - conn = self._get_shard_conn(shard_index) - shard_messages = conn.dequeue(size - collected) - messages.extend(shard_messages) - collected += len(shard_messages) - - return messages - - # 获取队列长度 - def get_queue_length(self) -> int: - """获取队列中待处理消息的数量 - - Returns: - int: 待处理消息数量 - """ - total = 0 - for i in range(self.shard_num): - conn = self._get_shard_conn(i) - total += conn.get_queue_length() - return total - - # 获取完成/失败的消息 - def get_completed_messages(self) -> List[dict]: - messages = [] - for i in range(self.shard_num): - conn = self._get_shard_conn(i) - messages.extend(conn.get_completed_messages()) - - return messages - - # 获取消息结果 - def get_result(self, message_id: str) -> Tuple[bool, Union[str, dict]]: - conn = self._get_shard_conn(self._get_shard_index(message_id)) - return conn.get_result(message_id) - - # 更新消息状态 - def update_status(self, message_id: str, status: MessageStatus): - conn = self._get_shard_conn(self._get_shard_index(message_id)) - conn.update_status(message_id, status) - - # 更新消息结果 - def update_result(self, message_id: str, result: str): - conn = self._get_shard_conn(self._get_shard_index(message_id)) - conn.update_result(message_id, result) - - # 删除消息 - def delete_message(self, message_id: str): - conn = self._get_shard_conn(self._get_shard_index(message_id)) - conn.delete_message(message_id) - - # 清理7天前的已完成/失败消息 - def clean_old_messages(self, shard_index: int, days: int = 7): - conn = self._get_shard_conn(shard_index) - conn.clean_old_messages(days) - - # 清理过期但未处理的消息 - def clean_expired_messages(self, shard_index: int): - conn = self._get_shard_conn(shard_index) - conn.clean_expired_messages() - - # 彻底删除消息 - def remove_expired_messages(self, shard_index: int, remove_days: int = 30): - conn = self._get_shard_conn(shard_index) - conn.remove_expired_messages(remove_days) diff --git a/src/queue_sqlite/scheduler/__init__.py b/src/queue_sqlite/scheduler/__init__.py index 29fce87..d54c891 100644 --- a/src/queue_sqlite/scheduler/__init__.py +++ b/src/queue_sqlite/scheduler/__init__.py @@ -1,3 +1,14 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : __init__.py +@Time : 2025-09-27 17:02:39 +@Author : chakcy +@Email : 947105045@qq.com +@description : 队列调度器 +""" + + from .standard import StandardQueueScheduler from ._async import AsyncQueueScheduler from .base import BaseScheduler diff --git a/src/queue_sqlite/scheduler/_async/__init__.py b/src/queue_sqlite/scheduler/_async/__init__.py index ba2aaae..7749d78 100644 --- a/src/queue_sqlite/scheduler/_async/__init__.py +++ b/src/queue_sqlite/scheduler/_async/__init__.py @@ -1,10 +1,21 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : __init__.py +@Time : 2025-09-27 17:04:34 +@Author : chakcy +@Email : 947105045@qq.com +@description : 异步调度器 +""" + + from .async_receive_scheduler import AsyncReceiveScheduler from .async_task_scheduler import AsyncTaskScheduler from .async_listen_data_scheduler import AsyncListenDataScheduler from ...model import MessageItem from ...queue_operation.listen_operation import ListenOperation from typing import Callable -from ...queue_operation import QueueOperation +from queue_sqlite_core import ShardedQueueOperation import os from ..base import BaseScheduler from ..cleanup_scheduler import CleanupScheduler @@ -12,8 +23,10 @@ from ..cleanup_scheduler import CleanupScheduler class AsyncQueueScheduler(BaseScheduler): - def __init__(self, receive_thread_num=1, task_thread_num=1, shard_num=4): - self.queue_operation = QueueOperation(shard_num=shard_num) + def __init__( + self, receive_thread_num=1, task_thread_num=1, shard_num=4, queue_name="default" + ): + self.queue_operation = ShardedQueueOperation(shard_num, queue_name=queue_name) self.receive_scheduler = AsyncReceiveScheduler( self.queue_operation, receive_thread_num ) diff --git a/src/queue_sqlite/scheduler/_async/async_listen_data_scheduler.py b/src/queue_sqlite/scheduler/_async/async_listen_data_scheduler.py index 48b7ab7..e5a27b8 100644 --- a/src/queue_sqlite/scheduler/_async/async_listen_data_scheduler.py +++ b/src/queue_sqlite/scheduler/_async/async_listen_data_scheduler.py @@ -1,3 +1,14 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : async_listen_data_scheduler.py +@Time : 2025-09-27 17:04:54 +@Author : chakcy +@Email : 947105045@qq.com +@description : 异步监听数据调度器 +""" + + from ...queue_operation.listen_operation import ListenOperation import asyncio from ...mounter.listen_mounter import ListenMounter diff --git a/src/queue_sqlite/scheduler/_async/async_receive_scheduler.py b/src/queue_sqlite/scheduler/_async/async_receive_scheduler.py index 17b82ae..98dd24c 100644 --- a/src/queue_sqlite/scheduler/_async/async_receive_scheduler.py +++ b/src/queue_sqlite/scheduler/_async/async_receive_scheduler.py @@ -1,13 +1,27 @@ -from ...queue_operation import QueueOperation +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : async_receive_scheduler.py +@Time : 2025-09-27 17:05:08 +@Author : chakcy +@Email : 947105045@qq.com +@description : 异步接收调度器 +""" + + +from queue_sqlite_core import ShardedQueueOperation from ...model import MessageItem from typing import Callable import asyncio import threading import concurrent.futures +import logging class AsyncReceiveScheduler: - def __init__(self, queue_operation: QueueOperation, receive_thread_num: int = 1): + def __init__( + self, queue_operation: ShardedQueueOperation, receive_thread_num: int = 1 + ): self.callbacks = dict() self.receive_thread_num = receive_thread_num self.is_running = False @@ -17,7 +31,7 @@ class AsyncReceiveScheduler: def send_message(self, message: MessageItem, callback: Callable = None): # type: ignore if callback is None: - callback = lambda message: print(message) + callback = lambda message: logging.info(message) # self.queue_operation.enqueue(message.to_dict()) self.queue_operation.enqueue(message.to_dict_by_core()) with self.lock: @@ -63,7 +77,7 @@ class AsyncReceiveScheduler: ) as executor: await loop.run_in_executor(executor, callback, message) except Exception as e: - print(f"回调执行错误: {str(e)}") + logging.error(f"回调执行错误: {str(e)}") finally: self.queue_operation.delete_message(message.id) @@ -77,7 +91,7 @@ class AsyncReceiveScheduler: # 运行异步接收函数直到停止 loop.run_until_complete(self.receive_message()) except Exception as e: - print(f"任务调度出错: {str(e)}") + logging.error(f"任务调度出错: {str(e)}") finally: loop.close() diff --git a/src/queue_sqlite/scheduler/_async/async_task_scheduler.py b/src/queue_sqlite/scheduler/_async/async_task_scheduler.py index ee1841f..7a1f8b8 100644 --- a/src/queue_sqlite/scheduler/_async/async_task_scheduler.py +++ b/src/queue_sqlite/scheduler/_async/async_task_scheduler.py @@ -1,4 +1,15 @@ -from ...queue_operation import QueueOperation +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : async_task_scheduler.py +@Time : 2025-09-27 17:05:22 +@Author : chakcy +@Email : 947105045@qq.com +@description : 异步任务调度器 +""" + + +from queue_sqlite_core import ShardedQueueOperation from ...model import MessageItem from ...constant import MessageStatus from ...cycle.async_task_cycle import AsyncTaskCycle @@ -7,13 +18,14 @@ import asyncio from datetime import datetime import threading import multiprocessing +import logging class AsyncTaskScheduler: def __init__( self, - queue_operation: QueueOperation, + queue_operation: ShardedQueueOperation, task_thread_num: int = multiprocessing.cpu_count() * 2, ): self.task_thread_num = task_thread_num @@ -58,7 +70,7 @@ class AsyncTaskScheduler: self.queue_operation.update_result(message.id, message.result) self.queue_operation.update_status(message.id, message.status) except Exception as e: - print(f"任务结果更新失败: {str(e)}") + logging.error(f"任务结果更新失败: {str(e)}") async def task_callback(self): """单一轮询线程，并行执行任务""" @@ -98,10 +110,10 @@ class AsyncTaskScheduler: message = task.result() self._process_messages(message) except Exception as e: - print(f"处理任务结果时出错: {str(e)}") + logging.error(f"处理任务结果时出错: {str(e)}") except Exception as e: - print(f"任务调度出错: {str(e)}") + logging.error(f"任务调度出错: {str(e)}") await asyncio.sleep(0.1) # 出错时短暂延迟 def _run_task_loop(self): @@ -114,7 +126,7 @@ class AsyncTaskScheduler: # 运行异步任务处理函数直到停止 loop.run_until_complete(self.task_callback()) except Exception as e: - print(f"任务调度出错: {str(e)}") + logging.error(f"任务调度出错: {str(e)}") finally: loop.close() diff --git a/src/queue_sqlite/scheduler/cleanup_scheduler.py b/src/queue_sqlite/scheduler/cleanup_scheduler.py index 175c340..c44a636 100644 --- a/src/queue_sqlite/scheduler/cleanup_scheduler.py +++ b/src/queue_sqlite/scheduler/cleanup_scheduler.py @@ -1,12 +1,27 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : cleanup_scheduler.py +@Time : 2025-09-27 17:02:55 +@Author : chakcy +@Email : 947105045@qq.com +@description : 清理调度器 +""" + + # 新建清理调度器 import threading import time -from ..queue_operation import QueueOperation +from queue_sqlite_core import ShardedQueueOperation +import logging class CleanupScheduler: def __init__( - self, queue_operation: QueueOperation, interval_minutes=60, remove_days=30 + self, + queue_operation: ShardedQueueOperation, + interval_minutes=60, + remove_days=30, ): self.queue_operation = queue_operation self.interval = interval_minutes * 60 # 转换为秒 @@ -14,22 +29,20 @@ class CleanupScheduler: self.cleanup_thread = None self.remove_days = remove_days - for i in range(self.queue_operation.shard_num): - # 清理过期但未处理的消息 - self.queue_operation.clean_expired_messages(i) - # 彻底删除30天前的消息 - self.queue_operation.remove_expired_messages(i, self.remove_days) + # for i in range(self.queue_operation.shard_num): + # 清理过期但未处理的消息 + self.queue_operation.clean_expired_messages() + # 彻底删除30天前的消息 + self.queue_operation.remove_expired_messages(self.remove_days) def cleanup_expired_messages(self): """清理过期消息""" while self.is_running: try: - for i in range(self.queue_operation.shard_num): - # 清理过期但未处理的消息 - self.queue_operation.clean_expired_messages(i) + self.queue_operation.clean_expired_messages() except Exception as e: - print(f"清理消息错误: {str(e)}") + logging.error(f"清理消息错误: {str(e)}") # 休眠等待下次清理 for _ in range(self.interval): diff --git a/src/queue_sqlite/scheduler/standard/__init__.py b/src/queue_sqlite/scheduler/standard/__init__.py index 6f3491e..04de86b 100644 --- a/src/queue_sqlite/scheduler/standard/__init__.py +++ b/src/queue_sqlite/scheduler/standard/__init__.py @@ -1,9 +1,20 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : __init__.py +@Time : 2025-09-27 17:03:16 +@Author : chakcy +@Email : 947105045@qq.com +@description : 标准调度器 +""" + + from .receive_scheduler import ReceiveScheduler from .task_scheduler import TaskScheduler from .listen_data_scheduler import ListenDataScheduler from ...model import MessageItem from typing import Callable -from ...queue_operation import QueueOperation +from queue_sqlite_core import ShardedQueueOperation from ...queue_operation.listen_operation import ListenOperation from ..cleanup_scheduler import CleanupScheduler import os @@ -11,8 +22,10 @@ from ..base import BaseScheduler class StandardQueueScheduler(BaseScheduler): - def __init__(self, receive_thread_num=1, task_thread_num=1, shard_num=4): - self.queue_operation = QueueOperation(shard_num) + def __init__( + self, receive_thread_num=1, task_thread_num=1, shard_num=4, queue_name="default" + ): + self.queue_operation = ShardedQueueOperation(shard_num, queue_name) self.listen_operation = ListenOperation( os.path.join(self.queue_operation.db_dir, "listen.db") ) diff --git a/src/queue_sqlite/scheduler/standard/listen_data_scheduler.py b/src/queue_sqlite/scheduler/standard/listen_data_scheduler.py index fbfe520..ecc0a4d 100644 --- a/src/queue_sqlite/scheduler/standard/listen_data_scheduler.py +++ b/src/queue_sqlite/scheduler/standard/listen_data_scheduler.py @@ -1,3 +1,14 @@ +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : listen_data_scheduler.py +@Time : 2025-09-27 17:03:39 +@Author : chakcy +@Email : 947105045@qq.com +@description : 监听数据调度器 +""" + + from ...queue_operation.listen_operation import ListenOperation from concurrent.futures import ThreadPoolExecutor from ...mounter.listen_mounter import ListenMounter diff --git a/src/queue_sqlite/scheduler/standard/receive_scheduler.py b/src/queue_sqlite/scheduler/standard/receive_scheduler.py index 3d8fc59..d691e9b 100644 --- a/src/queue_sqlite/scheduler/standard/receive_scheduler.py +++ b/src/queue_sqlite/scheduler/standard/receive_scheduler.py @@ -1,13 +1,27 @@ -from ...queue_operation import QueueOperation +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : receive_scheduler.py +@Time : 2025-09-27 17:03:55 +@Author : chakcy +@Email : 947105045@qq.com +@description : 接收调度器 +""" + + +from queue_sqlite_core import ShardedQueueOperation from ...model import MessageItem from typing import Callable from concurrent.futures import ThreadPoolExecutor import threading import time +import logging class ReceiveScheduler: - def __init__(self, queue_operation: QueueOperation, receive_thread_num: int = 1): + def __init__( + self, queue_operation: ShardedQueueOperation, receive_thread_num: int = 1 + ): self.callbacks = dict() self.receive_thread_num = receive_thread_num self.is_running = False @@ -20,7 +34,7 @@ class ReceiveScheduler: def send_message(self, message: MessageItem, callback: Callable = None): # type: ignore if callback is None: - callback = lambda message: print(message) + callback = lambda message: logging.info(f"receive message: {message.id}") # self.queue_operation.enqueue(message.to_dict()) self.queue_operation.enqueue(message.to_dict_by_core()) with self.lock: @@ -49,13 +63,8 @@ class ReceiveScheduler: """安全执行回调函数""" try: callback(message) - # with ThreadPoolExecutor(max_workers=1) as executor: - # future = executor.submit(callback, message) - # future.result(timeout=5.0) - # except TimeoutError: - # print(f"回调执行超时: {message.id}") except Exception as e: - print(f"回调执行错误: {str(e)}") + logging.error(f"回调执行错误: {str(e)}") finally: self.queue_operation.delete_message(message.id) diff --git a/src/queue_sqlite/scheduler/standard/task_scheduler.py b/src/queue_sqlite/scheduler/standard/task_scheduler.py index e11954a..5d2b5b0 100644 --- a/src/queue_sqlite/scheduler/standard/task_scheduler.py +++ b/src/queue_sqlite/scheduler/standard/task_scheduler.py @@ -1,4 +1,16 @@ -from ...queue_operation import QueueOperation +#!/usr/bin/env python +# -*- encoding: utf-8 -*- +""" +@File : task_scheduler.py +@Time : 2025-09-27 17:04:09 +@Author : chakcy +@Email : 947105045@qq.com +@description : 任务调度器 +""" + + +import logging +from queue_sqlite_core import ShardedQueueOperation from ...model import MessageItem from ...constant import MessageStatus from ...cycle.task_cycle import TaskCycle @@ -8,12 +20,13 @@ from datetime import datetime import time import threading import multiprocessing +import json class TaskScheduler: def __init__( self, - queue_operation: QueueOperation, + queue_operation: ShardedQueueOperation, task_thread_num: int = multiprocessing.cpu_count() * 2, ): self.task_thread_num = task_thread_num @@ -51,10 +64,10 @@ class TaskScheduler: def _update_result(self, message): """更新任务结果到数据库""" try: - self.queue_operation.update_result(message.id, message.result) + self.queue_operation.update_result(message.id, json.dumps(message.result)) self.queue_operation.update_status(message.id, message.status) except Exception as e: - print(f"任务结果更新失败: {str(e)}") + logging.error(f"任务结果更新失败: {str(e)}") def task_callback(self): """单一轮询线程，并行执行任务""" @@ -73,7 +86,7 @@ class TaskScheduler: else: time.sleep(0.1) # 适当休眠 except Exception as e: - print(f"任务调度错误: {str(e)}") + logging.error(f"任务调度错误: {str(e)}") time.sleep(1) def start_task_thread(self): diff --git a/src/core/.gitignore b/src/queue_sqlite_core/.gitignore similarity index 100% rename from src/core/.gitignore rename to src/queue_sqlite_core/.gitignore diff --git a/src/core/Cargo.lock b/src/queue_sqlite_core/Cargo.lock similarity index 73% rename from src/core/Cargo.lock rename to src/queue_sqlite_core/Cargo.lock index e3c4bea..2f772a8 100644 --- a/src/core/Cargo.lock +++ b/src/queue_sqlite_core/Cargo.lock @@ -2,6 +2,21 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "addr2line" +version = "0.24.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfbe277e56a376000877090da837660b4427aad530e3028d44e0bffe4f89a1c1" +dependencies = [ + "gimli", +] + +[[package]] +name = "adler2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" + [[package]] name = "ahash" version = "0.8.12" @@ -35,12 +50,36 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" +[[package]] +name = "backtrace" +version = "0.3.75" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6806a6321ec58106fea15becdad98371e28d92ccbc7c8f1b3b6dd724fe8f1002" +dependencies = [ + "addr2line", + "cfg-if", + "libc", + "miniz_oxide", + "object", + "rustc-demangle", + "windows-targets", +] + [[package]] name = "bitflags" version = "2.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1b8e56985ec62d17e9c1001dc89c88ecd7dc08e47eba5ec7c29c7b5eeecde967" +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + [[package]] name = "bumpalo" version = "3.19.0" @@ -76,23 +115,41 @@ dependencies = [ "windows-link", ] -[[package]] -name = "core" -version = "0.1.0" -dependencies = [ - "chrono", - "pyo3", - "r2d2", - "r2d2_sqlite", - "rusqlite", -] - [[package]] name = "core-foundation-sys" version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + +[[package]] +name = "crypto-common" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3" +dependencies = [ + "generic-array", + "typenum", +] + +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", +] + [[package]] name = "fallible-iterator" version = "0.3.0" @@ -105,6 +162,16 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a" +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + [[package]] name = "getrandom" version = "0.3.3" @@ -114,9 +181,15 @@ dependencies = [ "cfg-if", "libc", "r-efi", - "wasi", + "wasi 0.14.2+wasi-0.2.4", ] +[[package]] +name = "gimli" +version = "0.31.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07e28edb80900c19c28f1072f2e8aeca7fa06b23cd4169cefe1af5aa3260783f" + [[package]] name = "hashbrown" version = "0.14.5" @@ -171,6 +244,23 @@ version = "2.0.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f4c7245a08504955605670dbf141fceab975f15ca21570696aebe9d2e71576bd" +[[package]] +name = "io-uring" +version = "0.7.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "046fa2d4d00aea763528b4950358d0ead425372445dc8ff86312b3c69ff7727b" +dependencies = [ + "bitflags", + "cfg-if", + "libc", +] + +[[package]] +name = "itoa" +version = "1.0.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a5f13b858c8d314ee3e8f639011f7ccefe71f97f96e50151fb991f267928e2c" + [[package]] name = "js-sys" version = "0.3.77" @@ -214,6 +304,12 @@ version = "0.4.27" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13dc2df351e3202783a1fe0d44375f7295ffb4049267b0f3018346dc122a1d94" +[[package]] +name = "memchr" +version = "2.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f52b00d39961fc5b2736ea853c9cc86238e165017a493d1d5c8eac6bdc4cc273" + [[package]] name = "memoffset" version = "0.9.1" @@ -223,6 +319,26 @@ dependencies = [ "autocfg", ] +[[package]] +name = "miniz_oxide" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316" +dependencies = [ + "adler2", +] + +[[package]] +name = "mio" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78bed444cc8a2160f01cbcf811ef18cac863ad68ae8ca62092e8db51d51c761c" +dependencies = [ + "libc", + "wasi 0.11.1+wasi-snapshot-preview1", + "windows-sys", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -232,6 +348,15 @@ dependencies = [ "autocfg", ] +[[package]] +name = "object" +version = "0.36.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "62948e14d923ea95ea2c7c86c71013138b66525b86bdc08d2dcc262bdb497b87" +dependencies = [ + "memchr", +] + [[package]] name = "once_cell" version = "1.21.3" @@ -261,6 +386,12 @@ dependencies = [ "windows-targets", ] +[[package]] +name = "pin-project-lite" +version = "0.2.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b3cff922bd51709b605d9ead9aa71031d81447142d828eb4a6eba76fe619f9b" + [[package]] name = "pkg-config" version = "0.3.32" @@ -353,6 +484,24 @@ dependencies = [ "syn", ] +[[package]] +name = "queue_sqlite_core" +version = "0.2.0" +dependencies = [ + "chrono", + "pyo3", + "r2d2", + "r2d2_sqlite", + "rand", + "rusqlite", + "serde", + "serde_json", + "sha2", + "thiserror", + "tokio", + "uuid", +] + [[package]] name = "quote" version = "1.0.40" @@ -442,12 +591,24 @@ dependencies = [ "smallvec", ] +[[package]] +name = "rustc-demangle" +version = "0.1.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56f7d92ca342cea22a06f2121d944b4fd82af56988c270852495420f961d4ace" + [[package]] name = "rustversion" version = "1.0.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a0d197bd2c9dc6e53b84da9556a69ba4cdfab8619eb41a8bd1cc2027a0f6b1d" +[[package]] +name = "ryu" +version = "1.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "28d3b2b1366ec20994f1fd18c3c594f05c5dd4bc44d8bb0c1c632c8d6829481f" + [[package]] name = "scheduled-thread-pool" version = "0.2.7" @@ -463,12 +624,72 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" +[[package]] +name = "serde" +version = "1.0.226" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0dca6411025b24b60bfa7ec1fe1f8e710ac09782dca409ee8237ba74b51295fd" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.226" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba2ba63999edb9dac981fb34b3e5c0d111a69b0924e253ed29d83f7c99e966a4" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.226" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8db53ae22f34573731bafa1db20f04027b2d25e02d8205921b569171699cdb33" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.145" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "402a6f66d8c709116cf22f558eab210f5a50187f702eb4d7e5ef38d9a7f1c79c" +dependencies = [ + "itoa", + "memchr", + "ryu", + "serde", + "serde_core", +] + +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "shlex" version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +[[package]] +name = "slab" +version = "0.4.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7a2ae44ef20feb57a68b23d846850f861394c2e02dc425a50098ae8c90267589" + [[package]] name = "smallvec" version = "1.15.1" @@ -492,6 +713,46 @@ version = "0.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e502f78cdbb8ba4718f566c418c52bc729126ffd16baee5baa718cf25dd5a69a" +[[package]] +name = "thiserror" +version = "2.0.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3467d614147380f2e4e374161426ff399c91084acd2363eaf549172b3d5e60c0" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6c5e1be1c48b9172ee610da68fd9cd2770e7a4056cb3fc98710ee6906f0c7960" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tokio" +version = "1.47.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "89e49afdadebb872d3145a5638b59eb0691ea23e46ca484037cfab3b76b95038" +dependencies = [ + "backtrace", + "io-uring", + "libc", + "mio", + "pin-project-lite", + "slab", +] + +[[package]] +name = "typenum" +version = "1.18.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1dccffe3ce07af9386bfd29e80c0ab1a8205a2fc34e4bcd40364df902cfa8f3f" + [[package]] name = "unicode-ident" version = "1.0.18" @@ -506,9 +767,9 @@ checksum = "7264e107f553ccae879d21fbea1d6724ac785e8c3bfc762137959b5802826ef3" [[package]] name = "uuid" -version = "1.17.0" +version = "1.18.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3cf4199d1e5d15ddd86a694e4d0dffa9c323ce759fea589f00fef9d81cc1931d" +checksum = "2f87b8aa10b915a06587d0dec516c282ff295b475d94abf425d62b57710070a2" dependencies = [ "getrandom", "js-sys", @@ -528,6 +789,12 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + [[package]] name = "wasi" version = "0.14.2+wasi-0.2.4" @@ -654,6 +921,15 @@ dependencies = [ "windows-link", ] +[[package]] +name = "windows-sys" +version = "0.59.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" +dependencies = [ + "windows-targets", +] + [[package]] name = "windows-targets" version = "0.52.6" diff --git a/src/core/Cargo.toml b/src/queue_sqlite_core/Cargo.toml similarity index 50% rename from src/core/Cargo.toml rename to src/queue_sqlite_core/Cargo.toml index 77b073f..da53c01 100644 --- a/src/core/Cargo.toml +++ b/src/queue_sqlite_core/Cargo.toml @@ -1,11 +1,11 @@ [package] -name = "core" -version = "0.1.0" +name = "queue_sqlite_core" +version = "0.2.0" edition = "2021" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [lib] -name = "core" +name = "queue_sqlite_core" crate-type = ["cdylib"] [dependencies] @@ -13,4 +13,14 @@ chrono = "0.4.41" pyo3 = "0.25.0" r2d2 = "0.8" r2d2_sqlite = "0.25" +rand = "0.9.2" rusqlite = { version = "0.32", features = ["bundled"] } +serde = { version = "1.0", features = ["derive"] } +serde_json = "1.0.145" +sha2 = "0.10.9" +thiserror = "2.0.16" +tokio = "1.47.1" +uuid = "1.18.1" + +[package.metadata.maturin] +generate-importlib = true diff --git a/src/core/pyproject.toml b/src/queue_sqlite_core/pyproject.toml similarity index 52% rename from src/core/pyproject.toml rename to src/queue_sqlite_core/pyproject.toml index 1a19c88..abe82d3 100644 --- a/src/core/pyproject.toml +++ b/src/queue_sqlite_core/pyproject.toml @@ -1,9 +1,5 @@ -[build-system] -requires = ["maturin>=1.9,<2.0"] -build-backend = "maturin" - [project] -name = "core" +name = "queue_sqlite_core" requires-python = ">=3.7" classifiers = [ "Programming Language :: Rust", @@ -11,5 +7,25 @@ classifiers = [ "Programming Language :: Python :: Implementation :: PyPy", ] dynamic = ["version"] +description = "Python bindings for queue_sqlite_core Rust library" +dependencies = [] + +[build-system] +requires = ["maturin>=1.9,<2.0"] +build-backend = "maturin" + [tool.maturin] features = ["pyo3/extension-module"] + +[[tool.uv.index]] +default = true +url = "http://124.71.68.6:3141/root/pypi" + +[dependency-groups] +dev = [ + "twine>=4.0.2", + "maturin>=1.9.4", +] + +[[tool.uv.index]] +url = "https://pypi.tuna.tsinghua.edu.cn/simple" diff --git a/src/queue_sqlite_core/queue_sqlite_core.pyi b/src/queue_sqlite_core/queue_sqlite_core.pyi new file mode 100644 index 0000000..1a1ee37 --- /dev/null +++ b/src/queue_sqlite_core/queue_sqlite_core.pyi @@ -0,0 +1,38 @@ +from typing import Callable + +class QueueOperation: + def __init__(self, queue_path: str): ... + def init_db(self) -> None: ... + def enqueue(self, message: dict) -> str: ... + def dequeue(self, size: int = 1) -> list[dict]: ... + def get_queue_length(self) -> int: ... + def get_completed_messages(self) -> list[dict]: ... + def get_result(self, id: str) -> dict: ... + def update_status(self, id: str, status: str) -> None: ... + def update_result(self, id: str, result: str) -> None: ... + def delete_message(self, id: str) -> None: ... + def clean_expired_messages(self) -> None: ... + def clean_old_messages(self, days: int) -> None: ... + def remove_expired_messages(self, days: int) -> None: ... + +class ShardedQueueOperation: + @property + def db_dir(self) -> str: ... + def __init__(self, shard_num: int, queue_name: str): ... + def _get_shard_index(self, message_id: str) -> int: ... + def enqueue(self, message: dict) -> str: ... + def dequeue(self, size: int = 1) -> list[dict]: ... + def get_queue_length(self) -> int: ... + def get_completed_messages(self) -> list[dict]: ... + def get_result(self, id: str) -> dict: ... + def update_status(self, id: str, status: str) -> None: ... + def update_result(self, id: str, result: str) -> None: ... + def delete_message(self, id: str) -> None: ... + def clean_expired_messages(self) -> None: ... + def clean_old_messages(self, days: int) -> None: ... + def remove_expired_messages(self, days: int) -> None: ... + +class TaskMounter: + def __init__(self, task_mounter_class: type): ... + def get_task_list(self) -> list[str]: ... + def get_task_function(self, task_name: str) -> Callable: ... diff --git a/src/queue_sqlite_core/src/lib.rs b/src/queue_sqlite_core/src/lib.rs new file mode 100644 index 0000000..bed42b0 --- /dev/null +++ b/src/queue_sqlite_core/src/lib.rs @@ -0,0 +1,15 @@ +mod queue_operation; +mod task_mounter; + +use pyo3::prelude::*; +use queue_operation::{QueueOperation, ShardedQueueOperation}; +use task_mounter::TaskMounter; // 导入结构体 // 导入结构体 + +#[pymodule] +// #[pymodule(gil_used = false)] +fn queue_sqlite_core(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> { + m.add_class::()?; + m.add_class::()?; + m.add_class::()?; + Ok(()) +} diff --git a/src/queue_sqlite_core/src/queue_operation.rs b/src/queue_sqlite_core/src/queue_operation.rs new file mode 100644 index 0000000..412d8ea --- /dev/null +++ b/src/queue_sqlite_core/src/queue_operation.rs @@ -0,0 +1,814 @@ +// src/lib.rs +use chrono::{DateTime, Duration, Utc}; +use pyo3::prelude::*; +use pyo3::types::{PyDict, PyList}; +use r2d2::Pool; +use r2d2_sqlite::SqliteConnectionManager; +use rand::seq::SliceRandom; +use rusqlite::{params, Connection, Row}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use std::collections::HashMap; +use std::path::Path; +use std::sync::{Arc, Mutex}; + +/// 单个分片的队列操作 +#[pyclass] +pub struct QueueOperation { + pool: Pool, +} + +#[derive(Debug, Serialize, Deserialize)] +struct Message { + id: String, + message_type: String, + status: i32, + content: String, + createtime: String, + updatetime: String, + result: Option, + priority: i32, + source: String, + destination: String, + retry_count: i32, + expire_time: Option, + tags: Option, + metadata: Option, + is_deleted: i32, +} + +impl TryFrom<&Row<'_>> for Message { + type Error = rusqlite::Error; + + fn try_from(row: &Row) -> Result { + Ok(Message { + id: row.get("id")?, + message_type: row.get("type")?, + status: row.get("status")?, + content: row.get("content")?, + createtime: row.get("createtime")?, + updatetime: row.get("updatetime")?, + result: row.get("result")?, + priority: row.get("priority")?, + source: row.get("source")?, + destination: row.get("destination")?, + retry_count: row.get("retry_count")?, + expire_time: row.get("expire_time")?, + tags: row.get("tags")?, + metadata: row.get("metadata")?, + is_deleted: row.get("is_deleted")?, + }) + } +} + +#[pymethods] +impl QueueOperation { + #[new] + pub fn new(queue_path: String) -> PyResult { + let manager = SqliteConnectionManager::file(queue_path); + let pool = Pool::new(manager) + .map_err(|e| PyErr::new::(e.to_string()))?; + + Ok(QueueOperation { pool }) + } + + /// 初始化数据库和表 + /// Args: + /// - self: QueueOperation 实例 + /// Returns: + /// - None + pub fn init_db(&self) -> PyResult<()> { + let conn = self.get_connection()?; + + // 创建消息表 + conn.execute( + "CREATE TABLE IF NOT EXISTS messages ( + id TEXT PRIMARY KEY, + type TEXT NOT NULL, + status INTEGER NOT NULL, + content TEXT NOT NULL, + createtime DATETIME NOT NULL, + updatetime DATETIME NOT NULL, + result TEXT, + priority INTEGER NOT NULL, + source TEXT NOT NULL, + destination TEXT NOT NULL, + retry_count INTEGER NOT NULL, + expire_time DATETIME, + tags TEXT, + metadata TEXT, + is_deleted INTEGER NOT NULL DEFAULT 0 + )", + [], + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + + // 创建索引 + let indexes = [ + "CREATE INDEX IF NOT EXISTS idx_status ON messages(status)", + "CREATE INDEX IF NOT EXISTS idx_priority ON messages(priority)", + "CREATE INDEX IF NOT EXISTS idx_dequeue ON messages(status, priority DESC, createtime ASC) WHERE is_deleted = 0 AND status = 0", + "CREATE INDEX IF NOT EXISTS idx_expire_time ON messages(expire_time) WHERE expire_time IS NOT NULL", + ]; + + for index_sql in indexes.iter() { + conn.execute(index_sql, []) + .map_err(|e| PyErr::new::(e.to_string()))?; + } + + // 设置 SQLite 优化参数 + let pragmas = [ + "PRAGMA journal_mode=WAL", + "PRAGMA synchronous=NORMAL", + "PRAGMA cache_size=-20000", + "PRAGMA mmap_size=134217728", + "PRAGMA temp_store=MEMORY", + "PRAGMA busy_timeout=3000", + ]; + + for pragma in pragmas.iter() { + conn.execute_batch(pragma) + .map_err(|e| PyErr::new::(e.to_string()))?; + } + + Ok(()) + } + + /// 入队 + /// Args: + /// - self: QueueOperation 实例 + /// - message: PyDict 实例，包含消息内容 + /// Returns: + /// - String: 消息ID + pub fn enqueue(&self, message: &Bound<'_, PyDict>) -> PyResult { + let mut conn = self.get_connection()?; + + let required_fields = [ + "id", + "type", + "status", + "content", + "createtime", + "updatetime", + "priority", + "source", + "destination", + "retry_count", + ]; + + for field in required_fields.iter() { + if !message.contains(field)? { + return Err(PyErr::new::(format!( + "Missing required field: {}", + field + ))); + } + } + + let tx = conn + .transaction() + .map_err(|e| PyErr::new::(e.to_string()))?; + + // 从PyDict获取值 + let id: String = message.get_item("id")?.unwrap().extract()?; + let message_type: String = message.get_item("type")?.unwrap().extract()?; + let status: i32 = message.get_item("status")?.unwrap().extract()?; + let content: String = message.get_item("content")?.unwrap().extract()?; + let createtime: String = message.get_item("createtime")?.unwrap().extract()?; + let updatetime: String = message.get_item("updatetime")?.unwrap().extract()?; + let priority: i32 = message.get_item("priority")?.unwrap().extract()?; + let source: String = message.get_item("source")?.unwrap().extract()?; + let destination: String = message.get_item("destination")?.unwrap().extract()?; + let retry_count: i32 = message.get_item("retry_count")?.unwrap().extract()?; + + // 可选字段 + let result: Option = match message.get_item("result")? { + Some(item) => { + if item.is_none() { + None + } else { + Some(item.extract()?) + } + } + None => None, + }; + + let expire_time: Option = match message.get_item("expire_time")? { + Some(item) => { + if item.is_none() { + None + } else { + Some(item.extract()?) + } + } + None => None, + }; + + let tags: Option = match message.get_item("tags")? { + Some(item) => { + if item.is_none() { + None + } else { + Some(item.extract()?) + } + } + None => None, + }; + + let metadata: Option = match message.get_item("metadata")? { + Some(item) => { + if item.is_none() { + None + } else { + Some(item.extract()?) + } + } + None => None, + }; + + tx.execute( + "INSERT INTO messages ( + id, type, status, content, createtime, updatetime, result, + priority, source, destination, retry_count, expire_time, tags, metadata + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)", + params![ + id, + message_type, + status, + content, + createtime, + updatetime, + result, + priority, + source, + destination, + retry_count, + expire_time, + tags, + metadata, + ], + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + + tx.commit() + .map_err(|e| PyErr::new::(e.to_string()))?; + + Ok(id) + } + + /// 出队 + /// Args: + /// - self: QueueOperation 实例 + /// - size: i32, 获取消息数量 + /// Returns: + /// - Vec>: 获取的消息列表 + pub fn dequeue(&self, size: i32) -> PyResult>> { + let limited_size = std::cmp::min(size, 500); + let conn = self.get_connection()?; + let now = Utc::now().to_rfc3339(); + + let mut stmt = conn + .prepare( + "UPDATE messages + SET status = 1, updatetime = ?1 + WHERE id IN ( + SELECT id FROM messages + WHERE is_deleted = 0 + AND status = 0 + AND (expire_time IS NULL OR expire_time > ?1) + ORDER BY priority DESC, createtime ASC + LIMIT ?2 + ) + RETURNING *", + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + + let rows = stmt + .query_map(params![now, limited_size], |row| { + let message: Message = row.try_into()?; + let mut map = HashMap::new(); + map.insert("id".to_string(), message.id); + map.insert("type".to_string(), message.message_type); + map.insert("status".to_string(), message.status.to_string()); + map.insert("content".to_string(), message.content); + map.insert("createtime".to_string(), message.createtime); + map.insert("updatetime".to_string(), message.updatetime); + map.insert("result".to_string(), message.result.unwrap_or_default()); + map.insert("priority".to_string(), message.priority.to_string()); + map.insert("source".to_string(), message.source); + map.insert("destination".to_string(), message.destination); + map.insert("retry_count".to_string(), message.retry_count.to_string()); + map.insert( + "expire_time".to_string(), + message.expire_time.unwrap_or_default(), + ); + map.insert("tags".to_string(), message.tags.unwrap_or_default()); + map.insert("metadata".to_string(), message.metadata.unwrap_or_default()); + Ok(map) + }) + .map_err(|e| PyErr::new::(e.to_string()))?; + + let mut result = Vec::new(); + for row in rows { + result.push( + row.map_err(|e| PyErr::new::(e.to_string()))?, + ); + if result.len() > 1000 { + break; + } + } + + Ok(result) + } + + /// 获取队列长度 + /// Args: + /// - self: QueueOperation 实例 + /// Returns: + /// - i32: 队列长度 + pub fn get_queue_length(&self) -> PyResult { + let conn = self.get_connection()?; + let now = Utc::now().to_rfc3339(); + + let count: i32 = conn + .query_row( + "SELECT COUNT(*) FROM messages + WHERE is_deleted = 0 + AND status = 0 + AND (expire_time IS NULL OR expire_time > ?1)", + params![now], + |row| row.get(0), + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + + Ok(count) + } + + /// 获取已完成的消息 + /// Args: + /// - self: QueueOperation 实例 + /// Returns: + /// - Vec>: 已完成的消息列表 + pub fn get_completed_messages(&self) -> PyResult>> { + let conn = self.get_connection()?; + + let mut stmt = conn + .prepare( + "SELECT * FROM messages + WHERE is_deleted = 0 + AND (status = 2 OR status = 3) + LIMIT 10000", + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + + let rows = stmt + .query_map([], |row| { + let message: Message = row.try_into()?; + let mut map = HashMap::new(); + map.insert("id".to_string(), message.id); + map.insert("type".to_string(), message.message_type); + map.insert("status".to_string(), message.status.to_string()); + map.insert("content".to_string(), message.content); + map.insert("createtime".to_string(), message.createtime); + map.insert("updatetime".to_string(), message.updatetime); + map.insert("result".to_string(), message.result.unwrap_or_default()); + map.insert("priority".to_string(), message.priority.to_string()); + map.insert("source".to_string(), message.source); + map.insert("destination".to_string(), message.destination); + map.insert("retry_count".to_string(), message.retry_count.to_string()); + map.insert( + "expire_time".to_string(), + message.expire_time.unwrap_or_default(), + ); + map.insert("tags".to_string(), message.tags.unwrap_or_default()); + map.insert("metadata".to_string(), message.metadata.unwrap_or_default()); + Ok(map) + }) + .map_err(|e| PyErr::new::(e.to_string()))?; + + let mut result = Vec::new(); + for row in rows { + result.push( + row.map_err(|e| PyErr::new::(e.to_string()))?, + ); + if result.len() > 1000 { + break; + } + } + + Ok(result) + } + + /// 获取消息详情 + /// Args: + /// - self: QueueOperation 实例 + /// - id: String, 消息ID + /// Returns: + /// - HashMap: 消息详情 + pub fn get_result(&self, id: String) -> PyResult> { + let conn = self.get_connection()?; + + let mut stmt = conn + .prepare( + "SELECT * FROM messages + WHERE id = ?1 + AND (status = 2 OR status = 3)", + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + + let message: Message = stmt + .query_row(params![id], |row| row.try_into()) + .map_err(|e| { + // 更好地处理查询不到结果的情况 + if e == rusqlite::Error::QueryReturnedNoRows { + PyErr::new::( + "Message not found or not completed", + ) + } else { + PyErr::new::(e.to_string()) + } + })?; + + let mut result = HashMap::new(); + result.insert("id".to_string(), message.id); + result.insert("type".to_string(), message.message_type); + result.insert("status".to_string(), message.status.to_string()); + result.insert("content".to_string(), message.content); + result.insert("createtime".to_string(), message.createtime); + result.insert("updatetime".to_string(), message.updatetime); + result.insert("result".to_string(), message.result.unwrap_or_default()); + result.insert("priority".to_string(), message.priority.to_string()); + result.insert("source".to_string(), message.source); + result.insert("destination".to_string(), message.destination); + result.insert("retry_count".to_string(), message.retry_count.to_string()); + result.insert( + "expire_time".to_string(), + message.expire_time.unwrap_or_default(), + ); + result.insert("tags".to_string(), message.tags.unwrap_or_default()); + result.insert("metadata".to_string(), message.metadata.unwrap_or_default()); + + Ok(result) + } + + /// 更新消息状态 + /// Args: + /// - self: QueueOperation 实例 + /// - id: String, 消息ID + /// - status: i32, 新状态 + /// Returns: + /// - () : 无返回值 + pub fn update_status(&self, id: String, status: i32) -> PyResult<()> { + let conn = self.get_connection()?; + let now = Utc::now().to_rfc3339(); + + conn.execute( + "UPDATE messages SET status = ?1, updatetime = ?2 WHERE id = ?3", + params![status, now, id], + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + + Ok(()) + } + + /// 更新消息结果 + /// Args: + /// - self: QueueOperation 实例 + /// - id: String, 消息ID + /// - result: String, 新结果 + /// Returns: + /// - () : 无返回值 + pub fn update_result(&self, id: String, result: String) -> PyResult<()> { + let conn = self.get_connection()?; + let now = Utc::now().to_rfc3339(); + + conn.execute( + "UPDATE messages SET result = ?1, updatetime = ?2 WHERE id = ?3", + params![result, now, id], + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + + Ok(()) + } + + /// 删除消息 + /// Args: + /// - self: QueueOperation 实例 + /// - id: String, 删除的消息ID + /// Returns: + /// - () : 无返回值 + pub fn delete_message(&self, id: String) -> PyResult<()> { + let conn = self.get_connection()?; + + conn.execute( + "UPDATE messages SET is_deleted = 1 WHERE id = ?1", + params![id], + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + + Ok(()) + } + + /// 清理过期消息 + /// Args: + /// - self: QueueOperation 实例 + /// Returns: + /// - () : 无返回值 + pub fn clean_expired_messages(&self) -> PyResult<()> { + let conn = self.get_connection()?; + let now = Utc::now().to_rfc3339(); + + conn.execute( + "UPDATE messages SET is_deleted = 1 + WHERE is_deleted = 0 + AND expire_time IS NOT NULL + AND expire_time < ?1", + params![now], + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + + Ok(()) + } + + /// 清理旧消息 + /// Args: + /// - self: QueueOperation 实例 + /// - days: i64, 清理旧消息的天数 + /// Returns: + /// - () : 无返回值 + pub fn clean_old_messages(&self, days: i64) -> PyResult<()> { + let conn = self.get_connection()?; + let cutoff_time = (Utc::now() - Duration::days(days)).to_rfc3339(); + + conn.execute( + "UPDATE messages SET is_deleted = 1 + WHERE is_deleted = 0 + AND status IN (2, 3) + AND updatetime < ?1", + params![cutoff_time], + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + + Ok(()) + } + + /// 删除过期消息 + /// Args: + /// - self: QueueOperation 实例 + /// - days: i64, 删除过期消息的天数 + /// Returns: + /// - () : 无返回值 + pub fn remove_expired_messages(&self, days: i64) -> PyResult<()> { + let now: DateTime = Utc::now(); + let expire_time = now - Duration::days(days); + let iso_format = expire_time.to_rfc3339(); + + let conn = self + .pool + .get() + .map_err(|e| PyErr::new::(e.to_string()))?; + + conn.execute( + " + DELETE FROM messages + WHERE is_deleted = 0 + AND expire_time IS NOT NULL + AND expire_time < ?1 + ", + params![iso_format], + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + + Ok(()) + } +} + +/// 获取数据库连接 +impl QueueOperation { + fn get_connection(&self) -> PyResult> { + self.pool + .get() + .map_err(|e| PyErr::new::(e.to_string())) + } +} + +/// 分片队列操作 +#[pyclass] +pub struct ShardedQueueOperation { + shards: Vec, + #[pyo3(get)] + shard_num: usize, + #[pyo3(get)] + db_dir: String, +} + +#[pymethods] +impl ShardedQueueOperation { + #[new] + pub fn new(shard_num: usize, queue_name: String) -> PyResult { + // 创建缓存目录 + let db_dir = format!("cache/{}", queue_name); + std::fs::create_dir_all(&db_dir) + .map_err(|e| PyErr::new::(e.to_string()))?; + + let mut shards = Vec::with_capacity(shard_num); + + for i in 0..shard_num { + let db_path = format!("{}/queue_shard_{}.db", db_dir, i); + let queue_op = QueueOperation::new(db_path)?; + queue_op.init_db()?; + shards.push(queue_op); + } + + Ok(ShardedQueueOperation { + shards, + shard_num, + db_dir, + }) + } + + /// 获取分片索引 + /// Args: + /// - self: ShardedQueueOperation 实例 + /// - message_id: &str, 消息ID + /// Returns: + /// - usize: 分片索引 + fn _get_shard_index(&self, message_id: &str) -> usize { + let mut hasher = Sha256::new(); + hasher.update(message_id.as_bytes()); + let result = hasher.finalize(); + + // 将哈希值转换为 usize + let hash_bytes: [u8; 8] = result[..8].try_into().unwrap(); + let hash_value = u64::from_be_bytes(hash_bytes) as usize; + + hash_value % self.shard_num + } + + /// 添加消息 + /// Args: + /// - self: ShardedQueueOperation 实例 + /// - message: &Bound<'_, PyDict>, 待添加的消息 + /// Returns: + /// - String: 添加的消息ID + pub fn enqueue(&self, message: &Bound<'_, PyDict>) -> PyResult { + let message_id_bound = message + .get_item("id")? + .ok_or_else(|| PyErr::new::("Missing id field"))?; + let message_id: String = message_id_bound.extract()?; + + let shard_index = self._get_shard_index(&message_id); + self.shards[shard_index].enqueue(message) + } + + /// 获取消息 + /// Args: + /// - self: ShardedQueueOperation 实例 + /// - size: i32, 获取的消息数量 + /// Returns: + /// - Vec>: 获取的消息列表 + pub fn dequeue(&self, size: i32) -> PyResult>> { + let mut all_messages = Vec::new(); + let mut collected = 0; + + // 限制单次获取的最大消息数量 + let max_size = std::cmp::min(size, 1000); + + // 随机轮询分片顺序 + let mut shard_order: Vec = (0..self.shard_num).collect(); + let mut rng = rand::thread_rng(); + shard_order.shuffle(&mut rng); + + for shard_index in shard_order { + if collected >= max_size { + break; + } + + let remaining = size - collected; + let shard_messages = self.shards[shard_index].dequeue(remaining)?; + collected += shard_messages.len() as i32; + all_messages.extend(shard_messages); + + if all_messages.len() > 1000 { + break; + } + } + + Ok(all_messages) + } + + /// 获取队列长度 + /// Args: + /// - self: ShardedQueueOperation 实例 + /// Returns: + /// - i32: 队列长度 + pub fn get_queue_length(&self) -> PyResult { + let mut total = 0; + for shard in &self.shards { + total += shard.get_queue_length()?; + } + Ok(total) + } + + /// 获取已完成的消息 + /// Args: + /// - self: ShardedQueueOperation 实例 + /// Returns: + /// - Vec>: 已完成的消息列表 + pub fn get_completed_messages(&self) -> PyResult>> { + let mut all_messages = Vec::new(); + for shard in &self.shards { + let shard_messages = shard.get_completed_messages()?; + all_messages.extend(shard_messages); + + if all_messages.len() > 1000 { + break; + } + } + Ok(all_messages) + } + + /// 获取消息详情 + /// Args: + /// - self: ShardedQueueOperation 实例 + /// - message_id: String, 消息ID + /// Returns: + /// - HashMap: 消息详情 + pub fn get_result(&self, message_id: String) -> PyResult> { + let shard_index = self._get_shard_index(&message_id); + self.shards[shard_index].get_result(message_id) + } + + /// 更新消息状态 + /// Args: + /// - self: ShardedQueueOperation 实例 + /// - message_id: String, 消息ID + /// - status: i32, 新状态 + /// Returns: + /// - () : 无返回值 + pub fn update_status(&self, message_id: String, status: i32) -> PyResult<()> { + let shard_index = self._get_shard_index(&message_id); + self.shards[shard_index].update_status(message_id, status) + } + + /// 更新消息结果 + /// Args: + /// - self: ShardedQueueOperation 实例 + /// - message_id: String, 消息ID + /// - result: String, 新结果 + /// Returns: + /// - () : 无返回值 + pub fn update_result(&self, message_id: String, result: String) -> PyResult<()> { + let shard_index = self._get_shard_index(&message_id); + self.shards[shard_index].update_result(message_id, result) + } + + /// 删除消息 + /// Args: + /// - self: ShardedQueueOperation 实例 + /// - message_id: String, 待删除的消息ID + /// Returns: + /// - () : 无返回值 + pub fn delete_message(&self, message_id: String) -> PyResult<()> { + let shard_index = self._get_shard_index(&message_id); + self.shards[shard_index].delete_message(message_id) + } + + /// 清理过期消息 + /// Args: + /// - self: ShardedQueueOperation 实例 + /// Returns: + /// - () : 无返回值 + pub fn clean_expired_messages(&self) -> PyResult<()> { + for shard in &self.shards { + shard.clean_expired_messages()?; + } + Ok(()) + } + + /// 清理旧消息 + /// Args: + /// - self: ShardedQueueOperation 实例 + /// - days: i64, 清理旧消息的天数 + /// Returns: + /// - () : 无返回值 + pub fn remove_expired_messages(&self, days: i64) -> PyResult<()> { + for shard in &self.shards { + shard.remove_expired_messages(days)?; + } + Ok(()) + } + + /// 清理旧消息 + /// Args: + /// - self: ShardedQueueOperation 实例 + /// - days: i64, 清理旧消息的天数 + /// Returns: + /// - () : 无返回值 + pub fn clean_old_messages(&self, days: i64) -> PyResult<()> { + for shard in &self.shards { + shard.clean_old_messages(days)?; + } + Ok(()) + } +} diff --git a/src/core/src/task_mounter.rs b/src/queue_sqlite_core/src/task_mounter.rs similarity index 100% rename from src/core/src/task_mounter.rs rename to src/queue_sqlite_core/src/task_mounter.rs diff --git a/tests/listen/listen_data.py b/tests/listen/listen_data.py index 38dfcba..2f105b9 100644 --- a/tests/listen/listen_data.py +++ b/tests/listen/listen_data.py @@ -1,5 +1,6 @@ -from queue_sqlite.mounter.listen_mounter import ListenMounter +from queue_sqlite.mounter.listen_mounter import listener -@ListenMounter.listener() + +@listener() def key_1(data): - print(data) \ No newline at end of file + print(data) diff --git a/tests/test_async_scheduler.py b/tests/test_async_scheduler.py index f6db891..3ae9aef 100644 --- a/tests/test_async_scheduler.py +++ b/tests/test_async_scheduler.py @@ -25,7 +25,9 @@ class TestAsyncScheduler: async def _callback(cls, message_item: MessageItem): with cls.lock: cls.callback_counter += 1 - # print(f"callback: {message_item}") + # if cls.callback_counter % 100 == 0: # 每100个任务打印一次 + # print(f"callback: {cls.callback_counter}") + print(f"callback: {message_item}") # print(f"callback: {message_item.id}") # print(f"callback: {message_item.expire_time}") # print(f"callback: {message_item.tags}") diff --git a/tests/test_core.py b/tests/test_core.py index 657f1ec..7789388 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -1,4 +1,4 @@ -from queue_sqlite.core import core +import queue_sqlite_core as core from queue_sqlite.mounter.task_mounter import TaskMounter from queue_sqlite.model import MessageItem @@ -11,7 +11,7 @@ class TestCore: task_mounter = core.TaskMounter(TaskMounter) task_mounter.get_task_list() print( - task_mounter.get_task_function("task")({"num": 1})(lambda x: x + 1).meta[ + task_mounter.get_task_function("task")({"num": 1})(lambda x: x + 1).meta[ # type: ignore "num" ] ) diff --git a/tests/test_stress.py b/tests/test_stress.py index 25fcde4..4cde722 100644 --- a/tests/test_stress.py +++ b/tests/test_stress.py @@ -25,6 +25,9 @@ class TestStress: def _callback(cls, message_item: MessageItem): with cls.lock: cls.callback_counter += 1 + # if cls.callback_counter % 100 == 0: # 每100个任务打印一次 + # print(f"callback: {cls.callback_counter}") + print(message_item.id) # print(f"callback: {message_item}") # print(f"callback: {message_item.id}") # print(f"callback: {message_item.expire_time}")