From 4cf3e2550ac136137c4f725719dd6afd89ec4de4 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Sun, 1 Mar 2026 18:11:30 +0400 Subject: [PATCH 1/5] Update README --- README.md | 64 ++++++++++++++++++++++++++++++++++++++ python/fluxqueue/models.py | 2 +- 2 files changed, 65 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index fb2fd3c..b6f833a 100644 --- a/README.md +++ b/README.md @@ -29,6 +29,7 @@ FluxQueue is a task queue for Python that gets out of your way. The Rust core ma - **Multiple Queues**: Organize tasks across different queues - **Simple API**: Decorator-based interface that feels natural in Python - **Type Safe**: Full type hints support +- **Context Classes**: Access task metadata and manage thread-persistent resources with the Context class ## Requirements @@ -91,6 +92,69 @@ async def send_email(to_email: str, subject: str, body: str): Running the async function in an async context will also enqueue the task. +### Tasks with Context + +FluxQueue provides a `Context` class that gives you access to task metadata and allows you to manage thread-persistent resources. Use `task_with_context()` decorator to enable this feature: + +```python +from fluxqueue import FluxQueue, Context + +fluxqueue = FluxQueue() + +@fluxqueue.task_with_context() +def process_data(ctx: Context, data: str): + # Access task metadata + print(f"Task ID: {ctx.metadata.task_id}") + print(f"Retry count: {ctx.metadata.retries}") + + # Process the data + process(data) +``` + +You can also subclass `Context` to create custom contexts with domain-specific resources: + +```python +from contextlib import asynccontextmanager +from fluxqueue import FluxQueue, Context +from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession + +class DbContext(Context): + def __init__(self) -> None: + super().__init__() + + def _get_local_session(self): + if "session" not in self.thread_storage: + engine = create_async_engine(DATABASE_URL) + self.thread_storage["session"] = async_sessionmaker( + bind=engine, expire_on_commit=False + ) + return self.thread_storage["session"] + + @asynccontextmanager + async def session_context(self): + local_session = self._get_local_session() + async with local_session() as session: + try: + yield session + await session.commit() + except Exception: + await session.rollback() + raise + +@fluxqueue.task_with_context() +async def create_user(ctx: DbContext, email: str, username: str): + async with ctx.session_context() as db_session: + user = User(email=email, username=username) + db_session.add(user) +``` + +The context parameter is automatically injected by the worker and is not part of the function's public signature when enqueueing: + +```python +# Just call with your regular arguments +create_user("user@example.com", "johndoe") +``` + ## Installing the worker In order the tasks to be executed you need to run a FluxQueue worker. You need to install the worker on your system, recommended way of doing that is using `fluxqueue-cli` which comes with the `[cli]` extra: diff --git a/python/fluxqueue/models.py b/python/fluxqueue/models.py index 78f7d55..c3c74f6 100644 --- a/python/fluxqueue/models.py +++ b/python/fluxqueue/models.py @@ -10,7 +10,7 @@ class TaskMetadata: task_id: str """Unique identifier for the current task execution.""" - retry_count: int + retries: int """Number of times this task has been retried.""" max_retries: int """Maximum number of retry attempts allowed before failure.""" From ce3a8169a88e067e02aeb3be74c0e5f7c9848c0d Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Sun, 1 Mar 2026 20:24:10 +0400 Subject: [PATCH 2/5] Better examples --- README.md | 20 ++++---------------- 1 file changed, 4 insertions(+), 16 deletions(-) diff --git a/README.md b/README.md index b6f833a..d43fda9 100644 --- a/README.md +++ b/README.md @@ -52,15 +52,9 @@ from fluxqueue import FluxQueue fluxqueue = FluxQueue() @fluxqueue.task() -def send_email(to_email: str, subject: str, body: str): +def send_email(to_email: str): with email_context() as email_client: - message = EmailMessage() - message["From"] = "test@example.com" - message["To"] = to_email - message["Subject"] = subject - message.set_content(body) - - email_client.send_message(message) + send_email(to_email, email_client) ``` ### Enqueue Tasks @@ -79,15 +73,9 @@ FluxQueue supports async functions too. Just define an async function and use th ```python @fluxqueue.task() -async def send_email(to_email: str, subject: str, body: str): +async def send_email_task(to_email: str): async with email_context() as email_client: - message = EmailMessage() - message["From"] = "test@example.com" - message["To"] = to_email - message["Subject"] = subject - message.set_content(body) - - await email_client.send_message(message) + await send_email(to_email, email_client) ``` Running the async function in an async context will also enqueue the task. From 50aa2bc0ddb859edf8b4870a1b1eb58ee2ae6938 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Sun, 1 Mar 2026 20:29:17 +0400 Subject: [PATCH 3/5] Update overview --- README.md | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index d43fda9..8706866 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,7 @@ ## Overview -FluxQueue is a task queue for Python that gets out of your way. The Rust core makes the process fast with less overhead, least dependencies, and most importantly, less memory usage. Tasks are managed through Redis. +FluxQueue is a task queue for Python that gets out of your way. Built on a multi-threaded Tokio runtime, FluxQueue delivers high throughput while maintaining low memory usage. The Rust core ensures minimal overhead and dependencies, making it an efficient solution for background task processing. Tasks are managed through Redis. ## Key Features @@ -90,16 +90,15 @@ from fluxqueue import FluxQueue, Context fluxqueue = FluxQueue() @fluxqueue.task_with_context() -def process_data(ctx: Context, data: str): +def process_data_task(ctx: Context, data: str): # Access task metadata print(f"Task ID: {ctx.metadata.task_id}") print(f"Retry count: {ctx.metadata.retries}") - # Process the data - process(data) + process_data(data) ``` -You can also subclass `Context` to create custom contexts with domain-specific resources: +You can also subclass `Context` to create custom contexts with domain-specific resources. This example shows how to create a `DbContext` that manages database connections efficiently by reusing them across tasks in the same worker thread: ```python from contextlib import asynccontextmanager @@ -130,7 +129,7 @@ class DbContext(Context): raise @fluxqueue.task_with_context() -async def create_user(ctx: DbContext, email: str, username: str): +async def create_user_task(ctx: DbContext, email: str, username: str): async with ctx.session_context() as db_session: user = User(email=email, username=username) db_session.add(user) @@ -140,7 +139,7 @@ The context parameter is automatically injected by the worker and is not part of ```python # Just call with your regular arguments -create_user("user@example.com", "johndoe") +create_user_task("user@example.com", "johndoe") ``` ## Installing the worker From b4d5586d003dc62c177e7801861b270c394290e4 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Sun, 1 Mar 2026 21:02:59 +0400 Subject: [PATCH 4/5] Add docs and release notes --- docs/features/context.md | 154 +++++++++++++++++++++++++++++++++++++++ docs/features/index.md | 7 ++ docs/index.md | 3 +- docs/release-notes.md | 78 ++++++++++++++++++++ mkdocs.yml | 4 + 5 files changed, 245 insertions(+), 1 deletion(-) create mode 100644 docs/features/context.md create mode 100644 docs/features/index.md create mode 100644 docs/release-notes.md diff --git a/docs/features/context.md b/docs/features/context.md new file mode 100644 index 0000000..7600407 --- /dev/null +++ b/docs/features/context.md @@ -0,0 +1,154 @@ +# Context Classes + +The Context class provides access to task execution metadata and enables efficient resource management through thread-persistent storage. Use it when you need to share resources across multiple task executions in the same worker thread. + +## Overview + +The `Context` class provides a dual-layer storage system: + +- **Worker Layer (`thread_storage`)**: A dictionary that persists across all tasks executed by the same worker thread. Use this for heavy resource pooling (e.g., database engines, HTTP clients) to avoid re-initialization overhead. +- **Task Layer (`metadata`)**: Isolated metadata for each task execution via ContextVars. Provides read-only access to the current task's unique identity and execution state. + +## Basic Usage + +To use the Context feature, decorate your task function with `@fluxqueue.task_with_context()` instead of `@fluxqueue.task()`. The function must accept a context parameter as its first argument, typed as `Context` or a subclass of `Context`: + +```python +from fluxqueue import FluxQueue, Context + +fluxqueue = FluxQueue() + +@fluxqueue.task_with_context() +def process_data_task(ctx: Context, data: str): + # Access task metadata + print(f"Task ID: {ctx.metadata.task_id}") + print(f"Retry count: {ctx.metadata.retries}") + print(f"Max retries: {ctx.metadata.max_retries}") + print(f"Enqueued at: {ctx.metadata.enqueued_at}") + + # Process the data + process_data(data) +``` + +When you enqueue the task, the context parameter is automatically injected by the worker and is not part of the function's public signature: + +```python +# Just call with your regular arguments - no context needed! +process_data_task("some data") +``` + +## Task Metadata + +The `metadata` property provides read-only access to task execution information: + +- `task_id`: Unique identifier for the current task execution +- `retries`: Number of times this task has been retried +- `max_retries`: Maximum number of retry attempts allowed +- `enqueued_at`: ISO 8601 timestamp of when the task was originally enqueued + +This metadata is isolated per task using Python's `ContextVar`, ensuring data integrity even when multiple tasks execute concurrently on the same thread. + +## Custom Context Classes + +You can subclass `Context` to create domain-specific contexts that provide convenient access to resources. This is especially useful for database connections, HTTP clients, or other resources that benefit from pooling: + +```python +from contextlib import asynccontextmanager +from fluxqueue import FluxQueue, Context +from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession + +class DbContext(Context): + """ + Custom context for managing database connections. + + This context reuses database engines across tasks in the same worker thread, + significantly reducing connection overhead. + """ + def __init__(self) -> None: + super().__init__() + + def _get_local_session(self): + """ + Get or create a session maker for this worker thread. + """ + session = self.thread_storage.get("session") + if session is None: + engine = create_async_engine(DATABASE_URL) + session = async_sessionmaker( + bind=engine, expire_on_commit=False + ) + self.thread_storage["session"] = session + return session + + @asynccontextmanager + async def session_context(self): + """ + Context manager for database sessions with automatic commit/rollback. + """ + local_session = self._get_local_session() + async with local_session() as session: + try: + yield session + await session.commit() + except Exception: + await session.rollback() + raise + +# Use the custom context +@fluxqueue.task_with_context() +async def create_user_task(ctx: DbContext, email: str, username: str): + async with ctx.session_context() as db_session: + user = User(email=email, username=username) + db_session.add(user) + # Session commits automatically on success, rolls back on exception +``` + +## Benefits + +Using Context classes provides several key benefits: + +1. **Resource Efficiency**: Share expensive resources (like database engines) across multiple tasks in the same worker thread, reducing initialization overhead. + +2. **Task Isolation**: Each task gets its own isolated metadata via ContextVars, ensuring data integrity even with concurrent execution. + +3. **Type Safety**: Full type hint support means your IDE and type checkers understand the context structure. + +4. **Clean API**: The context parameter is automatically injected, so your task functions have a clean public interface without exposing implementation details. + +5. **Flexibility**: Subclass `Context` to create domain-specific contexts tailored to your application's needs. + +!!! tip "Avoiding Event Loop Issues in Multi-threaded Environments" + + FluxQueue's multi-threaded Tokio runtime can cause issues with async database libraries like `asyncpg` that throw errors such as "got future pending attached to a different loop" when resources are shared across threads. + + By using the Context class with `thread_storage`, each executor in the worker gets its own database engine instance. Combined with Python's context managers, you can safely acquire database sessions and run queries without event loop conflicts. This ensures that each worker thread maintains its own isolated database connection pool, preventing cross-thread resource sharing issues. + +## Async and Sync Support + +Context classes work seamlessly with both synchronous and asynchronous tasks: + +```python +# Synchronous task with context +@fluxqueue.task_with_context() +def sync_task(ctx: Context, data: str): + print(f"Processing {data} in task {ctx.metadata.task_id}") + process(data) + +# Async task with context +@fluxqueue.task_with_context() +async def async_task(ctx: Context, data: str): + print(f"Processing {data} in task {ctx.metadata.task_id}") + await process_async(data) +``` + +## Best Practices + +1. **Use thread_storage for expensive resources**: Database engines, HTTP clients, and connection pools are perfect candidates for thread storage. + +2. **Keep metadata read-only**: The metadata property is read-only for a reason - don't try to modify it. + +3. **Subclass for domain logic**: Create custom context classes when you have domain-specific resources or logic that multiple tasks share. + +4. **One context parameter per task**: Each task function should have exactly one context parameter, typed as `Context` or a subclass. + +5. **Initialize resources lazily**: Check if resources exist in `thread_storage` before creating them to avoid unnecessary initialization. diff --git a/docs/features/index.md b/docs/features/index.md new file mode 100644 index 0000000..16043dc --- /dev/null +++ b/docs/features/index.md @@ -0,0 +1,7 @@ +# Features + +Learn about FluxQueue's features and how to use them. + +## Feature Documentation + +- **[Context Classes](context.md)** - Access task metadata and manage thread-persistent resources diff --git a/docs/index.md b/docs/index.md index 1d45408..f5ff3bd 100644 --- a/docs/index.md +++ b/docs/index.md @@ -19,7 +19,7 @@ Welcome to FluxQueue documentation. FluxQueue is a lightweight, resource-efficie ## What is FluxQueue? -FluxQueue is a task queue for Python that gets out of your way. The Rust core makes the process fast with less overhead, least dependencies, and most importantly, less memory usage. Tasks are managed through Redis. +FluxQueue is a task queue for Python that gets out of your way. Built on a multi-threaded Tokio runtime, FluxQueue delivers high throughput while maintaining low memory usage. The Rust core ensures minimal overhead and dependencies, making it an efficient solution for background task processing. Tasks are managed through Redis. ## Key Features @@ -31,6 +31,7 @@ FluxQueue is a task queue for Python that gets out of your way. The Rust core ma - **Multiple Queues**: Organize tasks across different queues - **Simple API**: Decorator-based interface that feels natural in Python - **Type Safe**: Full type hints support +- **Context Classes**: Access task metadata and manage thread-persistent resources with the Context class ## Requirements diff --git a/docs/release-notes.md b/docs/release-notes.md new file mode 100644 index 0000000..640da5e --- /dev/null +++ b/docs/release-notes.md @@ -0,0 +1,78 @@ +# Release Notes + +All notable changes to FluxQueue will be documented in this file. + +The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), +and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). + +## [Unreleased] + +### Added + +- **Context Classes**: New `Context` class for accessing task metadata and managing thread-persistent resources ([#114](https://github.com/CCXLV/fluxqueue/pull/114)) + - `@fluxqueue.task_with_context()` decorator for tasks that need context access + - `Context.metadata` property for accessing task execution information (task ID, retry count, etc.) + - `Context.thread_storage` property for sharing resources across tasks in the same worker thread + - Support for custom context classes by subclassing `Context` + - Prevents event loop issues in multi-threaded environments (e.g., asyncpg "got future pending attached to a different loop" errors) + - Works with both synchronous and asynchronous tasks +- Client library version check before worker initialization ([#128](https://github.com/CCXLV/fluxqueue/pull/128)) +- Tests to cover the context feature ([#130](https://github.com/CCXLV/fluxqueue/pull/130)) + +### Changed + +- Moved function calls to dispatcher ([#116](https://github.com/CCXLV/fluxqueue/pull/116)) +- Improved executor startup speed ([#117](https://github.com/CCXLV/fluxqueue/pull/117)) +- Made startup logs more clear ([#118](https://github.com/CCXLV/fluxqueue/pull/118)) +- Improved examples documentation ([#119](https://github.com/CCXLV/fluxqueue/pull/119)) +- Fixed worker installation instructions in README ([#120](https://github.com/CCXLV/fluxqueue/pull/120)) +- Bumped pyo3 dependency versions ([#121](https://github.com/CCXLV/fluxqueue/pull/121)) +- Improved client docstrings ([#122](https://github.com/CCXLV/fluxqueue/pull/122)) +- Improved `_run_async_task` docstring for clarity ([#126](https://github.com/CCXLV/fluxqueue/pull/126)) +- Various fixes and updates ([#124](https://github.com/CCXLV/fluxqueue/pull/124)) + +### Fixed + +- Fixed tasks with the base Context class ([#123](https://github.com/CCXLV/fluxqueue/pull/123)) +- Fixed client library version check for older workers ([#129](https://github.com/CCXLV/fluxqueue/pull/129)) +- Reverted disallowing `_Context` as context name ([#127](https://github.com/CCXLV/fluxqueue/pull/127)) + +## [0.2.1] - 2026-02-21 + +### Changed + +- Moved examples to docs ([#110](https://github.com/CCXLV/fluxqueue/pull/110)) +- Moved core to crates directory ([#111](https://github.com/CCXLV/fluxqueue/pull/111)) +- Renamed Rust core module ([#112](https://github.com/CCXLV/fluxqueue/pull/112)) +- Improved task decorator type inference ([#113](https://github.com/CCXLV/fluxqueue/pull/113)) + +## [0.2.0] - 2026-02-17 + +### Added + +- Logs after task finishes ([#80](https://github.com/CCXLV/fluxqueue/pull/80)) +- Python version badges in README and classifiers/keywords in pyproject.toml ([#84](https://github.com/CCXLV/fluxqueue/pull/84)) +- Unit and integration tests ([#90](https://github.com/CCXLV/fluxqueue/pull/90)) +- Support for Windows and macOS ([#94](https://github.com/CCXLV/fluxqueue/pull/94)) +- Tests status badge in README ([#92](https://github.com/CCXLV/fluxqueue/pull/92)) + +### Changed + +- Updated way of running async task functions ([#83](https://github.com/CCXLV/fluxqueue/pull/83)) +- Improved README clarity and documentation ([#85](https://github.com/CCXLV/fluxqueue/pull/85), [#86](https://github.com/CCXLV/fluxqueue/pull/86), [#87](https://github.com/CCXLV/fluxqueue/pull/87)) +- Updated get_functions.py script to raise exception on duplication ([#78](https://github.com/CCXLV/fluxqueue/pull/78)) +- Bumped worker version to 0.2.0-beta.4 ([#95](https://github.com/CCXLV/fluxqueue/pull/95)) +- Updated pyproject description ([#108](https://github.com/CCXLV/fluxqueue/pull/108)) + +### Fixed + +- Fixed executors heartbeat flaw on startup ([#82](https://github.com/CCXLV/fluxqueue/pull/82)) +- Fixed test_path_to_module_path to work on Windows ([#93](https://github.com/CCXLV/fluxqueue/pull/93)) +- Fixed various workflow files (release-worker, build-worker, publish workflows) ([#96](https://github.com/CCXLV/fluxqueue/pull/96), [#97](https://github.com/CCXLV/fluxqueue/pull/97), [#98](https://github.com/CCXLV/fluxqueue/pull/98), [#99](https://github.com/CCXLV/fluxqueue/pull/99), [#100](https://github.com/CCXLV/fluxqueue/pull/100), [#101](https://github.com/CCXLV/fluxqueue/pull/101), [#102](https://github.com/CCXLV/fluxqueue/pull/102), [#105](https://github.com/CCXLV/fluxqueue/pull/105), [#106](https://github.com/CCXLV/fluxqueue/pull/106)) +- Fixed license name in pyproject.toml ([#107](https://github.com/CCXLV/fluxqueue/pull/107)) +- Fixed pyproject classifiers ([#109](https://github.com/CCXLV/fluxqueue/pull/109)) + +[Unreleased]: https://github.com/CCXLV/fluxqueue/compare/v0.3.0rc1...HEAD +[0.3.0-rc1]: https://github.com/CCXLV/fluxqueue/compare/v0.2.1...v0.3.0rc1 +[0.2.1]: https://github.com/CCXLV/fluxqueue/compare/v0.2.0...v0.2.1 +[0.2.0]: https://github.com/CCXLV/fluxqueue/compare/v0.2.0-beta.4...v0.2.0 diff --git a/mkdocs.yml b/mkdocs.yml index fabb4c6..575d057 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -60,6 +60,9 @@ plugins: nav: - FluxQueue: index.md + - Features: + - features/index.md + - features/context.md - Learn: - tutorial/index.md - tutorial/installation.md @@ -80,6 +83,7 @@ nav: - api/fluxqueue.md - api/environment.md - About: about.md + - Release Notes: release-notes.md markdown_extensions: - pymdownx.highlight: From d4eaf07829685af077dca7b396fb449d77a10eb1 Mon Sep 17 00:00:00 2001 From: Giorgi Merebashvili Date: Sun, 1 Mar 2026 21:05:05 +0400 Subject: [PATCH 5/5] Update versions --- Cargo.lock | 4 ++-- crates/fluxqueue-core/Cargo.toml | 2 +- crates/fluxqueue-worker/Cargo.toml | 2 +- docs/release-notes.md | 2 +- 4 files changed, 5 insertions(+), 5 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index cf90b63..76a3222 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -272,7 +272,7 @@ dependencies = [ [[package]] name = "fluxqueue-worker" -version = "0.3.0-rc1" +version = "0.3.0" dependencies = [ "anyhow", "chrono", @@ -297,7 +297,7 @@ dependencies = [ [[package]] name = "fluxqueue_core" -version = "0.3.0-rc1" +version = "0.3.0" dependencies = [ "deadpool-redis", "fluxqueue-common", diff --git a/crates/fluxqueue-core/Cargo.toml b/crates/fluxqueue-core/Cargo.toml index 53ab80a..92c11b8 100644 --- a/crates/fluxqueue-core/Cargo.toml +++ b/crates/fluxqueue-core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "fluxqueue_core" -version = "0.3.0-rc1" +version = "0.3.0" repository.workspace = true homepage.workspace = true authors.workspace = true diff --git a/crates/fluxqueue-worker/Cargo.toml b/crates/fluxqueue-worker/Cargo.toml index 662448d..c41977b 100644 --- a/crates/fluxqueue-worker/Cargo.toml +++ b/crates/fluxqueue-worker/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "fluxqueue-worker" -version = "0.3.0-rc1" +version = "0.3.0" edition = "2024" [lib] diff --git a/docs/release-notes.md b/docs/release-notes.md index 640da5e..3704705 100644 --- a/docs/release-notes.md +++ b/docs/release-notes.md @@ -5,7 +5,7 @@ All notable changes to FluxQueue will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [0.3.0] - 2026-03-01 ### Added