logging fsm

* feat: refactor logging with FSM * refactor: delete subscribers without dict * refactor: new kafka logging * refactor: new confluent logging * docs: generate API References * tests: fix tests * tests: fix in-memory mocks * confluent: fix self.logger usage * confluent: fix logger usage * confluent: fix logger usage * confluent: make laze logging * confluent: check producer before setup * confluent: fix producer * tests: correct setup call order * fix: remove Confluent producer logger * fix: remove useless option * tests: fix confluent * fix confluent --------- Co-authored-by: Lancetnik <[email protected]>
airtai · Sep 20, 2024 · 865185e · 865185e
1 parent 0a6b83b
commit 865185e
Show file tree

Hide file tree

Showing 59 changed files with 1,116 additions and 892 deletions.
diff --git a/docs/docs/SUMMARY.md b/docs/docs/SUMMARY.md
@@ -245,7 +245,7 @@ search:
                     - broker
                         - [KafkaBroker](api/faststream/confluent/broker/broker/KafkaBroker.md)
                     - logging
-                        - [KafkaLoggingBroker](api/faststream/confluent/broker/logging/KafkaLoggingBroker.md)
+                        - [KafkaParamsStorage](api/faststream/confluent/broker/logging/KafkaParamsStorage.md)
                     - registrator
                         - [KafkaRegistrator](api/faststream/confluent/broker/registrator/KafkaRegistrator.md)
                 - client
@@ -335,6 +335,7 @@ search:
                 - [FastStreamException](api/faststream/exceptions/FastStreamException.md)
                 - [HandlerException](api/faststream/exceptions/HandlerException.md)
                 - [IgnoredException](api/faststream/exceptions/IgnoredException.md)
+                - [IncorrectState](api/faststream/exceptions/IncorrectState.md)
                 - [NackMessage](api/faststream/exceptions/NackMessage.md)
                 - [OperationForbiddenError](api/faststream/exceptions/OperationForbiddenError.md)
                 - [RejectMessage](api/faststream/exceptions/RejectMessage.md)
@@ -358,7 +359,7 @@ search:
                     - broker
                         - [KafkaBroker](api/faststream/kafka/broker/broker/KafkaBroker.md)
                     - logging
-                        - [KafkaLoggingBroker](api/faststream/kafka/broker/logging/KafkaLoggingBroker.md)
+                        - [KafkaParamsStorage](api/faststream/kafka/broker/logging/KafkaParamsStorage.md)
                     - registrator
                         - [KafkaRegistrator](api/faststream/kafka/broker/registrator/KafkaRegistrator.md)
                 - fastapi
@@ -444,6 +445,7 @@ search:
                     - [ignore_handler](api/faststream/middlewares/exception/ignore_handler.md)
                 - logging
                     - [CriticalLogMiddleware](api/faststream/middlewares/logging/CriticalLogMiddleware.md)
+                    - [LoggingMiddleware](api/faststream/middlewares/logging/LoggingMiddleware.md)
             - nats
                 - [AckPolicy](api/faststream/nats/AckPolicy.md)
                 - [ConsumerConfig](api/faststream/nats/ConsumerConfig.md)
@@ -473,7 +475,7 @@ search:
                     - broker
                         - [NatsBroker](api/faststream/nats/broker/broker/NatsBroker.md)
                     - logging
-                        - [NatsLoggingBroker](api/faststream/nats/broker/logging/NatsLoggingBroker.md)
+                        - [NatsParamsStorage](api/faststream/nats/broker/logging/NatsParamsStorage.md)
                     - registrator
                         - [NatsRegistrator](api/faststream/nats/broker/registrator/NatsRegistrator.md)
                 - fastapi
@@ -617,7 +619,7 @@ search:
                     - broker
                         - [RabbitBroker](api/faststream/rabbit/broker/broker/RabbitBroker.md)
                     - logging
-                        - [RabbitLoggingBroker](api/faststream/rabbit/broker/logging/RabbitLoggingBroker.md)
+                        - [RabbitParamsStorage](api/faststream/rabbit/broker/logging/RabbitParamsStorage.md)
                     - registrator
                         - [RabbitRegistrator](api/faststream/rabbit/broker/registrator/RabbitRegistrator.md)
                 - fastapi
@@ -699,7 +701,7 @@ search:
                     - broker
                         - [RedisBroker](api/faststream/redis/broker/broker/RedisBroker.md)
                     - logging
-                        - [RedisLoggingBroker](api/faststream/redis/broker/logging/RedisLoggingBroker.md)
+                        - [RedisParamsStorage](api/faststream/redis/broker/logging/RedisParamsStorage.md)
                     - registrator
                         - [RedisRegistrator](api/faststream/redis/broker/registrator/RedisRegistrator.md)
                 - fastapi

diff --git a/...uent/broker/logging/KafkaLoggingBroker.md → ...uent/broker/logging/KafkaParamsStorage.md b/...uent/broker/logging/KafkaLoggingBroker.md → ...uent/broker/logging/KafkaParamsStorage.md
@@ -8,4 +8,4 @@ search:
   boost: 0.5
 ---
 
