zthreading

library0.1.19pypypi✓ verified 22d ago

Zthreading is a Python library offering a unified interface for managing both traditional Python threads and `asyncio` tasks. It provides wrapper classes for event broadcasting, task management, and decorators for common concurrency patterns. The library is actively maintained, with minor releases addressing bug fixes and new features every few months.

workflow

pip install zthreading

github ↗package ↗

INSTALL

IMPORT

SIG · ZTHREADING

zthreading

workflowpythonv0.1.19

Install

1.6s avg

Import

—

Disk

16MB

Pass rate

10/ 10

Env Coverage10 / 10

glibc

3.9–3.13

musl

3.9–3.13

Install & Compatibility

Where this runs

tested against v0.1.19 · pip install

no network on importno background threads

Install × environment matrix

Each cell = how many times install + import succeeded across repeated harness runs. Partial = flaky.
glibc = Debian/Ubuntu slim · musl = Alpine Linux

musl

py 3.10–3.95 runs

installs and imports cleanly · install 0.0s · import 0.000s · 18MB

glibc

py 3.10–3.95 runs

installs and imports cleanly · install 1.6s · import 0.000s · 18MB

16MB installed

● package 16MB

Code

Verified usage

Verified import paths — ran on the pinned version, not inferred.

ThreadManager

✓ import zthreading

✗ from zthreading import ThreadManager

This quickstart demonstrates how to use `ThreadManager` to add and manage tasks, both directly and using the `as_task` decorator. It shows how to retrieve results and properly wait for tasks to complete, then clean up the manager.

import time from zthreading.thread_manager import ThreadManager from zthreading.decorators import as_task def my_task(name): print(f"Task {name} started") time.sleep(1) # Simulate work print(f"Task {name} finished") return f"Result for {name}" # Initialize ThreadManager manager = ThreadManager() # Add tasks task1_handle = manager.add_task(my_task, args=("Task 1",)) task2_handle = manager.add_task(my_task, args=("Task 2",)) # Use decorator for another task @as_task(manager=manager) def decorated_task(name): print(f"Decorated task {name} started") time.sleep(0.5) print(f"Decorated task {name} finished") return f"Decorated result for {name}" decorated_task_handle = decorated_task("Decorated Task 3") # Wait for all tasks to complete manager.wait_for_tasks() print(f"\nTask 1 result: {task1_handle.get_result()}") print(f"Task 2 result: {task2_handle.get_result()}") print(f"Decorated task result: {decorated_task_handle.get_result()}") # Ensure all threads are properly joined/terminated manager.join_all()

Debug

Known issues

breakingThe signature for the `on_event` argument changed in version 0.1.11 from `on_event(name, *args, **kwargs)` to `on_event(evnt: Event)`. Event handlers now receive an `Event` object.

fix

Update event handler signatures to accept an `Event` object, and access event details (name, sender, args, kwargs) via its attributes (e.g., `evnt.name`, `evnt.args`).

affects: <0.1.11

breakingThe `predict` method signature for `wait_for_events` (and `wait_for`) was changed in version 0.1.13. It now takes `predict(handler, event)` instead of `predict(handler, name: str, *args, **kwargs)`.

fix

Adjust the `predict` method implementation to accept the `event` object directly, similar to the `on_event` changes. The `Event` object encapsulates all relevant event data.

affects: <0.1.13

gotchaOlder versions of `zthreading` (prior to 0.1.18) used `threading.currentThread` internally, which was deprecated in Python 3.10 and removed in Python 3.12. Running these older versions on Python 3.10+ will result in deprecation warnings or `AttributeError`.

fix

Upgrade `zthreading` to version 0.1.18 or newer to ensure compatibility with Python 3.10+ environments. The library now uses `threading.current_thread`.

affects: <0.1.18

gotchaWhen using `ThreadManager` or `AsyncioManager` with tasks, it's crucial to call `manager.wait_for_tasks()` or `manager.join_all()` to ensure all tasks complete and resources are cleaned up. Forgetting this can lead to unhandled tasks or processes remaining active.

fix

Always explicitly wait for tasks using `manager.wait_for_tasks()` (for results) and/or `manager.join_all()` (for full cleanup of threads/tasks) before your application exits or when you no longer need the manager.

affects: All versions

Upgrade

Version history

0.1.19latest on PyPI · released Jun 13, 2022

Audit

Dependencies

No dependency data recorded yet.

Agent activity

58 hits · last 30 days

node

OpenAI (training)

Resources

githubhttps://github.com/LamaAni/zthreading.py ↗

packagehttps://pypi.org/project/zthreading/ ↗