Skip to content

Operation

faststream.specification.asyncapi.v3_0_0.schema.Operation #

Bases: BaseModel

A class to represent an operation.

action instance-attribute #

action: Action

channel instance-attribute #

channel: Channel | Reference

summary class-attribute instance-attribute #

summary: str | None = None

description class-attribute instance-attribute #

description: str | None = None

bindings class-attribute instance-attribute #

bindings: OperationBinding | None = None

messages class-attribute instance-attribute #

messages: list[Reference] = Field(default_factory=list)

security class-attribute instance-attribute #

security: dict[str, list[str]] | None = None

tags class-attribute instance-attribute #

tags: list[Tag | dict[str, Any]] | None = None

model_config class-attribute instance-attribute #

model_config = {'extra': 'allow'}

Config #

extra class-attribute instance-attribute #

extra = 'allow'

from_sub classmethod #

from_sub(
    messages: list[Reference],
    channel: Reference,
    operation: Operation,
) -> Self
Source code in faststream/specification/asyncapi/v3_0_0/schema/operations.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@classmethod
def from_sub(
    cls,
    messages: list[Reference],
    channel: Reference,
    operation: OperationSpec,
) -> Self:
    return cls(
        action=Action.RECEIVE,
        messages=messages,
        channel=channel,
        bindings=OperationBinding.from_sub(operation.bindings),
        summary=None,
        description=None,
        security=None,
        tags=None,
    )

from_pub classmethod #

from_pub(
    messages: list[Reference],
    channel: Reference,
    operation: Operation,
) -> Self
Source code in faststream/specification/asyncapi/v3_0_0/schema/operations.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@classmethod
def from_pub(
    cls,
    messages: list[Reference],
    channel: Reference,
    operation: OperationSpec,
) -> Self:
    return cls(
        action=Action.SEND,
        messages=messages,
        channel=channel,
        bindings=OperationBinding.from_pub(operation.bindings),
        summary=None,
        description=None,
        security=None,
        tags=None,
    )