fix type hins in faststream/specification

airtai · Dec 13, 2024 · e75409d · e75409d
1 parent c032ecf
commit e75409d
Show file tree

Hide file tree

Showing 13 changed files with 49 additions and 33 deletions.
diff --git a/faststream/specification/asyncapi/factory.py b/faststream/specification/asyncapi/factory.py
@@ -1,4 +1,4 @@
-from collections.abc import Iterable
+from collections.abc import Sequence
 from typing import TYPE_CHECKING, Any, Literal, Optional, Union
 
 from faststream.specification.base.specification import Specification
@@ -28,7 +28,7 @@ def AsyncAPI(  # noqa: N802
     terms_of_service: Optional["AnyHttpUrl"] = None,
     license: Optional[Union["License", "LicenseDict", "AnyDict"]] = None,
     contact: Optional[Union["Contact", "ContactDict", "AnyDict"]] = None,
-    tags: Iterable[Union["Tag", "TagDict", "AnyDict"]] = (),
+    tags: Sequence[Union["Tag", "TagDict", "AnyDict"]] = (),
     external_docs: Optional[
         Union["ExternalDocs", "ExternalDocsDict", "AnyDict"]
     ] = None,

diff --git a/faststream/specification/asyncapi/message.py b/faststream/specification/asyncapi/message.py
@@ -18,7 +18,7 @@
 
 def parse_handler_params(call: "CallModel", prefix: str = "") -> AnyDict:
     """Parses the handler parameters."""
-    model = getattr(call, "serializer", call).model
+    model = getattr(call, "serializer", call).model  # type: ignore[union-attr]
     assert model  # nosec B101
 
     body = get_model_schema(

diff --git a/faststream/specification/asyncapi/v2_6_0/facade.py b/faststream/specification/asyncapi/v2_6_0/facade.py
@@ -1,4 +1,4 @@
-from collections.abc import Iterable
+from collections.abc import Sequence
 from typing import TYPE_CHECKING, Any, Optional, Union
 
 from faststream.specification.base.specification import Specification
@@ -34,7 +34,7 @@ def __init__(
         contact: Optional[Union["Contact", "ContactDict", "AnyDict"]] = None,
         license: Optional[Union["License", "LicenseDict", "AnyDict"]] = None,
         identifier: Optional[str] = None,
-        tags: Iterable[Union["Tag", "TagDict", "AnyDict"]] = (),
+        tags: Sequence[Union["Tag", "TagDict", "AnyDict"]] = (),
         external_docs: Optional[
             Union["ExternalDocs", "ExternalDocsDict", "AnyDict"]
         ] = None,

diff --git a/faststream/specification/asyncapi/v2_6_0/schema/bindings/sqs/channel.py b/faststream/specification/asyncapi/v2_6_0/schema/bindings/sqs/channel.py
@@ -22,12 +22,15 @@ class ChannelBinding(BaseModel):
     bindingVersion: str = "custom"
 
     @classmethod
-    def from_spec(cls, binding: sqs.ChannelBinding) -> Self:
+    def from_pub(cls, binding: sqs.ChannelBinding) -> Self:
         return cls(
             queue=binding.queue,
             bindingVersion=binding.bindingVersion,
         )
 
-
-def from_spec(binding: sqs.ChannelBinding) -> ChannelBinding:
-    return ChannelBinding.from_spec(binding)
+    @classmethod
+    def from_sub(cls, binding: sqs.ChannelBinding) -> Self:
+        return cls(
+            queue=binding.queue,
+            bindingVersion=binding.bindingVersion,
+        )
diff --git a/faststream/specification/asyncapi/v2_6_0/schema/bindings/sqs/operation.py b/faststream/specification/asyncapi/v2_6_0/schema/bindings/sqs/operation.py
@@ -24,12 +24,15 @@ class OperationBinding(BaseModel):
     bindingVersion: str = "custom"
 
     @classmethod
-    def from_spec(cls, binding: sqs.OperationBinding) -> Self:
+    def from_pub(cls, binding: sqs.OperationBinding) -> Self:
         return cls(
             replyTo=binding.replyTo,
             bindingVersion=binding.bindingVersion,
         )
 
-
-def from_spec(binding: sqs.OperationBinding) -> OperationBinding:
-    return OperationBinding.from_spec(binding)
+    @classmethod
+    def from_sub(cls, binding: sqs.OperationBinding) -> Self:
+        return cls(
+            replyTo=binding.replyTo,
+            bindingVersion=binding.bindingVersion,
+        )
diff --git a/faststream/specification/asyncapi/v2_6_0/schema/contact.py b/faststream/specification/asyncapi/v2_6_0/schema/contact.py
@@ -1,4 +1,4 @@
-from typing import Optional, Union, overload
+from typing import Optional, Union, cast, overload
 
 from pydantic import AnyHttpUrl, BaseModel
 from typing_extensions import Self
@@ -64,6 +64,7 @@ def from_spec(
                 email=contact.email,
             )
 
+        contact = cast(AnyDict, contact)
         contact_data, custom_data = filter_by_dict(ContactDict, contact)
 
         if custom_data:

diff --git a/faststream/specification/asyncapi/v2_6_0/schema/docs.py b/faststream/specification/asyncapi/v2_6_0/schema/docs.py
@@ -1,4 +1,4 @@
-from typing import Optional, Union, overload
+from typing import Optional, Union, cast, overload
 
 from pydantic import AnyHttpUrl, BaseModel
 from typing_extensions import Self
@@ -58,6 +58,7 @@ def from_spec(
         if isinstance(docs, SpecDocs):
             return cls(url=docs.url, description=docs.description)
 
+        docs = cast(AnyDict, docs)
         docs_data, custom_data = filter_by_dict(ExternalDocsDict, docs)
 
         if custom_data:

diff --git a/faststream/specification/asyncapi/v2_6_0/schema/license.py b/faststream/specification/asyncapi/v2_6_0/schema/license.py
@@ -1,4 +1,4 @@
-from typing import Optional, Union, overload
+from typing import Optional, Union, cast, overload
 
 from pydantic import AnyHttpUrl, BaseModel
 from typing_extensions import Self
@@ -64,6 +64,7 @@ def from_spec(
                 url=license.url,
             )
 
+        license = cast(AnyDict, license)
         license_data, custom_data = filter_by_dict(LicenseDict, license)
 
         if custom_data:

diff --git a/faststream/specification/asyncapi/v2_6_0/schema/tag.py b/faststream/specification/asyncapi/v2_6_0/schema/tag.py
@@ -1,4 +1,4 @@
-from typing import Optional, Union, overload
+from typing import Optional, Union, cast, overload
 
 from pydantic import BaseModel
 from typing_extensions import Self
@@ -56,6 +56,7 @@ def from_spec(cls, tag: Union[SpecTag, TagDict, AnyDict]) -> Union[Self, AnyDict
                 externalDocs=ExternalDocs.from_spec(tag.external_docs),
             )
 
+        tag = cast(AnyDict, tag)
         tag_data, custom_data = filter_by_dict(TagDict, tag)
 
         if custom_data:

diff --git a/faststream/specification/asyncapi/v3_0_0/facade.py b/faststream/specification/asyncapi/v3_0_0/facade.py
@@ -1,4 +1,4 @@
-from collections.abc import Iterable
+from collections.abc import Sequence
 from typing import TYPE_CHECKING, Any, Optional, Union
 
 from faststream.specification.base.specification import Specification
@@ -34,7 +34,7 @@ def __init__(
         contact: Optional[Union["Contact", "ContactDict", "AnyDict"]] = None,
         license: Optional[Union["License", "LicenseDict", "AnyDict"]] = None,
         identifier: Optional[str] = None,
-        tags: Iterable[Union["Tag", "TagDict", "AnyDict"]] = (),
+        tags: Sequence[Union["Tag", "TagDict", "AnyDict"]] = (),
         external_docs: Optional[
             Union["ExternalDocs", "ExternalDocsDict", "AnyDict"]
         ] = None,

diff --git a/faststream/specification/asyncapi/v3_0_0/generate.py b/faststream/specification/asyncapi/v3_0_0/generate.py
@@ -87,7 +87,7 @@ def get_app_schema(
             termsOfService=terms_of_service,
             contact=Contact.from_spec(contact),
             license=License.from_spec(license),
-            tags=[Tag.from_spec(tag) for tag in tags] or None,
+            tags=[Tag.from_spec(tag) for tag in tags] or None if tags else None,
             externalDocs=ExternalDocs.from_spec(external_docs),
         ),
         asyncapi=schema_version,
@@ -153,10 +153,10 @@ def get_broker_channels(
     operations = {}
 
     for sub in broker._subscribers:
-        for key, channel in sub.schema().items():
-            channel_obj = Channel.from_sub(key, channel)
+        for sub_key, sub_channel in sub.schema().items():
+            channel_obj = Channel.from_sub(sub_key, sub_channel)
 
-            channel_key = clear_key(key)
+            channel_key = clear_key(sub_key)
             # TODO: add duplication key warning
             channels[channel_key] = channel_obj
 
@@ -168,14 +168,14 @@ def get_broker_channels(
                     for msg_name in channel_obj.messages
                 ],
                 channel=Reference(**{"$ref": f"#/channels/{channel_key}"}),
-                operation=channel.operation,
+                operation=sub_channel.operation,
             )
 
     for pub in broker._publishers:
-        for key, channel in pub.schema().items():
-            channel_obj = Channel.from_pub(key, channel)
+        for pub_key, pub_channel in pub.schema().items():
+            channel_obj = Channel.from_pub(pub_key, pub_channel)
 
-            channel_key = clear_key(key)
+            channel_key = clear_key(pub_key)
             # TODO: add duplication key warning
             channels[channel_key] = channel_obj
 
@@ -187,7 +187,7 @@ def get_broker_channels(
                     for msg_name in channel_obj.messages
                 ],
                 channel=Reference(**{"$ref": f"#/channels/{channel_key}"}),
-                operation=channel.operation,
+                operation=pub_channel.operation,
             )
 
     return channels, operations

diff --git a/faststream/specification/asyncapi/v3_0_0/schema/__init__.py b/faststream/specification/asyncapi/v3_0_0/schema/__init__.py
@@ -1,3 +1,5 @@
+from faststream.specification.asyncapi.v2_6_0.schema import ServerVariable
+
 from .channels import Channel
 from .components import Components
 from .contact import Contact
@@ -7,7 +9,7 @@
 from .message import CorrelationId, Message
 from .operations import Operation
 from .schema import ApplicationSchema
-from .servers import Server, ServerVariable
+from .servers import Server
 from .tag import Tag
 from .utils import Parameter, Reference
 

diff --git a/faststream/specification/asyncapi/v3_0_0/schema/bindings/amqp/operation.py b/faststream/specification/asyncapi/v3_0_0/schema/bindings/amqp/operation.py
@@ -5,16 +5,20 @@
 
 from typing import Optional
 
+from pydantic import BaseModel, PositiveInt
 from typing_extensions import Self
 
-from faststream.specification.asyncapi.v2_6_0.schema.bindings.amqp import (
-    OperationBinding as V2Binding,
-)
 from faststream.specification.schema.bindings import amqp
 
 
-class OperationBinding(V2Binding):
+class OperationBinding(BaseModel):
     cc: Optional[list[str]] = None
+    ack: bool
+    replyTo: Optional[str] = None
+    deliveryMode: Optional[int] = None
+    mandatory: Optional[bool] = None
+    priority: Optional[PositiveInt] = None
+
     bindingVersion: str = "0.3.0"
 
     @classmethod