-::: faststream.confluent.broker.logging.KafkaLoggingBroker
+::: faststream.confluent.broker.logging.KafkaParamsStorage
diff --git a/.../nats/broker/logging/NatsLoggingBroker.md → ...i/faststream/exceptions/IncorrectState.md b/.../nats/broker/logging/NatsLoggingBroker.md → ...i/faststream/exceptions/IncorrectState.md
@@ -8,4 +8,4 @@ search:
   boost: 0.5
 ---
 
-::: faststream.nats.broker.logging.NatsLoggingBroker
+::: faststream.exceptions.IncorrectState
diff --git a/...edis/broker/logging/RedisLoggingBroker.md → ...afka/broker/logging/KafkaParamsStorage.md b/...edis/broker/logging/RedisLoggingBroker.md → ...afka/broker/logging/KafkaParamsStorage.md
@@ -8,4 +8,4 @@ search:
   boost: 0.5
 ---
 
-::: faststream.redis.broker.logging.RedisLoggingBroker
+::: faststream.kafka.broker.logging.KafkaParamsStorage
diff --git a/...afka/broker/logging/KafkaLoggingBroker.md → .../middlewares/logging/LoggingMiddleware.md b/...afka/broker/logging/KafkaLoggingBroker.md → .../middlewares/logging/LoggingMiddleware.md
@@ -8,4 +8,4 @@ search:
   boost: 0.5
 ---
 
-::: faststream.kafka.broker.logging.KafkaLoggingBroker
+::: faststream.middlewares.logging.LoggingMiddleware
diff --git a/...bit/broker/logging/RabbitLoggingBroker.md → .../nats/broker/logging/NatsParamsStorage.md b/...bit/broker/logging/RabbitLoggingBroker.md → .../nats/broker/logging/NatsParamsStorage.md
@@ -8,4 +8,4 @@ search:
   boost: 0.5
 ---
 
-::: faststream.rabbit.broker.logging.RabbitLoggingBroker
+::: faststream.nats.broker.logging.NatsParamsStorage
diff --git a/docs/docs/en/api/faststream/rabbit/broker/logging/RabbitParamsStorage.md b/docs/docs/en/api/faststream/rabbit/broker/logging/RabbitParamsStorage.md
@@ -0,0 +1,11 @@
+---
+# 0.5 - API
+# 2 - Release
+# 3 - Contributing
+# 5 - Template Page
+# 10 - Default
+search:
+  boost: 0.5
+---
+
+::: faststream.rabbit.broker.logging.RabbitParamsStorage
diff --git a/docs/docs/en/api/faststream/redis/broker/logging/RedisParamsStorage.md b/docs/docs/en/api/faststream/redis/broker/logging/RedisParamsStorage.md
@@ -0,0 +1,11 @@
+---
+# 0.5 - API
+# 2 - Release
+# 3 - Contributing
+# 5 - Template Page
+# 10 - Default
+search:
+  boost: 0.5
+---
+
+::: faststream.redis.broker.logging.RedisParamsStorage
diff --git a/faststream/_internal/broker/abc_broker.py b/faststream/_internal/broker/abc_broker.py
@@ -90,43 +90,37 @@ def include_router(
         for h in router._subscribers.values():
             h.add_prefix("".join((self.prefix, prefix)))
 
-            if (key := hash(h)) not in self._subscribers:
-                if include_in_schema is None:
-                    h.include_in_schema = self._solve_include_in_schema(
-                        h.include_in_schema
-                    )
-                else:
-                    h.include_in_schema = include_in_schema
-
-                h._broker_middlewares = (
-                    *self._middlewares,
-                    *middlewares,
-                    *h._broker_middlewares,
-                )
-                h._broker_dependencies = (
-                    *self._dependencies,
-                    *dependencies,
-                    *h._broker_dependencies,
-                )
-                self._subscribers = {**self._subscribers, key: h}
+            if include_in_schema is None:
+                h.include_in_schema = self._solve_include_in_schema(h.include_in_schema)
+            else:
+                h.include_in_schema = include_in_schema
+
+            h._broker_middlewares = (
+                *self._middlewares,
+                *middlewares,
+                *h._broker_middlewares,
+            )
+            h._broker_dependencies = (
+                *self._dependencies,
+                *dependencies,
+                *h._broker_dependencies,
+            )
+            self._subscribers = {**self._subscribers, hash(h): h}
 
         for p in router._publishers.values():
             p.add_prefix(self.prefix)
 
-            if (key := hash(p)) not in self._publishers:
-                if include_in_schema is None:
-                    p.include_in_schema = self._solve_include_in_schema(
-                        p.include_in_schema
-                    )
-                else:
-                    p.include_in_schema = include_in_schema
-
-                p._broker_middlewares = (
-                    *self._middlewares,
-                    *middlewares,
-                    *p._broker_middlewares,
-                )
-                self._publishers = {**self._publishers, key: p}
+            if include_in_schema is None:
+                p.include_in_schema = self._solve_include_in_schema(p.include_in_schema)
+            else:
+                p.include_in_schema = include_in_schema
+
+            p._broker_middlewares = (
+                *self._middlewares,
+                *middlewares,
+                *p._broker_middlewares,
+            )
+            self._publishers = {**self._publishers, hash(p): p}
 
     def include_routers(
         self,