Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -264,6 +264,22 @@ def includeme(config):

The user ID will now be accessible from `request.current_task.request.user_id`.

## Extending Tasks: Environment

You can pass in `environ` as a dictionary to the headers, which will populate `environ` in the task request object.
If desired, these can be added to every call via the `BeforeDeferTask` argument.
This can be particularly useful for integration testing,
where information about the test environment may need to be passed to the request.

```python
def mytask(request):
print(request.environ["foo"])

def invoke(request):
environ = {"foo": "bar"}
request.defer_task_with_options(mytask, headers={'environ': environ})
```

## Fork Safety

Celery by default uses a pre-fork worker model,
Expand Down
24 changes: 20 additions & 4 deletions pyramid_tasks/__init__.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,12 @@
import celery
import venusian
from pyramid.interfaces import PHASE1_CONFIG, PHASE2_CONFIG, PHASE3_CONFIG
from pyramid.interfaces import (
PHASE1_CONFIG,
PHASE2_CONFIG,
PHASE3_CONFIG,
IRequestFactory,
)
from pyramid.request import Request
from pyramid.scripting import prepare

from .events import BeforeDeferTask, CeleryWorkerProcessInit
Expand Down Expand Up @@ -76,6 +82,7 @@ def register():
handler,
name=name,
shared=False,
bind=True,
**kwargs,
)
config.registry["pyramid_tasks.task_map"][func] = task
Expand All @@ -91,13 +98,22 @@ def _make_task_handler(registry, func):

"""

def handler(*args, **kwargs):
with prepare(registry=registry) as env:
return func(env["request"], *args, **kwargs)
def handler(self, *args, **kwargs):
headers = self.request.headers
environ = headers.get("environ") if headers else None
request = _make_request(registry, environ)
with prepare(request=request, registry=registry):
return func(request, *args, **kwargs)

return handler


def _make_request(registry, environ):
request_factory = registry.queryUtility(IRequestFactory, default=Request)
request = request_factory.blank("/", environ=environ)
return request


def task(**kwargs):
"""
Decorator to register a new task.
Expand Down
30 changes: 22 additions & 8 deletions tests/test_integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,8 +64,8 @@ def test_task(request):
with make_request_with_worker(test_config) as request:
result = request.defer_task_with_options(
"test", headers={"foo": "bar"}
)
result.get() == "bar"
).get()
assert result == "bar"


def test_legacy_integration(test_config):
Expand All @@ -79,8 +79,8 @@ def add_task(request, x, y):

test_config.register_task(add_task, name="add")
with make_request_with_worker(test_config) as request:
result = request.delay_task("add", 2, 3)
assert result.get() == 5
result = request.delay_task("add", 2, 3).get()
assert result == 5


def test_task_by_function_integration(test_config):
Expand All @@ -89,15 +89,15 @@ def add_task(request, x, y):

test_config.register_task(add_task, name="add")
with make_request_with_worker(test_config) as request:
result = request.defer_task(add_task, 2, 3)
assert result.get() == 5
result = request.defer_task(add_task, 2, 3).get()
assert result == 5


def test_task_venusian_integration(test_config):
test_config.include("tests.pkgs.venusianapp")
with make_request_with_worker(test_config) as request:
result = request.defer_task("venusian_add_task", 2, 3)
assert result.get() == 5
result = request.defer_task("venusian_add_task", 2, 3).get()
assert result == 5


def test_get_task_result_integration(test_config):
Expand Down Expand Up @@ -132,6 +132,20 @@ def test_task_deriver_integration(test_config):
assert request.defer_task("increment", 4).get() == 7


def test_task_environ_integration(test_config):
def add_task(request):
return request.environ["foo"]

test_config.register_task(add_task, name="add")
environ = {"foo": "bar"}
with make_request_with_worker(test_config) as request:
result = request.defer_task_with_options(
"add",
headers={"environ": environ},
).get()
assert result == "bar"


def test_transaction_integration(test_config):
test_config.include("tests.pkgs.transactionapp")

Expand Down