Coverage for faststream / specification / asyncapi / v2_6_0 / schema / bindings / sqs / operation.py: 78%

1"""AsyncAPI SQS bindings. 

2 

3References: https://github.com/asyncapi/bindings/tree/master/sqs 

4""" 

5 

6from typing import Any (empty)

7 

8from pydantic import BaseModel (empty)

9from typing_extensions import Self (empty)

10 

11from faststream.specification.schema.bindings import sqs (empty)

12 

13 

14class OperationBinding(BaseModel): (empty)

15 """A class to represent an operation binding. 

16 

17 Attributes: 

18 replyTo : optional dictionary containing reply information 

19 bindingVersion : version of the binding, default is "custom" 

20 """ 

21 

22 replyTo: dict[str, Any] | None = None (empty)

23 bindingVersion: str = "custom" (empty)

24 

25 @classmethod (empty)

26 def from_pub(cls, binding: sqs.OperationBinding) -> Self: (empty)

27 return cls( 

28 replyTo=binding.replyTo, 

29 bindingVersion=binding.bindingVersion, 

30 ) 

31 

32 @classmethod (empty)

33 def from_sub(cls, binding: sqs.OperationBinding) -> Self: (empty)

34 return cls( 

35 replyTo=binding.replyTo, 

36 bindingVersion=binding.bindingVersion, 

37 )