Skip to content

AsyncAPI

faststream.specification.AsyncAPI #

AsyncAPI(
    broker: Optional[BrokerUsecase[Any, Any]] = None,
    /,
    title: str = "FastStream",
    version: str = "0.1.0",
    description: str | None = None,
    terms_of_service: Optional[AnyHttpUrl] = None,
    license: Union[License, dict[str, Any]] | None = None,
    contact: Union[Contact, dict[str, Any]] | None = None,
    tags: Sequence[Union[Tag, dict[str, Any]]] = (),
    external_docs: Union[ExternalDocs, dict[str, Any]]
    | None = None,
    identifier: str | None = None,
    schema_version: Literal["3.0.0", "2.6.0"]
    | str = "3.0.0",
)

Bases: SpecificationFactory

Source code in faststream/specification/asyncapi/factory.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def __init__(
    self,
    broker: Optional["BrokerUsecase[Any, Any]"] = None,
    /,
    title: str = "FastStream",
    version: str = "0.1.0",
    description: str | None = None,
    terms_of_service: Optional["AnyHttpUrl"] = None,
    license: Union["License", "dict[str, Any]"] | None = None,
    contact: Union["Contact", "dict[str, Any]"] | None = None,
    tags: Sequence[Union["Tag", "dict[str, Any]"]] = (),
    external_docs: Union["ExternalDocs", "dict[str, Any]"] | None = None,
    identifier: str | None = None,
    schema_version: Literal["3.0.0", "2.6.0"] | str = "3.0.0",
) -> None:
    self.title = title
    self.version = version
    self.description = description
    self.terms_of_service = terms_of_service
    self.license = license
    self.contact = contact
    self.tags = tags
    self.external_docs = external_docs
    self.identifier = identifier
    self.schema_version = schema_version

    self.brokers: list[BrokerUsecase[Any, Any]] = []
    if broker:
        self.add_broker(broker)

    self.http_handlers: list[tuple[str, HttpHandler]] = []

title instance-attribute #

title = title

version instance-attribute #

version = version

description instance-attribute #

description = description

terms_of_service instance-attribute #

terms_of_service = terms_of_service

license instance-attribute #

license = license

contact instance-attribute #

contact = contact

tags instance-attribute #

tags = tags

external_docs instance-attribute #

external_docs = external_docs

identifier instance-attribute #

identifier = identifier

schema_version instance-attribute #

schema_version = schema_version

brokers instance-attribute #

brokers: list[BrokerUsecase[Any, Any]] = []

http_handlers instance-attribute #

http_handlers: list[tuple[str, HttpHandler]] = []

add_broker #

add_broker(
    broker: BrokerUsecase[Any, Any],
) -> SpecificationFactory
Source code in faststream/specification/asyncapi/factory.py 
46
47
48
49
50
51
52
53
def add_broker(
    self,
    broker: "BrokerUsecase[Any, Any]",
    /,
) -> "SpecificationFactory":
    if broker not in self.brokers:
        self.brokers.append(broker)
    return self

add_http_route #

add_http_route(
    path: str, handler: HttpHandler
) -> SpecificationFactory
Source code in faststream/specification/asyncapi/factory.py 
55
56
57
58
59
60
61
def add_http_route(
    self,
    path: str,
    handler: "HttpHandler",
) -> "SpecificationFactory":
    self.http_handlers.append((path, handler))
    return self

to_specification #

to_specification() -> Specification
Source code in faststream/specification/asyncapi/factory.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def to_specification(self) -> Specification:
    if self.schema_version.startswith("3."):
        from .v3_0_0 import get_app_schema as schema_3_0

        return schema_3_0(
            self.brokers[0],
            title=self.title,
            app_version=self.version,
            schema_version=self.schema_version,
            description=self.description,
            terms_of_service=self.terms_of_service,
            contact=self.contact,
            license=self.license,
            identifier=self.identifier,
            tags=self.tags,
            external_docs=self.external_docs,
            http_handlers=self.http_handlers,
        )

    if self.schema_version.startswith("2.6."):
        from .v2_6_0 import get_app_schema as schema_2_6

        return schema_2_6(
            self.brokers[0],
            title=self.title,
            app_version=self.version,
            schema_version=self.schema_version,
            description=self.description,
            terms_of_service=self.terms_of_service,
            contact=self.contact,
            license=self.license,
            identifier=self.identifier,
            tags=self.tags,
            external_docs=self.external_docs,
            http_handlers=self.http_handlers,
        )

    msg = f"Unsupported schema version: {self.schema_version}"
    raise NotImplementedError(msg)