Skip to content

NatsToken

faststream.nats.security.NatsToken #

NatsToken(
    token: str | Callable[[], str],
    *,
    ssl_context: SSLContext | None = None,
    use_ssl: bool | None = None,
    tls_hostname: str | None = None,
    tls_handshake_first: bool = False,
)

Bases: NatsSecurity

NATS token authentication.

Source code in faststream/nats/security.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def __init__(
    self,
    token: str | Callable[[], str],
    *,
    ssl_context: "SSLContext | None" = None,
    use_ssl: bool | None = None,
    tls_hostname: str | None = None,
    tls_handshake_first: bool = False,
) -> None:
    if not callable(token) and not token:
        msg = "NATS token cannot be empty."
        raise ValueError(msg)

    super().__init__(
        ssl_context=ssl_context,
        use_ssl=use_ssl,
        tls_hostname=tls_hostname,
        tls_handshake_first=tls_handshake_first,
    )
    self.token = token

token instance-attribute #

token = token

ssl_context instance-attribute #

ssl_context: Optional[SSLContext] = ssl_context

use_ssl instance-attribute #

use_ssl: bool = use_ssl

tls_hostname instance-attribute #

tls_hostname = tls_hostname

tls_handshake_first instance-attribute #

tls_handshake_first = tls_handshake_first

get_requirement #

get_requirement() -> list[dict[str, Any]]

Get the AsyncAPI requirement for NATS token authentication.

Source code in faststream/nats/security.py 
79
80
81
82
@override
def get_requirement(self) -> list[dict[str, Any]]:
    """Get the AsyncAPI requirement for NATS token authentication."""
    return [{"nats-token": []}]

get_schema #

get_schema() -> dict[str, dict[str, Any]]

Get the AsyncAPI schema for NATS token authentication.

Source code in faststream/nats/security.py 
84
85
86
87
88
89
90
91
92
93
94
@override
def get_schema(self) -> dict[str, dict[str, Any]]:
    """Get the AsyncAPI schema for NATS token authentication."""
    return {
        "nats-token": {
            "type": "apiKey",
            "in": "user",
            "description": "NATS authentication token sent as CONNECT auth_token.",
            "x-nats-auth": "token",
        },
    }