Fix: pop schema_ from publishers dto

faststream/_internal/publisher/specified.py

@@ -1,10 +1,9 @@
 from inspect import Parameter, unwrap
-from typing import TYPE_CHECKING, Any, Callable, Union
+from typing import TYPE_CHECKING, Any, Callable, Optional, Union
 
 from fast_depends.core import build_call_model
 from fast_depends.pydantic._compat import create_model, get_config_base
 
-from faststream._internal.publisher.schemas import SpecificationPublisherOptions
 from faststream._internal.types import (
     MsgType,
     P_HandlerParams,
@@ -30,13 +29,13 @@ class SpecificationPublisher(EndpointSpecification[MsgType, PublisherSpec]):
     def __init__(
         self,
         *args: Any,
-        specification_options: SpecificationPublisherOptions,
+        schema_: Optional[Any],
         **kwargs: Any,
     ) -> None:
         self.calls: list[AnyCallable] = []
-        self.schema_ = specification_options.schema_
+        self.schema_ = schema_
         # Call next base class parent init
-        super().__init__(*args, specification_options=specification_options, **kwargs)
+        super().__init__(*args, **kwargs)
 
     def __call__(
         self,

faststream/confluent/publisher/factory.py

@@ -1,20 +1,12 @@
 from collections.abc import Sequence
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    Literal,
-    Optional,
-    Union,
-    overload,
-    cast
-)
+from typing import TYPE_CHECKING, Any, Literal, Optional, Union, cast, overload
 
 from faststream._internal.publisher.schemas import (
     PublisherUsecaseOptions,
-    SpecificationPublisherOptions,
 )
 from faststream.confluent.schemas.publisher import ConfluentPublisherBaseOptions
 from faststream.exceptions import SetupError
+from faststream.specification.schema.base import SpecificationOptions
 
 from .specified import SpecificationBatchPublisher, SpecificationDefaultPublisher
 
@@ -128,11 +120,10 @@ def create_publisher(
         reply_to=reply_to,
         internal_options=internal_options,
     )
-    specification_options = SpecificationPublisherOptions(
+    specification_options = SpecificationOptions(
         title_=title_,
         description_=description_,
         include_in_schema=include_in_schema,
-        schema_=schema_,
     )
 
     if batch:
@@ -141,9 +132,13 @@ def create_publisher(
             raise SetupError(msg)
 
         return SpecificationBatchPublisher(
-            specification_options=specification_options, base_options=base_options
+            specification_options=specification_options,
+            base_options=base_options,
+            schema_=schema_
         )
 
     return SpecificationDefaultPublisher(
-        specification_options=specification_options, base_options=base_options
+        specification_options=specification_options,
+        base_options=base_options,
+        schema_=schema_
     )

faststream/confluent/subscriber/factory.py

@@ -1,13 +1,6 @@
 import warnings
 from collections.abc import Iterable, Sequence
-from typing import (
-    TYPE_CHECKING,
-    Literal,
-    Optional,
-    Union,
-    overload,
-    cast
-)
+from typing import TYPE_CHECKING, Literal, Optional, Union, cast, overload
 
 from faststream._internal.constants import EMPTY
 from faststream._internal.subscriber.schemas import SubscriberUsecaseOptions

faststream/kafka/publisher/factory.py

@@ -10,10 +10,10 @@
 
 from faststream._internal.publisher.schemas import (
     PublisherUsecaseOptions,
-    SpecificationPublisherOptions,
 )
 from faststream.exceptions import SetupError
 from faststream.kafka.schemas.publishers import KafkaPublisherBaseOptions
+from faststream.specification.schema.base import SpecificationOptions
 
 from .specified import SpecificationBatchPublisher, SpecificationDefaultPublisher
 
@@ -121,11 +121,10 @@ def create_publisher(
         reply_to=reply_to,
         internal_options=internal_options,
     )
-    specification_options = SpecificationPublisherOptions(
+    specification_options = SpecificationOptions(
         title_=title_,
         description_=description_,
         include_in_schema=include_in_schema,
-        schema_=schema_,
     )
 
     if batch:
@@ -134,8 +133,12 @@ def create_publisher(
             raise SetupError(msg)
 
         return SpecificationBatchPublisher(
-            base_options=base_options, specification_options=specification_options
+            base_options=base_options,
+            specification_options=specification_options,
+            schema_=schema_
         )
     return SpecificationDefaultPublisher(
-        base_options=base_options, specification_options=specification_options
+        base_options=base_options,
+        specification_options=specification_options,
+        schema_=schema_
     )

faststream/kafka/subscriber/usecase.py

@@ -6,7 +6,6 @@
 import anyio
 from aiokafka import ConsumerRecord, TopicPartition
 from aiokafka.errors import ConsumerStoppedError, KafkaError
-from faststream.middlewares.acknowledgement.conf import AckPolicy
 from typing_extensions import override
 
 from faststream._internal.subscriber.mixins import ConcurrentMixin, TasksMixin
@@ -21,6 +20,7 @@
 from faststream.kafka.parser import AioKafkaBatchParser, AioKafkaParser
 from faststream.kafka.publisher.fake import KafkaFakePublisher
 from faststream.kafka.schemas.subscribers import KafkaSubscriberBaseOptions
+from faststream.middlewares.acknowledgement.conf import AckPolicy
 
 if TYPE_CHECKING:
     from aiokafka import AIOKafkaConsumer

faststream/nats/publisher/factory.py

@@ -3,9 +3,9 @@
 
 from faststream._internal.publisher.schemas import (
     PublisherUsecaseOptions,
-    SpecificationPublisherOptions,
 )
 from faststream.nats.schemas.publishers import NatsPublisherBaseOptions
+from faststream.specification.schema.base import SpecificationOptions
 
 from .specified import SpecificationPublisher
 
@@ -45,8 +45,7 @@ def create_publisher(
         internal_options=internal_options,
     )
 
-    specification_options = SpecificationPublisherOptions(
-        schema_=schema_,
+    specification_options = SpecificationOptions(
         title_=title_,
         description_=description_,
         include_in_schema=include_in_schema,
@@ -55,4 +54,5 @@ def create_publisher(
     return SpecificationPublisher(
         base_options=base_options,
         specification_options=specification_options,
+        schema_=schema_
     )

faststream/rabbit/publisher/factory.py

@@ -3,10 +3,10 @@
 
 from faststream._internal.publisher.schemas import (
     PublisherUsecaseOptions,
-    SpecificationPublisherOptions,
 )
 from faststream.rabbit.schemas.base import RabbitBaseOptions
 from faststream.rabbit.schemas.publishers import RabbitPublisherBaseOptions
+from faststream.specification.schema.base import SpecificationOptions
 
 from .specified import SpecificationPublisher
 
@@ -50,9 +50,8 @@ def create_publisher(
         queue=queue,
         exchange=exchange,
     )
-    # AsyncAPI options
-    specification_options = SpecificationPublisherOptions(
-        schema_=schema_,
+
+    specification_options = SpecificationOptions(
         title_=title_,
         description_=description_,
         include_in_schema=include_in_schema,
@@ -61,4 +60,5 @@ def create_publisher(
         base_options=base_options,
         rabbit_mq_base_options=rabbit_mq_base_options,
         specification_options=specification_options,
+        schema_=schema_,
     )

faststream/rabbit/publisher/specified.py

@@ -1,4 +1,5 @@
-from faststream._internal.publisher.schemas import SpecificationPublisherOptions
+from typing import Any, Optional
+
 from faststream._internal.publisher.specified import (
     SpecificationPublisher as SpecificationPublisherMixin,
 )
@@ -10,6 +11,7 @@
 from faststream.rabbit.utils import is_routing_exchange
 from faststream.specification.asyncapi.utils import resolve_payloads
 from faststream.specification.schema import Message, Operation, PublisherSpec
+from faststream.specification.schema.base import SpecificationOptions
 from faststream.specification.schema.bindings import (
     ChannelBinding,
     OperationBinding,
@@ -31,10 +33,12 @@ def __init__(
         *,
         base_options: RabbitPublisherBaseOptions,
         rabbit_mq_base_options: RabbitBaseOptions,
-        specification_options: SpecificationPublisherOptions,
+        specification_options: SpecificationOptions,
+        schema_: Optional["Any"],
     ) -> None:
         super().__init__(
             specification_options=specification_options,
+            schema_=schema_,
             # propagate to RMQSpecificationMixin
             rabbit_mq_options=rabbit_mq_base_options,
         )

faststream/redis/publisher/factory.py

@@ -5,12 +5,12 @@
 
 from faststream._internal.publisher.schemas import (
     PublisherUsecaseOptions,
-    SpecificationPublisherOptions,
 )
 from faststream.exceptions import SetupError
 from faststream.redis.schemas import INCORRECT_SETUP_MSG, ListSub, PubSub, StreamSub
 from faststream.redis.schemas.proto import validate_options
 from faststream.redis.schemas.publishers import RedisPublisherBaseOptions
+from faststream.specification.schema.base import SpecificationOptions
 
 from .specified import (
     SpecificationChannelPublisher,
@@ -58,8 +58,7 @@ def create_publisher(
         reply_to=reply_to, headers=headers, internal_options=internal_options
     )
 
-    specification_options = SpecificationPublisherOptions(
-        schema_=schema_,
+    specification_options = SpecificationOptions(
         title_=title_,
         description_=description_,
         include_in_schema=include_in_schema,
@@ -70,13 +69,15 @@ def create_publisher(
             channel=channel,
             base_options=base_options,
             specification_options=specification_options,
+            schema_=schema_,
         )
 
     if (stream := StreamSub.validate(stream)) is not None:
         return SpecificationStreamPublisher(
             stream=stream,
             base_options=base_options,
             specification_options=specification_options,
+            schema_=schema_,
         )
 
     if (list := ListSub.validate(list)) is not None:
@@ -85,11 +86,13 @@ def create_publisher(
                 list=list,
                 base_options=base_options,
                 specification_options=specification_options,
+                schema_=schema_,
             )
         return SpecificationListPublisher(
             list=list,
             base_options=base_options,
             specification_options=specification_options,
+            schema_=schema_,
         )
 
     raise SetupError(INCORRECT_SETUP_MSG)