# Taskiq + AioHTTP
This project is used to create mocked application and request
that you can use as a dependencies in your taskiq application.
It's useful because it runs all startup events of your application
and everything that you might expect in your application's state is
available inside of your tasks.
We suggest to use this library along with [taskiq-python/aiohttp-deps](https://github.com/taskiq-python/aiohttp-deps), because it might be super handy to reuse same dependencies of your application in your tasks.
To add an integration, you need to call the function `init` in your broker's module.
```python
import taskiq_aiohttp
broker = ...
taskiq_aiohttp.init(broker, "project.module:app")
```
## How does it work?
It adds startup functions to the broker, so it imports your aiohttp application and creates a single worker-wide Request and Application objects that you can depend on.
THIS REQUEST IS NOT RELATED TO THE ACTUAL REQUESTS IN AioHTTP! This request won't have actual data about the request you were handling while sending a task.
## Manual context updates
Sometimes it's required to update context manually. For example, for tests.
If you need to add context in your broker by hand, please use function populate_context.
Imagine, you want to use InMemoryBroker for testing and your broker file looks like this:
```python
broker = MyBroker()
if env == "pytest":
broker = InMemoryBroker()
```
In this case your context won't be updated, because inmemory brokers cannot run as workers.
To solve this issue, we have a populate context function. It's a bit complex and takes lots of
parmeters. But here's a fixture that creates aiohttp test client and populates context of inmemory broker.
```python
import asyncio
from typing import AsyncGenerator
import pytest
from aiohttp import web
from aiohttp.test_utils import BaseTestServer, TestClient, TestServer
from taskiq_aiohttp import populate_context
@pytest.fixture
async def test_client(
app: web.Application,
) -> AsyncGenerator[TestClient, None]:
"""
Create a test client.
This function creates a TestServer
and a test client for the application.
Also this fixture populates context
with needed variables.
:param app: current application.
:yield: ready to use client.
"""
loop = asyncio.get_running_loop()
server = TestServer(app)
client = TestClient(server, loop=loop)
await client.start_server()
# This is important part.
# Since InMemoryBroker doesn't
# run in worker_process, we have to populate
# broker's context manually.
populate_context(
broker=broker,
server=server.runner.server,
app=app,
loop=None,
)
yield client
await client.close()
```
Raw data
{
"_id": null,
"home_page": "",
"name": "taskiq-aiohttp",
"maintainer": "Taskiq team",
"docs_url": null,
"requires_python": ">=3.8.1,<4.0.0",
"maintainer_email": "taskiq@no-reply.com",
"keywords": "taskiq,tasks,distributed,async",
"author": "Taskiq team",
"author_email": "taskiq@no-reply.com",
"download_url": "https://files.pythonhosted.org/packages/a3/bd/5390c05baca78dc5529e712b3c4df5f3386050e1598976f24763ecd6defd/taskiq_aiohttp-0.3.1.tar.gz",
"platform": null,
"description": "# Taskiq + AioHTTP\n\nThis project is used to create mocked application and request\nthat you can use as a dependencies in your taskiq application.\n\n\nIt's useful because it runs all startup events of your application\nand everything that you might expect in your application's state is\navailable inside of your tasks.\n\nWe suggest to use this library along with [taskiq-python/aiohttp-deps](https://github.com/taskiq-python/aiohttp-deps), because it might be super handy to reuse same dependencies of your application in your tasks.\n\nTo add an integration, you need to call the function `init` in your broker's module.\n\n```python\nimport taskiq_aiohttp\n\nbroker = ...\n\ntaskiq_aiohttp.init(broker, \"project.module:app\")\n\n```\n\n## How does it work?\n\nIt adds startup functions to the broker, so it imports your aiohttp application and creates a single worker-wide Request and Application objects that you can depend on.\n\nTHIS REQUEST IS NOT RELATED TO THE ACTUAL REQUESTS IN AioHTTP! This request won't have actual data about the request you were handling while sending a task.\n\n\n## Manual context updates\n\nSometimes it's required to update context manually. For example, for tests.\nIf you need to add context in your broker by hand, please use function populate_context.\n\nImagine, you want to use InMemoryBroker for testing and your broker file looks like this:\n\n```python\nbroker = MyBroker()\n\nif env == \"pytest\":\n broker = InMemoryBroker()\n```\n\nIn this case your context won't be updated, because inmemory brokers cannot run as workers.\nTo solve this issue, we have a populate context function. It's a bit complex and takes lots of\nparmeters. But here's a fixture that creates aiohttp test client and populates context of inmemory broker.\n\n```python\nimport asyncio\nfrom typing import AsyncGenerator\n\nimport pytest\nfrom aiohttp import web\nfrom aiohttp.test_utils import BaseTestServer, TestClient, TestServer\nfrom taskiq_aiohttp import populate_context\n\n\n@pytest.fixture\nasync def test_client(\n app: web.Application,\n) -> AsyncGenerator[TestClient, None]:\n \"\"\"\n Create a test client.\n\n This function creates a TestServer\n and a test client for the application.\n\n Also this fixture populates context\n with needed variables.\n\n :param app: current application.\n :yield: ready to use client.\n \"\"\"\n loop = asyncio.get_running_loop()\n server = TestServer(app)\n client = TestClient(server, loop=loop)\n\n await client.start_server()\n\n # This is important part.\n # Since InMemoryBroker doesn't\n # run in worker_process, we have to populate\n # broker's context manually.\n populate_context(\n broker=broker,\n server=server.runner.server,\n app=app,\n loop=None,\n )\n\n yield client\n\n await client.close()\n\n```\n",
"bugtrack_url": null,
"license": "LICENSE",
"summary": "Taskiq integration with AioHTTP framework",
"version": "0.3.1",
"project_urls": null,
"split_keywords": [
"taskiq",
"tasks",
"distributed",
"async"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "9c591fc715741f664ee6c28f5ba913c44d76f4bcba7445c55e030513ea023b4a",
"md5": "07520e12b1925496a368e88353874117",
"sha256": "3e282575bc6418c4a47ab2abae5430d410b1935e4e0ff8ba1fded10083eeb80d"
},
"downloads": -1,
"filename": "taskiq_aiohttp-0.3.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "07520e12b1925496a368e88353874117",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8.1,<4.0.0",
"size": 5281,
"upload_time": "2023-11-30T11:38:42",
"upload_time_iso_8601": "2023-11-30T11:38:42.749699Z",
"url": "https://files.pythonhosted.org/packages/9c/59/1fc715741f664ee6c28f5ba913c44d76f4bcba7445c55e030513ea023b4a/taskiq_aiohttp-0.3.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "a3bd5390c05baca78dc5529e712b3c4df5f3386050e1598976f24763ecd6defd",
"md5": "d6ca8e6d0a83b948f1c90901027ccff9",
"sha256": "df62b137bc71fa7c6c470fca7e86b9afddbbadcc0bce7e16e2b9dc25ec45be9f"
},
"downloads": -1,
"filename": "taskiq_aiohttp-0.3.1.tar.gz",
"has_sig": false,
"md5_digest": "d6ca8e6d0a83b948f1c90901027ccff9",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8.1,<4.0.0",
"size": 4724,
"upload_time": "2023-11-30T11:38:43",
"upload_time_iso_8601": "2023-11-30T11:38:43.972878Z",
"url": "https://files.pythonhosted.org/packages/a3/bd/5390c05baca78dc5529e712b3c4df5f3386050e1598976f24763ecd6defd/taskiq_aiohttp-0.3.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-11-30 11:38:43",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "taskiq-aiohttp"
}
JSON