Added README.md.

Signed-off-by: Pavel Kirilin <win10@list.ru>

Author: s3rius

README.md

@@ -0,0 +1,264 @@
+# Taskiq
+
+Taskiq is an asynchronous distributed task queue.
+This project takes inspiration from big projects such as [Celery](https://docs.celeryq.dev) and [Dramatiq](https://dramatiq.io/).
+But taskiq can send and run both the sync and async functions.
+Also, we use [PEP-612](https://peps.python.org/pep-0612/) to provide the best autosuggestions possible. But since it's a new PEP, I encourage you to use taskiq with VS code because Pylance understands all types correctly.
+
+# Installation
+
+This project can be installed using pip:
+```bash
+pip install taskiq
+```
+
+Or it can be installed directly from git:
+
+```bash
+pip install git+https://github.com/taskiq-python/taskiq
+```
+
+# Usage
+
+Let's see the example with the in-memory broker:
+
+```python
+import asyncio
+
+from taskiq.brokers.inmemory_broker import InMemoryBroker
+
+
+# This is broker that can be used for
+# development or for demo purpose.
+# In production environment consider using
+# real distributed brokers, such as taskiq-aio-pika
+# for rabbitmq.
+broker = InMemoryBroker()
+
+
+# Or you can optionally
+# pass task_name as the parameter in the decorator.
+@broker.task
+async def my_async_task() -> None:
+    """My lovely task."""
+    await asyncio.sleep(1)
+    print("Hello")
+
+
+async def main():
+    # Kiq is the method that actually
+    # sends task over the network.
+    task = await my_async_task.kiq()
+    # Now we print result of execution.
+    print(await task.get_result())
+
+
+asyncio.run(main())
+
+```
+
+
+You can run it with python without any extra actions,
+since this script uses the `InMemoryBroker`.
+
+It won't send any data over the network,
+and you cannot use this type of broker in
+a real-world scenario, but it's useful for
+local development if you do not want to
+set up a taskiq worker.
+
+
+## Brokers
+
+Brokers are simple. They don't execute functions,
+but they can send messages and listen to new messages.
+
+Every broker implements the [taskiq.abc.broker.AsyncBroker](https://github.com/taskiq-python/taskiq/blob/master/taskiq/abc/broker.py#L50) abstract class. All tasks are assigned to brokers, so every time you call the `kiq` method, you send this task to the assigned broker. (This behavior can be changed, by using `Kicker` directly).
+
+Also you can add middlewares to brokers using `add_middlewares` method.
+
+Like this:
+
+```python
+from taskiq.brokers.inmemory_broker import InMemoryBroker
+from taskiq.middlewares.retry_middleware import SimpleRetryMiddleware
+
+# This is broker that can be used for
+# development or for demo purpose.
+# In production environment consider using
+# real distributed brokers, such as taskiq-aio-pika
+# for rabbitmq.
+broker = InMemoryBroker()
+broker.add_middlewares(
+    [
+        SimpleRetryMiddleware(
+            default_retry_count=4,
+        )
+    ]
+)
+```
+
+To run middlewares properly you must add them using the `add_middlewares` method.
+It lead to errors if you try to add them by modifying broker directly.
+
+Also brokers have formatters. You can change format
+of a message to be compitable with other task execution
+systems, so your migration to taskiq can be smoother.
+
+## Result backends
+
+After task is complete it will try to save the results of execution
+in result backends. By default brokers
+use `DummyResultBackend` wich doesn't do anything. It
+won't print the result in logs and it always returns
+`None` as the `return_value`, and 0 for `execution_time`.
+But some brokers can override it. For example `InMemoryBroker` by default uses `InMemoryResultBackend` and returns correct results.
+
+
+## CLI
+
+Taskiq has a command line interface to run workers.
+It's very simple to get it to work.
+
+You just have to provide path to your broker. As an example, if you want to start listen to new tasks
+with broker in module `my_project.broker` you just
+have to run:
+
+```
+taskiq my_project.broker:broker
+```
+
+taskiq can discover tasks modules to import,
+if you add the `-fsd` (file system discover) option.
+
+Let's assume we have project with the following structure:
+
+```
+test_project
+├── broker.py
+├── submodule
+│   └── tasks.py
+└── utils
+    └── tasks.py
+```
+
+You can specify all tasks modules to import manually.
+
+```bash
+taskiq test_project.broker:broker test_projec.submodule.tasks test_projec.utils.tasks
+```
+
+Or you can let taskiq find all python modules named tasks in current directory recursively.
+
+```bash
+taskiq test_project.broker:broker -fsd
+```
+
+You can always run `--help` to see all possible options.
+
+
+## Middlewares
+
+Middlewares are used to modify message, or take
+some actions after task is complete.
+
+You can write your own middlewares by subclassing
+the `taskiq.abc.middleware.TaskiqMiddleware`.
+
+Every hook can be sync or async. Taskiq will execute it.
+
+For example, this is a valid middleware.
+
+```python
+import asyncio
+
+from taskiq.abc.middleware import TaskiqMiddleware
+from taskiq.message import TaskiqMessage
+
+
+class MyMiddleware(TaskiqMiddleware):
+    async def pre_send(self, message: "TaskiqMessage") -> TaskiqMessage:
+        await asyncio.sleep(1)
+        message.labels["my_label"] = "my_value"
+        return message
+
+    def post_send(self, message: "TaskiqMessage") -> None:
+        print(f"Message {message} was sent.")
+
+```
+
+You can use sync or async hooks without changing aything, but adding async to the hook signature.
+
+Middlewares can store information in message.labels for
+later use. For example `SimpleRetryMiddleware` uses labels
+to remember number of failed attempts.
+
+## Messages
+
+Every message has labels. You can define labels
+using `task` decorator, or you can add them using kicker.
+
+For example:
+
+```python
+
+@broker.task(my_label=1, label2="something")
+async def my_async_task() -> None:
+    """My lovely task."""
+    await asyncio.sleep(1)
+    print("Hello")
+
+async def main():
+    await my_async_task.kiq()
+```
+
+It's equivalent to this
+
+```python
+
+@broker.task
+async def my_async_task() -> None:
+    """My lovely task."""
+    await asyncio.sleep(1)
+    print("Hello")
+
+async def main():
+    await my_async_task.kicker().with_labels(
+        my_label=1,
+        label2="something",
+    ).kiq()
+```
+
+## Kicker
+
+The kicker is the object that sends tasks.
+When you call kiq it generates a Kicker instance,
+remembering current broker and message labels.
+You can change the labels you want to use for this particular task or you can even change broker.
+
+For example:
+
+```python
+import asyncio
+
+from taskiq.brokers.inmemory_broker import InMemoryBroker
+
+broker = InMemoryBroker()
+second_broker = InMemoryBroker()
+
+
+@broker.task
+async def my_async_task() -> None:
+    """My lovely task."""
+    await asyncio.sleep(1)
+    print("Hello")
+
+
+async def main():
+    task = await my_async_task.kicker().with_broker(second_broker).kiq()
+    print(await task.get_result())
+
+
+asyncio.run(main())
+
+```

pyproject.toml

@@ -7,6 +7,21 @@ maintainers = ["Pavel Kirilin <win10@list.ru>"]
 readme = "README.md"
 repository = "https://github.com/taskiq-python/taskiq"
 license = "LICENSE"
+classifiers = [
+    "Typing :: Typed",
+    "Programming Language :: Python",
+    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3 :: Only",
+    "Programming Language :: Python :: 3.7",
+    "Programming Language :: Python :: 3.8",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.10",
+    "Topic :: Utilities",
+    "Topic :: System :: Networking",
+    "Operating System :: OS Independent",
+]
+homepage = "https://github.com/taskiq-python/taskiq"
+keywords = ["taskiq", "tasks", "distributed", "async"]
 
 [tool.poetry.dependencies]
 python = "^3.7"

taskiq/brokers/inmemory_broker.py

@@ -1,12 +1,12 @@
 import inspect
 from collections import OrderedDict
 from concurrent.futures import ThreadPoolExecutor
-from typing import AsyncGenerator, Optional, TypeVar
+from typing import Any, AsyncGenerator, Callable, Optional, TypeVar
 
 from taskiq.abc.broker import AsyncBroker
 from taskiq.abc.result_backend import AsyncResultBackend, TaskiqResult
 from taskiq.cli.async_task_runner import run_task
-from taskiq.exceptions import TaskiqError
+from taskiq.exceptions import ResultSetError, TaskiqError
 from taskiq.message import BrokerMessage
 
 _ReturnType = TypeVar("_ReturnType")
@@ -83,17 +83,22 @@ class InMemoryBroker(AsyncBroker):
     It's useful for local development, if you don't want to setup real broker.
     """
 
-    def __init__(
+    def __init__(  # noqa: WPS211
         self,
         sync_tasks_pool_size: int = 4,
         logs_format: Optional[str] = None,
         max_stored_results: int = 100,
         cast_types: bool = True,
+        result_backend: Optional[AsyncResultBackend[Any]] = None,
+        task_id_generator: Optional[Callable[[], str]] = None,
     ) -> None:
-        super().__init__(
-            InmemoryResultBackend(
+        if result_backend is None:
+            result_backend = InmemoryResultBackend(
                 max_stored_results=max_stored_results,
-            ),
+            )
+        super().__init__(
+            result_backend=result_backend,
+            task_id_generator=task_id_generator,
         )
         self.executor = ThreadPoolExecutor(max_workers=sync_tasks_pool_size)
         self.cast_types = cast_types
@@ -113,6 +118,8 @@ async def kick(self, message: BrokerMessage) -> None:
         This method just executes given task.
 
         :param message: incomming message.
+
+        :raises ResultSetError: if cannot save results in result backend.
         :raises TaskiqError: if someone wants to kick unknown task.
         """
         target_task = self.available_tasks.get(message.task_name)
@@ -127,7 +134,10 @@ async def kick(self, message: BrokerMessage) -> None:
             executor=self.executor,
             middlewares=self.middlewares,
         )
-        await self.result_backend.set_result(message.task_id, result)
+        try:
+            await self.result_backend.set_result(message.task_id, result)
+        except Exception as exc:
+            raise ResultSetError("Cannot set result.") from exc
 
     async def listen(self) -> AsyncGenerator[BrokerMessage, None]:  # type: ignore
         """