docs: add Context initial section (#1089)

* docs: add Context initial examples * docs: add orjson mention * chore: bump version * ruff format replaced with black --------- Co-authored-by: Davor Runje <[email protected]>
airtai · Dec 23, 2023 · 8ab9d30 · 8ab9d30
1 parent 1e5f52e
commit 8ab9d30
Show file tree

Hide file tree

Showing 32 changed files with 402 additions and 304 deletions.
diff --git a/docs/docs/en/getting-started/context/extra.md b/docs/docs/en/getting-started/context/extra.md
@@ -29,3 +29,9 @@ By default, context fields are **NOT CAST** to the type specified in their annot
 If you require this functionality, you can enable the appropriate flag.
 
 {! includes/getting_started/context/cast.md !}
+
+## Initial Value
+
+Also, `Context` provides you with a `initial` option to setup base context value without previous `set_global` call.
+
+{! includes/getting_started/context/initial.md !}
diff --git a/docs/docs/en/getting-started/subscription/annotation.md b/docs/docs/en/getting-started/subscription/annotation.md
@@ -38,3 +38,6 @@ But it doesn't looks like a correct message validation, does it?
 For this reason, **FastStream** supports per-argument message serialization: you can declare multiple arguments with various types and your message will unpack to them:
 
 {! includes/getting_started/subscription/annotation/3.md !}
+
+!!! tip
+  By default **FastStream** uses `json.loads` to decode and `json.dumps` to encode your messages. But if you prefer [**orjson**](https://github.com/ijl/orjson){.external-link target="_blank"}, just install it and framework will use it automatically.
diff --git a/docs/docs_src/getting_started/context/kafka/initial.py b/docs/docs_src/getting_started/context/kafka/initial.py
@@ -0,0 +1,11 @@
+from faststream import Context
+from faststream.kafka import KafkaBroker
+
+broker = KafkaBroker()
+
+@broker.subscriber("test-topic")
+async def handle(
+    msg: str,
+    collector: list[str] = Context(initial=list),
+):
+    collector.append(msg)
diff --git a/docs/docs_src/getting_started/context/nats/initial.py b/docs/docs_src/getting_started/context/nats/initial.py
@@ -0,0 +1,11 @@
+from faststream import Context
+from faststream.nats import NatsBroker
+
+broker = NatsBroker()
+
+@broker.subscriber("test-subject")
+async def handle(
+    msg: str,
+    collector: list[str] = Context(initial=list),
+):
+    collector.append(msg)
diff --git a/docs/docs_src/getting_started/context/rabbit/initial.py b/docs/docs_src/getting_started/context/rabbit/initial.py
@@ -0,0 +1,11 @@
+from faststream import Context
+from faststream.rabbit import RabbitBroker
+
+broker = RabbitBroker()
+
+@broker.subscriber("test-queue")
+async def handle(
+    msg: str,
+    collector: list[str] = Context(initial=list),
+):
+    collector.append(msg)
diff --git a/docs/docs_src/getting_started/context/redis/initial.py b/docs/docs_src/getting_started/context/redis/initial.py
@@ -0,0 +1,11 @@
+from faststream import Context
+from faststream.redis import RedisBroker
+
+broker = RedisBroker()
+
+@broker.subscriber("test-channel")
+async def handle(
+    msg: str,
+    collector: list[str] = Context(initial=list),
+):
+    collector.append(msg)
diff --git a/docs/includes/getting_started/context/initial.md b/docs/includes/getting_started/context/initial.md
@@ -0,0 +1,19 @@
+=== "Kafka"
+    ```python linenums="1" hl_lines="4 6"
+    {!> docs_src/getting_started/context/kafka/initial.py [ln:6-11] !}
+    ```
+
+=== "RabbitMQ"
+    ```python linenums="1" hl_lines="4 6"
+    {!> docs_src/getting_started/context/rabbit/initial.py [ln:6-11] !}
+    ```
+
+=== "NATS"
+    ```python linenums="1" hl_lines="4 6"
+    {!> docs_src/getting_started/context/nats/initial.py [ln:6-11] !}
+    ```
+
+=== "Redis"
+    ```python linenums="1" hl_lines="4 6"
+    {!> docs_src/getting_started/context/redis/initial.py [ln:6-11] !}
+    ```
diff --git a/faststream/__about__.py b/faststream/__about__.py
@@ -1,5 +1,5 @@
 """Simple and fast framework to create message brokers based microservices."""
-__version__ = "0.3.9"
+__version__ = "0.3.10"
 
 
 INSTALL_YAML = """

diff --git a/faststream/broker/core/asyncronous.py b/faststream/broker/core/asyncronous.py
@@ -160,10 +160,7 @@ def _process_message(
         func: Callable[[StreamMessage[MsgType]], Awaitable[T_HandlerReturn]],
         watcher: Callable[..., AsyncContextManager[None]],
         **kwargs: Any,
-    ) -> Callable[
-        [StreamMessage[MsgType]],
-        Awaitable[WrappedReturn[T_HandlerReturn]],
-    ]:
+    ) -> Callable[[StreamMessage[MsgType]], Awaitable[WrappedReturn[T_HandlerReturn]],]:
         """Process a message.
 
         Args:

diff --git a/faststream/broker/handler.py b/faststream/broker/handler.py
@@ -359,7 +359,9 @@ async def consume(self, msg: MsgType) -> SendableMessage:  # type: ignore[overri
                         if IS_OPTIMIZED:  # pragma: no cover
                             break
 
-            assert not self.running or processed, "You have to consume message"  # nosec B101
+            assert (
+                not self.running or processed
+            ), "You have to consume message"  # nosec B101
 
         context.reset_local("log_context", log_context_tag)
 

diff --git a/faststream/kafka/broker.py b/faststream/kafka/broker.py
@@ -202,10 +202,7 @@ def _process_message(
         func: Callable[[KafkaMessage], Awaitable[T_HandlerReturn]],
         watcher: Callable[..., AsyncContextManager[None]],
         **kwargs: Any,
-    ) -> Callable[
-        [KafkaMessage],
-        Awaitable[WrappedReturn[T_HandlerReturn]],
-    ]:
+    ) -> Callable[[KafkaMessage], Awaitable[WrappedReturn[T_HandlerReturn]],]:
         """Wrap a message processing function with a watcher and publisher.
 
         Args:

diff --git a/faststream/log/formatter.py b/faststream/log/formatter.py
@@ -24,7 +24,9 @@ class ColourizedFormatter(logging.Formatter):
 
     """
 
-    level_name_colors: DefaultDict[str, Callable[[str], str]] = defaultdict(  # noqa: RUF012
+    level_name_colors: DefaultDict[
+        str, Callable[[str], str]
+    ] = defaultdict(  # noqa: RUF012
         lambda: str,
         **{
             str(logging.DEBUG): lambda level_name: click.style(

diff --git a/faststream/nats/broker.py b/faststream/nats/broker.py
@@ -232,10 +232,7 @@ def _process_message(
         func: Callable[[StreamMessage[Msg]], Awaitable[T_HandlerReturn]],
         watcher: Callable[..., AsyncContextManager[None]],
         **kwargs: Any,
-    ) -> Callable[
-        [StreamMessage[Msg]],
-        Awaitable[WrappedReturn[T_HandlerReturn]],
-    ]:
+    ) -> Callable[[StreamMessage[Msg]], Awaitable[WrappedReturn[T_HandlerReturn]],]:
         @wraps(func)
         async def process_wrapper(
             message: StreamMessage[Msg],
@@ -411,11 +408,7 @@ def subscriber(  # type: ignore[override]
 
         def consumer_wrapper(
             func: Callable[P_HandlerParams, T_HandlerReturn],
-        ) -> HandlerCallWrapper[
-            Msg,
-            P_HandlerParams,
-            T_HandlerReturn,
-        ]:
+        ) -> HandlerCallWrapper[Msg, P_HandlerParams, T_HandlerReturn,]:
             handler_call, dependant = self._wrap_handler(
                 func,
                 extra_dependencies=dependencies,

diff --git a/faststream/nats/broker.pyi b/faststream/nats/broker.pyi
@@ -215,10 +215,7 @@ class NatsBroker(
         func: Callable[[StreamMessage[Msg]], Awaitable[T_HandlerReturn]],
         watcher: Callable[..., AsyncContextManager[None]],
         **kwargs: Any,
-    ) -> Callable[
-        [StreamMessage[Msg]],
-        Awaitable[WrappedReturn[T_HandlerReturn]],
-    ]: ...
+    ) -> Callable[[StreamMessage[Msg]], Awaitable[WrappedReturn[T_HandlerReturn]],]: ...
     def _log_connection_broken(
         self,
         error_cb: Optional[ErrorCallback] = None,

diff --git a/faststream/nats/parser.py b/faststream/nats/parser.py
@@ -38,10 +38,7 @@ async def parse_message(
 
     async def parse_message(
         self, message: Union[Msg, List[Msg]], *, path: Optional[AnyDict] = None
-    ) -> Union[
-        StreamMessage[Msg],
-        StreamMessage[List[Msg]],
-    ]:
+    ) -> Union[StreamMessage[Msg], StreamMessage[List[Msg]],]:
         if isinstance(message, list):
             return NatsMessage(
                 is_js=self.is_js,

diff --git a/faststream/redis/broker.py b/faststream/redis/broker.py
@@ -167,10 +167,7 @@ def _process_message(
         func: Callable[[StreamMessage[Any]], Awaitable[T_HandlerReturn]],
         watcher: Callable[..., AsyncContextManager[None]],
         **kwargs: Any,
-    ) -> Callable[
-        [StreamMessage[Any]],
-        Awaitable[WrappedReturn[T_HandlerReturn]],
-    ]:
+    ) -> Callable[[StreamMessage[Any]], Awaitable[WrappedReturn[T_HandlerReturn]],]:
         @wraps(func)
         async def process_wrapper(
             message: StreamMessage[Any],
@@ -256,11 +253,7 @@ def subscriber(  # type: ignore[override]
 
         def consumer_wrapper(
             func: Callable[P_HandlerParams, T_HandlerReturn],
-        ) -> HandlerCallWrapper[
-            AnyRedisDict,
-            P_HandlerParams,
-            T_HandlerReturn,
-        ]:
+        ) -> HandlerCallWrapper[AnyRedisDict, P_HandlerParams, T_HandlerReturn,]:
             handler_call, dependant = self._wrap_handler(
                 func,
                 extra_dependencies=dependencies,

diff --git a/faststream/redis/broker.pyi b/faststream/redis/broker.pyi
@@ -159,10 +159,7 @@ class RedisBroker(
         func: Callable[[StreamMessage[Any]], Awaitable[T_HandlerReturn]],
         watcher: Callable[..., AsyncContextManager[None]],
         **kwargs: Any,
-    ) -> Callable[
-        [StreamMessage[Any]],
-        Awaitable[WrappedReturn[T_HandlerReturn]],
-    ]: ...
+    ) -> Callable[[StreamMessage[Any]], Awaitable[WrappedReturn[T_HandlerReturn]],]: ...
     @override
     def subscriber(  # type: ignore[override]
         self,

diff --git a/faststream/redis/publisher.py b/faststream/redis/publisher.py
@@ -43,7 +43,9 @@ async def publish(  # type: ignore[override]
         list = ListSub.validate(list or self.list)
         stream = StreamSub.validate(stream or self.stream)
 
-        assert any((channel, list, stream)), "You have to specify outgoing channel"  # nosec B101
+        assert any(
+            (channel, list, stream)
+        ), "You have to specify outgoing channel"  # nosec B101
 
         headers_to_send = (self.headers or {}).copy()
         if headers is not None:

diff --git a/pyproject.toml b/pyproject.toml
@@ -318,3 +318,13 @@ omit = [
 ]
 
 [tool.bandit]
+
+[tool.black]
+line-length = 88
+
+extend-exclude = """
+/(
+    docs/docs_src
+    | some_other_dir
+)/
+"""
diff --git a/scripts/lint.sh b/scripts/lint.sh
@@ -6,5 +6,8 @@ pyup_dirs --py38-plus --recursive faststream examples tests docs
 echo "Running ruff linter (isort, flake, pyupgrade, etc. replacement)..."
 ruff check
 
-echo "Running ruff formater (black replacement)..."
-ruff format
+# echo "Running ruff formater (black replacement)..."
+# ruff format
+
+echo "Running black..."
+black faststream examples tests docs
diff --git a/tests/asyncapi/base/arguments.py b/tests/asyncapi/base/arguments.py
@@ -70,7 +70,7 @@ async def handle(msg: int):
         schema = get_app_schema(self.build_app(broker)).to_jsonable()
 
         payload = schema["components"]["schemas"]
-        assert tuple(schema["channels"].values())[0].get("description") is None  # noqa: RUF015
+        assert next(iter(schema["channels"].values())).get("description") is None
 
         for key, v in payload.items():
             assert key == "Handle:Message:Payload"
@@ -333,7 +333,7 @@ class User(pydantic.BaseModel):
             else:
 
                 class Config:
-                    schema_extra = {"examples": [{"name": "john", "id": 1}]}  # noqa: RUF012
+                    schema_extra = {"examples": [{"name": "john", "id": 1}]}
 
         broker = self.broker_class()
 
@@ -379,7 +379,11 @@ async def handle_default(msg):
         schema = get_app_schema(self.build_app(broker)).to_jsonable()
 
         assert (
-            len(list(schema["components"]["messages"].values())[0]["payload"]["oneOf"])  # noqa: RUF015
+            len(
+                next(iter(schema["components"]["messages"].values()))["payload"][
+                    "oneOf"
+                ]
+            )
             == 2
         )