commit 9edd812a7ee89238f93da08d9d4249f7268aa13a Author: chakcy <947105045@qq.com> Date: Fri Aug 8 16:03:02 2025 +0800 first commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..74752a4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,21 @@ +# Python-generated files +__pycache__/ +*.py[oc] +build/ +dist/ +wheels/ +*.egg-info + +# Virtual environments +.venv +output.md +.arts +uv.lock +cache/ + +# IDE files +.idea/ +.vscode/ + +# test files +.pytest_cache/ \ No newline at end of file diff --git a/.python-version b/.python-version new file mode 100644 index 0000000..24ee5b1 --- /dev/null +++ b/.python-version @@ -0,0 +1 @@ +3.13 diff --git a/README.md b/README.md new file mode 100644 index 0000000..e69de29 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..6178507 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,44 @@ +[project] +authors = [ + {name = "chakcy", email = "947105045@qq.com"}, +] +classifiers = [ + "Development Status :: 4 - Beta", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3.13", + "Topic :: System :: Distributed Computing", + "Typing :: Typed", + "Operating System :: Microsoft :: Windows", + "Operating System :: POSIX :: Linux", +] +dependencies = [] +description = "Add your description here" +name = "queue-sqlite" +readme = "README.md" +requires-python = ">=3.13" +version = "0.1.0" + +[[tool.uv.index]] +default = true +url = "https://pypi.tuna.tsinghua.edu.cn/simple" + +[build-system] +build-backend = "hatchling.build" +requires = ["hatchling"] + +[dependency-groups] +dev = [ + "maturin>=1.9.2", + "psutil>=7.0.0", + "pytest>=8.4.1", +] + +[tool.hatch.build.targets.wheel] +packages = ["src/queue_sqlite"] +platforms = ["windows_x86_64", "linux_x86_64"] + +[tool.hatch.build] +artifacts = [ + "src/queue_sqlite/core/*.pyd", +] diff --git a/scripts/clean_pycache.py b/scripts/clean_pycache.py new file mode 100644 index 0000000..070343f --- /dev/null +++ b/scripts/clean_pycache.py @@ -0,0 +1,12 @@ +import os +import shutil +# 需要遍历的目录 +root_dir = "./" +# 遍历目录 +for dirpath, dirnames, filenames in os.walk(root_dir): + if "__pycache__" in dirnames: + # 获取 __pycache__ 目录的全路径 + pycache_dir = os.path.join(dirpath, "__pycache__") + # 删除目录 + shutil.rmtree(pycache_dir) + print(f"Removed: {pycache_dir}") diff --git a/scripts/code_content.py b/scripts/code_content.py new file mode 100644 index 0000000..ae621af --- /dev/null +++ b/scripts/code_content.py @@ -0,0 +1,27 @@ +import os +import sys + +def generate_markdown_from_py_files(directory, output_file): + with open(output_file, 'w', encoding='utf-8') as md_file: + for root, dirs, files in os.walk(directory): + # 排除 venv 目录 + dirs[:] = [d for d in dirs if d != '.venv'] + dirs[:] = [d for d in dirs if d != '.vscode'] + dirs[:] = [d for d in dirs if d != 'scripts'] + dirs[:] = [d for d in dirs if d != 'build'] + for file in files: + if file.endswith('.py') or file.endswith('.rs'): + file_path = os.path.join(root, file) + md_file.write(f'`{file_path}`\n') + md_file.write('```python\n') + with open(file_path, 'r', encoding='utf-8') as py_file: + md_file.write(py_file.read()) + md_file.write('\n```\n\n') + +if __name__ == "__main__": + # 指定目录和输出文件名 + target_directory = sys.argv[1] # 替换为你的目标目录 + output_markdown_file = 'output.md' # 输出的 Markdown 文件名 + + generate_markdown_from_py_files(target_directory, output_markdown_file) + print(f'Markdown 文件已生成：{output_markdown_file}') diff --git a/src/core/.gitignore b/src/core/.gitignore new file mode 100644 index 0000000..c8f0442 --- /dev/null +++ b/src/core/.gitignore @@ -0,0 +1,72 @@ +/target + +# Byte-compiled / optimized / DLL files +__pycache__/ +.pytest_cache/ +*.py[cod] + +# C extensions +*.so + +# Distribution / packaging +.Python +.venv/ +env/ +bin/ +build/ +develop-eggs/ +dist/ +eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +include/ +man/ +venv/ +*.egg-info/ +.installed.cfg +*.egg + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt +pip-selfcheck.json + +# Unit test / coverage reports +htmlcov/ +.tox/ +.coverage +.cache +nosetests.xml +coverage.xml + +# Translations +*.mo + +# Mr Developer +.mr.developer.cfg +.project +.pydevproject + +# Rope +.ropeproject + +# Django stuff: +*.log +*.pot + +.DS_Store + +# Sphinx documentation +docs/_build/ + +# PyCharm +.idea/ + +# VSCode +.vscode/ + +# Pyenv +.python-version diff --git a/src/core/Cargo.lock b/src/core/Cargo.lock new file mode 100644 index 0000000..e3c4bea --- /dev/null +++ b/src/core/Cargo.lock @@ -0,0 +1,748 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "ahash" +version = "0.8.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" +dependencies = [ + "cfg-if", + "once_cell", + "version_check", + "zerocopy", +] + +[[package]] +name = "android-tzdata" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e999941b234f3131b00bc13c22d06e8c5ff726d1b6318ac7eb276997bbb4fef0" + +[[package]] +name = "android_system_properties" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311" +dependencies = [ + "libc", +] + +[[package]] +name = "autocfg" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" + +[[package]] +name = "bitflags" +version = "2.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1b8e56985ec62d17e9c1001dc89c88ecd7dc08e47eba5ec7c29c7b5eeecde967" + +[[package]] +name = "bumpalo" +version = "3.19.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "46c5e41b57b8bba42a04676d81cb89e9ee8e859a1a66f80a5a72e1cb76b34d43" + +[[package]] +name = "cc" +version = "1.2.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3a42d84bb6b69d3a8b3eaacf0d88f179e1929695e1ad012b6cf64d9caaa5fd2" +dependencies = [ + "shlex", +] + +[[package]] +name = "cfg-if" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9555578bc9e57714c812a1f84e4fc5b4d21fcb063490c624de019f7464c91268" + +[[package]] +name = "chrono" +version = "0.4.41" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c469d952047f47f91b68d1cba3f10d63c11d73e4636f24f08daf0278abf01c4d" +dependencies = [ + "android-tzdata", + "iana-time-zone", + "js-sys", + "num-traits", + "wasm-bindgen", + "windows-link", +] + +[[package]] +name = "core" +version = "0.1.0" +dependencies = [ + "chrono", + "pyo3", + "r2d2", + "r2d2_sqlite", + "rusqlite", +] + +[[package]] +name = "core-foundation-sys" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" + +[[package]] +name = "fallible-iterator" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649" + +[[package]] +name = "fallible-streaming-iterator" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a" + +[[package]] +name = "getrandom" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "26145e563e54f2cadc477553f1ec5ee650b00862f0a58bcd12cbdc5f0ea2d2f4" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "wasi", +] + +[[package]] +name = "hashbrown" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" +dependencies = [ + "ahash", +] + +[[package]] +name = "hashlink" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ba4ff7128dee98c7dc9794b6a411377e1404dba1c97deb8d1a55297bd25d8af" +dependencies = [ + "hashbrown", +] + +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" + +[[package]] +name = "iana-time-zone" +version = "0.1.63" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0c919e5debc312ad217002b8048a17b7d83f80703865bbfcfebb0458b0b27d8" +dependencies = [ + "android_system_properties", + "core-foundation-sys", + "iana-time-zone-haiku", + "js-sys", + "log", + "wasm-bindgen", + "windows-core", +] + +[[package]] +name = "iana-time-zone-haiku" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f" +dependencies = [ + "cc", +] + +[[package]] +name = "indoc" +version = "2.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f4c7245a08504955605670dbf141fceab975f15ca21570696aebe9d2e71576bd" + +[[package]] +name = "js-sys" +version = "0.3.77" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1cfaf33c695fc6e08064efbc1f72ec937429614f25eef83af942d0e227c3a28f" +dependencies = [ + "once_cell", + "wasm-bindgen", +] + +[[package]] +name = "libc" +version = "0.2.174" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1171693293099992e19cddea4e8b849964e9846f4acee11b3948bcc337be8776" + +[[package]] +name = "libsqlite3-sys" +version = "0.30.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e99fb7a497b1e3339bc746195567ed8d3e24945ecd636e3619d20b9de9e9149" +dependencies = [ + "cc", + "pkg-config", + "vcpkg", +] + +[[package]] +name = "lock_api" +version = "0.4.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96936507f153605bddfcda068dd804796c84324ed2510809e5b2a624c81da765" +dependencies = [ + "autocfg", + "scopeguard", +] + +[[package]] +name = "log" +version = "0.4.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13dc2df351e3202783a1fe0d44375f7295ffb4049267b0f3018346dc122a1d94" + +[[package]] +name = "memoffset" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "488016bfae457b036d996092f6cb448677611ce4449e970ceaf42695203f218a" +dependencies = [ + "autocfg", +] + +[[package]] +name = "num-traits" +version = "0.2.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" +dependencies = [ + "autocfg", +] + +[[package]] +name = "once_cell" +version = "1.21.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d" + +[[package]] +name = "parking_lot" +version = "0.12.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "70d58bf43669b5795d1576d0641cfb6fbb2057bf629506267a92807158584a13" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc838d2a56b5b1a6c25f55575dfc605fabb63bb2365f6c2353ef9159aa69e4a5" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-targets", +] + +[[package]] +name = "pkg-config" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" + +[[package]] +name = "portable-atomic" +version = "1.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f84267b20a16ea918e43c6a88433c2d54fa145c92a811b5b047ccbe153674483" + +[[package]] +name = "ppv-lite86" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" +dependencies = [ + "zerocopy", +] + +[[package]] +name = "proc-macro2" +version = "1.0.95" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "02b3e5e68a3a1a02aad3ec490a98007cbc13c37cbe84a3cd7b8e406d76e7f778" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "pyo3" +version = "0.25.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8970a78afe0628a3e3430376fc5fd76b6b45c4d43360ffd6cdd40bdde72b682a" +dependencies = [ + "indoc", + "libc", + "memoffset", + "once_cell", + "portable-atomic", + "pyo3-build-config", + "pyo3-ffi", + "pyo3-macros", + "unindent", +] + +[[package]] +name = "pyo3-build-config" +version = "0.25.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "458eb0c55e7ece017adeba38f2248ff3ac615e53660d7c71a238d7d2a01c7598" +dependencies = [ + "once_cell", + "target-lexicon", +] + +[[package]] +name = "pyo3-ffi" +version = "0.25.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7114fe5457c61b276ab77c5055f206295b812608083644a5c5b2640c3102565c" +dependencies = [ + "libc", + "pyo3-build-config", +] + +[[package]] +name = "pyo3-macros" +version = "0.25.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a8725c0a622b374d6cb051d11a0983786448f7785336139c3c94f5aa6bef7e50" +dependencies = [ + "proc-macro2", + "pyo3-macros-backend", + "quote", + "syn", +] + +[[package]] +name = "pyo3-macros-backend" +version = "0.25.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4109984c22491085343c05b0dbc54ddc405c3cf7b4374fc533f5c3313a572ccc" +dependencies = [ + "heck", + "proc-macro2", + "pyo3-build-config", + "quote", + "syn", +] + +[[package]] +name = "quote" +version = "1.0.40" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1885c039570dc00dcb4ff087a89e185fd56bae234ddc7f056a945bf36467248d" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + +[[package]] +name = "r2d2" +version = "0.8.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51de85fb3fb6524929c8a2eb85e6b6d363de4e8c48f9e2c2eac4944abc181c93" +dependencies = [ + "log", + "parking_lot", + "scheduled-thread-pool", +] + +[[package]] +name = "r2d2_sqlite" +version = "0.25.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eb14dba8247a6a15b7fdbc7d389e2e6f03ee9f184f87117706d509c092dfe846" +dependencies = [ + "r2d2", + "rusqlite", + "uuid", +] + +[[package]] +name = "rand" +version = "0.9.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6db2770f06117d490610c7488547d543617b21bfa07796d7a12f6f1bd53850d1" +dependencies = [ + "rand_chacha", + "rand_core", +] + +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.9.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "99d9a13982dcf210057a8a78572b2217b667c3beacbf3a0d8b454f6f82837d38" +dependencies = [ + "getrandom", +] + +[[package]] +name = "redox_syscall" +version = "0.5.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5407465600fb0548f1442edf71dd20683c6ed326200ace4b1ef0763521bb3b77" +dependencies = [ + "bitflags", +] + +[[package]] +name = "rusqlite" +version = "0.32.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7753b721174eb8ff87a9a0e799e2d7bc3749323e773db92e0984debb00019d6e" +dependencies = [ + "bitflags", + "fallible-iterator", + "fallible-streaming-iterator", + "hashlink", + "libsqlite3-sys", + "smallvec", +] + +[[package]] +name = "rustversion" +version = "1.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a0d197bd2c9dc6e53b84da9556a69ba4cdfab8619eb41a8bd1cc2027a0f6b1d" + +[[package]] +name = "scheduled-thread-pool" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3cbc66816425a074528352f5789333ecff06ca41b36b0b0efdfbb29edc391a19" +dependencies = [ + "parking_lot", +] + +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + +[[package]] +name = "shlex" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" + +[[package]] +name = "smallvec" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" + +[[package]] +name = "syn" +version = "2.0.104" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "17b6f705963418cdb9927482fa304bc562ece2fdd4f616084c50b7023b435a40" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "target-lexicon" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e502f78cdbb8ba4718f566c418c52bc729126ffd16baee5baa718cf25dd5a69a" + +[[package]] +name = "unicode-ident" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a5f39404a5da50712a4c1eecf25e90dd62b613502b7e925fd4e4d19b5c96512" + +[[package]] +name = "unindent" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7264e107f553ccae879d21fbea1d6724ac785e8c3bfc762137959b5802826ef3" + +[[package]] +name = "uuid" +version = "1.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3cf4199d1e5d15ddd86a694e4d0dffa9c323ce759fea589f00fef9d81cc1931d" +dependencies = [ + "getrandom", + "js-sys", + "rand", + "wasm-bindgen", +] + +[[package]] +name = "vcpkg" +version = "0.2.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426" + +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + +[[package]] +name = "wasi" +version = "0.14.2+wasi-0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9683f9a5a998d873c0d21fcbe3c083009670149a8fab228644b8bd36b2c48cb3" +dependencies = [ + "wit-bindgen-rt", +] + +[[package]] +name = "wasm-bindgen" +version = "0.2.100" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1edc8929d7499fc4e8f0be2262a241556cfc54a0bea223790e71446f2aab1ef5" +dependencies = [ + "cfg-if", + "once_cell", + "rustversion", + "wasm-bindgen-macro", +] + +[[package]] +name = "wasm-bindgen-backend" +version = "0.2.100" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f0a0651a5c2bc21487bde11ee802ccaf4c51935d0d3d42a6101f98161700bc6" +dependencies = [ + "bumpalo", + "log", + "proc-macro2", + "quote", + "syn", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.100" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7fe63fc6d09ed3792bd0897b314f53de8e16568c2b3f7982f468c0bf9bd0b407" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.100" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ae87ea40c9f689fc23f209965b6fb8a99ad69aeeb0231408be24920604395de" +dependencies = [ + "proc-macro2", + "quote", + "syn", + "wasm-bindgen-backend", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.100" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a05d73b933a847d6cccdda8f838a22ff101ad9bf93e33684f39c1f5f0eece3d" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "windows-core" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0fdd3ddb90610c7638aa2b3a3ab2904fb9e5cdbecc643ddb3647212781c4ae3" +dependencies = [ + "windows-implement", + "windows-interface", + "windows-link", + "windows-result", + "windows-strings", +] + +[[package]] +name = "windows-implement" +version = "0.60.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a47fddd13af08290e67f4acabf4b459f647552718f683a7b415d290ac744a836" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "windows-interface" +version = "0.59.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bd9211b69f8dcdfa817bfd14bf1c97c9188afa36f4750130fcdf3f400eca9fa8" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "windows-link" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e6ad25900d524eaabdbbb96d20b4311e1e7ae1699af4fb28c17ae66c80d798a" + +[[package]] +name = "windows-result" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56f42bd332cc6c8eac5af113fc0c1fd6a8fd2aa08a0119358686e5160d0586c6" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-strings" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56e6c93f3a0c3b36176cb1327a4958a0353d5d166c2a35cb268ace15e91d3b57" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + +[[package]] +name = "wit-bindgen-rt" +version = "0.39.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6f42320e61fe2cfd34354ecb597f86f413484a798ba44a8ca1165c58d42da6c1" +dependencies = [ + "bitflags", +] + +[[package]] +name = "zerocopy" +version = "0.8.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1039dd0d3c310cf05de012d8a39ff557cb0d23087fd44cad61df08fc31907a2f" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ecf5b4cc5364572d7f4c329661bcc82724222973f2cab6f050a4e5c22f75181" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] diff --git a/src/core/Cargo.toml b/src/core/Cargo.toml new file mode 100644 index 0000000..77b073f --- /dev/null +++ b/src/core/Cargo.toml @@ -0,0 +1,16 @@ +[package] +name = "core" +version = "0.1.0" +edition = "2021" + +# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html +[lib] +name = "core" +crate-type = ["cdylib"] + +[dependencies] +chrono = "0.4.41" +pyo3 = "0.25.0" +r2d2 = "0.8" +r2d2_sqlite = "0.25" +rusqlite = { version = "0.32", features = ["bundled"] } diff --git a/src/core/pyproject.toml b/src/core/pyproject.toml new file mode 100644 index 0000000..9fd6944 --- /dev/null +++ b/src/core/pyproject.toml @@ -0,0 +1,15 @@ +[build-system] +requires = ["maturin>=1.9,<2.0"] +build-backend = "maturin" + +[project] +name = "core" +requires-python = ">=3.8" +classifiers = [ + "Programming Language :: Rust", + "Programming Language :: Python :: Implementation :: CPython", + "Programming Language :: Python :: Implementation :: PyPy", +] +dynamic = ["version"] +[tool.maturin] +features = ["pyo3/extension-module"] diff --git a/src/core/src/lib.rs b/src/core/src/lib.rs new file mode 100644 index 0000000..a78e63d --- /dev/null +++ b/src/core/src/lib.rs @@ -0,0 +1,13 @@ +mod queue_operation; +mod task_mounter; + +use pyo3::prelude::*; +use queue_operation::QueueOperation; +use task_mounter::TaskMounter; // 导入结构体 // 导入结构体 + +#[pymodule] +fn core(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> { + m.add_class::()?; + m.add_class::()?; + Ok(()) +} diff --git a/src/core/src/queue_operation.rs b/src/core/src/queue_operation.rs new file mode 100644 index 0000000..c17d0d0 --- /dev/null +++ b/src/core/src/queue_operation.rs @@ -0,0 +1,507 @@ +use chrono::{DateTime, Duration, Utc}; +use pyo3::prelude::*; +use pyo3::types::*; +use r2d2::Pool; +use r2d2_sqlite::SqliteConnectionManager; +use rusqlite::params; + +#[pyclass] +pub struct QueueOperation { + pool: Pool, +} + +#[pymethods] +impl QueueOperation { + #[new] + pub fn new(queue_path: String) -> Self { + let manager = SqliteConnectionManager::file(queue_path.as_str()); + let pool = Pool::new(manager).unwrap(); + QueueOperation { pool } + } + + pub fn init_db(&self) -> PyResult<()> { + let conn = self + .pool + .get() + .map_err(|e| PyErr::new::(e.to_string()))?; + conn.execute( + " + CREATE TABLE IF NOT EXISTS messages ( + id TEXT PRIMARY KEY, + type TEXT NOT NULL, + status INTEGER NOT NULL, + content TEXT NOT NULL, + createtime DATETIME NOT NULL, + updatetime DATETIME NOT NULL, + result TEXT DEFAULT NULL, + priority INTEGER NOT NULL, + source TEXT NOT NULL, + destination TEXT NOT NULL, + retry_count INTEGER NOT NULL, + expire_time DATETIME DEFAULT NULL, + tags TEXT DEFAULT NULL, + metadata TEXT DEFAULT NULL, + is_deleted INTEGER NOT NULL DEFAULT 0 + ); + CREATE INDEX IF NOT EXISTS idx_status ON messages(status); + CREATE INDEX IF NOT EXISTS idx_priority ON messages(priority); + CREATE INDEX IF NOT EXISTS idx_dequeue ON messages( + status, + priority DESC, + createtime ASC + ) WHERE is_deleted = 0 AND status = 0; + ", + params![], + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + let pragmas = [ + "PRAGMA journal_mode=WAL", + "PRAGMA synchronous=NORMAL", + "PRAGMA cache_size=-20000", + "PRAGMA mmap_size=1073741824", + "PRAGMA temp_store=MEMORY", + "PRAGMA busy_timeout=5000", + ]; + for pragma in pragmas.iter() { + let mut stmt = conn + .prepare(pragma) + .map_err(|e| PyErr::new::(e.to_string()))?; + + let mut rows = stmt + .query([]) + .map_err(|e| PyErr::new::(e.to_string()))?; + + while let Some(_) = rows + .next() + .map_err(|e| PyErr::new::(e.to_string()))? + { + } + } + Ok(()) + } + + pub fn enqueue<'py>( + &self, + py: Python<'py>, + message: &Bound<'py, PyDict>, + ) -> PyResult> { + let mut conn = self + .pool + .get() + .map_err(|e| PyErr::new::(e.to_string()))?; + + let get_field = |key: &str| -> PyResult<_> { + message.get_item(key)?.ok_or_else(|| { + PyErr::new::(format!("{} not found", key)) + }) + }; + + let id = get_field("id")?.extract::()?; + let type_ = get_field("type")?.extract::()?; + let status = get_field("status")?.extract::()?; + let content = get_field("content")?.extract::()?; + let createtime = get_field("createtime")?.extract::()?; + let updatetime = get_field("updatetime")?.extract::()?; + let result = get_field("result")?.extract::()?; + let priority = get_field("priority")?.extract::()?; + let source = get_field("source")?.extract::()?; + let destination = get_field("destination")?.extract::()?; + let retry_count = get_field("retry_count")?.extract::()?; + let expire_time = get_field("expire_time")?.extract::()?; + let tags = get_field("tags")?.extract::()?; + let metadata = get_field("metadata")?.extract::()?; + + let tx = conn + .transaction() + .map_err(|e| PyErr::new::(e.to_string()))?; + tx.execute( + " + INSERT INTO messages ( + id, type, status, content, createtime, + updatetime, result, priority, source, + destination, retry_count, expire_time, tags, metadata + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14) + ", + params![ + id, + type_, + status, + content, + createtime, + updatetime, + result, + priority, + source, + destination, + retry_count, + expire_time, + tags, + metadata + ], + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + tx.commit() + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(Bound::from(PyString::new(py, &id))) + } + + pub fn dequeue<'py>(&self, py: Python<'py>, size: i32) -> PyResult> { + let conn = self + .pool + .get() + .map_err(|e| PyErr::new::(e.to_string()))?; + let now: DateTime = Utc::now(); + let iso_format = now.to_rfc3339(); + + // Fixed SQL query with proper subquery alias + let mut stmt = conn + .prepare( + "UPDATE messages + SET status = 1 + WHERE id IN ( + SELECT id + FROM ( + SELECT id, ROW_NUMBER() OVER (ORDER BY priority DESC, createtime ASC) AS rn + FROM messages + WHERE is_deleted = 0 + AND status = 0 + AND (expire_time IS NULL OR expire_time > ?1) + ) AS sub + WHERE rn <= ?2 + ) + RETURNING *;", + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + + let rows = PyList::empty(py); + let mapped_rows = stmt + .query_map(params![iso_format, size], |row| { + let dict = PyDict::new(py); + + // Helper to convert PyErr to rusqlite::Error + let to_rusqlite_error = + |e: PyErr| rusqlite::Error::ToSqlConversionFailure(Box::new(e)); + + // Extract and set each field with proper error handling + dict.set_item("id", row.get::<_, String>("id")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("type", row.get::<_, String>("type")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("status", row.get::<_, i32>("status")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("content", row.get::<_, String>("content")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("createtime", row.get::<_, String>("createtime")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("updatetime", row.get::<_, String>("updatetime")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("result", row.get::<_, Option>("result")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("priority", row.get::<_, i32>("priority")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("source", row.get::<_, Option>("source")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("destination", row.get::<_, Option>("destination")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("retry_count", row.get::<_, i32>("retry_count")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("expire_time", row.get::<_, Option>("expire_time")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("tags", row.get::<_, Option>("tags")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("metadata", row.get::<_, Option>("metadata")?) + .map_err(to_rusqlite_error)?; + + // Return the dictionary as Py + Ok(dict) + }) + .map_err(|e| PyErr::new::(e.to_string()))?; + + for row in mapped_rows { + let dict = + row.map_err(|e| PyErr::new::(e.to_string()))?; + rows.append(dict)?; + } + + Ok(rows.into()) + } + + pub fn get_queue_length(&self) -> PyResult { + let conn = self + .pool + .get() + .map_err(|e| PyErr::new::(e.to_string()))?; + let now = Utc::now().to_rfc3339(); + + // Using query_row for single result handling + let count: i32 = conn + .query_row( + "SELECT COUNT(*) FROM messages + WHERE is_deleted = 0 + AND status = ?1 + AND (expire_time IS NULL OR expire_time > ?2)", + params![0, now], + |row| row.get(0), + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + + Ok(count) + } + pub fn get_completed_messages<'py>(&self, py: Python<'py>) -> PyResult> { + let conn = self + .pool + .get() + .map_err(|e| PyErr::new::(e.to_string()))?; + let mut stmt = conn + .prepare( + "SELECT + * + FROM + messages + WHERE + is_deleted = 0 + AND (status = ?1 OR status = ?2)", + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + let rows = PyList::empty(py); + let mapped_rows = stmt + .query_map(params![2, 3], |row| { + let dict = PyDict::new(py); + + // Helper to convert PyErr to rusqlite::Error + let to_rusqlite_error = + |e: PyErr| rusqlite::Error::ToSqlConversionFailure(Box::new(e)); + + // Extract and set each field with proper error handling + dict.set_item("id", row.get::<_, String>("id")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("type", row.get::<_, String>("type")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("status", row.get::<_, i32>("status")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("content", row.get::<_, String>("content")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("createtime", row.get::<_, String>("createtime")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("updatetime", row.get::<_, String>("updatetime")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("result", row.get::<_, Option>("result")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("priority", row.get::<_, i32>("priority")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("source", row.get::<_, Option>("source")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("destination", row.get::<_, Option>("destination")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("retry_count", row.get::<_, i32>("retry_count")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("expire_time", row.get::<_, Option>("expire_time")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("tags", row.get::<_, Option>("tags")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("metadata", row.get::<_, Option>("metadata")?) + .map_err(to_rusqlite_error)?; + + // Return the dictionary as Py + Ok(dict) + }) + .map_err(|e| PyErr::new::(e.to_string()))?; + + for row in mapped_rows { + let dict = + row.map_err(|e| PyErr::new::(e.to_string()))?; + rows.append(dict)?; + } + + Ok(rows.into()) + } + + pub fn get_result<'py>(&self, py: Python<'py>, id: String) -> PyResult> { + let conn = self + .pool + .get() + .map_err(|e| PyErr::new::(e.to_string()))?; + let mut stmt = conn + .prepare( + "SELECT result, status FROM messages WHERE id = ?1 AND (status = ?2 OR status = ?3)", + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + let mut mapped_rows = stmt + .query_map(params![&id, 2, 3], |row| { + let dict = PyDict::new(py); + + // Helper to convert PyErr to rusqlite::Error + let to_rusqlite_error = + |e: PyErr| rusqlite::Error::ToSqlConversionFailure(Box::new(e)); + + // Extract and set each field with proper error handling + dict.set_item("id", row.get::<_, String>("id")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("type", row.get::<_, String>("type")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("status", row.get::<_, i32>("status")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("content", row.get::<_, String>("content")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("createtime", row.get::<_, String>("createtime")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("updatetime", row.get::<_, String>("updatetime")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("result", row.get::<_, Option>("result")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("priority", row.get::<_, i32>("priority")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("source", row.get::<_, Option>("source")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("destination", row.get::<_, Option>("destination")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("retry_count", row.get::<_, i32>("retry_count")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("expire_time", row.get::<_, Option>("expire_time")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("tags", row.get::<_, Option>("tags")?) + .map_err(to_rusqlite_error)?; + + dict.set_item("metadata", row.get::<_, Option>("metadata")?) + .map_err(to_rusqlite_error)?; + + // Return the dictionary as Py + Ok(dict) + }) + .map_err(|e| PyErr::new::(e.to_string()))?; + + let dict = mapped_rows + .next() + .transpose() + .map_err(|e| PyErr::new::(e.to_string()))?; + + Ok(dict.unwrap().into()) + } + + pub fn update_status(&self, id: String, status: i32) -> PyResult<()> { + let conn = self + .pool + .get() + .map_err(|e| PyErr::new::(e.to_string()))?; + conn.execute( + " + UPDATE messages SET status = ?1 WHERE id = ?2 + ", + params![status, id], + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(()) + } + + pub fn update_result(&self, id: String, result: String) -> PyResult<()> { + let conn = self + .pool + .get() + .map_err(|e| PyErr::new::(e.to_string()))?; + conn.execute( + " + UPDATE messages SET result = ?1 WHERE id = ?2 + ", + params![result, id], + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(()) + } + + pub fn delete_message(&self, id: String) -> PyResult<()> { + let conn = self + .pool + .get() + .map_err(|e| PyErr::new::(e.to_string()))?; + conn.execute( + " + UPDATE messages SET is_deleted = 1 WHERE id = ?1 + ", + params![id], + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(()) + } + + pub fn clean_expired_messages(&self) -> PyResult<()> { + let now: DateTime = Utc::now(); + let iso_format = now.to_rfc3339(); + let conn = self + .pool + .get() + .map_err(|e| PyErr::new::(e.to_string()))?; + conn.execute( + " + UPDATE messages + SET is_deleted = 1 + WHERE is_deleted = 0 + AND expire_time IS NOT NULL + AND expire_time < ?1 + ", + params![iso_format], + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(()) + } + + pub fn clean_old_messages(&self, days: i64) -> PyResult<()> { + // 获取七天前的时间 + let now: DateTime = Utc::now(); + let mut old_time = now - Duration::days(days); + let iso_format = old_time.to_rfc3339(); + let conn = self + .pool + .get() + .map_err(|e| PyErr::new::(e.to_string()))?; + conn.execute( + " + UPDATE messages + SET is_deleted = 1 + WHERE is_deleted = 0 + AND status IN (?1, ?2) + AND updatetime < ?3 + ", + params![2, 3, iso_format], + ) + .map_err(|e| PyErr::new::(e.to_string()))?; + Ok(()) + } +} diff --git a/src/core/src/task_mounter.rs b/src/core/src/task_mounter.rs new file mode 100644 index 0000000..fa04ff6 --- /dev/null +++ b/src/core/src/task_mounter.rs @@ -0,0 +1,27 @@ +use pyo3::prelude::*; + +#[pyclass] +pub struct TaskMounter { + obj: Py, +} + +#[pymethods] +impl TaskMounter { + #[new] + fn new(obj: Py) -> Self { + TaskMounter { obj } + } + + fn get_task_list(&self, py: Python<'_>) -> PyResult<()> { + let obj = self.obj.bind(py); + let result_obj = obj.call_method("get_task_list", (), None)?; + println!("Method result: {:?}", result_obj); + Ok(()) + } + + fn get_task_function(&self, py: Python<'_>, task_name: &str) -> PyResult { + let obj = self.obj.bind(py); + let result = obj.call_method1("get_task_function", (task_name,))?; + Ok(result.into()) + } +} diff --git a/src/queue_sqlite/__init__.py b/src/queue_sqlite/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/queue_sqlite/constant/__init__.py b/src/queue_sqlite/constant/__init__.py new file mode 100644 index 0000000..94a2196 --- /dev/null +++ b/src/queue_sqlite/constant/__init__.py @@ -0,0 +1,5 @@ +from .message_priority import MessagePriority +from .message_status import MessageStatus +from .message_type import MessageType + +__all__ = ["MessagePriority", "MessageStatus", "MessageType"] \ No newline at end of file diff --git a/src/queue_sqlite/constant/message_priority.py b/src/queue_sqlite/constant/message_priority.py new file mode 100644 index 0000000..1ca4c1f --- /dev/null +++ b/src/queue_sqlite/constant/message_priority.py @@ -0,0 +1,8 @@ +from enum import IntEnum + + +class MessagePriority(IntEnum): + LOW = 0 + NORMAL = 1 + HIGH = 2 + URGENT = 3 diff --git a/src/queue_sqlite/constant/message_status.py b/src/queue_sqlite/constant/message_status.py new file mode 100644 index 0000000..c4c16a1 --- /dev/null +++ b/src/queue_sqlite/constant/message_status.py @@ -0,0 +1,10 @@ +from enum import IntEnum + + +class MessageStatus(IntEnum): + PENDING = 0 + PROCESSING = 1 + COMPLETED = 2 + FAILED = 3 + RETRYING = 4 + diff --git a/src/queue_sqlite/constant/message_type.py b/src/queue_sqlite/constant/message_type.py new file mode 100644 index 0000000..323ad41 --- /dev/null +++ b/src/queue_sqlite/constant/message_type.py @@ -0,0 +1,7 @@ +from enum import Enum + + +class MessageType(Enum): + NOTIFICATION = "notification" + SYSTEM = "system" + TASK = "task" diff --git a/src/queue_sqlite/core/__init__.py b/src/queue_sqlite/core/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/queue_sqlite/core/__init__.pyi b/src/queue_sqlite/core/__init__.pyi new file mode 100644 index 0000000..209b35c --- /dev/null +++ b/src/queue_sqlite/core/__init__.pyi @@ -0,0 +1,52 @@ +from ..task_cycle import TaskMounter +from typing import Callable, List +from ..constant import MessageStatus + +class core: + class QueueOperation: + def __init__(self, db_path: str): + ... + + def init_db(self): + ... + + def enqueue(self, message_dict: 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, message_id: str): + ... + + def update_status(self, message_id: str, status: MessageStatus): + ... + + def update_result(self, message_id: str, result: str): + ... + + def delete_message(self, message_id: str): + ... + + def clean_old_messages(self, days: int): + ... + + def clean_expired_messages(self): + ... + + class TaskMounter: + def __init__(self, task_class: type[TaskMounter]): + ... + + def get_task_list(self) -> List[str]: + ... + + def get_task_function(self, name: str) -> Callable: + ... + \ No newline at end of file diff --git a/src/queue_sqlite/core/core.pyd b/src/queue_sqlite/core/core.pyd new file mode 100644 index 0000000..d787250 Binary files /dev/null and b/src/queue_sqlite/core/core.pyd differ diff --git a/src/queue_sqlite/model/__init__.py b/src/queue_sqlite/model/__init__.py new file mode 100644 index 0000000..dd25f13 --- /dev/null +++ b/src/queue_sqlite/model/__init__.py @@ -0,0 +1,3 @@ +from .message_item import MessageItem + +__all__ = ["MessageItem"] \ No newline at end of file diff --git a/src/queue_sqlite/model/message_item.py b/src/queue_sqlite/model/message_item.py new file mode 100644 index 0000000..0b24dc1 --- /dev/null +++ b/src/queue_sqlite/model/message_item.py @@ -0,0 +1,224 @@ +from dataclasses import dataclass, field +from typing import Optional, Dict, Any +from datetime import datetime +from ..constant import MessageStatus, MessagePriority, MessageType +import json +from enum import Enum +import uuid + + +@dataclass +class MessageItem: + # 必需字段 + content: dict = field( + metadata={"description": "消息内容，包含具体的任务数据或信息"} + ) + + # 可选字段 + status: MessageStatus = field( + default=MessageStatus.PENDING, + metadata={"description": "消息状态：PENDING(待处理)、PROCESSING(处理中)、COMPLETED(已完成)、FAILED(失败)"} + ) + type: MessageType = field( + default=MessageType.TASK, + metadata={"description": "消息类型：TASK(任务)、NOTIFICATION(通知)、RESPONSE(响应)"} + ) + createtime: datetime = field( + default_factory=datetime.now, # type: ignore + metadata={"description": "消息创建时间"} + ) + updatetime: datetime = field( + default_factory=datetime.now, # type: ignore + metadata={"description": "消息最后更新时间"} + ) + result: dict = field( + default_factory=dict, + metadata={"description": "消息处理结果，存储任务执行后的输出数据"} + ) + id: str = field( + default_factory=lambda: str(uuid.uuid4()), + metadata={"description": "消息唯一标识符"} + ) + priority: MessagePriority = field( + default=MessagePriority.NORMAL, + metadata={"description": "消息优先级：LOW(低)、NORMAL(普通)、HIGH(高)、URGENT(紧急)"} + ) + source: str = field( + default="client", + metadata={"description": "消息来源，标识发送方"} + ) + destination: str = field( + default="client", + metadata={"description": "消息目标，标识接收方(即task函数名)"} + ) + retry_count: int = field( + default=0, + metadata={"description": "重试次数，记录消息处理失败后的重试次数"} + ) + expire_time: Optional[datetime] = field( + default=None, + metadata={"description": "消息过期时间，超过此时间消息将不再处理"} + ) + tags: Optional[str] = field( + default=None, + metadata={"description": "消息标签，用于分类和筛选"} + ) + metadata: dict = field( + default_factory=dict, + metadata={"description": "额外元数据，存储自定义的扩展信息"} + ) + + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "MessageItem": + """从字典创建消息对象""" + # 处理日期时间字段 + datetime_fields = ['createtime', 'updatetime', 'expire_time'] + for field in datetime_fields: + if field in data: + if data[field] == 'null': + data[field] = None + continue + if isinstance(data[field], str): + data[field] = datetime.fromisoformat(data[field]) + elif isinstance(data[field], datetime): + continue + elif data[field] is None: + data[field] = None + else: + raise ValueError(f"请使用 'str' 或 'datetime' 类型设置 {field} 字段") + + # 处理枚举字段 + if 'type' in data: + if isinstance(data['type'], str): + data['type'] = MessageType(data['type']) # 使用枚举名称 + elif isinstance(data['type'], MessageType): + pass # 已经是枚举类型 + else: + raise ValueError(f"请使用 'str' 或 'MessageType' 类型设置 type 字段") + + if 'status' in data: + if isinstance(data['status'], str): + data['status'] = MessageStatus(data['status']) # 使用枚举名称 + elif isinstance(data['status'], int): + data['status'] = MessageStatus(data['status']) # 使用枚举值 + elif isinstance(data['status'], MessageStatus): + pass # 已经是枚举类型 + else: + raise ValueError(f"Invalid message status: {data['status']}") + + if 'priority' in data: + if isinstance(data['priority'], (int, str)): + data['priority'] = MessagePriority(data['priority']) # type: ignore + elif isinstance(data['priority'], MessagePriority): + pass + else: + raise ValueError("请使用 'int' 或 'str' 类型设置 priority 字段") + + return cls(**data) + + def to_dict(self) -> Dict[str, Any]: + """转换为字典 + + Returns: + Dict[str, Any]: 消息数据字典 + """ + return { + 'id': self.id, + 'type': self.type.value, + 'status': self.status.value, + 'content': self.content, + 'createtime': self.createtime.isoformat(), + 'updatetime': self.updatetime.isoformat(), + 'priority': self.priority.value, + 'source': self.source, + 'retry_count': self.retry_count, + 'expire_time': self.expire_time.isoformat() if self.expire_time else None, + 'tags': self.tags, + 'metadata': self.metadata, + 'result': self.result, + "destination": self.destination + } + + def to_dict_by_core(self) -> Dict[str, Any]: + """转换为核心库需要的字典格式""" + return { + 'id': self.id, + 'type': self.type.value, + 'status': self.status.value, + 'content': json.dumps(self.content), + 'createtime': self.createtime.isoformat(), + 'updatetime': self.updatetime.isoformat(), + 'priority': self.priority.value, + 'source': self.source, + 'retry_count': self.retry_count, + 'expire_time': self.expire_time.isoformat() if self.expire_time else 'null', + 'tags': self.tags if self.tags else 'null', + 'metadata': json.dumps(self.metadata), + 'result': json.dumps(self.result), + "destination": self.destination + } + + + def to_json(self) -> str: + """转换为json字符串""" + # 将枚举类型转换为字符串 + data = self.to_dict() + for key, value in data.items(): + if isinstance(value, Enum): + data[key] = value.value + elif isinstance(value, datetime): + data[key] = value.timestamp() + return json.dumps(data) + + def is_expired(self) -> bool: + """检查消息是否过期 + + Returns: + bool: 是否过期 + """ + if self.expire_time is None: + return False + return datetime.now() > self.expire_time + + def can_retry(self) -> bool: + """检查消息是否可以重试 + + Returns: + bool: 是否可以重试 + """ + return self.retry_count < 3 + + @classmethod + def get_field_descriptions(cls) -> Dict[str, str]: + """获取所有字段的描述信息 + + Returns: + Dict[str, str]: 字段名到描述信息的映射 + """ + from dataclasses import fields + + descriptions = {} + for field_info in fields(cls): + if 'description' in field_info.metadata: + descriptions[field_info.name] = field_info.metadata['description'] + else: + descriptions[field_info.name] = "无描述信息" + + return descriptions + + @classmethod + def print_field_info(cls): + """打印所有字段的信息，包括类型、默认值和描述""" + from dataclasses import fields + + print(f"=== {cls.__name__} 字段信息 ===") + for field_info in fields(cls): + print(f"字段名: {field_info.name}") + print(f"类型: {field_info.type}") + print(f"默认值: {field_info.default}") + if field_info.default_factory is not None: + print(f"默认工厂: {field_info.default_factory}") + if 'description' in field_info.metadata: + print(f"描述: {field_info.metadata['description']}") + print("-" * 50) diff --git a/src/queue_sqlite/py.typed b/src/queue_sqlite/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/src/queue_sqlite/queue_operation/__init__.py b/src/queue_sqlite/queue_operation/__init__.py new file mode 100644 index 0000000..b356f3b --- /dev/null +++ b/src/queue_sqlite/queue_operation/__init__.py @@ -0,0 +1,3 @@ +from .queue_operation import QueueOperation + +__all__ = ["QueueOperation"] \ No newline at end of file diff --git a/src/queue_sqlite/queue_operation/queue_operation.py b/src/queue_sqlite/queue_operation/queue_operation.py new file mode 100644 index 0000000..afa669b --- /dev/null +++ b/src/queue_sqlite/queue_operation/queue_operation.py @@ -0,0 +1,125 @@ +import os +import json +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: dict): + conn = self._get_shard_conn(self._get_shard_index(message_id)) + conn.update_result(message_id, json.dumps(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() + \ No newline at end of file diff --git a/src/queue_sqlite/scheduler/__init__.py b/src/queue_sqlite/scheduler/__init__.py new file mode 100644 index 0000000..67cdd1f --- /dev/null +++ b/src/queue_sqlite/scheduler/__init__.py @@ -0,0 +1,31 @@ +from .receive_scheduler import ReceiveScheduler +from .task_scheduler import TaskScheduler +from ..model import MessageItem +from typing import Callable +from ..queue_operation import QueueOperation +from .cleanup_scheduler import CleanupScheduler + +class QueueScheduler: + def __init__(self, receive_thread_num=1, task_thread_num=1, shard_num=4): + self.queue_operation = QueueOperation(shard_num) + self.receive_scheduler = ReceiveScheduler(self.queue_operation, receive_thread_num) + self.task_scheduler = TaskScheduler(self.queue_operation, task_thread_num) + self.cleanup_scheduler = CleanupScheduler(self.queue_operation) + + def send_message(self, message: MessageItem, callback: Callable): + self.receive_scheduler.send_message(message, callback) + + def start_queue_scheduler(self): + self.receive_scheduler.start_receive_thread() + self.task_scheduler.start_task_thread() + self.cleanup_scheduler.start_cleanup() + + def stop_queue_scheduler(self): + self.receive_scheduler.stop_receive_thread() + self.task_scheduler.stop_task_thread() + self.cleanup_scheduler.stop_cleanup() + # self.queue_operation.close_all_connections() + +__all__ = [ + "QueueScheduler" +] \ No newline at end of file diff --git a/src/queue_sqlite/scheduler/cleanup_scheduler.py b/src/queue_sqlite/scheduler/cleanup_scheduler.py new file mode 100644 index 0000000..0e02f1c --- /dev/null +++ b/src/queue_sqlite/scheduler/cleanup_scheduler.py @@ -0,0 +1,49 @@ +# 新建清理调度器 +import threading +import time +from ..queue_operation import QueueOperation + + +class CleanupScheduler: + def __init__(self, queue_operation: QueueOperation, interval_minutes=60): + self.queue_operation = queue_operation + self.interval = interval_minutes * 60 # 转换为秒 + self.is_running = False + self.cleanup_thread = None + + def cleanup_expired_messages(self): + """清理过期消息""" + while self.is_running: + try: + # 清理7天前的已完成/失败消息 + for i in range(self.queue_operation.shard_num): + self.queue_operation.clean_old_messages(i, 7) + + # 清理过期但未处理的消息 + for i in range(self.queue_operation.shard_num): + self.queue_operation.clean_expired_messages(i) + + except Exception as e: + print(f"清理消息错误: {str(e)}") + + # 休眠等待下次清理 + for _ in range(self.interval): + if not self.is_running: + break + time.sleep(1) + + def start_cleanup(self): + if self.is_running: + return + + self.is_running = True + self.cleanup_thread = threading.Thread(target=self.cleanup_expired_messages, daemon=True) + self.cleanup_thread.start() + + def stop_cleanup(self): + if not self.is_running: + return + + self.is_running = False + if self.cleanup_thread and self.cleanup_thread.is_alive(): + self.cleanup_thread.join(timeout=2.0) \ No newline at end of file diff --git a/src/queue_sqlite/scheduler/receive_scheduler.py b/src/queue_sqlite/scheduler/receive_scheduler.py new file mode 100644 index 0000000..1a54058 --- /dev/null +++ b/src/queue_sqlite/scheduler/receive_scheduler.py @@ -0,0 +1,78 @@ +from ..queue_operation import QueueOperation +from ..model import MessageItem +from typing import Callable +from concurrent.futures import ThreadPoolExecutor +import threading +import time + +class ReceiveScheduler: + def __init__(self, queue_operation: QueueOperation, receive_thread_num: int = 1): + self.callbacks = dict() + self.receive_thread_num = receive_thread_num + self.is_running = False + self.executor = ThreadPoolExecutor(max_workers=receive_thread_num) # 并行执行回调 + self.lock = threading.Lock() + self.queue_operation = queue_operation + self.receive_thread = None # 单一轮询线程 + + def send_message(self, message: MessageItem, callback: Callable = None): # type: ignore + if callback is None: + callback = lambda message: print(message) + # self.queue_operation.enqueue(message.to_dict()) + self.queue_operation.enqueue(message.to_dict_by_core()) + with self.lock: + self.callbacks[message.id] = callback + + def receive_message(self): + """单一轮询线程，并行执行回调""" + + while self.is_running: + message_list = self.queue_operation.get_completed_messages() + if message_list: + for message in message_list: + message = MessageItem.from_dict(message) + callback = None + with self.lock: + callback = self.callbacks.pop(message.id, None) + + if callback: + # 使用线程池并行执行回调 + self.executor.submit(self._safe_callback, callback, message) + else: + time.sleep(0.1) # 适当休眠避免CPU空转 + + def _safe_callback(self, callback, message): + """安全执行回调函数""" + try: + callback(message) + # with ThreadPoolExecutor(max_workers=1) as executor: + # future = executor.submit(callback, message) + # future.result(timeout=5.0) + # except TimeoutError: + # print(f"回调执行超时: {message.id}") + except Exception as e: + print(f"回调执行错误: {str(e)}") + finally: + self.queue_operation.delete_message(message.id) + + def start_receive_thread(self): + if self.is_running: + return + + self.is_running = True + # 创建单一轮询线程 + self.receive_thread = threading.Thread(target=self.receive_message, daemon=True) + self.receive_thread.start() + + def stop_receive_thread(self): + if not self.is_running: + return + + self.is_running = False + # 等待轮询线程结束 + if self.receive_thread and self.receive_thread.is_alive(): + self.receive_thread.join(timeout=2.0) + + # 关闭线程池 + self.executor.shutdown(wait=True) + \ No newline at end of file diff --git a/src/queue_sqlite/scheduler/task_scheduler.py b/src/queue_sqlite/scheduler/task_scheduler.py new file mode 100644 index 0000000..9927584 --- /dev/null +++ b/src/queue_sqlite/scheduler/task_scheduler.py @@ -0,0 +1,91 @@ +from ..queue_operation import QueueOperation +from ..model import MessageItem +from ..constant import MessageStatus +from ..task_cycle.task_cycle import TaskCycle +from ..task_cycle.task_mounter import TaskMounter +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime +import time +import threading + + +class TaskScheduler: + def __init__(self, queue_operation: QueueOperation, task_thread_num: int = 1): + self.task_thread_num = task_thread_num + self.is_running = False + self.executor = ThreadPoolExecutor(max_workers=task_thread_num) # 并行执行任务 + self.queue_operation = queue_operation + self.task_thread = None # 单一轮询线程 + + def _process_message(self, message: MessageItem): + """处理单个消息""" + try: + if message.destination == "client": + message.status = MessageStatus.COMPLETED + message.result = {"result": "success"} + message.updatetime = datetime.now() + return message + + task_function = TaskMounter.get_task_function(message.destination) + task_cycle = TaskCycle(message, task_function) + task_cycle.run() + + message.status = task_cycle.get_task_status() # type: ignore + if message.status == MessageStatus.FAILED: + message.result = {"error": task_cycle.get_task_error()} + else: + message.result = task_cycle.get_task_result() # type: ignore + message.updatetime = datetime.now() + except Exception as e: + message.status = MessageStatus.FAILED + message.result = {"error": str(e)} + message.updatetime = datetime.now() + + return message + + def _update_result(self, message): + """更新任务结果到数据库""" + try: + self.queue_operation.update_result(message.id, message.result) + self.queue_operation.update_status(message.id, message.status) + except Exception as e: + print(f"任务结果更新失败: {str(e)}") + + def task_callback(self): + """单一轮询线程，并行执行任务""" + while self.is_running: + try: + message_list = self.queue_operation.dequeue(size=self.task_thread_num * 2) + if message_list: + # 并行处理所有获取到的消息 + for message in message_list: + self.executor.submit( + lambda m: self._update_result(self._process_message(m)), + MessageItem.from_dict(message) + ) + else: + time.sleep(0.1) # 适当休眠 + except Exception as e: + print(f"任务调度错误: {str(e)}") + time.sleep(1) + + def start_task_thread(self): + if self.is_running: + return + + self.is_running = True + # 创建单一轮询线程 + self.task_thread = threading.Thread(target=self.task_callback, daemon=True) + self.task_thread.start() + + def stop_task_thread(self): + if not self.is_running: + return + + self.is_running = False + # 等待轮询线程结束 + if self.task_thread and self.task_thread.is_alive(): + self.task_thread.join(timeout=2.0) + + # 关闭线程池 + self.executor.shutdown(wait=True) \ No newline at end of file diff --git a/src/queue_sqlite/task_cycle/__init__.py b/src/queue_sqlite/task_cycle/__init__.py new file mode 100644 index 0000000..db346d3 --- /dev/null +++ b/src/queue_sqlite/task_cycle/__init__.py @@ -0,0 +1,5 @@ +from .task_mounter import TaskMounter + +__all__ = [ + "TaskMounter" +] \ No newline at end of file diff --git a/src/queue_sqlite/task_cycle/task_cycle.py b/src/queue_sqlite/task_cycle/task_cycle.py new file mode 100644 index 0000000..7cff5d0 --- /dev/null +++ b/src/queue_sqlite/task_cycle/task_cycle.py @@ -0,0 +1,40 @@ +from ..model import MessageItem +from typing import Callable, Optional +from ..constant import MessageStatus + +class TaskCycle: + def __init__(self, message_item: MessageItem, callback: Optional[Callable]): + self.message_item = message_item + self.callback = callback + self.task_result = None + self.task_status = None + self.task_error = None + + def run(self): + try: + task_result = self.callback(self.message_item) # type: ignore + except Exception as e: + self.task_result = None + self.task_status = MessageStatus.FAILED + self.task_error = str(e) + else: + self.task_result = task_result # type: ignore + self.task_status = MessageStatus.COMPLETED + self.task_error = None + + def get_task_result(self): + return self.task_result + + def get_task_status(self): + return self.task_status + + def get_task_error(self): + return self.task_error + + def get_task_message_item(self): + return self.message_item + + def get_task_callback(self): + return self.callback + + diff --git a/src/queue_sqlite/task_cycle/task_mounter.py b/src/queue_sqlite/task_cycle/task_mounter.py new file mode 100644 index 0000000..e2720cf --- /dev/null +++ b/src/queue_sqlite/task_cycle/task_mounter.py @@ -0,0 +1,42 @@ +from typing import Callable, List + + +class TaskMounter: + @classmethod + def mount_task(cls, function: Callable): + setattr(cls, function.__name__, function) + + @staticmethod + def task(meta: dict = {}): + """带参数的装饰器""" + def decorator(function: Callable): + # 使用自定义名称或函数原名 + function.meta = meta # type: ignore + setattr(TaskMounter, function.__name__, function) + return function + return decorator + + @classmethod + def get_task_function(cls, name: str): + return getattr(cls, name, None) + + @classmethod + def get_task_list(cls) -> List[str]: + """获取所有挂载的任务函数名称列表""" + task_list = [] + + # 遍历类属性字典 + for attr_name, attr_value in vars(cls).items(): + # 过滤条件： + # 1. 必须是可调用对象（函数） + # 2. 不是类自带的特殊方法（非双下划线开头） + # 3. 不是类方法本身（如 mount_task, get_task_list 等） + if ( + callable(attr_value) + and not attr_name.startswith("__") + and attr_name not in ["mount_task", "task", "get_task_function", "get_task_list"] + ): + task_list.append(attr_name) + + return task_list + \ No newline at end of file diff --git a/tests/tasks/__init__.py b/tests/tasks/__init__.py new file mode 100644 index 0000000..2bcfab1 --- /dev/null +++ b/tests/tasks/__init__.py @@ -0,0 +1,5 @@ +from .example import example + +__all__ = [ + "example" +] \ No newline at end of file diff --git a/tests/tasks/example.py b/tests/tasks/example.py new file mode 100644 index 0000000..fc2954b --- /dev/null +++ b/tests/tasks/example.py @@ -0,0 +1,19 @@ +from queue_sqlite.task_cycle.task_mounter import TaskMounter +from queue_sqlite.model import MessageItem + + +@TaskMounter.task(meta={"task_name": "test"}) +def example(message_item: MessageItem): + def fibonacci_generator(): + a, b = 0, 1 + while True: + yield a + a, b = b, a + b + + # 示例：获取前10项 + fib = fibonacci_generator() + message_item.result = {"fibonacci": [next(fib) for _ in range(500)]} + return message_item.to_json() + + +# 输出：[0, 1, 1, 2, 3, 5, 8, 13, 21, 34] \ No newline at end of file diff --git a/tests/test_core.py b/tests/test_core.py new file mode 100644 index 0000000..2c3dc99 --- /dev/null +++ b/tests/test_core.py @@ -0,0 +1,31 @@ +from queue_sqlite.core import core +from queue_sqlite.task_cycle import TaskMounter +from queue_sqlite.model import MessageItem + + +class TestCore: + queue_operation = core.QueueOperation("./cache/test.db") + + @classmethod + def test_task_mounter(cls): + task_mounter = core.TaskMounter(TaskMounter) + task_mounter.get_task_list() + print(task_mounter.get_task_function("task")({"num": 1})(lambda x: x+1).meta["num"]) + from tasks import example + task_mounter.get_task_list() + print(task_mounter.get_task_function("")) + + @classmethod + def test_queue_operation_init_db(cls): + cls.queue_operation.init_db() + + @classmethod + def test_queue_operation_enqueue(cls): + message_item = MessageItem(content={"num": 1}, destination="test") + cls.queue_operation.enqueue(message_item.to_dict_by_core()) + + @classmethod + def test_queue_operation_dequeue(cls): + message_item = MessageItem(content={"num": 1}, destination="test") + cls.queue_operation.enqueue(message_item.to_dict_by_core()) + print(cls.queue_operation.dequeue(1)) \ No newline at end of file diff --git a/tests/test_message_item.py b/tests/test_message_item.py new file mode 100644 index 0000000..7663c6e --- /dev/null +++ b/tests/test_message_item.py @@ -0,0 +1,18 @@ +from queue_sqlite.model import MessageItem + + +class TestMessageItem: + messageItem = MessageItem(content={"num": 1}, destination="test") + + @classmethod + def test_to_dict(cls): + print(cls.messageItem.to_dict()) + + @classmethod + def test_to_json(cls): + print(cls.messageItem.to_json()) + + @classmethod + def test_from_dict(cls): + messageItem = MessageItem.from_dict(cls.messageItem.to_dict()) + print(messageItem.to_dict()) \ No newline at end of file diff --git a/tests/test_stress.py b/tests/test_stress.py new file mode 100644 index 0000000..3ffaa32 --- /dev/null +++ b/tests/test_stress.py @@ -0,0 +1,80 @@ +from queue_sqlite.model import MessageItem +from queue_sqlite.scheduler import QueueScheduler +import time +from tasks import * +import threading +import psutil + + +class TestStress: + + callback_counter = 0 + lock = threading.Lock() + is_monitoring = False + + @classmethod + def monitor_resources(cls): + """监控资源使用情况""" + while cls.is_monitoring: + cpu = psutil.cpu_percent() + memory = psutil.virtual_memory().percent + print(f"CPU使用率: {cpu}%, 内存使用率: {memory}%") + time.sleep(0.5) + + @classmethod + def _callback(cls, message_item: MessageItem): + with cls.lock: + cls.callback_counter += 1 + # print(f"callback: {message_item}") + # print(f"callback: {message_item.id}") + # print(f"callback: {message_item.expire_time}") + # print(f"callback: {message_item.tags}") + + @classmethod + def test_stress(cls): + TASK_NUM = 1000 + messages = [] + for i in range(TASK_NUM): # 增加任务数量 + message_item = MessageItem(content={"num": i}, destination="example") + messages.append(message_item) + + cls.is_monitoring = True + monitor_thread = threading.Thread(target=cls.monitor_resources, daemon=True) + monitor_thread.start() + queue_scheduler = QueueScheduler(receive_thread_num=4, task_thread_num=8, shard_num=12) + + # threads = [] + # for message_item in messages: + # t = threading.Thread(target=queue_scheduler.send_message, args=(message_item, cls.test_callback)) + # threads.append(t) + + # 添加性能计时 + start_time = time.perf_counter() + + queue_scheduler.start_queue_scheduler() + + for message_item in messages: # 增加任务数量 + message_item: MessageItem + queue_scheduler.send_message(message_item, cls._callback) + + # # 多线程发送任务 + # for t in threads: + # t.start() + + # # 动态等待任务完成 + start_wait = time.time() + while (queue_scheduler.queue_operation.get_queue_length() > 0 or cls.callback_counter < TASK_NUM): # 30秒超时 + time.sleep(0.5) + print(f"当前队列长度：{queue_scheduler.queue_operation.get_queue_length()}") + print(f"当前回调数量：{cls.callback_counter}") + + queue_scheduler.stop_queue_scheduler() + + print(f"当前队列长度：{queue_scheduler.queue_operation.get_queue_length()}") + print(f"测试函数：斐波那契数列") + print(f"次数：{TASK_NUM}") + print(f"总耗时: {time.perf_counter() - start_time:.2f}秒") + + cls.is_monitoring = False + monitor_thread.join(timeout=1) + \ No newline at end of file