<!-- README.md -->
[](https://pypi.org/project/pynnex/)
[](https://github.com/nexconnectio/pynnex/blob/main/LICENSE)
[](https://github.com/nexconnectio/pynnex/actions)
[](https://pypi.org/project/pynnex/)
# PynneX
**Looking for a lightweight alternative to heavy Signals/Slots libraries in an asynchronous and multi-threaded environment?**
PynneX is a pure-Python (asyncio-based) library that streamlines event-driven concurrency without the overhead of larger GUI frameworks or external dependencies.
---
## Why PynneX?
Modern Python applications often blend async I/O and multithreading. Typical Signals/Slots solutions from GUI toolkits or external libraries can impose extra dependencies, especially when you only need concurrency handling rather than full UI features.
PynneX offers a **focused** approach:
- Decorator-based signals and slots for clean, event-driven code
- Built-in **thread-safety**, so you don’t manually deal with locks or queues
- Easy background task handling via `@nx_with_worker`
- Seamless integration with **asyncio** (async or sync slots)
- No external dependencies beyond Python 3.10+ (for improved asyncio support)
As a result, events flow safely across threads and coroutines without “callback spaghetti,” giving you a cleaner concurrency model in pure Python.
---
## Key Features
- **Pure Python**: No external dependencies needed
- **Async/Await Friendly**: Slots can be synchronous or asynchronous, integrating naturally with `asyncio`
- **Thread-Safe**: Automatically manages signal emissions and slot executions across thread boundaries
- **Flexible Connection Types**: Direct or queued connections, chosen based on whether caller/callee share the same thread
- **Worker Thread Pattern**: Decorator `@nx_with_worker` provides a dedicated thread & event loop, simplifying background tasks
- **Familiar Decorators**: `@nx_signal`, `@nx_slot`, `@nx_with_worker`; also available without `nx_` prefix
- **Thread-Safe Properties**: Use `@nx_property` to emit signals on value changes, with automatic thread dispatch
- **Weak Reference**: If you connect a slot with `weak=True`, the connection is removed automatically once the receiver is garbage-collected
### **Requires an Existing Event Loop**
PynneX depends on Python’s `asyncio`. You **must** have a running event loop (e.g., `asyncio.run(...)`) for certain features like async slots or cross-thread calls.
If no event loop is running, PynneX raises a `RuntimeError` instead of creating one behind the scenes—this ensures predictable concurrency behavior.
## Installation
```bash
pip install pynnex
```
PynneX requires **Python 3.10+**, leveraging newer asyncio improvements.
Alternatively, clone from GitHub and install locally:
```bash
git clone https://github.com/nexconnectio/pynnex.git
cd pynnex
pip install -e .
```
For development (includes tests and linting tools):
```
pip install -e ".[dev]
```
## Quick Hello (Signals/Slots)
Here’s the simplest “Hello, Signals/Slots” example. Once installed, run the snippet below:
```python
# hello_pynnex.py
from pynnex import with_signals, signal, slot
@with_signals
class Greeter:
@signal
def greet(self):
"""Signal emitted when greeting happens."""
pass
def say_hello(self):
self.greet.emit("Hello from PynneX!")
@with_signals
class Printer:
@slot
def on_greet(self, message):
print(message)
greeter = Greeter()
printer = Printer()
# Connect the signal to the slot
greeter.greet.connect(printer, printer.on_greet)
# Fire the signal
greeter.say_hello()
```
**Output:**
```
Hello from PynneX!
```
By simply defining `signal` and `slot`, you can set up intuitive event handling that also works smoothly in multithreaded contexts.
---
## Usage & Examples
Below are some brief examples. For more, see the [docs/](https://github.com/nexconnectio/pynnex/blob/main/docs/) directory.
### Basic Counter & Display
```python
from pynnex import with_signals, signal, slot
@with_signals
class Counter:
def __init__(self):
self.count = 0
@signal
def count_changed(self):
pass
def increment(self):
self.count += 1
self.count_changed.emit(self.count)
@with_signals
class Display:
@slot
async def on_count_changed(self, value):
print(f"Count is now: {value}")
# Connect and use
counter = Counter()
display = Display()
counter.count_changed.connect(display, display.on_count_changed)
counter.increment() # Will print: "Count is now: 1"
```
### Asynchronous Slot Example
```python
@with_signals
class AsyncDisplay:
@slot
async def on_count_changed(self, value):
await asyncio.sleep(1) # Simulate async operation
print(f"Count updated to: {value}")
# Usage in async context
async def main():
counter = Counter()
display = AsyncDisplay()
counter.count_changed.connect(display, display.on_count_changed)
counter.increment()
# Wait for async processing
await asyncio.sleep(1.1)
asyncio.run(main())
```
## Core Concepts
### Signals and Slots
- Signals: Declared with `@nx_signal`. Signals are attributes of a class that can be emitted to notify interested parties.
- Slots: Declared with `@nx_slot`. Slots are methods that respond to signals. Slots can be synchronous or async functions.
- Connections: Use `signal.connect(receiver, slot)` to link signals to slots. Connections can also be made directly to functions or lambdas.
### Thread Safety and Connection Types
PynneX automatically detects whether the signal emission and slot execution occur in the same thread or different threads:
- **Auto Connection**: When connection_type is AUTO_CONNECTION (default), PynneX checks whether the slot is a coroutine function or whether the caller and callee share the same thread affinity. If they are the same thread and slot is synchronous, it uses direct connection. Otherwise, it uses queued connection.
- **Direct Connection**: If signal and slot share the same thread affinity, the slot is invoked directly.
- **Queued Connection**: If they differ, the call is queued to the slot’s thread/event loop, ensuring thread safety.
This mechanism frees you from manually dispatching calls across threads.
### Thread-Safe Properties
The `@nx_property` decorator provides thread-safe property access with automatic signal emission:
```python
@with_signals
class Example:
def __init__(self):
super().__init__()
self._data = None
@signal
def updated(self):
"""Signal emitted when data changes."""
pass
@nx_property(notify=updated)
def data(self):
"""Thread-safe property with change notification."""
return self._data
@data.setter
def data(self, value):
self._data = value
e = Example()
e.data = 42 # Thread-safe property set; emits 'updated' signal on change
```
### Worker Threads
For background work, PynneX provides a `@nx_with_worker` decorator that:
- Spawns a dedicated event loop in a worker thread.
- Allows you to queue async tasks to this worker.
- Enables easy start/stop lifecycle management.
- Integrates with signals and slots for thread-safe updates to the main
**Worker Example**
```python
from pynnex import nx_with_worker, nx_signal
@with_worker
class DataProcessor:
@signal
def processing_done(self):
"""Emitted when processing completes"""
async def run(self, *args, **kwargs):
# The main entry point for the worker thread’s event loop
# Wait for tasks or stopping signal
await self.wait_for_stop()
async def process_data(self, data):
# Perform heavy computation in the worker thread
result = await heavy_computation(data)
self.processing_done.emit(result)
processor = DataProcessor()
processor.start()
# Queue a task to run in the worker thread:
processor.queue_task(processor.process_data(some_data))
# Stop the worker gracefully
processor.stop()
```
## Documentation and Example
- [Usage Guide](https://github.com/nexconnectio/pynnex/blob/main/docs/usage.md): Learn how to define signals/slots, manage threads, and structure your event-driven code.
- [API Reference](https://github.com/nexconnectio/pynnex/blob/main/docs/api.md): Detailed documentation of classes, decorators, and functions.
- [Examples](https://github.com/nexconnectio/pynnex/blob/main/docs/examples.md): Practical use cases, including UI integration, async operations, and worker pattern usage.
- [Logging Guidelines](https://github.com/nexconnectio/pynnex/blob/main/docs/logging.md): Configure logging levels and handlers for debugging.
- [Testing Guide](https://github.com/nexconnectio/pynnex/blob/main/docs/testing.md): earn how to run tests and contribute safely.
## Logging
Configure logging to diagnose issues:
```python
import logging
logging.getLogger('pynnex').setLevel(logging.DEBUG)
```
For more details, see the [Logging Guidelines](https://github.com/nexconnectio/pynnex/blob/main/docs/logging.md).
## Testing
Pynnex uses `pytest` for testing:
```bash
# Run all tests
pytest
# Run with verbose output
pytest -v
# Run specific test file
pytest tests/unit/test_signal.py
```
See the [Testing Guide](https://github.com/nexconnectio/pynnex/blob/main/docs/testing.md) for more details.
## Contributing
We welcome contributions! Please read our [Contributing Guidelines](https://github.com/nexconnectio/pynnex/blob/main/CONTRIBUTING.md) before submitting PRs.
## Sponsorship & Donations
If PynneX has helped simplify your async/multithreaded workflows, please consider [sponsoring us](https://github.com/nexconnectio/pynnex/blob/main/.github/FUNDING.yml). All funds go toward infrastructure, documentation, and future development.
Please note that financial contributions support only the project's maintenance and do not grant financial rewards to individual contributors.
## License
`PynneX` is licensed under the MIT License. See [LICENSE](https://github.com/nexconnectio/pynnex/blob/main/LICENSE) for details.
Raw data
{
"_id": null,
"home_page": null,
"name": "pynnex",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": "signal-slot, decorator, multithreading, asyncio",
"author": null,
"author_email": "San Kim <nexconnect.io@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/8b/85/3a6c241935aca58a7fbe7e0f99596a202142c3b8a232a87ec0910e35a960/pynnex-1.0.1.tar.gz",
"platform": null,
"description": "<!-- README.md -->\r\n\r\n[](https://pypi.org/project/pynnex/)\r\n[](https://github.com/nexconnectio/pynnex/blob/main/LICENSE)\r\n[](https://github.com/nexconnectio/pynnex/actions)\r\n[](https://pypi.org/project/pynnex/)\r\n\r\n# PynneX\r\n\r\n**Looking for a lightweight alternative to heavy Signals/Slots libraries in an asynchronous and multi-threaded environment?** \r\nPynneX is a pure-Python (asyncio-based) library that streamlines event-driven concurrency without the overhead of larger GUI frameworks or external dependencies.\r\n\r\n---\r\n\r\n## Why PynneX?\r\n\r\nModern Python applications often blend async I/O and multithreading. Typical Signals/Slots solutions from GUI toolkits or external libraries can impose extra dependencies, especially when you only need concurrency handling rather than full UI features.\r\n\r\nPynneX offers a **focused** approach:\r\n- Decorator-based signals and slots for clean, event-driven code\r\n- Built-in **thread-safety**, so you don\u2019t manually deal with locks or queues\r\n- Easy background task handling via `@nx_with_worker`\r\n- Seamless integration with **asyncio** (async or sync slots)\r\n- No external dependencies beyond Python 3.10+ (for improved asyncio support)\r\n\r\nAs a result, events flow safely across threads and coroutines without \u201ccallback spaghetti,\u201d giving you a cleaner concurrency model in pure Python.\r\n\r\n---\r\n\r\n## Key Features\r\n\r\n- **Pure Python**: No external dependencies needed \r\n- **Async/Await Friendly**: Slots can be synchronous or asynchronous, integrating naturally with `asyncio` \r\n- **Thread-Safe**: Automatically manages signal emissions and slot executions across thread boundaries \r\n- **Flexible Connection Types**: Direct or queued connections, chosen based on whether caller/callee share the same thread \r\n- **Worker Thread Pattern**: Decorator `@nx_with_worker` provides a dedicated thread & event loop, simplifying background tasks \r\n- **Familiar Decorators**: `@nx_signal`, `@nx_slot`, `@nx_with_worker`; also available without `nx_` prefix \r\n- **Thread-Safe Properties**: Use `@nx_property` to emit signals on value changes, with automatic thread dispatch \r\n- **Weak Reference**: If you connect a slot with `weak=True`, the connection is removed automatically once the receiver is garbage-collected\r\n\r\n### **Requires an Existing Event Loop**\r\n\r\nPynneX depends on Python\u2019s `asyncio`. You **must** have a running event loop (e.g., `asyncio.run(...)`) for certain features like async slots or cross-thread calls. \r\nIf no event loop is running, PynneX raises a `RuntimeError` instead of creating one behind the scenes\u2014this ensures predictable concurrency behavior.\r\n\r\n## Installation\r\n\r\n```bash\r\npip install pynnex\r\n```\r\n\r\nPynneX requires **Python 3.10+**, leveraging newer asyncio improvements.\r\nAlternatively, clone from GitHub and install locally: \r\n\r\n```bash\r\ngit clone https://github.com/nexconnectio/pynnex.git\r\ncd pynnex\r\npip install -e .\r\n```\r\n\r\nFor development (includes tests and linting tools):\r\n```\r\npip install -e \".[dev]\r\n```\r\n\r\n## Quick Hello (Signals/Slots)\r\n\r\nHere\u2019s the simplest \u201cHello, Signals/Slots\u201d example. Once installed, run the snippet below:\r\n\r\n```python\r\n# hello_pynnex.py\r\nfrom pynnex import with_signals, signal, slot\r\n\r\n@with_signals\r\nclass Greeter:\r\n @signal\r\n def greet(self):\r\n \"\"\"Signal emitted when greeting happens.\"\"\"\r\n pass\r\n\r\n def say_hello(self):\r\n self.greet.emit(\"Hello from PynneX!\")\r\n\r\n@with_signals\r\nclass Printer:\r\n @slot\r\n def on_greet(self, message):\r\n print(message)\r\n\r\ngreeter = Greeter()\r\nprinter = Printer()\r\n\r\n# Connect the signal to the slot\r\ngreeter.greet.connect(printer, printer.on_greet)\r\n\r\n# Fire the signal\r\ngreeter.say_hello()\r\n```\r\n\r\n**Output:**\r\n```\r\nHello from PynneX!\r\n```\r\n\r\nBy simply defining `signal` and `slot`, you can set up intuitive event handling that also works smoothly in multithreaded contexts.\r\n\r\n---\r\n\r\n## Usage & Examples\r\n\r\nBelow are some brief examples. For more, see the [docs/](https://github.com/nexconnectio/pynnex/blob/main/docs/) directory.\r\n\r\n### Basic Counter & Display\r\n```python\r\nfrom pynnex import with_signals, signal, slot\r\n\r\n@with_signals\r\nclass Counter:\r\n def __init__(self):\r\n self.count = 0\r\n \r\n @signal\r\n def count_changed(self):\r\n pass\r\n \r\n def increment(self):\r\n self.count += 1\r\n self.count_changed.emit(self.count)\r\n\r\n@with_signals\r\nclass Display:\r\n @slot\r\n async def on_count_changed(self, value):\r\n print(f\"Count is now: {value}\")\r\n\r\n# Connect and use\r\ncounter = Counter()\r\ndisplay = Display()\r\ncounter.count_changed.connect(display, display.on_count_changed)\r\ncounter.increment() # Will print: \"Count is now: 1\"\r\n```\r\n\r\n### Asynchronous Slot Example\r\n```python\r\n@with_signals\r\nclass AsyncDisplay:\r\n @slot\r\n async def on_count_changed(self, value):\r\n await asyncio.sleep(1) # Simulate async operation\r\n print(f\"Count updated to: {value}\")\r\n\r\n# Usage in async context\r\nasync def main():\r\n counter = Counter()\r\n display = AsyncDisplay()\r\n \r\n counter.count_changed.connect(display, display.on_count_changed)\r\n counter.increment()\r\n \r\n # Wait for async processing\r\n await asyncio.sleep(1.1)\r\n\r\nasyncio.run(main())\r\n```\r\n\r\n## Core Concepts\r\n\r\n### Signals and Slots\r\n- Signals: Declared with `@nx_signal`. Signals are attributes of a class that can be emitted to notify interested parties.\r\n- Slots: Declared with `@nx_slot`. Slots are methods that respond to signals. Slots can be synchronous or async functions.\r\n- Connections: Use `signal.connect(receiver, slot)` to link signals to slots. Connections can also be made directly to functions or lambdas.\r\n\r\n### Thread Safety and Connection Types\r\nPynneX automatically detects whether the signal emission and slot execution occur in the same thread or different threads:\r\n\r\n- **Auto Connection**: When connection_type is AUTO_CONNECTION (default), PynneX checks whether the slot is a coroutine function or whether the caller and callee share the same thread affinity. If they are the same thread and slot is synchronous, it uses direct connection. Otherwise, it uses queued connection.\r\n- **Direct Connection**: If signal and slot share the same thread affinity, the slot is invoked directly.\r\n- **Queued Connection**: If they differ, the call is queued to the slot\u2019s thread/event loop, ensuring thread safety.\r\n\r\nThis mechanism frees you from manually dispatching calls across threads.\r\n\r\n### Thread-Safe Properties\r\nThe `@nx_property` decorator provides thread-safe property access with automatic signal emission:\r\n\r\n```python\r\n@with_signals\r\nclass Example:\r\n def __init__(self):\r\n super().__init__()\r\n self._data = None\r\n \r\n @signal\r\n def updated(self):\r\n \"\"\"Signal emitted when data changes.\"\"\"\r\n pass\r\n \r\n @nx_property(notify=updated)\r\n def data(self):\r\n \"\"\"Thread-safe property with change notification.\"\"\"\r\n return self._data\r\n \r\n @data.setter\r\n def data(self, value):\r\n self._data = value\r\n\r\ne = Example()\r\ne.data = 42 # Thread-safe property set; emits 'updated' signal on change\r\n```\r\n\r\n### Worker Threads\r\nFor background work, PynneX provides a `@nx_with_worker` decorator that:\r\n\r\n- Spawns a dedicated event loop in a worker thread.\r\n- Allows you to queue async tasks to this worker.\r\n- Enables easy start/stop lifecycle management.\r\n- Integrates with signals and slots for thread-safe updates to the main \r\n\r\n**Worker Example**\r\n```python\r\nfrom pynnex import nx_with_worker, nx_signal\r\n\r\n@with_worker\r\nclass DataProcessor:\r\n @signal\r\n def processing_done(self):\r\n \"\"\"Emitted when processing completes\"\"\"\r\n\r\n async def run(self, *args, **kwargs):\r\n # The main entry point for the worker thread\u2019s event loop\r\n # Wait for tasks or stopping signal\r\n await self.wait_for_stop()\r\n\r\n async def process_data(self, data):\r\n # Perform heavy computation in the worker thread\r\n result = await heavy_computation(data)\r\n self.processing_done.emit(result)\r\n\r\nprocessor = DataProcessor()\r\nprocessor.start()\r\n\r\n# Queue a task to run in the worker thread:\r\nprocessor.queue_task(processor.process_data(some_data))\r\n\r\n# Stop the worker gracefully\r\nprocessor.stop()\r\n```\r\n\r\n## Documentation and Example\r\n- [Usage Guide](https://github.com/nexconnectio/pynnex/blob/main/docs/usage.md): Learn how to define signals/slots, manage threads, and structure your event-driven code.\r\n- [API Reference](https://github.com/nexconnectio/pynnex/blob/main/docs/api.md): Detailed documentation of classes, decorators, and functions.\r\n- [Examples](https://github.com/nexconnectio/pynnex/blob/main/docs/examples.md): Practical use cases, including UI integration, async operations, and worker pattern usage.\r\n- [Logging Guidelines](https://github.com/nexconnectio/pynnex/blob/main/docs/logging.md): Configure logging levels and handlers for debugging.\r\n- [Testing Guide](https://github.com/nexconnectio/pynnex/blob/main/docs/testing.md): earn how to run tests and contribute safely.\r\n\r\n## Logging\r\nConfigure logging to diagnose issues:\r\n\r\n```python\r\nimport logging\r\nlogging.getLogger('pynnex').setLevel(logging.DEBUG)\r\n```\r\n\r\nFor more details, see the [Logging Guidelines](https://github.com/nexconnectio/pynnex/blob/main/docs/logging.md).\r\n\r\n## Testing\r\n\r\nPynnex uses `pytest` for testing:\r\n\r\n```bash\r\n# Run all tests\r\npytest\r\n\r\n# Run with verbose output\r\npytest -v\r\n\r\n# Run specific test file\r\npytest tests/unit/test_signal.py\r\n```\r\n\r\nSee the [Testing Guide](https://github.com/nexconnectio/pynnex/blob/main/docs/testing.md) for more details.\r\n\r\n## Contributing\r\nWe welcome contributions! Please read our [Contributing Guidelines](https://github.com/nexconnectio/pynnex/blob/main/CONTRIBUTING.md) before submitting PRs.\r\n\r\n## Sponsorship & Donations\r\nIf PynneX has helped simplify your async/multithreaded workflows, please consider [sponsoring us](https://github.com/nexconnectio/pynnex/blob/main/.github/FUNDING.yml). All funds go toward infrastructure, documentation, and future development.\r\n\r\nPlease note that financial contributions support only the project's maintenance and do not grant financial rewards to individual contributors.\r\n\r\n## License\r\n`PynneX` is licensed under the MIT License. See [LICENSE](https://github.com/nexconnectio/pynnex/blob/main/LICENSE) for details.\r\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A Python Signal-Slot library",
"version": "1.0.1",
"project_urls": {
"Documentation": "https://github.com/nexconnectio/pynnex#readme",
"Homepage": "https://github.com/nexconnectio/pynnex",
"Issues": "https://github.com/nexconnectio/pynnex/issues",
"Repository": "https://github.com/nexconnectio/pynnex"
},
"split_keywords": [
"signal-slot",
" decorator",
" multithreading",
" asyncio"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "8c07dd56e0d7590612e998f00b6c1c55d1eaad932b6fd46ea487a8b9dad8e969",
"md5": "1fd538b78a1be1e6a14b218df10f8bfd",
"sha256": "28aa34b9b433e69b932c3985745b377064e6cb7545bb9991d0123e8eecf5f9cf"
},
"downloads": -1,
"filename": "pynnex-1.0.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "1fd538b78a1be1e6a14b218df10f8bfd",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 20857,
"upload_time": "2025-01-02T06:09:08",
"upload_time_iso_8601": "2025-01-02T06:09:08.630155Z",
"url": "https://files.pythonhosted.org/packages/8c/07/dd56e0d7590612e998f00b6c1c55d1eaad932b6fd46ea487a8b9dad8e969/pynnex-1.0.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "8b853a6c241935aca58a7fbe7e0f99596a202142c3b8a232a87ec0910e35a960",
"md5": "1b980d509ebcd6762b06f9031d052002",
"sha256": "57177c96b449aa2c2dbe7706161505b00187c2750028fb4ceb05e9a552538065"
},
"downloads": -1,
"filename": "pynnex-1.0.1.tar.gz",
"has_sig": false,
"md5_digest": "1b980d509ebcd6762b06f9031d052002",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 23299,
"upload_time": "2025-01-02T06:09:11",
"upload_time_iso_8601": "2025-01-02T06:09:11.335683Z",
"url": "https://files.pythonhosted.org/packages/8b/85/3a6c241935aca58a7fbe7e0f99596a202142c3b8a232a87ec0910e35a960/pynnex-1.0.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-01-02 06:09:11",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "nexconnectio",
"github_project": "pynnex#readme",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "pynnex"
}
JSON
