chakcy_code_repository/queue_sqlite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

modify 版本更新 0.2.0

Browse Source
This commit is contained in:
chakcy
2025-09-27 21:18:06 +08:00
parent cfc4028a51
commit e6291fe3a4
46 changed files with 1650 additions and 953 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.python-version
+1 -1
View File
@@ -1 +1 @@
3.13t 3.10
pyproject.toml
+5 -9
View File
@@ -6,18 +6,20 @@ classifiers = [
"Development Status :: 4 - Beta", "Development Status :: 4 - Beta",
"Intended Audience :: Developers", "Intended Audience :: Developers",
"License :: OSI Approved :: MIT License", "License :: OSI Approved :: MIT License",
"Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3",
"Topic :: System :: Distributed Computing", "Topic :: System :: Distributed Computing",
"Typing :: Typed", "Typing :: Typed",
"Operating System :: Microsoft :: Windows", "Operating System :: Microsoft :: Windows",
"Operating System :: POSIX :: Linux", "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." 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" name = "queue-sqlite"
readme = "README.md" readme = "README.md"
requires-python = ">=3.7" requires-python = ">=3.7"
version = "0.1.0" version = "0.2.0"
[project.optional-dependencies] [project.optional-dependencies]
pyside6 = ["PySide6"] pyside6 = ["PySide6"]
@@ -41,9 +43,3 @@ dev = [
[tool.hatch.build.targets.wheel] [tool.hatch.build.targets.wheel]
packages = ["src/queue_sqlite"] packages = ["src/queue_sqlite"]
platforms = ["windows_x86_64", "linux_x86_64"]
[tool.hatch.build]
artifacts = [
"src/queue_sqlite/core/*.pyd",
]
scripts/code_content.py
+15 -13
View File
@@ -1,27 +1,29 @@
import os import os
import sys import sys
def generate_markdown_from_py_files(directory, output_file): 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): for root, dirs, files in os.walk(directory):
# 排除 venv 目录 # 排除 venv 目录
dirs[:] = [d for d in dirs if d != '.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 != ".vscode"]
dirs[:] = [d for d in dirs if d != 'scripts'] 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 != "build"]
for file in files: 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) file_path = os.path.join(root, file)
md_file.write(f'`{file_path}`\n') md_file.write(f"`{file_path}`\n")
md_file.write('```python\n') md_file.write("```python\n")
with open(file_path, 'r', encoding='utf-8') as py_file: with open(file_path, "r", encoding="utf-8") as py_file:
md_file.write(py_file.read()) md_file.write(py_file.read())
md_file.write('\n```\n\n') md_file.write("\n```\n\n")
if __name__ == "__main__": if __name__ == "__main__":
# 指定目录和输出文件名 # 指定目录和输出文件名
target_directory = sys.argv[1] # 替换为你的目标目录 target_directory = sys.argv[1] # 替换为你的目标目录
output_markdown_file = 'output.md' # 输出的 Markdown 文件名 output_markdown_file = "output.md" # 输出的 Markdown 文件名
generate_markdown_from_py_files(target_directory, output_markdown_file) generate_markdown_from_py_files(target_directory, output_markdown_file)
print(f'Markdown 文件已生成：{output_markdown_file}') print(f"Markdown 文件已生成：{output_markdown_file}")
scripts/to_json_utils.py
+18 -8
View File
@@ -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 sqlite3
import json import json
import tkinter as tk import tkinter as tk
@@ -21,11 +31,11 @@ class SQLiteToJSONConverter:
def create_widgets(self): def create_widgets(self):
# 主框架 # 主框架
main_frame = ttk.Frame(self.root, padding="10") 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 = 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( ttk.Entry(db_frame, textvariable=self.db_path, width=70).grid(
row=0, column=0, padx=5 row=0, column=0, padx=5
@@ -39,13 +49,13 @@ class SQLiteToJSONConverter:
# 表选择部分 # 表选择部分
table_frame = ttk.LabelFrame(main_frame, text="表选择", padding="5") 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) ttk.Label(table_frame, text="选择表:").grid(row=0, column=0, sticky=tk.W)
self.table_combo = ttk.Combobox( self.table_combo = ttk.Combobox(
table_frame, textvariable=self.selected_table, state="readonly" 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( ttk.Button(table_frame, text="加载数据", command=self.load_table_data).grid(
row=0, column=2, padx=5 row=0, column=2, padx=5
) )
@@ -53,7 +63,7 @@ class SQLiteToJSONConverter:
# 数据显示部分 # 数据显示部分
data_frame = ttk.LabelFrame(main_frame, text="表数据", padding="5") data_frame = ttk.LabelFrame(main_frame, text="表数据", padding="5")
data_frame.grid( 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显示数据 # 创建Treeview显示数据
@@ -63,9 +73,9 @@ class SQLiteToJSONConverter:
hsb = ttk.Scrollbar(data_frame, orient="horizontal", command=self.tree.xview) hsb = ttk.Scrollbar(data_frame, orient="horizontal", command=self.tree.xview)
self.tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set) 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)) 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)) vsb.grid(row=0, column=1, sticky=(tk.N, tk.S)) # type: ignore
hsb.grid(row=1, column=0, sticky=(tk.W, tk.E)) hsb.grid(row=1, column=0, sticky=(tk.W, tk.E)) # type: ignore
# 按钮部分 # 按钮部分
button_frame = ttk.Frame(main_frame) button_frame = ttk.Frame(main_frame)
src/core/src/lib.rs
-13
View File
@@ -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::<TaskMounter>()?;
m.add_class::<QueueOperation>()?;
Ok(())
}
src/core/src/queue_operation.rs
-530
View File
@@ -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<SqliteConnectionManager>,
}
#[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::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
let mut rows = stmt
.query([])
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
while let Some(_) = rows
.next()
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?
{
}
}
Ok(())
}
pub fn enqueue<'py>(
&self,
py: Python<'py>,
message: &Bound<'py, PyDict>,
) -> PyResult<Bound<'py, PyString>> {
let mut conn = self
.pool
.get()
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
let get_field = |key: &str| -> PyResult<_> {
message.get_item(key)?.ok_or_else(|| {
PyErr::new::<pyo3::exceptions::PyKeyError, _>(format!("{} not found", key))
})
};
let id = get_field("id")?.extract::<String>()?;
let type_ = get_field("type")?.extract::<String>()?;
let status = get_field("status")?.extract::<i32>()?;
let content = get_field("content")?.extract::<String>()?;
let createtime = get_field("createtime")?.extract::<String>()?;
let updatetime = get_field("updatetime")?.extract::<String>()?;
let result = get_field("result")?.extract::<String>()?;
let priority = get_field("priority")?.extract::<i32>()?;
let source = get_field("source")?.extract::<String>()?;
let destination = get_field("destination")?.extract::<String>()?;
let retry_count = get_field("retry_count")?.extract::<i32>()?;
let expire_time = get_field("expire_time")?.extract::<String>()?;
let tags = get_field("tags")?.extract::<String>()?;
let metadata = get_field("metadata")?.extract::<String>()?;
let tx = conn
.transaction()
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
tx.commit()
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
Ok(Bound::from(PyString::new(py, &id)))
}
pub fn dequeue<'py>(&self, py: Python<'py>, size: i32) -> PyResult<Bound<'py, PyList>> {
let conn = self
.pool
.get()
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
let now: DateTime<Utc> = 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::<pyo3::exceptions::PyRuntimeError, _>(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<String>>("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<String>>("source")?)
.map_err(to_rusqlite_error)?;
dict.set_item("destination", row.get::<_, Option<String>>("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<String>>("expire_time")?)
.map_err(to_rusqlite_error)?;
dict.set_item("tags", row.get::<_, Option<String>>("tags")?)
.map_err(to_rusqlite_error)?;
dict.set_item("metadata", row.get::<_, Option<String>>("metadata")?)
.map_err(to_rusqlite_error)?;
// Return the dictionary as Py<PyDict>
Ok(dict)
})
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
for row in mapped_rows {
let dict =
row.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
rows.append(dict)?;
}
Ok(rows.into())
}
pub fn get_queue_length(&self) -> PyResult<i32> {
let conn = self
.pool
.get()
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
Ok(count)
}
pub fn get_completed_messages<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
let conn = self
.pool
.get()
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(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<String>>("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<String>>("source")?)
.map_err(to_rusqlite_error)?;
dict.set_item("destination", row.get::<_, Option<String>>("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<String>>("expire_time")?)
.map_err(to_rusqlite_error)?;
dict.set_item("tags", row.get::<_, Option<String>>("tags")?)
.map_err(to_rusqlite_error)?;
dict.set_item("metadata", row.get::<_, Option<String>>("metadata")?)
.map_err(to_rusqlite_error)?;
// Return the dictionary as Py<PyDict>
Ok(dict)
})
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
for row in mapped_rows {
let dict =
row.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
rows.append(dict)?;
}
Ok(rows.into())
}
pub fn get_result<'py>(&self, py: Python<'py>, id: String) -> PyResult<Bound<'py, PyDict>> {
let conn = self
.pool
.get()
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(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<String>>("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<String>>("source")?)
.map_err(to_rusqlite_error)?;
dict.set_item("destination", row.get::<_, Option<String>>("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<String>>("expire_time")?)
.map_err(to_rusqlite_error)?;
dict.set_item("tags", row.get::<_, Option<String>>("tags")?)
.map_err(to_rusqlite_error)?;
dict.set_item("metadata", row.get::<_, Option<String>>("metadata")?)
.map_err(to_rusqlite_error)?;
// Return the dictionary as Py<PyDict>
Ok(dict)
})
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
let dict = mapped_rows
.next()
.transpose()
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
conn.execute(
"
UPDATE messages SET status = ?1 WHERE id = ?2
",
params![status, id],
)
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
Ok(())
}
pub fn update_result(&self, id: String, result: String) -> PyResult<()> {
let conn = self
.pool
.get()
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
conn.execute(
"
UPDATE messages SET result = ?1 WHERE id = ?2
",
params![result, id],
)
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
Ok(())
}
pub fn delete_message(&self, id: String) -> PyResult<()> {
let conn = self
.pool
.get()
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
conn.execute(
"
UPDATE messages SET is_deleted = 1 WHERE id = ?1
",
params![id],
)
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
Ok(())
}
pub fn remove_expired_messages(&self, days: i64) -> PyResult<()> {
let now: DateTime<Utc> = 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::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
Ok(())
}
pub fn clean_expired_messages(&self) -> PyResult<()> {
let now: DateTime<Utc> = Utc::now();
let iso_format = now.to_rfc3339();
let conn = self
.pool
.get()
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
Ok(())
}
pub fn clean_old_messages(&self, days: i64) -> PyResult<()> {
// 获取七天前的时间
let now: DateTime<Utc> = 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::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
Ok(())
}
}
src/queue_sqlite/__init__.py
+46
View File
@@ -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__,
}
src/queue_sqlite/constant/__init__.py
+11
View File
@@ -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_priority import MessagePriority
from .message_status import MessageStatus from .message_status import MessageStatus
from .message_type import MessageType from .message_type import MessageType
src/queue_sqlite/constant/message_priority.py
+11
View File
@@ -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 from enum import IntEnum
src/queue_sqlite/constant/message_status.py
+11
View File
@@ -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 from enum import IntEnum
src/queue_sqlite/constant/message_type.py
+11
View File
@@ -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 from enum import Enum
src/queue_sqlite/core/__init__.py
View File
src/queue_sqlite/core/__init__.pyi
-167
View File
@@ -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.
"""
...
src/queue_sqlite/cycle/__init__.py
+9
View File
@@ -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 : 任务周期模块
"""
src/queue_sqlite/cycle/async_task_cycle.py
+11
View File
@@ -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 ..model import MessageItem
from ..constant import MessageStatus from ..constant import MessageStatus
import json import json
src/queue_sqlite/cycle/task_cycle.py
+11
View File
@@ -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 ..model import MessageItem
from typing import Callable, Optional from typing import Callable, Optional
from ..constant import MessageStatus from ..constant import MessageStatus
src/queue_sqlite/model/__init__.py
+11
View File
@@ -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 from .message_item import MessageItem
src/queue_sqlite/model/message_item.py
+61 -12
View File
@@ -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 dataclasses import dataclass, field
from typing import Optional, Dict, Any from typing import Optional, Dict, Any
from datetime import datetime from datetime import datetime
@@ -5,6 +16,7 @@ from ..constant import MessageStatus, MessagePriority, MessageType
import json import json
from enum import Enum from enum import Enum
import uuid import uuid
import logging
@dataclass @dataclass
@@ -115,7 +127,13 @@ class MessageItem:
# 处理枚举字段 # 处理枚举字段
if "type" in data: if "type" in data:
if isinstance(data["type"], str): 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): elif isinstance(data["type"], MessageType):
pass # 已经是枚举类型 pass # 已经是枚举类型
else: else:
@@ -123,7 +141,13 @@ class MessageItem:
if "status" in data: if "status" in data:
if isinstance(data["status"], str): 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): elif isinstance(data["status"], int):
data["status"] = MessageStatus(data["status"]) # 使用枚举值 data["status"] = MessageStatus(data["status"]) # 使用枚举值
elif isinstance(data["status"], MessageStatus): elif isinstance(data["status"], MessageStatus):
@@ -132,13 +156,38 @@ class MessageItem:
raise ValueError(f"Invalid message status: {data['status']}") raise ValueError(f"Invalid message status: {data['status']}")
if "priority" in data: if "priority" in data:
if isinstance(data["priority"], (int, str)): if isinstance(data["priority"], str):
data["priority"] = MessagePriority(data["priority"]) # type: ignore # 尝试将字符串转换为整数，如果失败则使用字符串作为枚举名称
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): elif isinstance(data["priority"], MessagePriority):
pass pass # 已经是枚举类型
else: else:
raise ValueError("请使用 'int' 或 'str' 类型设置 priority 字段") 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) return cls(**data)
def to_dict(self) -> Dict[str, Any]: def to_dict(self) -> Dict[str, Any]:
@@ -235,13 +284,13 @@ class MessageItem:
"""打印所有字段的信息，包括类型、默认值和描述""" """打印所有字段的信息，包括类型、默认值和描述"""
from dataclasses import fields from dataclasses import fields
print(f"=== {cls.__name__} 字段信息 ===") logging.info(f"=== {cls.__name__} 字段信息 ===")
for field_info in fields(cls): for field_info in fields(cls):
print(f"字段名: {field_info.name}") logging.info(f"字段名: {field_info.name}")
print(f"类型: {field_info.type}") logging.info(f"类型: {field_info.type}")
print(f"默认值: {field_info.default}") logging.info(f"默认值: {field_info.default}")
if field_info.default_factory is not None: 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: if "description" in field_info.metadata:
print(f"描述: {field_info.metadata['description']}") logging.info(f"描述: {field_info.metadata['description']}")
print("-" * 50) logging.info("-" * 50)
src/queue_sqlite/mounter/__init__.py
+11
View File
@@ -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 .task_mounter import task
from .listen_mounter import listener from .listen_mounter import listener
src/queue_sqlite/mounter/listen_mounter.py
+11
View File
@@ -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 from typing import Callable, List
src/queue_sqlite/mounter/task_mounter.py
+11
View File
@@ -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 from typing import Callable, List
src/queue_sqlite/queue_operation/__init__.py
+9 -3
View File
@@ -1,3 +1,9 @@
from .queue_operation import QueueOperation #!/usr/bin/env python
# -*- encoding: utf-8 -*-
__all__ = ["QueueOperation"] """
@File : __init__.py
@Time : 2025-09-27 17:01:59
@Author : chakcy
@Email : 947105045@qq.com
@description : 队列操作模块
"""
src/queue_sqlite/queue_operation/listen_operation.py
+11
View File
@@ -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 from ..mounter.listen_mounter import ListenMounter
import sqlite3 import sqlite3
from typing import List, Tuple, Union from typing import List, Tuple, Union
src/queue_sqlite/queue_operation/queue_operation.py
-127
View File
@@ -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)
src/queue_sqlite/scheduler/__init__.py
+11
View File
@@ -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 .standard import StandardQueueScheduler
from ._async import AsyncQueueScheduler from ._async import AsyncQueueScheduler
from .base import BaseScheduler from .base import BaseScheduler
src/queue_sqlite/scheduler/_async/__init__.py
+16 -3
View File
@@ -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_receive_scheduler import AsyncReceiveScheduler
from .async_task_scheduler import AsyncTaskScheduler from .async_task_scheduler import AsyncTaskScheduler
from .async_listen_data_scheduler import AsyncListenDataScheduler from .async_listen_data_scheduler import AsyncListenDataScheduler
from ...model import MessageItem from ...model import MessageItem
from ...queue_operation.listen_operation import ListenOperation from ...queue_operation.listen_operation import ListenOperation
from typing import Callable from typing import Callable
from ...queue_operation import QueueOperation from queue_sqlite_core import ShardedQueueOperation
import os import os
from ..base import BaseScheduler from ..base import BaseScheduler
from ..cleanup_scheduler import CleanupScheduler from ..cleanup_scheduler import CleanupScheduler
@@ -12,8 +23,10 @@ from ..cleanup_scheduler import CleanupScheduler
class AsyncQueueScheduler(BaseScheduler): class AsyncQueueScheduler(BaseScheduler):
def __init__(self, receive_thread_num=1, task_thread_num=1, shard_num=4): def __init__(
self.queue_operation = QueueOperation(shard_num=shard_num) 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.receive_scheduler = AsyncReceiveScheduler(
self.queue_operation, receive_thread_num self.queue_operation, receive_thread_num
) )
src/queue_sqlite/scheduler/_async/async_listen_data_scheduler.py
+11
View File
@@ -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 from ...queue_operation.listen_operation import ListenOperation
import asyncio import asyncio
from ...mounter.listen_mounter import ListenMounter from ...mounter.listen_mounter import ListenMounter
src/queue_sqlite/scheduler/_async/async_receive_scheduler.py
+19 -5
View File
@@ -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 ...model import MessageItem
from typing import Callable from typing import Callable
import asyncio import asyncio
import threading import threading
import concurrent.futures import concurrent.futures
import logging
class AsyncReceiveScheduler: 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.callbacks = dict()
self.receive_thread_num = receive_thread_num self.receive_thread_num = receive_thread_num
self.is_running = False self.is_running = False
@@ -17,7 +31,7 @@ class AsyncReceiveScheduler:
def send_message(self, message: MessageItem, callback: Callable = None): # type: ignore def send_message(self, message: MessageItem, callback: Callable = None): # type: ignore
if callback is None: 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())
self.queue_operation.enqueue(message.to_dict_by_core()) self.queue_operation.enqueue(message.to_dict_by_core())
with self.lock: with self.lock:
@@ -63,7 +77,7 @@ class AsyncReceiveScheduler:
) as executor: ) as executor:
await loop.run_in_executor(executor, callback, message) await loop.run_in_executor(executor, callback, message)
except Exception as e: except Exception as e:
print(f"回调执行错误: {str(e)}") logging.error(f"回调执行错误: {str(e)}")
finally: finally:
self.queue_operation.delete_message(message.id) self.queue_operation.delete_message(message.id)
@@ -77,7 +91,7 @@ class AsyncReceiveScheduler:
# 运行异步接收函数直到停止 # 运行异步接收函数直到停止
loop.run_until_complete(self.receive_message()) loop.run_until_complete(self.receive_message())
except Exception as e: except Exception as e:
print(f"任务调度出错: {str(e)}") logging.error(f"任务调度出错: {str(e)}")
finally: finally:
loop.close() loop.close()
src/queue_sqlite/scheduler/_async/async_task_scheduler.py
+18 -6
View File
@@ -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 ...model import MessageItem
from ...constant import MessageStatus from ...constant import MessageStatus
from ...cycle.async_task_cycle import AsyncTaskCycle from ...cycle.async_task_cycle import AsyncTaskCycle
@@ -7,13 +18,14 @@ import asyncio
from datetime import datetime from datetime import datetime
import threading import threading
import multiprocessing import multiprocessing
import logging
class AsyncTaskScheduler: class AsyncTaskScheduler:
def __init__( def __init__(
self, self,
queue_operation: QueueOperation, queue_operation: ShardedQueueOperation,
task_thread_num: int = multiprocessing.cpu_count() * 2, task_thread_num: int = multiprocessing.cpu_count() * 2,
): ):
self.task_thread_num = task_thread_num 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_result(message.id, message.result)
self.queue_operation.update_status(message.id, message.status) self.queue_operation.update_status(message.id, message.status)
except Exception as e: except Exception as e:
print(f"任务结果更新失败: {str(e)}") logging.error(f"任务结果更新失败: {str(e)}")
async def task_callback(self): async def task_callback(self):
"""单一轮询线程，并行执行任务""" """单一轮询线程，并行执行任务"""
@@ -98,10 +110,10 @@ class AsyncTaskScheduler:
message = task.result() message = task.result()
self._process_messages(message) self._process_messages(message)
except Exception as e: except Exception as e:
print(f"处理任务结果时出错: {str(e)}") logging.error(f"处理任务结果时出错: {str(e)}")
except Exception as e: except Exception as e:
print(f"任务调度出错: {str(e)}") logging.error(f"任务调度出错: {str(e)}")
await asyncio.sleep(0.1) # 出错时短暂延迟 await asyncio.sleep(0.1) # 出错时短暂延迟
def _run_task_loop(self): def _run_task_loop(self):
@@ -114,7 +126,7 @@ class AsyncTaskScheduler:
# 运行异步任务处理函数直到停止 # 运行异步任务处理函数直到停止
loop.run_until_complete(self.task_callback()) loop.run_until_complete(self.task_callback())
except Exception as e: except Exception as e:
print(f"任务调度出错: {str(e)}") logging.error(f"任务调度出错: {str(e)}")
finally: finally:
loop.close() loop.close()
src/queue_sqlite/scheduler/cleanup_scheduler.py
+24 -11
View File
@@ -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 threading
import time import time
from ..queue_operation import QueueOperation from queue_sqlite_core import ShardedQueueOperation
import logging
class CleanupScheduler: class CleanupScheduler:
def __init__( 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.queue_operation = queue_operation
self.interval = interval_minutes * 60 # 转换为秒 self.interval = interval_minutes * 60 # 转换为秒
@@ -14,22 +29,20 @@ class CleanupScheduler:
self.cleanup_thread = None self.cleanup_thread = None
self.remove_days = remove_days self.remove_days = remove_days
for i in range(self.queue_operation.shard_num): # for i in range(self.queue_operation.shard_num):
# 清理过期但未处理的消息 # 清理过期但未处理的消息
self.queue_operation.clean_expired_messages(i) self.queue_operation.clean_expired_messages()
# 彻底删除30天前的消息 # 彻底删除30天前的消息
self.queue_operation.remove_expired_messages(i, self.remove_days) self.queue_operation.remove_expired_messages(self.remove_days)
def cleanup_expired_messages(self): def cleanup_expired_messages(self):
"""清理过期消息""" """清理过期消息"""
while self.is_running: while self.is_running:
try: try:
for i in range(self.queue_operation.shard_num): self.queue_operation.clean_expired_messages()
# 清理过期但未处理的消息
self.queue_operation.clean_expired_messages(i)
except Exception as e: except Exception as e:
print(f"清理消息错误: {str(e)}") logging.error(f"清理消息错误: {str(e)}")
# 休眠等待下次清理 # 休眠等待下次清理
for _ in range(self.interval): for _ in range(self.interval):
src/queue_sqlite/scheduler/standard/__init__.py
+16 -3
View File
@@ -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 .receive_scheduler import ReceiveScheduler
from .task_scheduler import TaskScheduler from .task_scheduler import TaskScheduler
from .listen_data_scheduler import ListenDataScheduler from .listen_data_scheduler import ListenDataScheduler
from ...model import MessageItem from ...model import MessageItem
from typing import Callable from typing import Callable
from ...queue_operation import QueueOperation from queue_sqlite_core import ShardedQueueOperation
from ...queue_operation.listen_operation import ListenOperation from ...queue_operation.listen_operation import ListenOperation
from ..cleanup_scheduler import CleanupScheduler from ..cleanup_scheduler import CleanupScheduler
import os import os
@@ -11,8 +22,10 @@ from ..base import BaseScheduler
class StandardQueueScheduler(BaseScheduler): class StandardQueueScheduler(BaseScheduler):
def __init__(self, receive_thread_num=1, task_thread_num=1, shard_num=4): def __init__(
self.queue_operation = QueueOperation(shard_num) 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( self.listen_operation = ListenOperation(
os.path.join(self.queue_operation.db_dir, "listen.db") os.path.join(self.queue_operation.db_dir, "listen.db")
) )
src/queue_sqlite/scheduler/standard/listen_data_scheduler.py
+11
View File
@@ -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 ...queue_operation.listen_operation import ListenOperation
from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ThreadPoolExecutor
from ...mounter.listen_mounter import ListenMounter from ...mounter.listen_mounter import ListenMounter
src/queue_sqlite/scheduler/standard/receive_scheduler.py
+18 -9
View File
@@ -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 ...model import MessageItem
from typing import Callable from typing import Callable
from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ThreadPoolExecutor
import threading import threading
import time import time
import logging
class ReceiveScheduler: 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.callbacks = dict()
self.receive_thread_num = receive_thread_num self.receive_thread_num = receive_thread_num
self.is_running = False self.is_running = False
@@ -20,7 +34,7 @@ class ReceiveScheduler:
def send_message(self, message: MessageItem, callback: Callable = None): # type: ignore def send_message(self, message: MessageItem, callback: Callable = None): # type: ignore
if callback is None: 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())
self.queue_operation.enqueue(message.to_dict_by_core()) self.queue_operation.enqueue(message.to_dict_by_core())
with self.lock: with self.lock:
@@ -49,13 +63,8 @@ class ReceiveScheduler:
"""安全执行回调函数""" """安全执行回调函数"""
try: try:
callback(message) 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: except Exception as e:
print(f"回调执行错误: {str(e)}") logging.error(f"回调执行错误: {str(e)}")
finally: finally:
self.queue_operation.delete_message(message.id) self.queue_operation.delete_message(message.id)
src/queue_sqlite/scheduler/standard/task_scheduler.py
+18 -5
View File
@@ -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 ...model import MessageItem
from ...constant import MessageStatus from ...constant import MessageStatus
from ...cycle.task_cycle import TaskCycle from ...cycle.task_cycle import TaskCycle
@@ -8,12 +20,13 @@ from datetime import datetime
import time import time
import threading import threading
import multiprocessing import multiprocessing
import json
class TaskScheduler: class TaskScheduler:
def __init__( def __init__(
self, self,
queue_operation: QueueOperation, queue_operation: ShardedQueueOperation,
task_thread_num: int = multiprocessing.cpu_count() * 2, task_thread_num: int = multiprocessing.cpu_count() * 2,
): ):
self.task_thread_num = task_thread_num self.task_thread_num = task_thread_num
@@ -51,10 +64,10 @@ class TaskScheduler:
def _update_result(self, message): def _update_result(self, message):
"""更新任务结果到数据库""" """更新任务结果到数据库"""
try: 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) self.queue_operation.update_status(message.id, message.status)
except Exception as e: except Exception as e:
print(f"任务结果更新失败: {str(e)}") logging.error(f"任务结果更新失败: {str(e)}")
def task_callback(self): def task_callback(self):
"""单一轮询线程，并行执行任务""" """单一轮询线程，并行执行任务"""
@@ -73,7 +86,7 @@ class TaskScheduler:
else: else:
time.sleep(0.1) # 适当休眠 time.sleep(0.1) # 适当休眠
except Exception as e: except Exception as e:
print(f"任务调度错误: {str(e)}") logging.error(f"任务调度错误: {str(e)}")
time.sleep(1) time.sleep(1)
def start_task_thread(self): def start_task_thread(self):
src/core/.gitignore → src/queue_sqlite_core/.gitignore
View File
src/core/Cargo.lock → src/queue_sqlite_core/Cargo.lock
Generated
+290 -14
View File
@@ -2,6 +2,21 @@
# It is not intended for manual editing. # It is not intended for manual editing.
version = 4 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]] [[package]]
name = "ahash" name = "ahash"
version = "0.8.12" version = "0.8.12"
@@ -35,12 +50,36 @@ version = "1.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" 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]] [[package]]
name = "bitflags" name = "bitflags"
version = "2.9.1" version = "2.9.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1b8e56985ec62d17e9c1001dc89c88ecd7dc08e47eba5ec7c29c7b5eeecde967" 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]] [[package]]
name = "bumpalo" name = "bumpalo"
version = "3.19.0" version = "3.19.0"
@@ -76,23 +115,41 @@ dependencies = [
"windows-link", "windows-link",
] ]
[[package]]
name = "core"
version = "0.1.0"
dependencies = [
"chrono",
"pyo3",
"r2d2",
"r2d2_sqlite",
"rusqlite",
]
[[package]] [[package]]
name = "core-foundation-sys" name = "core-foundation-sys"
version = "0.8.7" version = "0.8.7"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" 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]] [[package]]
name = "fallible-iterator" name = "fallible-iterator"
version = "0.3.0" version = "0.3.0"
@@ -105,6 +162,16 @@ version = "0.1.9"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a" 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]] [[package]]
name = "getrandom" name = "getrandom"
version = "0.3.3" version = "0.3.3"
@@ -114,9 +181,15 @@ dependencies = [
"cfg-if", "cfg-if",
"libc", "libc",
"r-efi", "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]] [[package]]
name = "hashbrown" name = "hashbrown"
version = "0.14.5" version = "0.14.5"
@@ -171,6 +244,23 @@ version = "2.0.6"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f4c7245a08504955605670dbf141fceab975f15ca21570696aebe9d2e71576bd" 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]] [[package]]
name = "js-sys" name = "js-sys"
version = "0.3.77" version = "0.3.77"
@@ -214,6 +304,12 @@ version = "0.4.27"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "13dc2df351e3202783a1fe0d44375f7295ffb4049267b0f3018346dc122a1d94" checksum = "13dc2df351e3202783a1fe0d44375f7295ffb4049267b0f3018346dc122a1d94"
[[package]]
name = "memchr"
version = "2.7.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f52b00d39961fc5b2736ea853c9cc86238e165017a493d1d5c8eac6bdc4cc273"
[[package]] [[package]]
name = "memoffset" name = "memoffset"
version = "0.9.1" version = "0.9.1"
@@ -223,6 +319,26 @@ dependencies = [
"autocfg", "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]] [[package]]
name = "num-traits" name = "num-traits"
version = "0.2.19" version = "0.2.19"
@@ -232,6 +348,15 @@ dependencies = [
"autocfg", "autocfg",
] ]
[[package]]
name = "object"
version = "0.36.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "62948e14d923ea95ea2c7c86c71013138b66525b86bdc08d2dcc262bdb497b87"
dependencies = [
"memchr",
]
[[package]] [[package]]
name = "once_cell" name = "once_cell"
version = "1.21.3" version = "1.21.3"
@@ -261,6 +386,12 @@ dependencies = [
"windows-targets", "windows-targets",
] ]
[[package]]
name = "pin-project-lite"
version = "0.2.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3b3cff922bd51709b605d9ead9aa71031d81447142d828eb4a6eba76fe619f9b"
[[package]] [[package]]
name = "pkg-config" name = "pkg-config"
version = "0.3.32" version = "0.3.32"
@@ -353,6 +484,24 @@ dependencies = [
"syn", "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]] [[package]]
name = "quote" name = "quote"
version = "1.0.40" version = "1.0.40"
@@ -442,12 +591,24 @@ dependencies = [
"smallvec", "smallvec",
] ]
[[package]]
name = "rustc-demangle"
version = "0.1.26"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "56f7d92ca342cea22a06f2121d944b4fd82af56988c270852495420f961d4ace"
[[package]] [[package]]
name = "rustversion" name = "rustversion"
version = "1.0.21" version = "1.0.21"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8a0d197bd2c9dc6e53b84da9556a69ba4cdfab8619eb41a8bd1cc2027a0f6b1d" checksum = "8a0d197bd2c9dc6e53b84da9556a69ba4cdfab8619eb41a8bd1cc2027a0f6b1d"
[[package]]
name = "ryu"
version = "1.0.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "28d3b2b1366ec20994f1fd18c3c594f05c5dd4bc44d8bb0c1c632c8d6829481f"
[[package]] [[package]]
name = "scheduled-thread-pool" name = "scheduled-thread-pool"
version = "0.2.7" version = "0.2.7"
@@ -463,12 +624,72 @@ version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" 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]] [[package]]
name = "shlex" name = "shlex"
version = "1.3.0" version = "1.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64"
[[package]]
name = "slab"
version = "0.4.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7a2ae44ef20feb57a68b23d846850f861394c2e02dc425a50098ae8c90267589"
[[package]] [[package]]
name = "smallvec" name = "smallvec"
version = "1.15.1" version = "1.15.1"
@@ -492,6 +713,46 @@ version = "0.13.2"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e502f78cdbb8ba4718f566c418c52bc729126ffd16baee5baa718cf25dd5a69a" 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]] [[package]]
name = "unicode-ident" name = "unicode-ident"
version = "1.0.18" version = "1.0.18"
@@ -506,9 +767,9 @@ checksum = "7264e107f553ccae879d21fbea1d6724ac785e8c3bfc762137959b5802826ef3"
[[package]] [[package]]
name = "uuid" name = "uuid"
version = "1.17.0" version = "1.18.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3cf4199d1e5d15ddd86a694e4d0dffa9c323ce759fea589f00fef9d81cc1931d" checksum = "2f87b8aa10b915a06587d0dec516c282ff295b475d94abf425d62b57710070a2"
dependencies = [ dependencies = [
"getrandom", "getrandom",
"js-sys", "js-sys",
@@ -528,6 +789,12 @@ version = "0.9.5"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" 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]] [[package]]
name = "wasi" name = "wasi"
version = "0.14.2+wasi-0.2.4" version = "0.14.2+wasi-0.2.4"
@@ -654,6 +921,15 @@ dependencies = [
"windows-link", "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]] [[package]]
name = "windows-targets" name = "windows-targets"
version = "0.52.6" version = "0.52.6"
src/core/Cargo.toml → src/queue_sqlite_core/Cargo.toml
+13 -3
View File
@@ -1,11 +1,11 @@
[package] [package]
name = "core" name = "queue_sqlite_core"
version = "0.1.0" version = "0.2.0"
edition = "2021" edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[lib] [lib]
name = "core" name = "queue_sqlite_core"
crate-type = ["cdylib"] crate-type = ["cdylib"]
[dependencies] [dependencies]
@@ -13,4 +13,14 @@ chrono = "0.4.41"
pyo3 = "0.25.0" pyo3 = "0.25.0"
r2d2 = "0.8" r2d2 = "0.8"
r2d2_sqlite = "0.25" r2d2_sqlite = "0.25"
rand = "0.9.2"
rusqlite = { version = "0.32", features = ["bundled"] } 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
src/core/pyproject.toml → src/queue_sqlite_core/pyproject.toml
+21 -5
View File
@@ -1,9 +1,5 @@
[build-system]
requires = ["maturin>=1.9,<2.0"]
build-backend = "maturin"
[project] [project]
name = "core" name = "queue_sqlite_core"
requires-python = ">=3.7" requires-python = ">=3.7"
classifiers = [ classifiers = [
"Programming Language :: Rust", "Programming Language :: Rust",
@@ -11,5 +7,25 @@ classifiers = [
"Programming Language :: Python :: Implementation :: PyPy", "Programming Language :: Python :: Implementation :: PyPy",
] ]
dynamic = ["version"] 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] [tool.maturin]
features = ["pyo3/extension-module"] 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"
src/queue_sqlite_core/queue_sqlite_core.pyi
+38
View File
@@ -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: ...
src/queue_sqlite_core/src/lib.rs
+15
View File
@@ -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::<TaskMounter>()?;
m.add_class::<QueueOperation>()?;
m.add_class::<ShardedQueueOperation>()?;
Ok(())
}
src/queue_sqlite_core/src/queue_operation.rs
+814
View File
@@ -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<SqliteConnectionManager>,
}
#[derive(Debug, Serialize, Deserialize)]
struct Message {
id: String,
message_type: String,
status: i32,
content: String,
createtime: String,
updatetime: String,
result: Option<String>,
priority: i32,
source: String,
destination: String,
retry_count: i32,
expire_time: Option<String>,
tags: Option<String>,
metadata: Option<String>,
is_deleted: i32,
}
impl TryFrom<&Row<'_>> for Message {
type Error = rusqlite::Error;
fn try_from(row: &Row) -> Result<Self, Self::Error> {
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<Self> {
let manager = SqliteConnectionManager::file(queue_path);
let pool = Pool::new(manager)
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
}
Ok(())
}
/// 入队
/// Args:
/// - self: QueueOperation 实例
/// - message: PyDict 实例，包含消息内容
/// Returns:
/// - String: 消息ID
pub fn enqueue(&self, message: &Bound<'_, PyDict>) -> PyResult<String> {
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::<pyo3::exceptions::PyKeyError, _>(format!(
"Missing required field: {}",
field
)));
}
}
let tx = conn
.transaction()
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(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<String> = match message.get_item("result")? {
Some(item) => {
if item.is_none() {
None
} else {
Some(item.extract()?)
}
}
None => None,
};
let expire_time: Option<String> = match message.get_item("expire_time")? {
Some(item) => {
if item.is_none() {
None
} else {
Some(item.extract()?)
}
}
None => None,
};
let tags: Option<String> = match message.get_item("tags")? {
Some(item) => {
if item.is_none() {
None
} else {
Some(item.extract()?)
}
}
None => None,
};
let metadata: Option<String> = 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::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
tx.commit()
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
Ok(id)
}
/// 出队
/// Args:
/// - self: QueueOperation 实例
/// - size: i32, 获取消息数量
/// Returns:
/// - Vec<HashMap<String, String>>: 获取的消息列表
pub fn dequeue(&self, size: i32) -> PyResult<Vec<HashMap<String, String>>> {
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::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
let mut result = Vec::new();
for row in rows {
result.push(
row.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?,
);
if result.len() > 1000 {
break;
}
}
Ok(result)
}
/// 获取队列长度
/// Args:
/// - self: QueueOperation 实例
/// Returns:
/// - i32: 队列长度
pub fn get_queue_length(&self) -> PyResult<i32> {
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::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
Ok(count)
}
/// 获取已完成的消息
/// Args:
/// - self: QueueOperation 实例
/// Returns:
/// - Vec<HashMap<String, String>>: 已完成的消息列表
pub fn get_completed_messages(&self) -> PyResult<Vec<HashMap<String, String>>> {
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::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
let mut result = Vec::new();
for row in rows {
result.push(
row.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?,
);
if result.len() > 1000 {
break;
}
}
Ok(result)
}
/// 获取消息详情
/// Args:
/// - self: QueueOperation 实例
/// - id: String, 消息ID
/// Returns:
/// - HashMap<String, String>: 消息详情
pub fn get_result(&self, id: String) -> PyResult<HashMap<String, String>> {
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::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyValueError, _>(
"Message not found or not completed",
)
} else {
PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
Ok(())
}
/// 删除过期消息
/// Args:
/// - self: QueueOperation 实例
/// - days: i64, 删除过期消息的天数
/// Returns:
/// - () : 无返回值
pub fn remove_expired_messages(&self, days: i64) -> PyResult<()> {
let now: DateTime<Utc> = 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::<pyo3::exceptions::PyRuntimeError, _>(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::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))?;
Ok(())
}
}
/// 获取数据库连接
impl QueueOperation {
fn get_connection(&self) -> PyResult<r2d2::PooledConnection<SqliteConnectionManager>> {
self.pool
.get()
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(e.to_string()))
}
}
/// 分片队列操作
#[pyclass]
pub struct ShardedQueueOperation {
shards: Vec<QueueOperation>,
#[pyo3(get)]
shard_num: usize,
#[pyo3(get)]
db_dir: String,
}
#[pymethods]
impl ShardedQueueOperation {
#[new]
pub fn new(shard_num: usize, queue_name: String) -> PyResult<Self> {
// 创建缓存目录
let db_dir = format!("cache/{}", queue_name);
std::fs::create_dir_all(&db_dir)
.map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(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<String> {
let message_id_bound = message
.get_item("id")?
.ok_or_else(|| PyErr::new::<pyo3::exceptions::PyKeyError, _>("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<HashMap<String, String>>: 获取的消息列表
pub fn dequeue(&self, size: i32) -> PyResult<Vec<HashMap<String, String>>> {
let mut all_messages = Vec::new();
let mut collected = 0;
// 限制单次获取的最大消息数量
let max_size = std::cmp::min(size, 1000);
// 随机轮询分片顺序
let mut shard_order: Vec<usize> = (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<i32> {
let mut total = 0;
for shard in &self.shards {
total += shard.get_queue_length()?;
}
Ok(total)
}
/// 获取已完成的消息
/// Args:
/// - self: ShardedQueueOperation 实例
/// Returns:
/// - Vec<HashMap<String, String>>: 已完成的消息列表
pub fn get_completed_messages(&self) -> PyResult<Vec<HashMap<String, String>>> {
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<String, String>: 消息详情
pub fn get_result(&self, message_id: String) -> PyResult<HashMap<String, String>> {
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(())
}
}
src/core/src/task_mounter.rs → src/queue_sqlite_core/src/task_mounter.rs
View File
tests/listen/listen_data.py
+3 -2
View File
@@ -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): def key_1(data):
print(data) print(data)
tests/test_async_scheduler.py
+3 -1
View File
@@ -25,7 +25,9 @@ class TestAsyncScheduler:
async def _callback(cls, message_item: MessageItem): async def _callback(cls, message_item: MessageItem):
with cls.lock: with cls.lock:
cls.callback_counter += 1 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.id}")
# print(f"callback: {message_item.expire_time}") # print(f"callback: {message_item.expire_time}")
# print(f"callback: {message_item.tags}") # print(f"callback: {message_item.tags}")
tests/test_core.py
+2 -2
View File
@@ -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.mounter.task_mounter import TaskMounter
from queue_sqlite.model import MessageItem from queue_sqlite.model import MessageItem
@@ -11,7 +11,7 @@ class TestCore:
task_mounter = core.TaskMounter(TaskMounter) task_mounter = core.TaskMounter(TaskMounter)
task_mounter.get_task_list() task_mounter.get_task_list()
print( 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" "num"
] ]
) )
tests/test_stress.py
+3
View File
@@ -25,6 +25,9 @@ class TestStress:
def _callback(cls, message_item: MessageItem): def _callback(cls, message_item: MessageItem):
with cls.lock: with cls.lock:
cls.callback_counter += 1 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}")
# print(f"callback: {message_item.id}") # print(f"callback: {message_item.id}")
# print(f"callback: {message_item.expire_time}") # print(f"callback: {message_item.expire_time}")