KafkaRouter

faststream.confluent.fastapi.KafkaRouter #

KafkaRouter(
    bootstrap_servers: str | Iterable[str] = "localhost",
    *,
    request_timeout_ms: int = 40 * 1000,
    retry_backoff_ms: int = 100,
    metadata_max_age_ms: int = 5 * 60 * 1000,
    connections_max_idle_ms: int = 9 * 60 * 1000,
    client_id: str | None = SERVICE_NAME,
    allow_auto_create_topics: bool = True,
    config: Optional[ConfluentConfig] = None,
    acks: Literal[0, 1, -1, "all"] = EMPTY,
    compression_type: Literal[
        "gzip", "snappy", "lz4", "zstd"
    ]
    | None = None,
    partitioner: str
    | Callable[
        [bytes, list[Partition], list[Partition]], Partition
    ] = "consistent_random",
    max_request_size: int = 1024 * 1024,
    linger_ms: int = 0,
    enable_idempotence: bool = False,
    transactional_id: str | None = None,
    transaction_timeout_ms: int = 60 * 1000,
    graceful_timeout: float | None = 15.0,
    decoder: Optional[CustomCallable] = None,
    parser: Optional[CustomCallable] = None,
    middlewares: Sequence[BrokerMiddleware[Any, Any]] = (),
    specification: Optional[SpecificationFactory] = None,
    security: Optional[BaseSecurity] = None,
    specification_url: str | None = None,
    protocol: str | None = None,
    protocol_version: str = "auto",
    description: str | None = None,
    specification_tags: Iterable[Union[Tag, TagDict]] = (),
    logger: Optional[LoggerProto] = EMPTY,
    log_level: int = INFO,
    setup_state: bool = True,
    context: ContextRepo | None = None,
    schema_url: str | None = "/asyncapi",
    prefix: str = "",
    tags: list[Union[str, Enum]] | None = None,
    dependencies: Sequence[Depends] | None = None,
    default_response_class: type[Response] = Default(
        JSONResponse
    ),
    responses: dict[int | str, dict[str, Any]]
    | None = None,
    callbacks: list[BaseRoute] | None = None,
    routes: list[BaseRoute] | None = None,
    redirect_slashes: bool = True,
    default: Optional[ASGIApp] = None,
    dependency_overrides_provider: Any | None = None,
    route_class: type[APIRoute] = APIRoute,
    on_startup: Sequence[Callable[[], Any]] | None = None,
    on_shutdown: Sequence[Callable[[], Any]] | None = None,
    lifespan: Optional[Lifespan[Any]] = None,
    deprecated: bool | None = None,
    include_in_schema: bool = True,
    generate_unique_id_function: Callable[
        [APIRoute], str
    ] = Default(generate_unique_id),
)

Bases: StreamRouter[Message | tuple[Message, ...]]

A class to represent a Kafka router.

Initialize the KafkaRouter.

PARAMETER	DESCRIPTION
`bootstrap_servers`	A `host[:port]` string (or list of `host[:port]` strings) that the consumer should contact to bootstrap initial cluster metadata. This does not have to be the full node list. It just needs to have at least one broker that will respond to a Metadata API Request. Default port is 9092. TYPE: `str \| Iterable[str]` DEFAULT: `'localhost'`
`request_timeout_ms`	Client request timeout in milliseconds. TYPE: `int` DEFAULT: `40 * 1000`
`retry_backoff_ms`	Milliseconds to backoff when retrying on errors. TYPE: `int` DEFAULT: `100`
`metadata_max_age_ms`	The period of time in milliseconds after which we force a refresh of metadata even if we haven't seen any partition leadership changes to proactively discover any new brokers or partitions. TYPE: `int` DEFAULT: `5 * 60 * 1000`
`connections_max_idle_ms`	Close idle connections after the number of milliseconds specified by this config. Specifying `None` will disable idle checks. TYPE: `int` DEFAULT: `9 * 60 * 1000`
`client_id`	A name for this client. This string is passed in each request to servers and can be used to identify specific server-side log entries that correspond to this client. Also submitted to :class:`~.consumer.group_coordinator.GroupCoordinator` for logging with respect to consumer group administration. TYPE: `str \| None` DEFAULT: `SERVICE_NAME`
`allow_auto_create_topics`	Allow automatic topic creation on the broker when subscribing to or assigning non-existent topics. TYPE: `bool` DEFAULT: `True`
`config`	Extra configuration for the confluent-kafka-python producer/consumer. See `confluent_kafka.Config <https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#kafka-client-configuration>`_. TYPE: `Optional[ConfluentConfig]` DEFAULT: `None`
`acks`	One of `0`, `1`, `all`. The number of acknowledgments the producer requires the leader to have received before considering a request complete. This controls the durability of records that are sent. The following settings are common: `0`: Producer will not wait for any acknowledgment from the server at all. The message will immediately be added to the socket buffer and considered sent. No guarantee can be made that the server has received the record in this case, and the retries configuration will not take effect (as the client won't generally know of any failures). The offset given back for each record will always be set to -1. `1`: The broker leader will write the record to its local log but will respond without awaiting full acknowledgement from all followers. In this case should the leader fail immediately after acknowledging the record but before the followers have replicated it then the record will be lost. `all`: The broker leader will wait for the full set of in-sync replicas to acknowledge the record. This guarantees that the record will not be lost as long as at least one in-sync replica remains alive. This is the strongest available guarantee. If unset, defaults to `acks=1`. If `enable_idempotence` is :data:`True` defaults to `acks=all`. TYPE: `Literal[0, 1, -1, 'all']` DEFAULT: `EMPTY`
`compression_type`	The compression type for all data generated bythe producer. Compression is of full batches of data, so the efficacy of batching will also impact the compression ratio (more batching means better compression). TYPE: `Literal['gzip', 'snappy', 'lz4', 'zstd'] \| None` DEFAULT: `None`
`partitioner`	Callable used to determine which partition each message is assigned to. Called (after key serialization): `partitioner(key_bytes, all_partitions, available_partitions)`. The default partitioner implementation hashes each non-None key using the same murmur2 algorithm as the Java client so that messages with the same key are assigned to the same partition. When a key is :data:`None`, the message is delivered to a random partition (filtered to partitions with available leaders only, if possible). TYPE: `str \| Callable[[bytes, list[Partition], list[Partition]], Partition]` DEFAULT: `'consistent_random'`
`max_request_size`	The maximum size of a request. This is also effectively a cap on the maximum record size. Note that the server has its own cap on record size which may be different from this. This setting will limit the number of record batches the producer will send in a single request to avoid sending huge requests. TYPE: `int` DEFAULT: `1024 * 1024`
`linger_ms`	The producer groups together any records that arrive in between request transmissions into a single batched request. Normally this occurs only under load when records arrive faster than they can be sent out. However in some circumstances the client may want to reduce the number of requests even under moderate load. This setting accomplishes this by adding a small amount of artificial delay; that is, if first request is processed faster, than `linger_ms`, producer will wait `linger_ms - process_time`. TYPE: `int` DEFAULT: `0`
`enable_idempotence`	When set to `True`, the producer will ensure that exactly one copy of each message is written in the stream. If `False`, producer retries due to broker failures, etc., may write duplicates of the retried message in the stream. Note that enabling idempotence acks to set to `all`. If it is not explicitly set by the user it will be chosen. TYPE: `bool` DEFAULT: `False`
`transactional_id`	Transactional ID for the producer. TYPE: `str \| None` DEFAULT: `None`
`transaction_timeout_ms`	Transaction timeout in milliseconds. TYPE: `int` DEFAULT: `60 * 1000`
`graceful_timeout`	Graceful shutdown timeout. Broker waits for all running subscribers completion before shut down. TYPE: `float \| None` DEFAULT: `15.0`
`decoder`	Custom decoder object. TYPE: `Optional[CustomCallable]` DEFAULT: `None`
`parser`	Custom parser object. TYPE: `Optional[CustomCallable]` DEFAULT: `None`
`middlewares`	Middlewares to apply to all broker publishers/subscribers. TYPE: `Sequence[BrokerMiddleware[Any, Any]]` DEFAULT: `()`
`specification`	Optional specification factory to generate the AsyncAPI schema for the broker. If provided, this will be used to customize or override the default schema generation. TYPE: `Optional[SpecificationFactory]` DEFAULT: `None`
`security`	Security options to connect broker and generate Specification server security information. TYPE: `Optional[BaseSecurity]` DEFAULT: `None`
`specification_url`	Specification hardcoded server addresses. Use `servers` if not specified. TYPE: `str \| None` DEFAULT: `None`
`protocol`	Specification server protocol. TYPE: `str \| None` DEFAULT: `None`
`protocol_version`	Specification server protocol version. TYPE: `str` DEFAULT: `'auto'`
`description`	Specification server description. TYPE: `str \| None` DEFAULT: `None`
`specification_tags`	Specification server tags. TYPE: `Iterable[Union[Tag, TagDict]]` DEFAULT: `()`
`context`	faststream.ContextRepo object to store application injections. TYPE: `ContextRepo \| None` DEFAULT: `None`
`logger`	User specified logger to pass into Context and log service messages. TYPE: `Optional[LoggerProto]` DEFAULT: `EMPTY`
`log_level`	Service messages log level. TYPE: `int` DEFAULT: `INFO`
`setup_state`	Whether to add broker to app scope in lifespan. You should disable this option at old ASGI servers. TYPE: `bool` DEFAULT: `True`
`schema_url`	The URL path where the AsyncAPI schema will be served (e.g. "/asyncapi"). If set to None, the schema endpoint will not be added automatically. TYPE: `str \| None` DEFAULT: `'/asyncapi'`
`prefix`	An optional path prefix for the router. TYPE: `str` DEFAULT: `''`
`tags`	A list of tags to be applied to all the path operations in this router. It will be added to the generated OpenAPI (e.g. visible at `/docs`). Read more about it in the FastAPI docs for Path Operation Configuration: https://fastapi.tiangolo.com/tutorial/path-operation-configuration/ TYPE: `list[Union[str, Enum]] \| None` DEFAULT: `None`
`dependencies`	A list of dependencies (using `Depends()`) to be applied to all the path operations in this router. Read more about it in the FastAPI docs for Bigger Applications - Multiple Files: https://fastapi.tiangolo.com/tutorial/bigger-applications/#include-an-apirouter-with-a-custom-prefix-tags-responses-and-dependencies TYPE: `Sequence[Depends] \| None` DEFAULT: `None`
`default_response_class`	The default response class to be used. Read more in the FastAPI docs for Custom Response - HTML, Stream, File, others: https://fastapi.tiangolo.com/advanced/custom-response/#default-response-class TYPE: `type[Response]` DEFAULT: `Default(JSONResponse)`
`responses`	Additional responses to be shown in OpenAPI. It will be added to the generated OpenAPI (e.g. visible at `/docs`). Read more about it in the FastAPI docs for Additional Responses in OpenAPI: https://fastapi.tiangolo.com/advanced/additional-responses/ And in the FastAPI docs for Bigger Applications: https://fastapi.tiangolo.com/tutorial/bigger-applications/#include-an-apirouter-with-a-custom-prefix-tags-responses-and-dependencies TYPE: `dict[int \| str, dict[str, Any]] \| None` DEFAULT: `None`
`callbacks`	OpenAPI callbacks that should apply to all path operations in this router. It will be added to the generated OpenAPI (e.g. visible at `/docs`). Read more about it in the FastAPI docs for OpenAPI Callbacks: https://fastapi.tiangolo.com/advanced/openapi-callbacks/ TYPE: `list[BaseRoute] \| None` DEFAULT: `None`
`routes`	Note: you probably shouldn't use this parameter, it is inherited from Starlette and supported for compatibility. A list of routes to serve incoming HTTP and WebSocket requests. You normally wouldn't use this parameter with FastAPI, it is inherited from Starlette and supported for compatibility. In FastAPI, you normally would use the path operation methods, like `router.get()`, `router.post()`, etc. TYPE: `list[BaseRoute] \| None` DEFAULT: `None`
`redirect_slashes`	Whether to detect and redirect slashes in URLs when the client doesn't use the same format. TYPE: `bool` DEFAULT: `True`
`default`	Default function handler for this router. Used to handle 404 Not Found errors. TYPE: `Optional[ASGIApp]` DEFAULT: `None`
`dependency_overrides_provider`	Only used internally by FastAPI to handle dependency overrides. You shouldn't need to use it. It normally points to the `FastAPI` app object. TYPE: `Any \| None` DEFAULT: `None`
`route_class`	Custom route (path operation) class to be used by this router. Read more about it in the FastAPI docs for Custom Request and APIRoute class: https://fastapi.tiangolo.com/how-to/custom-request-and-route/#custom-apiroute-class-in-a-router TYPE: `type[APIRoute]` DEFAULT: `APIRoute`
`on_startup`	A list of startup event handler functions. You should instead use the `lifespan` handlers. Read more in the FastAPI docs for `lifespan`: https://fastapi.tiangolo.com/advanced/events/ TYPE: `Sequence[Callable[[], Any]] \| None` DEFAULT: `None`
`on_shutdown`	A list of shutdown event handler functions. You should instead use the `lifespan` handlers. Read more in the FastAPI docs for `lifespan`: https://fastapi.tiangolo.com/advanced/events/ TYPE: `Sequence[Callable[[], Any]] \| None` DEFAULT: `None`
`lifespan`	A `Lifespan` context manager handler. This replaces `startup` and `shutdown` functions with a single context manager. Read more in the FastAPI docs for `lifespan`: https://fastapi.tiangolo.com/advanced/events/ TYPE: `Optional[Lifespan[Any]]` DEFAULT: `None`
`deprecated`	Mark all path operations in this router as deprecated. It will be added to the generated OpenAPI (e.g. visible at `/docs`). Read more about it in the FastAPI docs for Path Operation Configuration: https://fastapi.tiangolo.com/tutorial/path-operation-configuration/ TYPE: `bool \| None` DEFAULT: `None`
`include_in_schema`	To include (or not) all the path operations in this router in the generated OpenAPI. This affects the generated OpenAPI (e.g. visible at `/docs`). Read more about it in the FastAPI docs for Query Parameters and String Validations: https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi TYPE: `bool` DEFAULT: `True`
`generate_unique_id_function`	Customize the function used to generate unique IDs for the path operations shown in the generated OpenAPI. This is particularly useful when automatically generating clients or SDKs for your API. Read more about it in the FastAPI docs about how to Generate Clients: https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function TYPE: `Callable[[APIRoute], str]` DEFAULT: `Default(generate_unique_id)`

Source code in faststream/confluent/fastapi/fastapi.py

def __init__(
    self,
    bootstrap_servers: str | Iterable[str] = "localhost",
    *,
    # both
    request_timeout_ms: int = 40 * 1000,
    retry_backoff_ms: int = 100,
    metadata_max_age_ms: int = 5 * 60 * 1000,
    connections_max_idle_ms: int = 9 * 60 * 1000,
    client_id: str | None = SERVICE_NAME,
    allow_auto_create_topics: bool = True,
    config: Optional["ConfluentConfig"] = None,
    # publisher args
    acks: Literal[0, 1, -1, "all"] = EMPTY,
    compression_type: Literal["gzip", "snappy", "lz4", "zstd"] | None = None,
    partitioner: str
    | Callable[
        [bytes, list[Partition], list[Partition]],
        Partition,
    ] = "consistent_random",
    max_request_size: int = 1024 * 1024,
    linger_ms: int = 0,
    enable_idempotence: bool = False,
    transactional_id: str | None = None,
    transaction_timeout_ms: int = 60 * 1000,
    # broker base args
    graceful_timeout: float | None = 15.0,
    decoder: Optional["CustomCallable"] = None,
    parser: Optional["CustomCallable"] = None,
    middlewares: Sequence["BrokerMiddleware[Any, Any]"] = (),
    # Specification args
    specification: Optional["SpecificationFactory"] = None,
    security: Optional["BaseSecurity"] = None,
    specification_url: str | None = None,
    protocol: str | None = None,
    protocol_version: str = "auto",
    description: str | None = None,
    specification_tags: Iterable[Union["Tag", "TagDict"]] = (),
    # logging args
    logger: Optional["LoggerProto"] = EMPTY,
    log_level: int = logging.INFO,
    # StreamRouter options
    setup_state: bool = True,
    context: ContextRepo | None = None,
    schema_url: str | None = "/asyncapi",
    # FastAPI args
    prefix: str = "",
    tags: list[Union[str, "Enum"]] | None = None,
    dependencies: Sequence["params.Depends"] | None = None,
    default_response_class: type["Response"] = Default(JSONResponse),
    responses: dict[int | str, dict[str, Any]] | None = None,
    callbacks: list[BaseRoute] | None = None,
    routes: list[BaseRoute] | None = None,
    redirect_slashes: bool = True,
    default: Optional["ASGIApp"] = None,
    dependency_overrides_provider: Any | None = None,
    route_class: type["APIRoute"] = APIRoute,
    on_startup: Sequence[Callable[[], Any]] | None = None,
    on_shutdown: Sequence[Callable[[], Any]] | None = None,
    lifespan: Optional["Lifespan[Any]"] = None,
    deprecated: bool | None = None,
    include_in_schema: bool = True,
    generate_unique_id_function: Callable[["APIRoute"], str] = Default(
        generate_unique_id,
    ),
) -> None:
    """Initialize the KafkaRouter.

    Args:
        bootstrap_servers: A `host[:port]` string (or list of `host[:port]` strings) that the consumer should contact to bootstrap
            initial cluster metadata.

            This does not have to be the full node list.
            It just needs to have at least one broker that will respond to a
            Metadata API Request. Default port is 9092.
        request_timeout_ms: Client request timeout in milliseconds.
        retry_backoff_ms: Milliseconds to backoff when retrying on errors.
        metadata_max_age_ms: The period of time in milliseconds after
            which we force a refresh of metadata even if we haven't seen any
            partition leadership changes to proactively discover any new
            brokers or partitions.
        connections_max_idle_ms: Close idle connections after the number
            of milliseconds specified by this config. Specifying `None` will
            disable idle checks.
        client_id: A name for this client. This string is passed in
            each request to servers and can be used to identify specific
            server-side log entries that correspond to this client. Also
            submitted to :class:`~.consumer.group_coordinator.GroupCoordinator`
            for logging with respect to consumer group administration.
        allow_auto_create_topics: Allow automatic topic creation on the broker when subscribing to or assigning non-existent topics.
        config: Extra configuration for the confluent-kafka-python
            producer/consumer. See `confluent_kafka.Config <https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#kafka-client-configuration>`_.
        acks: One of ``0``, ``1``, ``all``. The number of acknowledgments
            the producer requires the leader to have received before considering a
            request complete. This controls the durability of records that are
            sent. The following settings are common:

            * ``0``: Producer will not wait for any acknowledgment from the server
              at all. The message will immediately be added to the socket
              buffer and considered sent. No guarantee can be made that the
              server has received the record in this case, and the retries
              configuration will not take effect (as the client won't
              generally know of any failures). The offset given back for each
              record will always be set to -1.
            * ``1``: The broker leader will write the record to its local log but
              will respond without awaiting full acknowledgement from all
              followers. In this case should the leader fail immediately
              after acknowledging the record but before the followers have
              replicated it then the record will be lost.
            * ``all``: The broker leader will wait for the full set of in-sync
              replicas to acknowledge the record. This guarantees that the
              record will not be lost as long as at least one in-sync replica
              remains alive. This is the strongest available guarantee.

            If unset, defaults to ``acks=1``. If `enable_idempotence` is
            :data:`True` defaults to ``acks=all``.
        compression_type: The compression type for all data generated bythe producer.
            Compression is of full batches of data, so the efficacy of batching
            will also impact the compression ratio (more batching means better
            compression).
        partitioner: Callable used to determine which partition
            each message is assigned to. Called (after key serialization):
            ``partitioner(key_bytes, all_partitions, available_partitions)``.
            The default partitioner implementation hashes each non-None key
            using the same murmur2 algorithm as the Java client so that
            messages with the same key are assigned to the same partition.
            When a key is :data:`None`, the message is delivered to a random partition
            (filtered to partitions with available leaders only, if possible).
        max_request_size: The maximum size of a request. This is also
            effectively a cap on the maximum record size. Note that the server
            has its own cap on record size which may be different from this.
            This setting will limit the number of record batches the producer
            will send in a single request to avoid sending huge requests.
        linger_ms: The producer groups together any records that arrive
            in between request transmissions into a single batched request.
            Normally this occurs only under load when records arrive faster
            than they can be sent out. However in some circumstances the client
            may want to reduce the number of requests even under moderate load.
            This setting accomplishes this by adding a small amount of
            artificial delay; that is, if first request is processed faster,
            than `linger_ms`, producer will wait ``linger_ms - process_time``.
        enable_idempotence: When set to `True`, the producer will
            ensure that exactly one copy of each message is written in the
            stream. If `False`, producer retries due to broker failures,
            etc., may write duplicates of the retried message in the stream.
            Note that enabling idempotence acks to set to ``all``. If it is not
            explicitly set by the user it will be chosen.
        transactional_id: Transactional ID for the producer.
        transaction_timeout_ms: Transaction timeout in milliseconds.
        graceful_timeout: Graceful shutdown timeout. Broker waits for all running subscribers completion before shut down.
        decoder: Custom decoder object.
        parser: Custom parser object.
        middlewares: Middlewares to apply to all broker publishers/subscribers.
        specification: Optional specification factory to generate the AsyncAPI schema for the broker.
            If provided, this will be used to customize or override the default schema generation.
        security: Security options to connect broker and generate Specification server security information.
        specification_url: Specification hardcoded server addresses. Use `servers` if not specified.
        protocol: Specification server protocol.
        protocol_version: Specification server protocol version.
        description: Specification server description.
        specification_tags: Specification server tags.
        context: faststream.ContextRepo object to store application injections.
        logger: User specified logger to pass into Context and log service messages.
        log_level: Service messages log level.
        setup_state: Whether to add broker to app scope in lifespan.
            You should disable this option at old ASGI servers.
        schema_url: The URL path where the AsyncAPI schema will be served (e.g. "/asyncapi").
            If set to None, the schema endpoint will not be added automatically.

        prefix: An optional path prefix for the router.
        tags: A list of tags to be applied to all the *path operations* in this
            router.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            FastAPI docs for Path Operation Configuration:
            https://fastapi.tiangolo.com/tutorial/path-operation-configuration/
        dependencies: A list of dependencies (using `Depends()`) to be applied to all the
            *path operations* in this router.

            Read more about it in the
            FastAPI docs for Bigger Applications - Multiple Files:
            https://fastapi.tiangolo.com/tutorial/bigger-applications/#include-an-apirouter-with-a-custom-prefix-tags-responses-and-dependencies
        default_response_class: The default response class to be used.

            Read more in the
            FastAPI docs for Custom Response - HTML, Stream, File, others:
            https://fastapi.tiangolo.com/advanced/custom-response/#default-response-class
        responses: Additional responses to be shown in OpenAPI.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            FastAPI docs for Additional Responses in OpenAPI:
            https://fastapi.tiangolo.com/advanced/additional-responses/

            And in the
            FastAPI docs for Bigger Applications:
            https://fastapi.tiangolo.com/tutorial/bigger-applications/#include-an-apirouter-with-a-custom-prefix-tags-responses-and-dependencies
        callbacks: OpenAPI callbacks that should apply to all *path operations* in this
            router.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            FastAPI docs for OpenAPI Callbacks:
            https://fastapi.tiangolo.com/advanced/openapi-callbacks/
        routes: **Note**: you probably shouldn't use this parameter, it is inherited
            from Starlette and supported for compatibility.

            ---

            A list of routes to serve incoming HTTP and WebSocket requests.

            You normally wouldn't use this parameter with FastAPI, it is inherited
            from Starlette and supported for compatibility.

            In FastAPI, you normally would use the *path operation methods*,
            like `router.get()`, `router.post()`, etc.
        redirect_slashes: Whether to detect and redirect slashes in URLs when the client doesn't
            use the same format.
        default: Default function handler for this router. Used to handle
            404 Not Found errors.
        dependency_overrides_provider: Only used internally by FastAPI to handle dependency overrides.

            You shouldn't need to use it. It normally points to the `FastAPI` app
            object.
        route_class: Custom route (*path operation*) class to be used by this router.

            Read more about it in the
            FastAPI docs for Custom Request and APIRoute class:
            https://fastapi.tiangolo.com/how-to/custom-request-and-route/#custom-apiroute-class-in-a-router
        on_startup: A list of startup event handler functions.

            You should instead use the `lifespan` handlers.

            Read more in the FastAPI docs for `lifespan`:
            https://fastapi.tiangolo.com/advanced/events/
        on_shutdown: A list of shutdown event handler functions.

            You should instead use the `lifespan` handlers.

            Read more in the
            FastAPI docs for `lifespan`:
            https://fastapi.tiangolo.com/advanced/events/
        lifespan: A `Lifespan` context manager handler. This replaces `startup` and
            `shutdown` functions with a single context manager.

            Read more in the
            FastAPI docs for `lifespan`:
            https://fastapi.tiangolo.com/advanced/events/
        deprecated: Mark all *path operations* in this router as deprecated.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            FastAPI docs for Path Operation Configuration:
            https://fastapi.tiangolo.com/tutorial/path-operation-configuration/
        include_in_schema: To include (or not) all the *path operations* in this router in the
            generated OpenAPI.

            This affects the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            FastAPI docs for Query Parameters and String Validations:
            https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi
        generate_unique_id_function: Customize the function used to generate unique IDs for the *path
            operations* shown in the generated OpenAPI.

            This is particularly useful when automatically generating clients or
            SDKs for your API.

            Read more about it in the
            FastAPI docs about how to Generate Clients:
            https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function
    """
    super().__init__(
        bootstrap_servers=bootstrap_servers,
        client_id=client_id,
        request_timeout_ms=request_timeout_ms,
        retry_backoff_ms=retry_backoff_ms,
        metadata_max_age_ms=metadata_max_age_ms,
        connections_max_idle_ms=connections_max_idle_ms,
        allow_auto_create_topics=allow_auto_create_topics,
        acks=acks,
        config=config,
        compression_type=compression_type,
        partitioner=partitioner,
        max_request_size=max_request_size,
        linger_ms=linger_ms,
        enable_idempotence=enable_idempotence,
        transactional_id=transactional_id,
        transaction_timeout_ms=transaction_timeout_ms,
        # broker args
        graceful_timeout=graceful_timeout,
        decoder=decoder,
        parser=parser,
        middlewares=middlewares,
        schema_url=schema_url,
        setup_state=setup_state,
        context=context,
        # logger options
        logger=logger,
        log_level=log_level,
        # Specification options
        security=security,
        protocol=protocol,
        description=description,
        protocol_version=protocol_version,
        specification_tags=specification_tags,
        specification_url=specification_url,
        # FastAPI kwargs
        prefix=prefix,
        tags=tags,
        dependencies=dependencies,
        default_response_class=default_response_class,
        responses=responses,
        callbacks=callbacks,
        routes=routes,
        redirect_slashes=redirect_slashes,
        default=default,
        dependency_overrides_provider=dependency_overrides_provider,
        route_class=route_class,
        on_startup=on_startup,
        on_shutdown=on_shutdown,
        deprecated=deprecated,
        include_in_schema=include_in_schema,
        lifespan=lifespan,
        generate_unique_id_function=generate_unique_id_function,
        specification=specification,
    )

context `property` #

context: ContextRepo

routes `instance-attribute` #

routes = [] if routes is None else list(routes)

redirect_slashes `instance-attribute` #

redirect_slashes = redirect_slashes

default `instance-attribute` #

default = self.not_found if default is None else default

lifespan_context `instance-attribute` #

lifespan_context = lifespan_context

middleware_stack `instance-attribute` #

middleware_stack = cls(
    self.middleware_stack, *args, **kwargs
)

on_startup `instance-attribute` #

on_startup: list[Callable[[], Any]] = (
    [] if on_startup is None else list(on_startup)
)

on_shutdown `instance-attribute` #

on_shutdown: list[Callable[[], Any]] = (
    [] if on_shutdown is None else list(on_shutdown)
)

prefix `instance-attribute` #

prefix = prefix

tags `instance-attribute` #

tags: list[str | Enum] = tags or []

dependencies `instance-attribute` #

dependencies = list(dependencies or [])

deprecated `instance-attribute` #

deprecated = deprecated

include_in_schema `instance-attribute` #

include_in_schema = include_in_schema

responses `instance-attribute` #

responses = responses or {}

callbacks `instance-attribute` #

callbacks = callbacks or []

dependency_overrides_provider `instance-attribute` #

dependency_overrides_provider = (
    dependency_overrides_provider
)

route_class `instance-attribute` #

route_class = route_class

default_response_class `instance-attribute` #

default_response_class = default_response_class

generate_unique_id_function `instance-attribute` #

generate_unique_id_function = generate_unique_id_function

strict_content_type `instance-attribute` #

strict_content_type = strict_content_type

docs_router `instance-attribute` #

docs_router: APIRouter | None

title `instance-attribute` #

title: str

description `instance-attribute` #

description: str

version `instance-attribute` #

version: str

license `instance-attribute` #

license: dict[str, Any] | None

contact `instance-attribute` #

contact: dict[str, Any] | None

setup_state `instance-attribute` #

setup_state = setup_state

fastapi_config `instance-attribute` #

fastapi_config = FastAPIConfig(
    dependency_overrides_provider=dependency_overrides_provider
)

broker_class `class-attribute` `instance-attribute` #

broker_class = KB

broker `instance-attribute` #

broker: KafkaBroker

set_broker #

set_broker(broker: BrokerUsecase[Any, Any]) -> None

Set already existed App object broker.

Useful then you create/init broker in on_startup hook.

Source code in faststream/_internal/application.py

@deprecated(
    "This method is deprecated and will be removed in 0.8.0 Use `add_broker` instead."
)
def set_broker(self, broker: "BrokerUsecase[Any, Any]") -> None:
    """Set already existed App object broker.

    Useful then you create/init broker in `on_startup` hook.
    """
    self.add_broker(broker)

add_broker #

add_broker(broker: BrokerUsecase[Any, Any]) -> None

Source code in faststream/_internal/application.py

def add_broker(self, broker: "BrokerUsecase[Any, Any]") -> None:
    if broker in self.brokers:
        msg = f"Broker {broker} is already added"
        raise SetupError(msg)

    self.brokers.append(broker)
    self.schema.add_broker(broker)
    broker._update_fd_config(self.config)

not_found `async` #

not_found(
    scope: Scope, receive: Receive, send: Send
) -> None

Source code in starlette/routing.py

async def not_found(self, scope: Scope, receive: Receive, send: Send) -> None:
    if scope["type"] == "websocket":
        websocket_close = WebSocketClose()
        await websocket_close(scope, receive, send)
        return

    # If we're running inside a starlette application then raise an
    # exception, so that the configurable exception handler can deal with
    # returning the response. For plain ASGI apps, just return the response.
    if "app" in scope:
        raise HTTPException(status_code=404)
    else:
        response = PlainTextResponse("Not Found", status_code=404)
    await response(scope, receive, send)

url_path_for #

url_path_for(name: str, /, **path_params: Any) -> URLPath

Source code in starlette/routing.py

def url_path_for(self, name: str, /, **path_params: Any) -> URLPath:
    for route in self.routes:
        try:
            return route.url_path_for(name, **path_params)
        except NoMatchFound:
            pass
    raise NoMatchFound(name, path_params)

lifespan `async` #

lifespan(
    scope: Scope, receive: Receive, send: Send
) -> None

Handle ASGI lifespan messages, which allows us to manage application startup and shutdown events.

Source code in starlette/routing.py

async def lifespan(self, scope: Scope, receive: Receive, send: Send) -> None:
    """
    Handle ASGI lifespan messages, which allows us to manage application
    startup and shutdown events.
    """
    started = False
    app: Any = scope.get("app")
    await receive()
    try:
        async with self.lifespan_context(app) as maybe_state:
            if maybe_state is not None:
                if "state" not in scope:
                    raise RuntimeError('The server does not support "state" in the lifespan scope.')
                scope["state"].update(maybe_state)
            await send({"type": "lifespan.startup.complete"})
            started = True
            await receive()
    except BaseException:
        exc_text = traceback.format_exc()
        if started:
            await send({"type": "lifespan.shutdown.failed", "message": exc_text})
        else:
            await send({"type": "lifespan.startup.failed", "message": exc_text})
        raise
    else:
        await send({"type": "lifespan.shutdown.complete"})

app `async` #

app(scope: Scope, receive: Receive, send: Send) -> None

Source code in fastapi/routing.py

async def app(self, scope: Scope, receive: Receive, send: Send) -> None:
    assert scope["type"] in ("http", "websocket", "lifespan")

    if "router" not in scope:
        scope["router"] = self

    if scope["type"] == "lifespan":
        await self.lifespan(scope, receive, send)
        return

    partial: tuple[BaseRoute, Scope] | None = None
    for route in self.routes:
        match, child_scope = route.matches(scope)
        if match == Match.FULL:
            scope.update(child_scope)
            await route.handle(scope, receive, send)
            return
        if match == Match.PARTIAL and partial is None:
            partial = (route, child_scope)

    if partial is not None:
        route, child_scope = partial
        scope.update(child_scope)
        await route.handle(scope, receive, send)
        return

    route_path = get_route_path(scope)
    if scope["type"] == "http" and self.redirect_slashes and route_path != "/":
        redirect_scope = dict(scope)
        if route_path.endswith("/"):
            redirect_scope["path"] = redirect_scope["path"].rstrip("/")
        else:
            redirect_scope["path"] = redirect_scope["path"] + "/"

        for route in self.routes:
            match, _ = route.matches(redirect_scope)
            if match != Match.NONE:
                redirect_url = URL(scope=redirect_scope)
                response = RedirectResponse(url=str(redirect_url))
                await response(scope, receive, send)
                return

    (
        low_priority_match,
        low_priority_scope,
        low_priority_route,
        low_priority_context,
    ) = self._match_low_priority(scope)
    if low_priority_match != Match.NONE and low_priority_route is not None:
        _update_scope(scope, low_priority_scope)
        if low_priority_context is not None:
            _get_fastapi_scope(scope)[_FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY] = (
                low_priority_context
            )
            original_route = low_priority_context.original_route
            if isinstance(original_route, APIRoute):
                scope["route"] = original_route
                await original_route.handle(scope, receive, send)
                return
        await low_priority_route.handle(scope, receive, send)
        return

    await self.default(scope, receive, send)

mount #

mount(
    path: str, app: ASGIApp, name: str | None = None
) -> None

Source code in starlette/routing.py

def mount(self, path: str, app: ASGIApp, name: str | None = None) -> None:  # pragma: no cover
    route = Mount(path, app=app, name=name)
    self.routes.append(route)

host #

host(
    host: str, app: ASGIApp, name: str | None = None
) -> None

Source code in starlette/routing.py

def host(self, host: str, app: ASGIApp, name: str | None = None) -> None:  # pragma: no cover
    route = Host(host, app=app, name=name)
    self.routes.append(route)

add_route #

add_route(
    path: str,
    endpoint: Callable[
        [Request], Awaitable[Response] | Response
    ],
    methods: Collection[str] | None = None,
    name: str | None = None,
    include_in_schema: bool = True,
) -> None

Source code in fastapi/routing.py

def add_route(
    self,
    path: str,
    endpoint: Callable[[Request], Awaitable[Response] | Response],
    methods: Collection[str] | None = None,
    name: str | None = None,
    include_in_schema: bool = True,
) -> None:
    super().add_route(
        path,
        endpoint,
        methods=methods,
        name=name,
        include_in_schema=include_in_schema,
    )
    self._mark_routes_changed()

add_websocket_route #

add_websocket_route(
    path: str,
    endpoint: Callable[[WebSocket], Awaitable[None]],
    name: str | None = None,
) -> None

Source code in fastapi/routing.py

def add_websocket_route(
    self,
    path: str,
    endpoint: Callable[[WebSocket], Awaitable[None]],
    name: str | None = None,
) -> None:
    super().add_websocket_route(path, endpoint, name=name)
    self._mark_routes_changed()

frontend #

frontend(
    path: str,
    *,
    directory: str | PathLike[str],
    fallback: Literal["auto", "index.html", "404.html"]
    | None = "auto",
    check_dir: bool = True,
) -> None

Serve a static frontend build as low-priority routes.

Use this for frontend tools that build static files into a directory, such as dist. FastAPI path operations are checked first, and the frontend files are checked only if no normal route matched.

A typical project could look like this:

.
├── pyproject.toml
├── app
│   ├── __init__.py
│   └── main.py
└── dist
    ├── index.html
    └── assets
        └── app.js

Then in app/main.py:

from fastapi import APIRouter, FastAPI

app = FastAPI()
router = APIRouter()
router.frontend("/", directory="dist")
app.include_router(router)

Source code in fastapi/routing.py

def frontend(
    self,
    path: Annotated[
        str,
        Doc(
            """
            The URL path prefix where the frontend build should be served.
            """
        ),
    ],
    *,
    directory: Annotated[
        str | os.PathLike[str],
        Doc(
            """
            The directory containing the static frontend build output.
            """
        ),
    ],
    fallback: Annotated[
        Literal["auto", "index.html", "404.html"] | None,
        Doc(
            """
            The fallback file behavior for missing frontend paths.
            """
        ),
    ] = "auto",
    check_dir: Annotated[
        bool,
        Doc(
            """
            Check that the frontend directory exists when the app is created.
            """
        ),
    ] = True,
) -> None:
    """
    Serve a static frontend build as low-priority routes.

    Use this for frontend tools that build static files into a directory,
    such as `dist`. **FastAPI** path operations are checked first, and
    the frontend files are checked only if no normal route matched.

    A typical project could look like this:

    ```text
    .
    ├── pyproject.toml
    ├── app
    │   ├── __init__.py
    │   └── main.py
    └── dist
        ├── index.html
        └── assets
            └── app.js
    ```

    Then in `app/main.py`:

    ```python
    from fastapi import APIRouter, FastAPI

    app = FastAPI()
    router = APIRouter()
    router.frontend("/", directory="dist")
    app.include_router(router)
    ```
    """
    normalized_path = _normalize_frontend_path(path)
    if self._frontend_routes is None:
        self._frontend_routes = _FrontendRouteGroup(
            dependencies=self.dependencies,
            dependency_overrides_provider=self.dependency_overrides_provider,
        )
        self._low_priority_routes.append(self._frontend_routes)
    self._frontend_routes.add_frontend_route(
        _join_frontend_paths(self.prefix, normalized_path),
        directory=directory,
        fallback=fallback,
        check_dir=check_dir,
    )
    self._mark_routes_changed()

handle `async` #

handle(scope: Scope, receive: Receive, send: Send) -> None

Source code in fastapi/routing.py

async def handle(self, scope: Scope, receive: Receive, send: Send) -> None:
    included_router = _get_scope_included_router(scope)
    if (
        isinstance(included_router, _IncludedRouter)
        and included_router.original_router is self
    ):
        await included_router._handle_selected(scope, receive, send)
        return
    await self.app(scope, receive, send)

matches #

matches(scope: Scope) -> tuple[Match, Scope]

Source code in fastapi/routing.py

def matches(self, scope: Scope) -> tuple[Match, Scope]:
    included_router = _get_scope_included_router(scope)
    if (
        isinstance(included_router, _IncludedRouter)
        and included_router.original_router is self
    ):
        match, child_scope, _, _ = included_router._match(scope)
        return match, child_scope
    return Match.NONE, {}

route #

route(
    path: str,
    methods: Collection[str] | None = None,
    name: str | None = None,
    include_in_schema: bool = True,
) -> Callable[[DecoratedCallable], DecoratedCallable]

Source code in fastapi/routing.py

def route(
    self,
    path: str,
    methods: Collection[str] | None = None,
    name: str | None = None,
    include_in_schema: bool = True,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
    def decorator(func: DecoratedCallable) -> DecoratedCallable:
        self.add_route(
            path,
            func,
            methods=methods,
            name=name,
            include_in_schema=include_in_schema,
        )
        return func

    return decorator

add_api_route #

add_api_route(
    path: str,
    endpoint: Callable[..., Any],
    *,
    response_model: Any = Default(None),
    status_code: int | None = None,
    tags: list[str | Enum] | None = None,
    dependencies: Sequence[Depends] | None = None,
    summary: str | None = None,
    description: str | None = None,
    response_description: str = "Successful Response",
    responses: dict[int | str, dict[str, Any]]
    | None = None,
    deprecated: bool | None = None,
    methods: set[str] | list[str] | None = None,
    operation_id: str | None = None,
    response_model_include: IncEx | None = None,
    response_model_exclude: IncEx | None = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
    include_in_schema: bool = True,
    response_class: type[Response]
    | DefaultPlaceholder = Default(JSONResponse),
    name: str | None = None,
    route_class_override: type[APIRoute] | None = None,
    callbacks: list[BaseRoute] | None = None,
    openapi_extra: dict[str, Any] | None = None,
    generate_unique_id_function: Callable[[APIRoute], str]
    | DefaultPlaceholder = Default(generate_unique_id),
    strict_content_type: bool
    | DefaultPlaceholder = Default(True),
) -> None

Source code in fastapi/routing.py

def add_api_route(
    self,
    path: str,
    endpoint: Callable[..., Any],
    *,
    response_model: Any = Default(None),
    status_code: int | None = None,
    tags: list[str | Enum] | None = None,
    dependencies: Sequence[params.Depends] | None = None,
    summary: str | None = None,
    description: str | None = None,
    response_description: str = "Successful Response",
    responses: dict[int | str, dict[str, Any]] | None = None,
    deprecated: bool | None = None,
    methods: set[str] | list[str] | None = None,
    operation_id: str | None = None,
    response_model_include: IncEx | None = None,
    response_model_exclude: IncEx | None = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
    include_in_schema: bool = True,
    response_class: type[Response] | DefaultPlaceholder = Default(JSONResponse),
    name: str | None = None,
    route_class_override: type[APIRoute] | None = None,
    callbacks: list[BaseRoute] | None = None,
    openapi_extra: dict[str, Any] | None = None,
    generate_unique_id_function: Callable[[APIRoute], str]
    | DefaultPlaceholder = Default(generate_unique_id),
    strict_content_type: bool | DefaultPlaceholder = Default(True),
) -> None:
    route_class = route_class_override or self.route_class
    responses = responses or {}
    combined_responses = {**self.responses, **responses}
    current_response_class = get_value_or_default(
        response_class, self.default_response_class
    )
    current_tags = self.tags.copy()
    if tags:
        current_tags.extend(tags)
    current_dependencies = self.dependencies.copy()
    if dependencies:
        current_dependencies.extend(dependencies)
    current_callbacks = self.callbacks.copy()
    if callbacks:
        current_callbacks.extend(callbacks)
    current_generate_unique_id = get_value_or_default(
        generate_unique_id_function, self.generate_unique_id_function
    )
    route = route_class(
        self.prefix + path,
        endpoint=endpoint,
        response_model=response_model,
        status_code=status_code,
        tags=current_tags,
        dependencies=current_dependencies,
        summary=summary,
        description=description,
        response_description=response_description,
        responses=combined_responses,
        deprecated=deprecated or self.deprecated,
        methods=methods,
        operation_id=operation_id,
        response_model_include=response_model_include,
        response_model_exclude=response_model_exclude,
        response_model_by_alias=response_model_by_alias,
        response_model_exclude_unset=response_model_exclude_unset,
        response_model_exclude_defaults=response_model_exclude_defaults,
        response_model_exclude_none=response_model_exclude_none,
        include_in_schema=include_in_schema and self.include_in_schema,
        response_class=current_response_class,
        name=name,
        dependency_overrides_provider=self.dependency_overrides_provider,
        callbacks=current_callbacks,
        openapi_extra=openapi_extra,
        generate_unique_id_function=current_generate_unique_id,
        strict_content_type=get_value_or_default(
            strict_content_type, self.strict_content_type
        ),
    )
    self.routes.append(route)
    self._mark_routes_changed()

api_route #

api_route(
    path: str,
    *,
    response_model: Any = Default(None),
    status_code: int | None = None,
    tags: list[str | Enum] | None = None,
    dependencies: Sequence[Depends] | None = None,
    summary: str | None = None,
    description: str | None = None,
    response_description: str = "Successful Response",
    responses: dict[int | str, dict[str, Any]]
    | None = None,
    deprecated: bool | None = None,
    methods: list[str] | None = None,
    operation_id: str | None = None,
    response_model_include: IncEx | None = None,
    response_model_exclude: IncEx | None = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
    include_in_schema: bool = True,
    response_class: type[Response] = Default(JSONResponse),
    name: str | None = None,
    callbacks: list[BaseRoute] | None = None,
    openapi_extra: dict[str, Any] | None = None,
    generate_unique_id_function: Callable[
        [APIRoute], str
    ] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]

Source code in fastapi/routing.py

def api_route(
    self,
    path: str,
    *,
    response_model: Any = Default(None),
    status_code: int | None = None,
    tags: list[str | Enum] | None = None,
    dependencies: Sequence[params.Depends] | None = None,
    summary: str | None = None,
    description: str | None = None,
    response_description: str = "Successful Response",
    responses: dict[int | str, dict[str, Any]] | None = None,
    deprecated: bool | None = None,
    methods: list[str] | None = None,
    operation_id: str | None = None,
    response_model_include: IncEx | None = None,
    response_model_exclude: IncEx | None = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
    include_in_schema: bool = True,
    response_class: type[Response] = Default(JSONResponse),
    name: str | None = None,
    callbacks: list[BaseRoute] | None = None,
    openapi_extra: dict[str, Any] | None = None,
    generate_unique_id_function: Callable[[APIRoute], str] = Default(
        generate_unique_id
    ),
) -> Callable[[DecoratedCallable], DecoratedCallable]:
    def decorator(func: DecoratedCallable) -> DecoratedCallable:
        self.add_api_route(
            path,
            func,
            response_model=response_model,
            status_code=status_code,
            tags=tags,
            dependencies=dependencies,
            summary=summary,
            description=description,
            response_description=response_description,
            responses=responses,
            deprecated=deprecated,
            methods=methods,
            operation_id=operation_id,
            response_model_include=response_model_include,
            response_model_exclude=response_model_exclude,
            response_model_by_alias=response_model_by_alias,
            response_model_exclude_unset=response_model_exclude_unset,
            response_model_exclude_defaults=response_model_exclude_defaults,
            response_model_exclude_none=response_model_exclude_none,
            include_in_schema=include_in_schema,
            response_class=response_class,
            name=name,
            callbacks=callbacks,
            openapi_extra=openapi_extra,
            generate_unique_id_function=generate_unique_id_function,
        )
        return func

    return decorator

add_api_websocket_route #

add_api_websocket_route(
    path: str,
    endpoint: Callable[..., Any],
    name: str | None = None,
    *,
    dependencies: Sequence[Depends] | None = None,
) -> None

Source code in fastapi/routing.py

def add_api_websocket_route(
    self,
    path: str,
    endpoint: Callable[..., Any],
    name: str | None = None,
    *,
    dependencies: Sequence[params.Depends] | None = None,
) -> None:
    current_dependencies = self.dependencies.copy()
    if dependencies:
        current_dependencies.extend(dependencies)

    route = APIWebSocketRoute(
        self.prefix + path,
        endpoint=endpoint,
        name=name,
        dependencies=current_dependencies,
        dependency_overrides_provider=self.dependency_overrides_provider,
    )
    self.routes.append(route)
    self._mark_routes_changed()

websocket #

websocket(
    path: str,
    name: str | None = None,
    *,
    dependencies: Sequence[Depends] | None = None,
) -> Callable[[DecoratedCallable], DecoratedCallable]

Decorate a WebSocket function.

Read more about it in the FastAPI docs for WebSockets.

Example

Example#

from fastapi import APIRouter, FastAPI, WebSocket

app = FastAPI()
router = APIRouter()

@router.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message text was: {data}")

app.include_router(router)

Source code in fastapi/routing.py

def websocket(
    self,
    path: Annotated[
        str,
        Doc(
            """
            WebSocket path.
            """
        ),
    ],
    name: Annotated[
        str | None,
        Doc(
            """
            A name for the WebSocket. Only used internally.
            """
        ),
    ] = None,
    *,
    dependencies: Annotated[
        Sequence[params.Depends] | None,
        Doc(
            """
            A list of dependencies (using `Depends()`) to be used for this
            WebSocket.

            Read more about it in the
            [FastAPI docs for WebSockets](https://fastapi.tiangolo.com/advanced/websockets/).
            """
        ),
    ] = None,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
    """
    Decorate a WebSocket function.

    Read more about it in the
    [FastAPI docs for WebSockets](https://fastapi.tiangolo.com/advanced/websockets/).

    **Example**

    ## Example

    ```python
    from fastapi import APIRouter, FastAPI, WebSocket

    app = FastAPI()
    router = APIRouter()

    @router.websocket("/ws")
    async def websocket_endpoint(websocket: WebSocket):
        await websocket.accept()
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message text was: {data}")

    app.include_router(router)
    ```
    """

    def decorator(func: DecoratedCallable) -> DecoratedCallable:
        self.add_api_websocket_route(
            path, func, name=name, dependencies=dependencies
        )
        return func

    return decorator

websocket_route #

websocket_route(
    path: str, name: str | None = None
) -> Callable[[DecoratedCallable], DecoratedCallable]

Source code in fastapi/routing.py

def websocket_route(
    self, path: str, name: str | None = None
) -> Callable[[DecoratedCallable], DecoratedCallable]:
    def decorator(func: DecoratedCallable) -> DecoratedCallable:
        self.add_websocket_route(path, func, name=name)
        return func

    return decorator

include_router #

include_router(
    router: Union[
        StreamRouter[MsgType], BrokerRouter[MsgType]
    ],
    *,
    prefix: str = "",
    tags: list[str | Enum] | None = None,
    dependencies: Sequence[Depends] | None = None,
    default_response_class: type[Response] = Default(
        JSONResponse
    ),
    responses: dict[int | str, dict[str, Any]]
    | None = None,
    callbacks: list[BaseRoute] | None = None,
    deprecated: bool | None = None,
    include_in_schema: bool = True,
    generate_unique_id_function: Callable[
        [APIRoute], str
    ] = Default(generate_unique_id),
) -> None

Includes a router in the API.

Source code in faststream/_internal/fastapi/router.py

def include_router(  # type: ignore[override]
    self,
    router: Union["StreamRouter[MsgType]", "BrokerRouter[MsgType]"],
    *,
    prefix: str = "",
    tags: list[str | Enum] | None = None,
    dependencies: Sequence["params.Depends"] | None = None,
    default_response_class: type[Response] = Default(JSONResponse),
    responses: dict[int | str, dict[str, Any]] | None = None,
    callbacks: list["BaseRoute"] | None = None,
    deprecated: bool | None = None,
    include_in_schema: bool = True,
    generate_unique_id_function: Callable[["APIRoute"], str] = Default(
        generate_unique_id,
    ),
) -> None:
    """Includes a router in the API."""
    if isinstance(router, BrokerRouter):
        for sub in router.subscribers:
            sub._call_decorators = (
                self._subscriber_compatibility_wrapper(),
                *sub._call_decorators,
            )

        self.broker.include_router(router)
        return

    msg = (
        "Including a StreamRouter into another StreamRouter is not supported "
        "and may cause subtle context issues (e.g. message dependencies "
        "returning EmptyPlaceholder). "
        "Use a regular broker router (e.g. KafkaRouter, RabbitRouter, etc.) "
        "for grouping subscribers and include that into the StreamRouter instead. "
        "See: https://faststream.ag2.ai/latest/getting-started/integrations/fastapi/#multiple-routers"
    )
    raise TypeError(msg)

get #

get(
    path: str,
    *,
    response_model: Any = Default(None),
    status_code: int | None = None,
    tags: list[str | Enum] | None = None,
    dependencies: Sequence[Depends] | None = None,
    summary: str | None = None,
    description: str | None = None,
    response_description: str = "Successful Response",
    responses: dict[int | str, dict[str, Any]]
    | None = None,
    deprecated: bool | None = None,
    operation_id: str | None = None,
    response_model_include: IncEx | None = None,
    response_model_exclude: IncEx | None = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
    include_in_schema: bool = True,
    response_class: type[Response] = Default(JSONResponse),
    name: str | None = None,
    callbacks: list[BaseRoute] | None = None,
    openapi_extra: dict[str, Any] | None = None,
    generate_unique_id_function: Callable[
        [APIRoute], str
    ] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]

Add a path operation using an HTTP GET operation.

Example#

from fastapi import APIRouter, FastAPI

app = FastAPI()
router = APIRouter()

@router.get("/items/")
def read_items():
    return [{"name": "Empanada"}, {"name": "Arepa"}]

app.include_router(router)

Source code in fastapi/routing.py

def get(
    self,
    path: Annotated[
        str,
        Doc(
            """
            The URL path to be used for this *path operation*.

            For example, in `http://example.com/items`, the path is `/items`.
            """
        ),
    ],
    *,
    response_model: Annotated[
        Any,
        Doc(
            """
            The type to use for the response.

            It could be any valid Pydantic *field* type. So, it doesn't have to
            be a Pydantic model, it could be other things, like a `list`, `dict`,
            etc.

            It will be used for:

            * Documentation: the generated OpenAPI (and the UI at `/docs`) will
                show it as the response (JSON Schema).
            * Serialization: you could return an arbitrary object and the
                `response_model` would be used to serialize that object into the
                corresponding JSON.
            * Filtering: the JSON sent to the client will only contain the data
                (fields) defined in the `response_model`. If you returned an object
                that contains an attribute `password` but the `response_model` does
                not include that field, the JSON sent to the client would not have
                that `password`.
            * Validation: whatever you return will be serialized with the
                `response_model`, converting any data as necessary to generate the
                corresponding JSON. But if the data in the object returned is not
                valid, that would mean a violation of the contract with the client,
                so it's an error from the API developer. So, FastAPI will raise an
                error and return a 500 error code (Internal Server Error).

            Read more about it in the
            [FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/).
            """
        ),
    ] = Default(None),
    status_code: Annotated[
        int | None,
        Doc(
            """
            The default status code to be used for the response.

            You could override the status code by returning a response directly.

            Read more about it in the
            [FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/).
            """
        ),
    ] = None,
    tags: Annotated[
        list[str | Enum] | None,
        Doc(
            """
            A list of tags to be applied to the *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags).
            """
        ),
    ] = None,
    dependencies: Annotated[
        Sequence[params.Depends] | None,
        Doc(
            """
            A list of dependencies (using `Depends()`) to be applied to the
            *path operation*.

            Read more about it in the
            [FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/).
            """
        ),
    ] = None,
    summary: Annotated[
        str | None,
        Doc(
            """
            A summary for the *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    description: Annotated[
        str | None,
        Doc(
            """
            A description for the *path operation*.

            If not provided, it will be extracted automatically from the docstring
            of the *path operation function*.

            It can contain Markdown.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    response_description: Annotated[
        str,
        Doc(
            """
            The description for the default response.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = "Successful Response",
    responses: Annotated[
        dict[int | str, dict[str, Any]] | None,
        Doc(
            """
            Additional responses that could be returned by this *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = None,
    deprecated: Annotated[
        bool | None,
        Doc(
            """
            Mark this *path operation* as deprecated.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = None,
    operation_id: Annotated[
        str | None,
        Doc(
            """
            Custom operation ID to be used by this *path operation*.

            By default, it is generated automatically.

            If you provide a custom operation ID, you need to make sure it is
            unique for the whole API.

            You can customize the
            operation ID generation with the parameter
            `generate_unique_id_function` in the `FastAPI` class.

            Read more about it in the
            [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
            """
        ),
    ] = None,
    response_model_include: Annotated[
        IncEx | None,
        Doc(
            """
            Configuration passed to Pydantic to include only certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = None,
    response_model_exclude: Annotated[
        IncEx | None,
        Doc(
            """
            Configuration passed to Pydantic to exclude certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = None,
    response_model_by_alias: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response model
            should be serialized by alias when an alias is used.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = True,
    response_model_exclude_unset: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that were not set and
            have their default values. This is different from
            `response_model_exclude_defaults` in that if the fields are set,
            they will be included in the response, even if the value is the same
            as the default.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
            """
        ),
    ] = False,
    response_model_exclude_defaults: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that have the same value
            as the default. This is different from `response_model_exclude_unset`
            in that if the fields are set but contain the same default values,
            they will be excluded from the response.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
            """
        ),
    ] = False,
    response_model_exclude_none: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data should
            exclude fields set to `None`.

            This is much simpler (less smart) than `response_model_exclude_unset`
            and `response_model_exclude_defaults`. You probably want to use one of
            those two instead of this one, as those allow returning `None` values
            when it makes sense.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none).
            """
        ),
    ] = False,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Include this *path operation* in the generated OpenAPI schema.

            This affects the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi).
            """
        ),
    ] = True,
    response_class: Annotated[
        type[Response],
        Doc(
            """
            Response class to be used for this *path operation*.

            This will not be used if you return a response directly.

            Read more about it in the
            [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse).
            """
        ),
    ] = Default(JSONResponse),
    name: Annotated[
        str | None,
        Doc(
            """
            Name for this *path operation*. Only used internally.
            """
        ),
    ] = None,
    callbacks: Annotated[
        list[BaseRoute] | None,
        Doc(
            """
            List of *path operations* that will be used as OpenAPI callbacks.

            This is only for OpenAPI documentation, the callbacks won't be used
            directly.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/).
            """
        ),
    ] = None,
    openapi_extra: Annotated[
        dict[str, Any] | None,
        Doc(
            """
            Extra metadata to be included in the OpenAPI schema for this *path
            operation*.

            Read more about it in the
            [FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema).
            """
        ),
    ] = None,
    generate_unique_id_function: Annotated[
        Callable[[APIRoute], str],
        Doc(
            """
            Customize the function used to generate unique IDs for the *path
            operations* shown in the generated OpenAPI.

            This is particularly useful when automatically generating clients or
            SDKs for your API.

            Read more about it in the
            [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
            """
        ),
    ] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]:
    """
    Add a *path operation* using an HTTP GET operation.

    ## Example

    ```python
    from fastapi import APIRouter, FastAPI

    app = FastAPI()
    router = APIRouter()

    @router.get("/items/")
    def read_items():
        return [{"name": "Empanada"}, {"name": "Arepa"}]

    app.include_router(router)
    ```
    """
    return self.api_route(
        path=path,
        response_model=response_model,
        status_code=status_code,
        tags=tags,
        dependencies=dependencies,
        summary=summary,
        description=description,
        response_description=response_description,
        responses=responses,
        deprecated=deprecated,
        methods=["GET"],
        operation_id=operation_id,
        response_model_include=response_model_include,
        response_model_exclude=response_model_exclude,
        response_model_by_alias=response_model_by_alias,
        response_model_exclude_unset=response_model_exclude_unset,
        response_model_exclude_defaults=response_model_exclude_defaults,
        response_model_exclude_none=response_model_exclude_none,
        include_in_schema=include_in_schema,
        response_class=response_class,
        name=name,
        callbacks=callbacks,
        openapi_extra=openapi_extra,
        generate_unique_id_function=generate_unique_id_function,
    )

put #

put(
    path: str,
    *,
    response_model: Any = Default(None),
    status_code: int | None = None,
    tags: list[str | Enum] | None = None,
    dependencies: Sequence[Depends] | None = None,
    summary: str | None = None,
    description: str | None = None,
    response_description: str = "Successful Response",
    responses: dict[int | str, dict[str, Any]]
    | None = None,
    deprecated: bool | None = None,
    operation_id: str | None = None,
    response_model_include: IncEx | None = None,
    response_model_exclude: IncEx | None = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
    include_in_schema: bool = True,
    response_class: type[Response] = Default(JSONResponse),
    name: str | None = None,
    callbacks: list[BaseRoute] | None = None,
    openapi_extra: dict[str, Any] | None = None,
    generate_unique_id_function: Callable[
        [APIRoute], str
    ] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]

Add a path operation using an HTTP PUT operation.

Example#

from fastapi import APIRouter, FastAPI
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str | None = None

app = FastAPI()
router = APIRouter()

@router.put("/items/{item_id}")
def replace_item(item_id: str, item: Item):
    return {"message": "Item replaced", "id": item_id}

app.include_router(router)

Source code in fastapi/routing.py

def put(
    self,
    path: Annotated[
        str,
        Doc(
            """
            The URL path to be used for this *path operation*.

            For example, in `http://example.com/items`, the path is `/items`.
            """
        ),
    ],
    *,
    response_model: Annotated[
        Any,
        Doc(
            """
            The type to use for the response.

            It could be any valid Pydantic *field* type. So, it doesn't have to
            be a Pydantic model, it could be other things, like a `list`, `dict`,
            etc.

            It will be used for:

            * Documentation: the generated OpenAPI (and the UI at `/docs`) will
                show it as the response (JSON Schema).
            * Serialization: you could return an arbitrary object and the
                `response_model` would be used to serialize that object into the
                corresponding JSON.
            * Filtering: the JSON sent to the client will only contain the data
                (fields) defined in the `response_model`. If you returned an object
                that contains an attribute `password` but the `response_model` does
                not include that field, the JSON sent to the client would not have
                that `password`.
            * Validation: whatever you return will be serialized with the
                `response_model`, converting any data as necessary to generate the
                corresponding JSON. But if the data in the object returned is not
                valid, that would mean a violation of the contract with the client,
                so it's an error from the API developer. So, FastAPI will raise an
                error and return a 500 error code (Internal Server Error).

            Read more about it in the
            [FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/).
            """
        ),
    ] = Default(None),
    status_code: Annotated[
        int | None,
        Doc(
            """
            The default status code to be used for the response.

            You could override the status code by returning a response directly.

            Read more about it in the
            [FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/).
            """
        ),
    ] = None,
    tags: Annotated[
        list[str | Enum] | None,
        Doc(
            """
            A list of tags to be applied to the *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags).
            """
        ),
    ] = None,
    dependencies: Annotated[
        Sequence[params.Depends] | None,
        Doc(
            """
            A list of dependencies (using `Depends()`) to be applied to the
            *path operation*.

            Read more about it in the
            [FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/).
            """
        ),
    ] = None,
    summary: Annotated[
        str | None,
        Doc(
            """
            A summary for the *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    description: Annotated[
        str | None,
        Doc(
            """
            A description for the *path operation*.

            If not provided, it will be extracted automatically from the docstring
            of the *path operation function*.

            It can contain Markdown.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    response_description: Annotated[
        str,
        Doc(
            """
            The description for the default response.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = "Successful Response",
    responses: Annotated[
        dict[int | str, dict[str, Any]] | None,
        Doc(
            """
            Additional responses that could be returned by this *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = None,
    deprecated: Annotated[
        bool | None,
        Doc(
            """
            Mark this *path operation* as deprecated.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = None,
    operation_id: Annotated[
        str | None,
        Doc(
            """
            Custom operation ID to be used by this *path operation*.

            By default, it is generated automatically.

            If you provide a custom operation ID, you need to make sure it is
            unique for the whole API.

            You can customize the
            operation ID generation with the parameter
            `generate_unique_id_function` in the `FastAPI` class.

            Read more about it in the
            [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
            """
        ),
    ] = None,
    response_model_include: Annotated[
        IncEx | None,
        Doc(
            """
            Configuration passed to Pydantic to include only certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = None,
    response_model_exclude: Annotated[
        IncEx | None,
        Doc(
            """
            Configuration passed to Pydantic to exclude certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = None,
    response_model_by_alias: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response model
            should be serialized by alias when an alias is used.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = True,
    response_model_exclude_unset: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that were not set and
            have their default values. This is different from
            `response_model_exclude_defaults` in that if the fields are set,
            they will be included in the response, even if the value is the same
            as the default.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
            """
        ),
    ] = False,
    response_model_exclude_defaults: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that have the same value
            as the default. This is different from `response_model_exclude_unset`
            in that if the fields are set but contain the same default values,
            they will be excluded from the response.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
            """
        ),
    ] = False,
    response_model_exclude_none: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data should
            exclude fields set to `None`.

            This is much simpler (less smart) than `response_model_exclude_unset`
            and `response_model_exclude_defaults`. You probably want to use one of
            those two instead of this one, as those allow returning `None` values
            when it makes sense.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none).
            """
        ),
    ] = False,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Include this *path operation* in the generated OpenAPI schema.

            This affects the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi).
            """
        ),
    ] = True,
    response_class: Annotated[
        type[Response],
        Doc(
            """
            Response class to be used for this *path operation*.

            This will not be used if you return a response directly.

            Read more about it in the
            [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse).
            """
        ),
    ] = Default(JSONResponse),
    name: Annotated[
        str | None,
        Doc(
            """
            Name for this *path operation*. Only used internally.
            """
        ),
    ] = None,
    callbacks: Annotated[
        list[BaseRoute] | None,
        Doc(
            """
            List of *path operations* that will be used as OpenAPI callbacks.

            This is only for OpenAPI documentation, the callbacks won't be used
            directly.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/).
            """
        ),
    ] = None,
    openapi_extra: Annotated[
        dict[str, Any] | None,
        Doc(
            """
            Extra metadata to be included in the OpenAPI schema for this *path
            operation*.

            Read more about it in the
            [FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema).
            """
        ),
    ] = None,
    generate_unique_id_function: Annotated[
        Callable[[APIRoute], str],
        Doc(
            """
            Customize the function used to generate unique IDs for the *path
            operations* shown in the generated OpenAPI.

            This is particularly useful when automatically generating clients or
            SDKs for your API.

            Read more about it in the
            [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
            """
        ),
    ] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]:
    """
    Add a *path operation* using an HTTP PUT operation.

    ## Example

    ```python
    from fastapi import APIRouter, FastAPI
    from pydantic import BaseModel

    class Item(BaseModel):
        name: str
        description: str | None = None

    app = FastAPI()
    router = APIRouter()

    @router.put("/items/{item_id}")
    def replace_item(item_id: str, item: Item):
        return {"message": "Item replaced", "id": item_id}

    app.include_router(router)
    ```
    """
    return self.api_route(
        path=path,
        response_model=response_model,
        status_code=status_code,
        tags=tags,
        dependencies=dependencies,
        summary=summary,
        description=description,
        response_description=response_description,
        responses=responses,
        deprecated=deprecated,
        methods=["PUT"],
        operation_id=operation_id,
        response_model_include=response_model_include,
        response_model_exclude=response_model_exclude,
        response_model_by_alias=response_model_by_alias,
        response_model_exclude_unset=response_model_exclude_unset,
        response_model_exclude_defaults=response_model_exclude_defaults,
        response_model_exclude_none=response_model_exclude_none,
        include_in_schema=include_in_schema,
        response_class=response_class,
        name=name,
        callbacks=callbacks,
        openapi_extra=openapi_extra,
        generate_unique_id_function=generate_unique_id_function,
    )

post #

post(
    path: str,
    *,
    response_model: Any = Default(None),
    status_code: int | None = None,
    tags: list[str | Enum] | None = None,
    dependencies: Sequence[Depends] | None = None,
    summary: str | None = None,
    description: str | None = None,
    response_description: str = "Successful Response",
    responses: dict[int | str, dict[str, Any]]
    | None = None,
    deprecated: bool | None = None,
    operation_id: str | None = None,
    response_model_include: IncEx | None = None,
    response_model_exclude: IncEx | None = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
    include_in_schema: bool = True,
    response_class: type[Response] = Default(JSONResponse),
    name: str | None = None,
    callbacks: list[BaseRoute] | None = None,
    openapi_extra: dict[str, Any] | None = None,
    generate_unique_id_function: Callable[
        [APIRoute], str
    ] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]

Add a path operation using an HTTP POST operation.

Example#

from fastapi import APIRouter, FastAPI
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str | None = None

app = FastAPI()
router = APIRouter()

@router.post("/items/")
def create_item(item: Item):
    return {"message": "Item created"}

app.include_router(router)

Source code in fastapi/routing.py

def post(
    self,
    path: Annotated[
        str,
        Doc(
            """
            The URL path to be used for this *path operation*.

            For example, in `http://example.com/items`, the path is `/items`.
            """
        ),
    ],
    *,
    response_model: Annotated[
        Any,
        Doc(
            """
            The type to use for the response.

            It could be any valid Pydantic *field* type. So, it doesn't have to
            be a Pydantic model, it could be other things, like a `list`, `dict`,
            etc.

            It will be used for:

            * Documentation: the generated OpenAPI (and the UI at `/docs`) will
                show it as the response (JSON Schema).
            * Serialization: you could return an arbitrary object and the
                `response_model` would be used to serialize that object into the
                corresponding JSON.
            * Filtering: the JSON sent to the client will only contain the data
                (fields) defined in the `response_model`. If you returned an object
                that contains an attribute `password` but the `response_model` does
                not include that field, the JSON sent to the client would not have
                that `password`.
            * Validation: whatever you return will be serialized with the
                `response_model`, converting any data as necessary to generate the
                corresponding JSON. But if the data in the object returned is not
                valid, that would mean a violation of the contract with the client,
                so it's an error from the API developer. So, FastAPI will raise an
                error and return a 500 error code (Internal Server Error).

            Read more about it in the
            [FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/).
            """
        ),
    ] = Default(None),
    status_code: Annotated[
        int | None,
        Doc(
            """
            The default status code to be used for the response.

            You could override the status code by returning a response directly.

            Read more about it in the
            [FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/).
            """
        ),
    ] = None,
    tags: Annotated[
        list[str | Enum] | None,
        Doc(
            """
            A list of tags to be applied to the *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags).
            """
        ),
    ] = None,
    dependencies: Annotated[
        Sequence[params.Depends] | None,
        Doc(
            """
            A list of dependencies (using `Depends()`) to be applied to the
            *path operation*.

            Read more about it in the
            [FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/).
            """
        ),
    ] = None,
    summary: Annotated[
        str | None,
        Doc(
            """
            A summary for the *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    description: Annotated[
        str | None,
        Doc(
            """
            A description for the *path operation*.

            If not provided, it will be extracted automatically from the docstring
            of the *path operation function*.

            It can contain Markdown.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    response_description: Annotated[
        str,
        Doc(
            """
            The description for the default response.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = "Successful Response",
    responses: Annotated[
        dict[int | str, dict[str, Any]] | None,
        Doc(
            """
            Additional responses that could be returned by this *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = None,
    deprecated: Annotated[
        bool | None,
        Doc(
            """
            Mark this *path operation* as deprecated.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = None,
    operation_id: Annotated[
        str | None,
        Doc(
            """
            Custom operation ID to be used by this *path operation*.

            By default, it is generated automatically.

            If you provide a custom operation ID, you need to make sure it is
            unique for the whole API.

            You can customize the
            operation ID generation with the parameter
            `generate_unique_id_function` in the `FastAPI` class.

            Read more about it in the
            [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
            """
        ),
    ] = None,
    response_model_include: Annotated[
        IncEx | None,
        Doc(
            """
            Configuration passed to Pydantic to include only certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = None,
    response_model_exclude: Annotated[
        IncEx | None,
        Doc(
            """
            Configuration passed to Pydantic to exclude certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = None,
    response_model_by_alias: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response model
            should be serialized by alias when an alias is used.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = True,
    response_model_exclude_unset: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that were not set and
            have their default values. This is different from
            `response_model_exclude_defaults` in that if the fields are set,
            they will be included in the response, even if the value is the same
            as the default.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
            """
        ),
    ] = False,
    response_model_exclude_defaults: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that have the same value
            as the default. This is different from `response_model_exclude_unset`
            in that if the fields are set but contain the same default values,
            they will be excluded from the response.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
            """
        ),
    ] = False,
    response_model_exclude_none: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data should
            exclude fields set to `None`.

            This is much simpler (less smart) than `response_model_exclude_unset`
            and `response_model_exclude_defaults`. You probably want to use one of
            those two instead of this one, as those allow returning `None` values
            when it makes sense.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none).
            """
        ),
    ] = False,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Include this *path operation* in the generated OpenAPI schema.

            This affects the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi).
            """
        ),
    ] = True,
    response_class: Annotated[
        type[Response],
        Doc(
            """
            Response class to be used for this *path operation*.

            This will not be used if you return a response directly.

            Read more about it in the
            [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse).
            """
        ),
    ] = Default(JSONResponse),
    name: Annotated[
        str | None,
        Doc(
            """
            Name for this *path operation*. Only used internally.
            """
        ),
    ] = None,
    callbacks: Annotated[
        list[BaseRoute] | None,
        Doc(
            """
            List of *path operations* that will be used as OpenAPI callbacks.

            This is only for OpenAPI documentation, the callbacks won't be used
            directly.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/).
            """
        ),
    ] = None,
    openapi_extra: Annotated[
        dict[str, Any] | None,
        Doc(
            """
            Extra metadata to be included in the OpenAPI schema for this *path
            operation*.

            Read more about it in the
            [FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema).
            """
        ),
    ] = None,
    generate_unique_id_function: Annotated[
        Callable[[APIRoute], str],
        Doc(
            """
            Customize the function used to generate unique IDs for the *path
            operations* shown in the generated OpenAPI.

            This is particularly useful when automatically generating clients or
            SDKs for your API.

            Read more about it in the
            [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
            """
        ),
    ] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]:
    """
    Add a *path operation* using an HTTP POST operation.

    ## Example

    ```python
    from fastapi import APIRouter, FastAPI
    from pydantic import BaseModel

    class Item(BaseModel):
        name: str
        description: str | None = None

    app = FastAPI()
    router = APIRouter()

    @router.post("/items/")
    def create_item(item: Item):
        return {"message": "Item created"}

    app.include_router(router)
    ```
    """
    return self.api_route(
        path=path,
        response_model=response_model,
        status_code=status_code,
        tags=tags,
        dependencies=dependencies,
        summary=summary,
        description=description,
        response_description=response_description,
        responses=responses,
        deprecated=deprecated,
        methods=["POST"],
        operation_id=operation_id,
        response_model_include=response_model_include,
        response_model_exclude=response_model_exclude,
        response_model_by_alias=response_model_by_alias,
        response_model_exclude_unset=response_model_exclude_unset,
        response_model_exclude_defaults=response_model_exclude_defaults,
        response_model_exclude_none=response_model_exclude_none,
        include_in_schema=include_in_schema,
        response_class=response_class,
        name=name,
        callbacks=callbacks,
        openapi_extra=openapi_extra,
        generate_unique_id_function=generate_unique_id_function,
    )

delete #

delete(
    path: str,
    *,
    response_model: Any = Default(None),
    status_code: int | None = None,
    tags: list[str | Enum] | None = None,
    dependencies: Sequence[Depends] | None = None,
    summary: str | None = None,
    description: str | None = None,
    response_description: str = "Successful Response",
    responses: dict[int | str, dict[str, Any]]
    | None = None,
    deprecated: bool | None = None,
    operation_id: str | None = None,
    response_model_include: IncEx | None = None,
    response_model_exclude: IncEx | None = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
    include_in_schema: bool = True,
    response_class: type[Response] = Default(JSONResponse),
    name: str | None = None,
    callbacks: list[BaseRoute] | None = None,
    openapi_extra: dict[str, Any] | None = None,
    generate_unique_id_function: Callable[
        [APIRoute], str
    ] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]

Add a path operation using an HTTP DELETE operation.

Example#

from fastapi import APIRouter, FastAPI

app = FastAPI()
router = APIRouter()

@router.delete("/items/{item_id}")
def delete_item(item_id: str):
    return {"message": "Item deleted"}

app.include_router(router)

Source code in fastapi/routing.py

def delete(
    self,
    path: Annotated[
        str,
        Doc(
            """
            The URL path to be used for this *path operation*.

            For example, in `http://example.com/items`, the path is `/items`.
            """
        ),
    ],
    *,
    response_model: Annotated[
        Any,
        Doc(
            """
            The type to use for the response.

            It could be any valid Pydantic *field* type. So, it doesn't have to
            be a Pydantic model, it could be other things, like a `list`, `dict`,
            etc.

            It will be used for:

            * Documentation: the generated OpenAPI (and the UI at `/docs`) will
                show it as the response (JSON Schema).
            * Serialization: you could return an arbitrary object and the
                `response_model` would be used to serialize that object into the
                corresponding JSON.
            * Filtering: the JSON sent to the client will only contain the data
                (fields) defined in the `response_model`. If you returned an object
                that contains an attribute `password` but the `response_model` does
                not include that field, the JSON sent to the client would not have
                that `password`.
            * Validation: whatever you return will be serialized with the
                `response_model`, converting any data as necessary to generate the
                corresponding JSON. But if the data in the object returned is not
                valid, that would mean a violation of the contract with the client,
                so it's an error from the API developer. So, FastAPI will raise an
                error and return a 500 error code (Internal Server Error).

            Read more about it in the
            [FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/).
            """
        ),
    ] = Default(None),
    status_code: Annotated[
        int | None,
        Doc(
            """
            The default status code to be used for the response.

            You could override the status code by returning a response directly.

            Read more about it in the
            [FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/).
            """
        ),
    ] = None,
    tags: Annotated[
        list[str | Enum] | None,
        Doc(
            """
            A list of tags to be applied to the *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags).
            """
        ),
    ] = None,
    dependencies: Annotated[
        Sequence[params.Depends] | None,
        Doc(
            """
            A list of dependencies (using `Depends()`) to be applied to the
            *path operation*.

            Read more about it in the
            [FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/).
            """
        ),
    ] = None,
    summary: Annotated[
        str | None,
        Doc(
            """
            A summary for the *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    description: Annotated[
        str | None,
        Doc(
            """
            A description for the *path operation*.

            If not provided, it will be extracted automatically from the docstring
            of the *path operation function*.

            It can contain Markdown.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    response_description: Annotated[
        str,
        Doc(
            """
            The description for the default response.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = "Successful Response",
    responses: Annotated[
        dict[int | str, dict[str, Any]] | None,
        Doc(
            """
            Additional responses that could be returned by this *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = None,
    deprecated: Annotated[
        bool | None,
        Doc(
            """
            Mark this *path operation* as deprecated.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = None,
    operation_id: Annotated[
        str | None,
        Doc(
            """
            Custom operation ID to be used by this *path operation*.

            By default, it is generated automatically.

            If you provide a custom operation ID, you need to make sure it is
            unique for the whole API.

            You can customize the
            operation ID generation with the parameter
            `generate_unique_id_function` in the `FastAPI` class.

            Read more about it in the
            [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
            """
        ),
    ] = None,
    response_model_include: Annotated[
        IncEx | None,
        Doc(
            """
            Configuration passed to Pydantic to include only certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = None,
    response_model_exclude: Annotated[
        IncEx | None,
        Doc(
            """
            Configuration passed to Pydantic to exclude certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = None,
    response_model_by_alias: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response model
            should be serialized by alias when an alias is used.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = True,
    response_model_exclude_unset: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that were not set and
            have their default values. This is different from
            `response_model_exclude_defaults` in that if the fields are set,
            they will be included in the response, even if the value is the same
            as the default.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
            """
        ),
    ] = False,
    response_model_exclude_defaults: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that have the same value
            as the default. This is different from `response_model_exclude_unset`
            in that if the fields are set but contain the same default values,
            they will be excluded from the response.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
            """
        ),
    ] = False,
    response_model_exclude_none: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data should
            exclude fields set to `None`.

            This is much simpler (less smart) than `response_model_exclude_unset`
            and `response_model_exclude_defaults`. You probably want to use one of
            those two instead of this one, as those allow returning `None` values
            when it makes sense.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none).
            """
        ),
    ] = False,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Include this *path operation* in the generated OpenAPI schema.

            This affects the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi).
            """
        ),
    ] = True,
    response_class: Annotated[
        type[Response],
        Doc(
            """
            Response class to be used for this *path operation*.

            This will not be used if you return a response directly.

            Read more about it in the
            [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse).
            """
        ),
    ] = Default(JSONResponse),
    name: Annotated[
        str | None,
        Doc(
            """
            Name for this *path operation*. Only used internally.
            """
        ),
    ] = None,
    callbacks: Annotated[
        list[BaseRoute] | None,
        Doc(
            """
            List of *path operations* that will be used as OpenAPI callbacks.

            This is only for OpenAPI documentation, the callbacks won't be used
            directly.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/).
            """
        ),
    ] = None,
    openapi_extra: Annotated[
        dict[str, Any] | None,
        Doc(
            """
            Extra metadata to be included in the OpenAPI schema for this *path
            operation*.

            Read more about it in the
            [FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema).
            """
        ),
    ] = None,
    generate_unique_id_function: Annotated[
        Callable[[APIRoute], str],
        Doc(
            """
            Customize the function used to generate unique IDs for the *path
            operations* shown in the generated OpenAPI.

            This is particularly useful when automatically generating clients or
            SDKs for your API.

            Read more about it in the
            [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
            """
        ),
    ] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]:
    """
    Add a *path operation* using an HTTP DELETE operation.

    ## Example

    ```python
    from fastapi import APIRouter, FastAPI

    app = FastAPI()
    router = APIRouter()

    @router.delete("/items/{item_id}")
    def delete_item(item_id: str):
        return {"message": "Item deleted"}

    app.include_router(router)
    ```
    """
    return self.api_route(
        path=path,
        response_model=response_model,
        status_code=status_code,
        tags=tags,
        dependencies=dependencies,
        summary=summary,
        description=description,
        response_description=response_description,
        responses=responses,
        deprecated=deprecated,
        methods=["DELETE"],
        operation_id=operation_id,
        response_model_include=response_model_include,
        response_model_exclude=response_model_exclude,
        response_model_by_alias=response_model_by_alias,
        response_model_exclude_unset=response_model_exclude_unset,
        response_model_exclude_defaults=response_model_exclude_defaults,
        response_model_exclude_none=response_model_exclude_none,
        include_in_schema=include_in_schema,
        response_class=response_class,
        name=name,
        callbacks=callbacks,
        openapi_extra=openapi_extra,
        generate_unique_id_function=generate_unique_id_function,
    )

options #

options(
    path: str,
    *,
    response_model: Any = Default(None),
    status_code: int | None = None,
    tags: list[str | Enum] | None = None,
    dependencies: Sequence[Depends] | None = None,
    summary: str | None = None,
    description: str | None = None,
    response_description: str = "Successful Response",
    responses: dict[int | str, dict[str, Any]]
    | None = None,
    deprecated: bool | None = None,
    operation_id: str | None = None,
    response_model_include: IncEx | None = None,
    response_model_exclude: IncEx | None = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
    include_in_schema: bool = True,
    response_class: type[Response] = Default(JSONResponse),
    name: str | None = None,
    callbacks: list[BaseRoute] | None = None,
    openapi_extra: dict[str, Any] | None = None,
    generate_unique_id_function: Callable[
        [APIRoute], str
    ] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]

Add a path operation using an HTTP OPTIONS operation.

Example#

from fastapi import APIRouter, FastAPI

app = FastAPI()
router = APIRouter()

@router.options("/items/")
def get_item_options():
    return {"additions": ["Aji", "Guacamole"]}

app.include_router(router)

Source code in fastapi/routing.py

def options(
    self,
    path: Annotated[
        str,
        Doc(
            """
            The URL path to be used for this *path operation*.

            For example, in `http://example.com/items`, the path is `/items`.
            """
        ),
    ],
    *,
    response_model: Annotated[
        Any,
        Doc(
            """
            The type to use for the response.

            It could be any valid Pydantic *field* type. So, it doesn't have to
            be a Pydantic model, it could be other things, like a `list`, `dict`,
            etc.

            It will be used for:

            * Documentation: the generated OpenAPI (and the UI at `/docs`) will
                show it as the response (JSON Schema).
            * Serialization: you could return an arbitrary object and the
                `response_model` would be used to serialize that object into the
                corresponding JSON.
            * Filtering: the JSON sent to the client will only contain the data
                (fields) defined in the `response_model`. If you returned an object
                that contains an attribute `password` but the `response_model` does
                not include that field, the JSON sent to the client would not have
                that `password`.
            * Validation: whatever you return will be serialized with the
                `response_model`, converting any data as necessary to generate the
                corresponding JSON. But if the data in the object returned is not
                valid, that would mean a violation of the contract with the client,
                so it's an error from the API developer. So, FastAPI will raise an
                error and return a 500 error code (Internal Server Error).

            Read more about it in the
            [FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/).
            """
        ),
    ] = Default(None),
    status_code: Annotated[
        int | None,
        Doc(
            """
            The default status code to be used for the response.

            You could override the status code by returning a response directly.

            Read more about it in the
            [FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/).
            """
        ),
    ] = None,
    tags: Annotated[
        list[str | Enum] | None,
        Doc(
            """
            A list of tags to be applied to the *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags).
            """
        ),
    ] = None,
    dependencies: Annotated[
        Sequence[params.Depends] | None,
        Doc(
            """
            A list of dependencies (using `Depends()`) to be applied to the
            *path operation*.

            Read more about it in the
            [FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/).
            """
        ),
    ] = None,
    summary: Annotated[
        str | None,
        Doc(
            """
            A summary for the *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    description: Annotated[
        str | None,
        Doc(
            """
            A description for the *path operation*.

            If not provided, it will be extracted automatically from the docstring
            of the *path operation function*.

            It can contain Markdown.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    response_description: Annotated[
        str,
        Doc(
            """
            The description for the default response.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = "Successful Response",
    responses: Annotated[
        dict[int | str, dict[str, Any]] | None,
        Doc(
            """
            Additional responses that could be returned by this *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = None,
    deprecated: Annotated[
        bool | None,
        Doc(
            """
            Mark this *path operation* as deprecated.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = None,
    operation_id: Annotated[
        str | None,
        Doc(
            """
            Custom operation ID to be used by this *path operation*.

            By default, it is generated automatically.

            If you provide a custom operation ID, you need to make sure it is
            unique for the whole API.

            You can customize the
            operation ID generation with the parameter
            `generate_unique_id_function` in the `FastAPI` class.

            Read more about it in the
            [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
            """
        ),
    ] = None,
    response_model_include: Annotated[
        IncEx | None,
        Doc(
            """
            Configuration passed to Pydantic to include only certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = None,
    response_model_exclude: Annotated[
        IncEx | None,
        Doc(
            """
            Configuration passed to Pydantic to exclude certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = None,
    response_model_by_alias: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response model
            should be serialized by alias when an alias is used.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = True,
    response_model_exclude_unset: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that were not set and
            have their default values. This is different from
            `response_model_exclude_defaults` in that if the fields are set,
            they will be included in the response, even if the value is the same
            as the default.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
            """
        ),
    ] = False,
    response_model_exclude_defaults: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that have the same value
            as the default. This is different from `response_model_exclude_unset`
            in that if the fields are set but contain the same default values,
            they will be excluded from the response.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
            """
        ),
    ] = False,
    response_model_exclude_none: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data should
            exclude fields set to `None`.

            This is much simpler (less smart) than `response_model_exclude_unset`
            and `response_model_exclude_defaults`. You probably want to use one of
            those two instead of this one, as those allow returning `None` values
            when it makes sense.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none).
            """
        ),
    ] = False,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Include this *path operation* in the generated OpenAPI schema.

            This affects the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi).
            """
        ),
    ] = True,
    response_class: Annotated[
        type[Response],
        Doc(
            """
            Response class to be used for this *path operation*.

            This will not be used if you return a response directly.

            Read more about it in the
            [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse).
            """
        ),
    ] = Default(JSONResponse),
    name: Annotated[
        str | None,
        Doc(
            """
            Name for this *path operation*. Only used internally.
            """
        ),
    ] = None,
    callbacks: Annotated[
        list[BaseRoute] | None,
        Doc(
            """
            List of *path operations* that will be used as OpenAPI callbacks.

            This is only for OpenAPI documentation, the callbacks won't be used
            directly.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/).
            """
        ),
    ] = None,
    openapi_extra: Annotated[
        dict[str, Any] | None,
        Doc(
            """
            Extra metadata to be included in the OpenAPI schema for this *path
            operation*.

            Read more about it in the
            [FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema).
            """
        ),
    ] = None,
    generate_unique_id_function: Annotated[
        Callable[[APIRoute], str],
        Doc(
            """
            Customize the function used to generate unique IDs for the *path
            operations* shown in the generated OpenAPI.

            This is particularly useful when automatically generating clients or
            SDKs for your API.

            Read more about it in the
            [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
            """
        ),
    ] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]:
    """
    Add a *path operation* using an HTTP OPTIONS operation.

    ## Example

    ```python
    from fastapi import APIRouter, FastAPI

    app = FastAPI()
    router = APIRouter()

    @router.options("/items/")
    def get_item_options():
        return {"additions": ["Aji", "Guacamole"]}

    app.include_router(router)
    ```
    """
    return self.api_route(
        path=path,
        response_model=response_model,
        status_code=status_code,
        tags=tags,
        dependencies=dependencies,
        summary=summary,
        description=description,
        response_description=response_description,
        responses=responses,
        deprecated=deprecated,
        methods=["OPTIONS"],
        operation_id=operation_id,
        response_model_include=response_model_include,
        response_model_exclude=response_model_exclude,
        response_model_by_alias=response_model_by_alias,
        response_model_exclude_unset=response_model_exclude_unset,
        response_model_exclude_defaults=response_model_exclude_defaults,
        response_model_exclude_none=response_model_exclude_none,
        include_in_schema=include_in_schema,
        response_class=response_class,
        name=name,
        callbacks=callbacks,
        openapi_extra=openapi_extra,
        generate_unique_id_function=generate_unique_id_function,
    )

head #

head(
    path: str,
    *,
    response_model: Any = Default(None),
    status_code: int | None = None,
    tags: list[str | Enum] | None = None,
    dependencies: Sequence[Depends] | None = None,
    summary: str | None = None,
    description: str | None = None,
    response_description: str = "Successful Response",
    responses: dict[int | str, dict[str, Any]]
    | None = None,
    deprecated: bool | None = None,
    operation_id: str | None = None,
    response_model_include: IncEx | None = None,
    response_model_exclude: IncEx | None = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
    include_in_schema: bool = True,
    response_class: type[Response] = Default(JSONResponse),
    name: str | None = None,
    callbacks: list[BaseRoute] | None = None,
    openapi_extra: dict[str, Any] | None = None,
    generate_unique_id_function: Callable[
        [APIRoute], str
    ] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]

Add a path operation using an HTTP HEAD operation.

Example#

from fastapi import APIRouter, FastAPI
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str | None = None

app = FastAPI()
router = APIRouter()

@router.head("/items/", status_code=204)
def get_items_headers(response: Response):
    response.headers["X-Cat-Dog"] = "Alone in the world"

app.include_router(router)

Source code in fastapi/routing.py

def head(
    self,
    path: Annotated[
        str,
        Doc(
            """
            The URL path to be used for this *path operation*.

            For example, in `http://example.com/items`, the path is `/items`.
            """
        ),
    ],
    *,
    response_model: Annotated[
        Any,
        Doc(
            """
            The type to use for the response.

            It could be any valid Pydantic *field* type. So, it doesn't have to
            be a Pydantic model, it could be other things, like a `list`, `dict`,
            etc.

            It will be used for:

            * Documentation: the generated OpenAPI (and the UI at `/docs`) will
                show it as the response (JSON Schema).
            * Serialization: you could return an arbitrary object and the
                `response_model` would be used to serialize that object into the
                corresponding JSON.
            * Filtering: the JSON sent to the client will only contain the data
                (fields) defined in the `response_model`. If you returned an object
                that contains an attribute `password` but the `response_model` does
                not include that field, the JSON sent to the client would not have
                that `password`.
            * Validation: whatever you return will be serialized with the
                `response_model`, converting any data as necessary to generate the
                corresponding JSON. But if the data in the object returned is not
                valid, that would mean a violation of the contract with the client,
                so it's an error from the API developer. So, FastAPI will raise an
                error and return a 500 error code (Internal Server Error).

            Read more about it in the
            [FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/).
            """
        ),
    ] = Default(None),
    status_code: Annotated[
        int | None,
        Doc(
            """
            The default status code to be used for the response.

            You could override the status code by returning a response directly.

            Read more about it in the
            [FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/).
            """
        ),
    ] = None,
    tags: Annotated[
        list[str | Enum] | None,
        Doc(
            """
            A list of tags to be applied to the *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags).
            """
        ),
    ] = None,
    dependencies: Annotated[
        Sequence[params.Depends] | None,
        Doc(
            """
            A list of dependencies (using `Depends()`) to be applied to the
            *path operation*.

            Read more about it in the
            [FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/).
            """
        ),
    ] = None,
    summary: Annotated[
        str | None,
        Doc(
            """
            A summary for the *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    description: Annotated[
        str | None,
        Doc(
            """
            A description for the *path operation*.

            If not provided, it will be extracted automatically from the docstring
            of the *path operation function*.

            It can contain Markdown.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    response_description: Annotated[
        str,
        Doc(
            """
            The description for the default response.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = "Successful Response",
    responses: Annotated[
        dict[int | str, dict[str, Any]] | None,
        Doc(
            """
            Additional responses that could be returned by this *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = None,
    deprecated: Annotated[
        bool | None,
        Doc(
            """
            Mark this *path operation* as deprecated.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = None,
    operation_id: Annotated[
        str | None,
        Doc(
            """
            Custom operation ID to be used by this *path operation*.

            By default, it is generated automatically.

            If you provide a custom operation ID, you need to make sure it is
            unique for the whole API.

            You can customize the
            operation ID generation with the parameter
            `generate_unique_id_function` in the `FastAPI` class.

            Read more about it in the
            [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
            """
        ),
    ] = None,
    response_model_include: Annotated[
        IncEx | None,
        Doc(
            """
            Configuration passed to Pydantic to include only certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = None,
    response_model_exclude: Annotated[
        IncEx | None,
        Doc(
            """
            Configuration passed to Pydantic to exclude certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = None,
    response_model_by_alias: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response model
            should be serialized by alias when an alias is used.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = True,
    response_model_exclude_unset: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that were not set and
            have their default values. This is different from
            `response_model_exclude_defaults` in that if the fields are set,
            they will be included in the response, even if the value is the same
            as the default.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
            """
        ),
    ] = False,
    response_model_exclude_defaults: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that have the same value
            as the default. This is different from `response_model_exclude_unset`
            in that if the fields are set but contain the same default values,
            they will be excluded from the response.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
            """
        ),
    ] = False,
    response_model_exclude_none: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data should
            exclude fields set to `None`.

            This is much simpler (less smart) than `response_model_exclude_unset`
            and `response_model_exclude_defaults`. You probably want to use one of
            those two instead of this one, as those allow returning `None` values
            when it makes sense.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none).
            """
        ),
    ] = False,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Include this *path operation* in the generated OpenAPI schema.

            This affects the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi).
            """
        ),
    ] = True,
    response_class: Annotated[
        type[Response],
        Doc(
            """
            Response class to be used for this *path operation*.

            This will not be used if you return a response directly.

            Read more about it in the
            [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse).
            """
        ),
    ] = Default(JSONResponse),
    name: Annotated[
        str | None,
        Doc(
            """
            Name for this *path operation*. Only used internally.
            """
        ),
    ] = None,
    callbacks: Annotated[
        list[BaseRoute] | None,
        Doc(
            """
            List of *path operations* that will be used as OpenAPI callbacks.

            This is only for OpenAPI documentation, the callbacks won't be used
            directly.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/).
            """
        ),
    ] = None,
    openapi_extra: Annotated[
        dict[str, Any] | None,
        Doc(
            """
            Extra metadata to be included in the OpenAPI schema for this *path
            operation*.

            Read more about it in the
            [FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema).
            """
        ),
    ] = None,
    generate_unique_id_function: Annotated[
        Callable[[APIRoute], str],
        Doc(
            """
            Customize the function used to generate unique IDs for the *path
            operations* shown in the generated OpenAPI.

            This is particularly useful when automatically generating clients or
            SDKs for your API.

            Read more about it in the
            [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
            """
        ),
    ] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]:
    """
    Add a *path operation* using an HTTP HEAD operation.

    ## Example

    ```python
    from fastapi import APIRouter, FastAPI
    from pydantic import BaseModel

    class Item(BaseModel):
        name: str
        description: str | None = None

    app = FastAPI()
    router = APIRouter()

    @router.head("/items/", status_code=204)
    def get_items_headers(response: Response):
        response.headers["X-Cat-Dog"] = "Alone in the world"

    app.include_router(router)
    ```
    """
    return self.api_route(
        path=path,
        response_model=response_model,
        status_code=status_code,
        tags=tags,
        dependencies=dependencies,
        summary=summary,
        description=description,
        response_description=response_description,
        responses=responses,
        deprecated=deprecated,
        methods=["HEAD"],
        operation_id=operation_id,
        response_model_include=response_model_include,
        response_model_exclude=response_model_exclude,
        response_model_by_alias=response_model_by_alias,
        response_model_exclude_unset=response_model_exclude_unset,
        response_model_exclude_defaults=response_model_exclude_defaults,
        response_model_exclude_none=response_model_exclude_none,
        include_in_schema=include_in_schema,
        response_class=response_class,
        name=name,
        callbacks=callbacks,
        openapi_extra=openapi_extra,
        generate_unique_id_function=generate_unique_id_function,
    )

patch #

patch(
    path: str,
    *,
    response_model: Any = Default(None),
    status_code: int | None = None,
    tags: list[str | Enum] | None = None,
    dependencies: Sequence[Depends] | None = None,
    summary: str | None = None,
    description: str | None = None,
    response_description: str = "Successful Response",
    responses: dict[int | str, dict[str, Any]]
    | None = None,
    deprecated: bool | None = None,
    operation_id: str | None = None,
    response_model_include: IncEx | None = None,
    response_model_exclude: IncEx | None = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
    include_in_schema: bool = True,
    response_class: type[Response] = Default(JSONResponse),
    name: str | None = None,
    callbacks: list[BaseRoute] | None = None,
    openapi_extra: dict[str, Any] | None = None,
    generate_unique_id_function: Callable[
        [APIRoute], str
    ] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]

Add a path operation using an HTTP PATCH operation.

Example#

from fastapi import APIRouter, FastAPI
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str | None = None

app = FastAPI()
router = APIRouter()

@router.patch("/items/")
def update_item(item: Item):
    return {"message": "Item updated in place"}

app.include_router(router)

Source code in fastapi/routing.py

def patch(
    self,
    path: Annotated[
        str,
        Doc(
            """
            The URL path to be used for this *path operation*.

            For example, in `http://example.com/items`, the path is `/items`.
            """
        ),
    ],
    *,
    response_model: Annotated[
        Any,
        Doc(
            """
            The type to use for the response.

            It could be any valid Pydantic *field* type. So, it doesn't have to
            be a Pydantic model, it could be other things, like a `list`, `dict`,
            etc.

            It will be used for:

            * Documentation: the generated OpenAPI (and the UI at `/docs`) will
                show it as the response (JSON Schema).
            * Serialization: you could return an arbitrary object and the
                `response_model` would be used to serialize that object into the
                corresponding JSON.
            * Filtering: the JSON sent to the client will only contain the data
                (fields) defined in the `response_model`. If you returned an object
                that contains an attribute `password` but the `response_model` does
                not include that field, the JSON sent to the client would not have
                that `password`.
            * Validation: whatever you return will be serialized with the
                `response_model`, converting any data as necessary to generate the
                corresponding JSON. But if the data in the object returned is not
                valid, that would mean a violation of the contract with the client,
                so it's an error from the API developer. So, FastAPI will raise an
                error and return a 500 error code (Internal Server Error).

            Read more about it in the
            [FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/).
            """
        ),
    ] = Default(None),
    status_code: Annotated[
        int | None,
        Doc(
            """
            The default status code to be used for the response.

            You could override the status code by returning a response directly.

            Read more about it in the
            [FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/).
            """
        ),
    ] = None,
    tags: Annotated[
        list[str | Enum] | None,
        Doc(
            """
            A list of tags to be applied to the *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags).
            """
        ),
    ] = None,
    dependencies: Annotated[
        Sequence[params.Depends] | None,
        Doc(
            """
            A list of dependencies (using `Depends()`) to be applied to the
            *path operation*.

            Read more about it in the
            [FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/).
            """
        ),
    ] = None,
    summary: Annotated[
        str | None,
        Doc(
            """
            A summary for the *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    description: Annotated[
        str | None,
        Doc(
            """
            A description for the *path operation*.

            If not provided, it will be extracted automatically from the docstring
            of the *path operation function*.

            It can contain Markdown.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    response_description: Annotated[
        str,
        Doc(
            """
            The description for the default response.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = "Successful Response",
    responses: Annotated[
        dict[int | str, dict[str, Any]] | None,
        Doc(
            """
            Additional responses that could be returned by this *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = None,
    deprecated: Annotated[
        bool | None,
        Doc(
            """
            Mark this *path operation* as deprecated.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = None,
    operation_id: Annotated[
        str | None,
        Doc(
            """
            Custom operation ID to be used by this *path operation*.

            By default, it is generated automatically.

            If you provide a custom operation ID, you need to make sure it is
            unique for the whole API.

            You can customize the
            operation ID generation with the parameter
            `generate_unique_id_function` in the `FastAPI` class.

            Read more about it in the
            [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
            """
        ),
    ] = None,
    response_model_include: Annotated[
        IncEx | None,
        Doc(
            """
            Configuration passed to Pydantic to include only certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = None,
    response_model_exclude: Annotated[
        IncEx | None,
        Doc(
            """
            Configuration passed to Pydantic to exclude certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = None,
    response_model_by_alias: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response model
            should be serialized by alias when an alias is used.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = True,
    response_model_exclude_unset: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that were not set and
            have their default values. This is different from
            `response_model_exclude_defaults` in that if the fields are set,
            they will be included in the response, even if the value is the same
            as the default.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
            """
        ),
    ] = False,
    response_model_exclude_defaults: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that have the same value
            as the default. This is different from `response_model_exclude_unset`
            in that if the fields are set but contain the same default values,
            they will be excluded from the response.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
            """
        ),
    ] = False,
    response_model_exclude_none: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data should
            exclude fields set to `None`.

            This is much simpler (less smart) than `response_model_exclude_unset`
            and `response_model_exclude_defaults`. You probably want to use one of
            those two instead of this one, as those allow returning `None` values
            when it makes sense.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none).
            """
        ),
    ] = False,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Include this *path operation* in the generated OpenAPI schema.

            This affects the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi).
            """
        ),
    ] = True,
    response_class: Annotated[
        type[Response],
        Doc(
            """
            Response class to be used for this *path operation*.

            This will not be used if you return a response directly.

            Read more about it in the
            [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse).
            """
        ),
    ] = Default(JSONResponse),
    name: Annotated[
        str | None,
        Doc(
            """
            Name for this *path operation*. Only used internally.
            """
        ),
    ] = None,
    callbacks: Annotated[
        list[BaseRoute] | None,
        Doc(
            """
            List of *path operations* that will be used as OpenAPI callbacks.

            This is only for OpenAPI documentation, the callbacks won't be used
            directly.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/).
            """
        ),
    ] = None,
    openapi_extra: Annotated[
        dict[str, Any] | None,
        Doc(
            """
            Extra metadata to be included in the OpenAPI schema for this *path
            operation*.

            Read more about it in the
            [FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema).
            """
        ),
    ] = None,
    generate_unique_id_function: Annotated[
        Callable[[APIRoute], str],
        Doc(
            """
            Customize the function used to generate unique IDs for the *path
            operations* shown in the generated OpenAPI.

            This is particularly useful when automatically generating clients or
            SDKs for your API.

            Read more about it in the
            [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
            """
        ),
    ] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]:
    """
    Add a *path operation* using an HTTP PATCH operation.

    ## Example

    ```python
    from fastapi import APIRouter, FastAPI
    from pydantic import BaseModel

    class Item(BaseModel):
        name: str
        description: str | None = None

    app = FastAPI()
    router = APIRouter()

    @router.patch("/items/")
    def update_item(item: Item):
        return {"message": "Item updated in place"}

    app.include_router(router)
    ```
    """
    return self.api_route(
        path=path,
        response_model=response_model,
        status_code=status_code,
        tags=tags,
        dependencies=dependencies,
        summary=summary,
        description=description,
        response_description=response_description,
        responses=responses,
        deprecated=deprecated,
        methods=["PATCH"],
        operation_id=operation_id,
        response_model_include=response_model_include,
        response_model_exclude=response_model_exclude,
        response_model_by_alias=response_model_by_alias,
        response_model_exclude_unset=response_model_exclude_unset,
        response_model_exclude_defaults=response_model_exclude_defaults,
        response_model_exclude_none=response_model_exclude_none,
        include_in_schema=include_in_schema,
        response_class=response_class,
        name=name,
        callbacks=callbacks,
        openapi_extra=openapi_extra,
        generate_unique_id_function=generate_unique_id_function,
    )

trace #

trace(
    path: str,
    *,
    response_model: Any = Default(None),
    status_code: int | None = None,
    tags: list[str | Enum] | None = None,
    dependencies: Sequence[Depends] | None = None,
    summary: str | None = None,
    description: str | None = None,
    response_description: str = "Successful Response",
    responses: dict[int | str, dict[str, Any]]
    | None = None,
    deprecated: bool | None = None,
    operation_id: str | None = None,
    response_model_include: IncEx | None = None,
    response_model_exclude: IncEx | None = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
    include_in_schema: bool = True,
    response_class: type[Response] = Default(JSONResponse),
    name: str | None = None,
    callbacks: list[BaseRoute] | None = None,
    openapi_extra: dict[str, Any] | None = None,
    generate_unique_id_function: Callable[
        [APIRoute], str
    ] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]

Add a path operation using an HTTP TRACE operation.

Example#

from fastapi import APIRouter, FastAPI
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str | None = None

app = FastAPI()
router = APIRouter()

@router.trace("/items/{item_id}")
def trace_item(item_id: str):
    return None

app.include_router(router)

Source code in fastapi/routing.py

def trace(
    self,
    path: Annotated[
        str,
        Doc(
            """
            The URL path to be used for this *path operation*.

            For example, in `http://example.com/items`, the path is `/items`.
            """
        ),
    ],
    *,
    response_model: Annotated[
        Any,
        Doc(
            """
            The type to use for the response.

            It could be any valid Pydantic *field* type. So, it doesn't have to
            be a Pydantic model, it could be other things, like a `list`, `dict`,
            etc.

            It will be used for:

            * Documentation: the generated OpenAPI (and the UI at `/docs`) will
                show it as the response (JSON Schema).
            * Serialization: you could return an arbitrary object and the
                `response_model` would be used to serialize that object into the
                corresponding JSON.
            * Filtering: the JSON sent to the client will only contain the data
                (fields) defined in the `response_model`. If you returned an object
                that contains an attribute `password` but the `response_model` does
                not include that field, the JSON sent to the client would not have
                that `password`.
            * Validation: whatever you return will be serialized with the
                `response_model`, converting any data as necessary to generate the
                corresponding JSON. But if the data in the object returned is not
                valid, that would mean a violation of the contract with the client,
                so it's an error from the API developer. So, FastAPI will raise an
                error and return a 500 error code (Internal Server Error).

            Read more about it in the
            [FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/).
            """
        ),
    ] = Default(None),
    status_code: Annotated[
        int | None,
        Doc(
            """
            The default status code to be used for the response.

            You could override the status code by returning a response directly.

            Read more about it in the
            [FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/).
            """
        ),
    ] = None,
    tags: Annotated[
        list[str | Enum] | None,
        Doc(
            """
            A list of tags to be applied to the *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags).
            """
        ),
    ] = None,
    dependencies: Annotated[
        Sequence[params.Depends] | None,
        Doc(
            """
            A list of dependencies (using `Depends()`) to be applied to the
            *path operation*.

            Read more about it in the
            [FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/).
            """
        ),
    ] = None,
    summary: Annotated[
        str | None,
        Doc(
            """
            A summary for the *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    description: Annotated[
        str | None,
        Doc(
            """
            A description for the *path operation*.

            If not provided, it will be extracted automatically from the docstring
            of the *path operation function*.

            It can contain Markdown.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
            """
        ),
    ] = None,
    response_description: Annotated[
        str,
        Doc(
            """
            The description for the default response.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = "Successful Response",
    responses: Annotated[
        dict[int | str, dict[str, Any]] | None,
        Doc(
            """
            Additional responses that could be returned by this *path operation*.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = None,
    deprecated: Annotated[
        bool | None,
        Doc(
            """
            Mark this *path operation* as deprecated.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).
            """
        ),
    ] = None,
    operation_id: Annotated[
        str | None,
        Doc(
            """
            Custom operation ID to be used by this *path operation*.

            By default, it is generated automatically.

            If you provide a custom operation ID, you need to make sure it is
            unique for the whole API.

            You can customize the
            operation ID generation with the parameter
            `generate_unique_id_function` in the `FastAPI` class.

            Read more about it in the
            [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
            """
        ),
    ] = None,
    response_model_include: Annotated[
        IncEx | None,
        Doc(
            """
            Configuration passed to Pydantic to include only certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = None,
    response_model_exclude: Annotated[
        IncEx | None,
        Doc(
            """
            Configuration passed to Pydantic to exclude certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = None,
    response_model_by_alias: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response model
            should be serialized by alias when an alias is used.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
            """
        ),
    ] = True,
    response_model_exclude_unset: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that were not set and
            have their default values. This is different from
            `response_model_exclude_defaults` in that if the fields are set,
            they will be included in the response, even if the value is the same
            as the default.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
            """
        ),
    ] = False,
    response_model_exclude_defaults: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that have the same value
            as the default. This is different from `response_model_exclude_unset`
            in that if the fields are set but contain the same default values,
            they will be excluded from the response.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
            """
        ),
    ] = False,
    response_model_exclude_none: Annotated[
        bool,
        Doc(
            """
            Configuration passed to Pydantic to define if the response data should
            exclude fields set to `None`.

            This is much simpler (less smart) than `response_model_exclude_unset`
            and `response_model_exclude_defaults`. You probably want to use one of
            those two instead of this one, as those allow returning `None` values
            when it makes sense.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none).
            """
        ),
    ] = False,
    include_in_schema: Annotated[
        bool,
        Doc(
            """
            Include this *path operation* in the generated OpenAPI schema.

            This affects the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi).
            """
        ),
    ] = True,
    response_class: Annotated[
        type[Response],
        Doc(
            """
            Response class to be used for this *path operation*.

            This will not be used if you return a response directly.

            Read more about it in the
            [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse).
            """
        ),
    ] = Default(JSONResponse),
    name: Annotated[
        str | None,
        Doc(
            """
            Name for this *path operation*. Only used internally.
            """
        ),
    ] = None,
    callbacks: Annotated[
        list[BaseRoute] | None,
        Doc(
            """
            List of *path operations* that will be used as OpenAPI callbacks.

            This is only for OpenAPI documentation, the callbacks won't be used
            directly.

            It will be added to the generated OpenAPI (e.g. visible at `/docs`).

            Read more about it in the
            [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/).
            """
        ),
    ] = None,
    openapi_extra: Annotated[
        dict[str, Any] | None,
        Doc(
            """
            Extra metadata to be included in the OpenAPI schema for this *path
            operation*.

            Read more about it in the
            [FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema).
            """
        ),
    ] = None,
    generate_unique_id_function: Annotated[
        Callable[[APIRoute], str],
        Doc(
            """
            Customize the function used to generate unique IDs for the *path
            operations* shown in the generated OpenAPI.

            This is particularly useful when automatically generating clients or
            SDKs for your API.

            Read more about it in the
            [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
            """
        ),
    ] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]:
    """
    Add a *path operation* using an HTTP TRACE operation.

    ## Example

    ```python
    from fastapi import APIRouter, FastAPI
    from pydantic import BaseModel

    class Item(BaseModel):
        name: str
        description: str | None = None

    app = FastAPI()
    router = APIRouter()

    @router.trace("/items/{item_id}")
    def trace_item(item_id: str):
        return None

    app.include_router(router)
    ```
    """
    return self.api_route(
        path=path,
        response_model=response_model,
        status_code=status_code,
        tags=tags,
        dependencies=dependencies,
        summary=summary,
        description=description,
        response_description=response_description,
        responses=responses,
        deprecated=deprecated,
        methods=["TRACE"],
        operation_id=operation_id,
        response_model_include=response_model_include,
        response_model_exclude=response_model_exclude,
        response_model_by_alias=response_model_by_alias,
        response_model_exclude_unset=response_model_exclude_unset,
        response_model_exclude_defaults=response_model_exclude_defaults,
        response_model_exclude_none=response_model_exclude_none,
        include_in_schema=include_in_schema,
        response_class=response_class,
        name=name,
        callbacks=callbacks,
        openapi_extra=openapi_extra,
        generate_unique_id_function=generate_unique_id_function,
    )

add_event_handler #

add_event_handler(
    event_type: str, func: Callable[[], Any]
) -> None

Add an event handler function for startup or shutdown.

This method is kept for backward compatibility after Starlette removed support for on_startup/on_shutdown handlers.

Ref: https://github.com/Kludex/starlette/pull/3117

Source code in fastapi/routing.py

def add_event_handler(
    self,
    event_type: str,
    func: Callable[[], Any],
) -> None:
    """
    Add an event handler function for startup or shutdown.

    This method is kept for backward compatibility after Starlette removed
    support for on_startup/on_shutdown handlers.

    Ref: https://github.com/Kludex/starlette/pull/3117
    """
    assert event_type in ("startup", "shutdown")
    if event_type == "startup":
        self.on_startup.append(func)
    else:
        self.on_shutdown.append(func)

on_event #

on_event(
    event_type: str,
) -> Callable[[DecoratedCallable], DecoratedCallable]

Add an event handler for the router.

on_event is deprecated, use lifespan event handlers instead.

Read more about it in the FastAPI docs for Lifespan Events.

Source code in fastapi/routing.py

@deprecated(
    """
    on_event is deprecated, use lifespan event handlers instead.

    Read more about it in the
    [FastAPI docs for Lifespan Events](https://fastapi.tiangolo.com/advanced/events/).
    """
)
def on_event(
    self,
    event_type: Annotated[
        str,
        Doc(
            """
            The type of event. `startup` or `shutdown`.
            """
        ),
    ],
) -> Callable[[DecoratedCallable], DecoratedCallable]:
    """
    Add an event handler for the router.

    `on_event` is deprecated, use `lifespan` event handlers instead.

    Read more about it in the
    [FastAPI docs for Lifespan Events](https://fastapi.tiangolo.com/advanced/events/#alternative-events-deprecated).
    """

    def decorator(func: DecoratedCallable) -> DecoratedCallable:
        self.add_event_handler(event_type, func)
        return func

    return decorator

after_startup #

after_startup(
    func: Callable[[AppType], Mapping[str, Any]],
) -> Callable[[AppType], Mapping[str, Any]]

after_startup(
    func: Callable[[AppType], Awaitable[Mapping[str, Any]]],
) -> Callable[[AppType], Awaitable[Mapping[str, Any]]]

after_startup(
    func: Callable[[AppType], None],
) -> Callable[[AppType], None]

after_startup(
    func: Callable[[AppType], Awaitable[None]],
) -> Callable[[AppType], Awaitable[None]]

after_startup(
    func: Callable[[AppType], Mapping[str, Any]]
    | Callable[[AppType], Awaitable[Mapping[str, Any]]]
    | Callable[[AppType], None]
    | Callable[[AppType], Awaitable[None]],
) -> (
    Callable[[AppType], Mapping[str, Any]]
    | Callable[[AppType], Awaitable[Mapping[str, Any]]]
    | Callable[[AppType], None]
    | Callable[[AppType], Awaitable[None]]
)

Register a function to be executed after startup.

Source code in faststream/_internal/fastapi/router.py

def after_startup(
    self,
    func: Callable[["AppType"], Mapping[str, Any]]
    | Callable[["AppType"], Awaitable[Mapping[str, Any]]]
    | Callable[["AppType"], None]
    | Callable[["AppType"], Awaitable[None]],
) -> (
    Callable[["AppType"], Mapping[str, Any]]
    | Callable[["AppType"], Awaitable[Mapping[str, Any]]]
    | Callable[["AppType"], None]
    | Callable[["AppType"], Awaitable[None]]
):
    """Register a function to be executed after startup."""
    self._after_startup_hooks.append(to_async(func))
    return func

on_broker_shutdown #

on_broker_shutdown(
    func: Callable[[AppType], None],
) -> Callable[[AppType], None]

on_broker_shutdown(
    func: Callable[[AppType], Awaitable[None]],
) -> Callable[[AppType], Awaitable[None]]

on_broker_shutdown(
    func: Callable[[AppType], None]
    | Callable[[AppType], Awaitable[None]],
) -> (
    Callable[[AppType], None]
    | Callable[[AppType], Awaitable[None]]
)

Register a function to be executed before broker stop.

Source code in faststream/_internal/fastapi/router.py

def on_broker_shutdown(
    self,
    func: Callable[["AppType"], None] | Callable[["AppType"], Awaitable[None]],
) -> Callable[["AppType"], None] | Callable[["AppType"], Awaitable[None]]:
    """Register a function to be executed before broker stop."""
    self._on_shutdown_hooks.append(to_async(func))
    return func

subscriber #

subscriber(
    *topics: str,
    partitions: Sequence[TopicPartition] = (),
    polling_interval: float = 0.1,
    group_id: str | None = None,
    group_instance_id: str | None = None,
    fetch_max_wait_ms: int = 500,
    fetch_max_bytes: int = 50 * 1024 * 1024,
    fetch_min_bytes: int = 1,
    max_partition_fetch_bytes: int = 1 * 1024 * 1024,
    auto_offset_reset: Literal[
        "latest", "earliest", "none"
    ] = "latest",
    auto_commit_interval_ms: int = 5 * 1000,
    check_crcs: bool = True,
    partition_assignment_strategy: Sequence[str] = (
        "roundrobin",
    ),
    max_poll_interval_ms: int = 5 * 60 * 1000,
    session_timeout_ms: int = 10 * 1000,
    heartbeat_interval_ms: int = 3 * 1000,
    isolation_level: Literal[
        "read_uncommitted", "read_committed"
    ] = "read_uncommitted",
    on_assign: Callable[..., None] | None = None,
    on_revoke: Callable[..., None] | None = None,
    on_lost: Callable[..., None] | None = None,
    batch: Literal[False] = False,
    max_records: int | None = None,
    dependencies: Iterable[Depends] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    max_workers: None = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    response_model: Any = Default(None),
    response_model_include: Optional[IncEx] = None,
    response_model_exclude: Optional[IncEx] = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
) -> DefaultSubscriber

subscriber(
    *topics: str,
    partitions: Sequence[TopicPartition] = (),
    polling_interval: float = 0.1,
    group_id: str | None = None,
    group_instance_id: str | None = None,
    fetch_max_wait_ms: int = 500,
    fetch_max_bytes: int = 50 * 1024 * 1024,
    fetch_min_bytes: int = 1,
    max_partition_fetch_bytes: int = 1 * 1024 * 1024,
    auto_offset_reset: Literal[
        "latest", "earliest", "none"
    ] = "latest",
    auto_commit_interval_ms: int = 5 * 1000,
    check_crcs: bool = True,
    partition_assignment_strategy: Sequence[str] = (
        "roundrobin",
    ),
    max_poll_interval_ms: int = 5 * 60 * 1000,
    session_timeout_ms: int = 10 * 1000,
    heartbeat_interval_ms: int = 3 * 1000,
    isolation_level: Literal[
        "read_uncommitted", "read_committed"
    ] = "read_uncommitted",
    on_assign: Callable[..., None] | None = None,
    on_revoke: Callable[..., None] | None = None,
    on_lost: Callable[..., None] | None = None,
    batch: Literal[False] = False,
    max_records: int | None = None,
    dependencies: Iterable[Depends] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    max_workers: int = ...,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    response_model: Any = Default(None),
    response_model_include: Optional[IncEx] = None,
    response_model_exclude: Optional[IncEx] = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
) -> ConcurrentDefaultSubscriber

subscriber(
    *topics: str,
    partitions: Sequence[TopicPartition] = (),
    polling_interval: float = 0.1,
    group_id: str | None = None,
    group_instance_id: str | None = None,
    fetch_max_wait_ms: int = 500,
    fetch_max_bytes: int = 50 * 1024 * 1024,
    fetch_min_bytes: int = 1,
    max_partition_fetch_bytes: int = 1 * 1024 * 1024,
    auto_offset_reset: Literal[
        "latest", "earliest", "none"
    ] = "latest",
    auto_commit_interval_ms: int = 5 * 1000,
    check_crcs: bool = True,
    partition_assignment_strategy: Sequence[str] = (
        "roundrobin",
    ),
    max_poll_interval_ms: int = 5 * 60 * 1000,
    session_timeout_ms: int = 10 * 1000,
    heartbeat_interval_ms: int = 3 * 1000,
    isolation_level: Literal[
        "read_uncommitted", "read_committed"
    ] = "read_uncommitted",
    on_assign: Callable[..., None] | None = None,
    on_revoke: Callable[..., None] | None = None,
    on_lost: Callable[..., None] | None = None,
    batch: Literal[True] = ...,
    max_records: int | None = None,
    dependencies: Iterable[Depends] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    max_workers: None = None,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    response_model: Any = Default(None),
    response_model_include: Optional[IncEx] = None,
    response_model_exclude: Optional[IncEx] = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
) -> BatchSubscriber

subscriber(
    *topics: str,
    partitions: Sequence[TopicPartition] = (),
    polling_interval: float = 0.1,
    group_id: str | None = None,
    group_instance_id: str | None = None,
    fetch_max_wait_ms: int = 500,
    fetch_max_bytes: int = 50 * 1024 * 1024,
    fetch_min_bytes: int = 1,
    max_partition_fetch_bytes: int = 1 * 1024 * 1024,
    auto_offset_reset: Literal[
        "latest", "earliest", "none"
    ] = "latest",
    auto_commit_interval_ms: int = 5 * 1000,
    check_crcs: bool = True,
    partition_assignment_strategy: Sequence[str] = (
        "roundrobin",
    ),
    max_poll_interval_ms: int = 5 * 60 * 1000,
    session_timeout_ms: int = 10 * 1000,
    heartbeat_interval_ms: int = 3 * 1000,
    isolation_level: Literal[
        "read_uncommitted", "read_committed"
    ] = "read_uncommitted",
    on_assign: Callable[..., None] | None = None,
    on_revoke: Callable[..., None] | None = None,
    on_lost: Callable[..., None] | None = None,
    batch: bool = False,
    max_records: int | None = None,
    dependencies: Iterable[Depends] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    max_workers: int | None = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    response_model: Any = Default(None),
    response_model_include: Optional[IncEx] = None,
    response_model_exclude: Optional[IncEx] = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
) -> Union[
    BatchSubscriber,
    DefaultSubscriber,
    ConcurrentDefaultSubscriber,
]

subscriber(
    *topics: str,
    partitions: Sequence[TopicPartition] = (),
    polling_interval: float = 0.1,
    group_id: str | None = None,
    group_instance_id: str | None = None,
    fetch_max_wait_ms: int = 500,
    fetch_max_bytes: int = 50 * 1024 * 1024,
    fetch_min_bytes: int = 1,
    max_partition_fetch_bytes: int = 1 * 1024 * 1024,
    auto_offset_reset: Literal[
        "latest", "earliest", "none"
    ] = "latest",
    auto_commit_interval_ms: int = 5 * 1000,
    check_crcs: bool = True,
    partition_assignment_strategy: Sequence[str] = (
        "roundrobin",
    ),
    max_poll_interval_ms: int = 5 * 60 * 1000,
    session_timeout_ms: int = 10 * 1000,
    heartbeat_interval_ms: int = 3 * 1000,
    isolation_level: Literal[
        "read_uncommitted", "read_committed"
    ] = "read_uncommitted",
    on_assign: Callable[..., None] | None = None,
    on_revoke: Callable[..., None] | None = None,
    on_lost: Callable[..., None] | None = None,
    batch: bool = False,
    max_records: int | None = None,
    dependencies: Iterable[Depends] = (),
    parser: Optional[CustomCallable] = None,
    decoder: Optional[CustomCallable] = None,
    max_workers: int | None = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    response_model: Any = Default(None),
    response_model_include: Optional[IncEx] = None,
    response_model_exclude: Optional[IncEx] = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
) -> Union[
    BatchSubscriber,
    DefaultSubscriber,
    ConcurrentDefaultSubscriber,
]

Create a subscriber for Kafka topics.

PARAMETER	DESCRIPTION
`*topics`	Kafka topics to consume messages from. TYPE: `str` DEFAULT: `()`
`partitions`	Sequence of topic partitions. TYPE: `Sequence[TopicPartition]` DEFAULT: `()`
`polling_interval`	Polling interval in seconds. TYPE: `float` DEFAULT: `0.1`
`group_id`	Name of the consumer group to join for dynamic partition assignment (if enabled), and to use for fetching and committing offsets. If `None`, auto-partition assignment (via group coordinator) and offset commits are disabled. TYPE: `str \| None` DEFAULT: `None`
`group_instance_id`	A unique string that identifies the consumer instance. If set, the consumer is treated as a static member of the group and does not participate in consumer group management (e.g. partition assignment, rebalances). This can be used to assign partitions to specific consumers, rather than letting the group assign partitions based on consumer metadata. TYPE: `str \| None` DEFAULT: `None`
`fetch_max_wait_ms`	The maximum amount of time in milliseconds the server will block before answering the fetch request if there isn't sufficient data to immediately satisfy the requirement given by `fetch_min_bytes`. TYPE: `int` DEFAULT: `500`
`fetch_max_bytes`	The maximum amount of data the server should return for a fetch request. This is not an absolute maximum, if the first message in the first non-empty partition of the fetch is larger than this value, the message will still be returned to ensure that the consumer can make progress. NOTE: consumer performs fetches to multiple brokers in parallel so memory usage will depend on the number of brokers containing partitions for the topic. TYPE: `int` DEFAULT: `50 * 1024 * 1024`
`fetch_min_bytes`	Minimum amount of data the server should return for a fetch request, otherwise wait up to `fetch_max_wait_ms` for more data to accumulate. TYPE: `int` DEFAULT: `1`
`max_partition_fetch_bytes`	The maximum amount of data per-partition the server will return. The maximum total memory used for a request `= #partitions * max_partition_fetch_bytes`. This size must be at least as large as the maximum message size the server allows or else it is possible for the producer to send messages larger than the consumer can fetch. If that happens, the consumer can get stuck trying to fetch a large message on a certain partition. TYPE: `int` DEFAULT: `1 * 1024 * 1024`
`auto_offset_reset`	A policy for resetting offsets on `OffsetOutOfRangeError` errors: `earliest` will move to the oldest available message `latest` will move to the most recent `none` will raise an exception so you can handle this case TYPE: `Literal['latest', 'earliest', 'none']` DEFAULT: `'latest'`
`auto_commit`	If `True` the consumer's offset will be periodically committed in the background.
`auto_commit_interval_ms`	Milliseconds between automatic offset commits, if `auto_commit` is `True`. TYPE: `int` DEFAULT: `5 * 1000`
`check_crcs`	Automatically check the CRC32 of the records consumed. This ensures no on-the-wire or on-disk corruption to the messages occurred. This check adds some overhead, so it may be disabled in cases seeking extreme performance. TYPE: `bool` DEFAULT: `True`
`partition_assignment_strategy`	List of objects to use to distribute partition ownership amongst consumer instances when group management is used. This preference is implicit in the order of the strategies in the list. When assignment strategy changes: to support a change to the assignment strategy, new versions must enable support both for the old assignment strategy and the new one. The coordinator will choose the old assignment strategy until all members have been updated. Then it will choose the new strategy. TYPE: `Sequence[str]` DEFAULT: `('roundrobin',)`
`max_poll_interval_ms`	Maximum allowed time between calls to consume messages in batches. If this interval is exceeded the consumer is considered failed and the group will rebalance in order to reassign the partitions to another consumer group member. If API methods block waiting for messages, that time does not count against this timeout. TYPE: `int` DEFAULT: `5 * 60 * 1000`
`session_timeout_ms`	Client group session and failure detection timeout. The consumer sends periodic heartbeats (`heartbeat.interval.ms`) to indicate its liveness to the broker. If no hearts are received by the broker for a group member within the session timeout, the broker will remove the consumer from the group and trigger a rebalance. The allowed range is configured with the broker configuration properties `group.min.session.timeout.ms` and `group.max.session.timeout.ms`. TYPE: `int` DEFAULT: `10 * 1000`
`heartbeat_interval_ms`	The expected time in milliseconds between heartbeats to the consumer coordinator when using Kafka's group management feature. Heartbeats are used to ensure that the consumer's session stays active and to facilitate rebalancing when new consumers join or leave the group. The value must be set lower than `session_timeout_ms`, but typically should be set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances. TYPE: `int` DEFAULT: `3 * 1000`
`isolation_level`	Controls how to read messages written transactionally. `read_committed`, batch consumer will only return transactional messages which have been committed. `read_uncommitted` (the default), batch consumer will return all messages, even transactional messages which have been aborted. Non-transactional messages will be returned unconditionally in either mode. Messages will always be returned in offset order. Hence, in `read_committed` mode, batch consumer will only return messages up to the last stable offset (ALSO), which is the one less than the offset of the first open transaction. In particular any messages appearing after messages belonging to ongoing transactions will be withheld until the relevant transaction has been completed. As a result, `read_committed` consumers will not be able to read up to the high watermark when there are in flight transactions. Further, when in `read_committed` the seek_to_end method will return the ALSO. See method docs below. TYPE: `Literal['read_uncommitted', 'read_committed']` DEFAULT: `'read_uncommitted'`
`batch`	Whether to consume messages in batches or not. TYPE: `bool` DEFAULT: `False`
`max_records`	Number of messages to consume as one batch. TYPE: `int \| None` DEFAULT: `None`
`on_assign`	Callback called when partitions are assigned to the consumer during a rebalance. Receives `(consumer, partitions)` arguments. TYPE: `Callable[..., None] \| None` DEFAULT: `None`
`on_revoke`	Callback called when partitions are revoked from the consumer during a rebalance. Receives `(consumer, partitions)` arguments. TYPE: `Callable[..., None] \| None` DEFAULT: `None`
`on_lost`	Callback called when partitions are lost (e.g., due to session timeout). Receives `(consumer, partitions)` arguments. TYPE: `Callable[..., None] \| None` DEFAULT: `None`
`dependencies`	Dependencies list (`[Dependant(),]`) to apply to the subscriber. TYPE: `Iterable[Depends]` DEFAULT: `()`
`parser`	Parser to map original Message object to FastStream one. TYPE: `Optional[CustomCallable]` DEFAULT: `None`
`decoder`	Function to decode FastStream msg bytes body to python objects. TYPE: `Optional[CustomCallable]` DEFAULT: `None`
`middlewares`	Subscriber middlewares to wrap incoming message processing.
`no_ack`	Whether to disable FastStream auto acknowledgement logic or not.
`ack_policy`	Acknowledgement policy for the subscriber. TYPE: `AckPolicy` DEFAULT: `EMPTY`
`no_reply`	Whether to disable FastStream RPC and Reply To auto responses or not. TYPE: `bool` DEFAULT: `False`
`title`	Specification subscriber object title. TYPE: `str \| None` DEFAULT: `None`
`description`	Specification subscriber object description. Uses decorated docstring as default. TYPE: `str \| None` DEFAULT: `None`
`include_in_schema`	Whether to include operation in Specification schema or not. TYPE: `bool` DEFAULT: `True`
`max_workers`	Number of workers to process messages concurrently. TYPE: `int \| None` DEFAULT: `None`
`response_model`	The type to use for the response. It could be any valid Pydantic field type. So, it doesn't have to be a Pydantic model, it could be other things, like a `list`, `dict`, etc. It will be used for: Documentation: the generated OpenAPI (and the UI at `/docs`) will show it as the response (JSON Schema). Serialization: you could return an arbitrary object and the `response_model` would be used to serialize that object into the corresponding JSON. Filtering: the JSON sent to the client will only contain the data (fields) defined in the `response_model`. If you returned an object that contains an attribute `password` but the `response_model` does not include that field, the JSON sent to the client would not have that `password`. Validation: whatever you return will be serialized with the `response_model`, converting any data as necessary to generate the corresponding JSON. But if the data in the object returned is not valid, that would mean a violation of the contract with the client, so it's an error from the API developer. So, FastAPI will raise an error and return a 500 error code (Internal Server Error). Read more about it in the FastAPI docs for Response Model. TYPE: `Any` DEFAULT: `Default(None)`
`response_model_include`	Configuration passed to Pydantic to include only certain fields in the response data. Read more about it in the FastAPI docs for Response Model - Return Type. TYPE: `Optional[IncEx]` DEFAULT: `None`
`response_model_exclude`	Configuration passed to Pydantic to exclude certain fields in the response data. Read more about it in the FastAPI docs for Response Model - Return Type. TYPE: `Optional[IncEx]` DEFAULT: `None`
`response_model_by_alias`	Configuration passed to Pydantic to define if the response model should be serialized by alias when an alias is used. Read more about it in the FastAPI docs for Response Model - Return Type. TYPE: `bool` DEFAULT: `True`
`response_model_exclude_unset`	Configuration passed to Pydantic to define if the response data should have all the fields, including the ones that were not set and have their default values. This is different from `response_model_exclude_defaults` in that if the fields are set, they will be included in the response, even if the value is the same as the default. When `True`, default values are omitted from the response. Read more about it in the FastAPI docs for Response Model - Return Type. TYPE: `bool` DEFAULT: `False`
`response_model_exclude_defaults`	Configuration passed to Pydantic to define if the response data should have all the fields, including the ones that have the same value as the default. This is different from `response_model_exclude_unset` in that if the fields are set but contain the same default values, they will be excluded from the response. When `True`, default values are omitted from the response. Read more about it in the FastAPI docs for Response Model - Return Type. TYPE: `bool` DEFAULT: `False`
`response_model_exclude_none`	Configuration passed to Pydantic to define if the response data should exclude fields set to `None`. This is much simpler (less smart) than `response_model_exclude_unset` and `response_model_exclude_defaults`. You probably want to use one of those two instead of this one, as those allow returning `None` values when it makes sense. Read more about it in the FastAPI docs for Response Model - Return Type. TYPE: `bool` DEFAULT: `False`

Source code in faststream/confluent/fastapi/fastapi.py

@override
def subscriber(
    self,
    *topics: str,
    partitions: Sequence["TopicPartition"] = (),
    polling_interval: float = 0.1,
    group_id: str | None = None,
    group_instance_id: str | None = None,
    fetch_max_wait_ms: int = 500,
    fetch_max_bytes: int = 50 * 1024 * 1024,
    fetch_min_bytes: int = 1,
    max_partition_fetch_bytes: int = 1 * 1024 * 1024,
    auto_offset_reset: Literal["latest", "earliest", "none"] = "latest",
    auto_commit_interval_ms: int = 5 * 1000,
    check_crcs: bool = True,
    partition_assignment_strategy: Sequence[str] = ("roundrobin",),
    max_poll_interval_ms: int = 5 * 60 * 1000,
    session_timeout_ms: int = 10 * 1000,
    heartbeat_interval_ms: int = 3 * 1000,
    isolation_level: Literal[
        "read_uncommitted",
        "read_committed",
    ] = "read_uncommitted",
    # rebalance callbacks
    on_assign: Callable[..., None] | None = None,
    on_revoke: Callable[..., None] | None = None,
    on_lost: Callable[..., None] | None = None,
    batch: bool = False,
    max_records: int | None = None,
    # broker args
    dependencies: Iterable["params.Depends"] = (),
    parser: Optional["CustomCallable"] = None,
    decoder: Optional["CustomCallable"] = None,
    max_workers: int | None = None,
    ack_policy: AckPolicy = EMPTY,
    no_reply: bool = False,
    # Specification args
    title: str | None = None,
    description: str | None = None,
    include_in_schema: bool = True,
    # FastAPI args
    response_model: Any = Default(None),
    response_model_include: Optional["IncEx"] = None,
    response_model_exclude: Optional["IncEx"] = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
) -> Union["BatchSubscriber", "DefaultSubscriber", "ConcurrentDefaultSubscriber"]:
    """Create a subscriber for Kafka topics.

    Args:
        *topics: Kafka topics to consume messages from.
        partitions: Sequence of topic partitions.
        polling_interval: Polling interval in seconds.
        group_id: Name of the consumer group to join for dynamic
            partition assignment (if enabled), and to use for fetching and
            committing offsets. If `None`, auto-partition assignment (via
            group coordinator) and offset commits are disabled.
        group_instance_id: A unique string that identifies the consumer instance.
            If set, the consumer is treated as a static member of the group
            and does not participate in consumer group management (e.g.
            partition assignment, rebalances). This can be used to assign
            partitions to specific consumers, rather than letting the group
            assign partitions based on consumer metadata.
        fetch_max_wait_ms: The maximum amount of time in milliseconds
            the server will block before answering the fetch request if
            there isn't sufficient data to immediately satisfy the
            requirement given by `fetch_min_bytes`.
        fetch_max_bytes: The maximum amount of data the server should
            return for a fetch request. This is not an absolute maximum, if
            the first message in the first non-empty partition of the fetch
            is larger than this value, the message will still be returned
            to ensure that the consumer can make progress. NOTE: consumer
            performs fetches to multiple brokers in parallel so memory
            usage will depend on the number of brokers containing
            partitions for the topic.
        fetch_min_bytes: Minimum amount of data the server should
            return for a fetch request, otherwise wait up to
            `fetch_max_wait_ms` for more data to accumulate.
        max_partition_fetch_bytes: The maximum amount of data
            per-partition the server will return. The maximum total memory
            used for a request ``= #partitions * max_partition_fetch_bytes``.
            This size must be at least as large as the maximum message size
            the server allows or else it is possible for the producer to
            send messages larger than the consumer can fetch. If that
            happens, the consumer can get stuck trying to fetch a large
            message on a certain partition.
        auto_offset_reset: A policy for resetting offsets on `OffsetOutOfRangeError` errors:

            * `earliest` will move to the oldest available message
            * `latest` will move to the most recent
            * `none` will raise an exception so you can handle this case
        auto_commit: If `True` the consumer's offset will be
            periodically committed in the background.
        auto_commit_interval_ms: Milliseconds between automatic
            offset commits, if `auto_commit` is `True`.
        check_crcs: Automatically check the CRC32 of the records
            consumed. This ensures no on-the-wire or on-disk corruption to
            the messages occurred. This check adds some overhead, so it may
            be disabled in cases seeking extreme performance.
        partition_assignment_strategy: List of objects to use to
            distribute partition ownership amongst consumer instances when
            group management is used. This preference is implicit in the order
            of the strategies in the list. When assignment strategy changes:
            to support a change to the assignment strategy, new versions must
            enable support both for the old assignment strategy and the new
            one. The coordinator will choose the old assignment strategy until
            all members have been updated. Then it will choose the new
            strategy.
        max_poll_interval_ms: Maximum allowed time between calls to
            consume messages in batches. If this interval
            is exceeded the consumer is considered failed and the group will
            rebalance in order to reassign the partitions to another consumer
            group member. If API methods block waiting for messages, that time
            does not count against this timeout.
        session_timeout_ms: Client group session and failure detection
            timeout. The consumer sends periodic heartbeats
            (`heartbeat.interval.ms`) to indicate its liveness to the broker.
            If no hearts are received by the broker for a group member within
            the session timeout, the broker will remove the consumer from the
            group and trigger a rebalance. The allowed range is configured with
            the **broker** configuration properties
            `group.min.session.timeout.ms` and `group.max.session.timeout.ms`.
        heartbeat_interval_ms: The expected time in milliseconds
            between heartbeats to the consumer coordinator when using
            Kafka's group management feature. Heartbeats are used to ensure
            that the consumer's session stays active and to facilitate
            rebalancing when new consumers join or leave the group. The
            value must be set lower than `session_timeout_ms`, but typically
            should be set no higher than 1/3 of that value. It can be
            adjusted even lower to control the expected time for normal
            rebalances.
        isolation_level: Controls how to read messages written
            transactionally.

            * `read_committed`, batch consumer will only return
            transactional messages which have been committed.

            * `read_uncommitted` (the default), batch consumer will
            return all messages, even transactional messages which have been
            aborted.

            Non-transactional messages will be returned unconditionally in
            either mode.

            Messages will always be returned in offset order. Hence, in
            `read_committed` mode, batch consumer will only return
            messages up to the last stable offset (ALSO), which is the one less
            than the offset of the first open transaction. In particular any
            messages appearing after messages belonging to ongoing transactions
            will be withheld until the relevant transaction has been completed.
            As a result, `read_committed` consumers will not be able to read up
            to the high watermark when there are in flight transactions.
            Further, when in `read_committed` the seek_to_end method will
            return the ALSO. See method docs below.
        batch: Whether to consume messages in batches or not.
        max_records: Number of messages to consume as one batch.
        on_assign: Callback called when partitions are assigned to the consumer
            during a rebalance. Receives ``(consumer, partitions)`` arguments.
        on_revoke: Callback called when partitions are revoked from the consumer
            during a rebalance. Receives ``(consumer, partitions)`` arguments.
        on_lost: Callback called when partitions are lost (e.g., due to session
            timeout). Receives ``(consumer, partitions)`` arguments.
        dependencies: Dependencies list (`[Dependant(),]`) to apply to the subscriber.
        parser: Parser to map original **Message** object to FastStream one.
        decoder: Function to decode FastStream msg bytes body to python objects.
        middlewares: Subscriber middlewares to wrap incoming message processing.
        no_ack: Whether to disable **FastStream** auto acknowledgement logic or not.
        ack_policy: Acknowledgement policy for the subscriber.
        no_reply: Whether to disable **FastStream** RPC and Reply To auto responses or not.
        title: Specification subscriber object title.
        description: Specification subscriber object description.
            Uses decorated docstring as default.
        include_in_schema: Whether to include operation in Specification schema or not.
        max_workers: Number of workers to process messages concurrently.
        response_model: The type to use for the response.

            It could be any valid Pydantic *field* type. So, it doesn't have to
            be a Pydantic model, it could be other things, like a `list`, `dict`,
            etc.

            It will be used for:

            * Documentation: the generated OpenAPI (and the UI at `/docs`) will
                show it as the response (JSON Schema).
            * Serialization: you could return an arbitrary object and the
                `response_model` would be used to serialize that object into the
                corresponding JSON.
            * Filtering: the JSON sent to the client will only contain the data
                (fields) defined in the `response_model`. If you returned an object
                that contains an attribute `password` but the `response_model` does
                not include that field, the JSON sent to the client would not have
                that `password`.
            * Validation: whatever you return will be serialized with the
                `response_model`, converting any data as necessary to generate the
                corresponding JSON. But if the data in the object returned is not
                valid, that would mean a violation of the contract with the client,
                so it's an error from the API developer. So, FastAPI will raise an
                error and return a 500 error code (Internal Server Error).

            Read more about it in the
            [FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/).

        response_model_include: Configuration passed to Pydantic to include only certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).

        response_model_exclude: Configuration passed to Pydantic to exclude certain fields in the
            response data.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).

        response_model_by_alias: Configuration passed to Pydantic to define if the response model
            should be serialized by alias when an alias is used.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).

        response_model_exclude_unset: Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that were not set and
            have their default values. This is different from
            `response_model_exclude_defaults` in that if the fields are set,
            they will be included in the response, even if the value is the same
            as the default.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).

        response_model_exclude_defaults: Configuration passed to Pydantic to define if the response data
            should have all the fields, including the ones that have the same value
            as the default. This is different from `response_model_exclude_unset`
            in that if the fields are set but contain the same default values,
            they will be excluded from the response.

            When `True`, default values are omitted from the response.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).

        response_model_exclude_none: Configuration passed to Pydantic to define if the response data should
            exclude fields set to `None`.

            This is much simpler (less smart) than `response_model_exclude_unset`
            and `response_model_exclude_defaults`. You probably want to use one of
            those two instead of this one, as those allow returning `None` values
            when it makes sense.

            Read more about it in the
            [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none).
    """
    subscriber = super().subscriber(
        *topics,
        polling_interval=polling_interval,
        max_workers=max_workers,
        partitions=partitions,
        group_id=group_id,
        group_instance_id=group_instance_id,
        fetch_max_wait_ms=fetch_max_wait_ms,
        fetch_max_bytes=fetch_max_bytes,
        fetch_min_bytes=fetch_min_bytes,
        max_partition_fetch_bytes=max_partition_fetch_bytes,
        auto_offset_reset=auto_offset_reset,
        auto_commit_interval_ms=auto_commit_interval_ms,
        check_crcs=check_crcs,
        partition_assignment_strategy=partition_assignment_strategy,
        max_poll_interval_ms=max_poll_interval_ms,
        session_timeout_ms=session_timeout_ms,
        heartbeat_interval_ms=heartbeat_interval_ms,
        isolation_level=isolation_level,
        on_assign=on_assign,
        on_revoke=on_revoke,
        on_lost=on_lost,
        batch=batch,
        max_records=max_records,
        # broker args
        dependencies=dependencies,
        parser=parser,
        decoder=decoder,
        ack_policy=ack_policy,
        no_reply=no_reply,
        title=title,
        description=description,
        include_in_schema=include_in_schema,
        # FastAPI args
        response_model=response_model,
        response_model_include=response_model_include,
        response_model_exclude=response_model_exclude,
        response_model_by_alias=response_model_by_alias,
        response_model_exclude_unset=response_model_exclude_unset,
        response_model_exclude_defaults=response_model_exclude_defaults,
        response_model_exclude_none=response_model_exclude_none,
    )

    workers = max_workers or 1

    if batch:
        return cast("BatchSubscriber", subscriber)
    if workers > 1:
        return cast("ConcurrentDefaultSubscriber", subscriber)
    return cast("DefaultSubscriber", subscriber)

publisher #

publisher(
    topic: str,
    *,
    key: bytes | str | None = None,
    partition: int | None = None,
    headers: dict[str, str] | None = None,
    reply_to: str = "",
    batch: Literal[False] = False,
    title: str | None = None,
    description: str | None = None,
    schema: Any | None = None,
    include_in_schema: bool = True,
) -> DefaultPublisher

publisher(
    topic: str,
    *,
    key: bytes | str | None = None,
    partition: int | None = None,
    headers: dict[str, str] | None = None,
    reply_to: str = "",
    batch: Literal[True] = ...,
    title: str | None = None,
    description: str | None = None,
    schema: Any | None = None,
    include_in_schema: bool = True,
) -> BatchPublisher

publisher(
    topic: str,
    *,
    key: bytes | str | None = None,
    partition: int | None = None,
    headers: dict[str, str] | None = None,
    reply_to: str = "",
    batch: bool = False,
    title: str | None = None,
    description: str | None = None,
    schema: Any | None = None,
    include_in_schema: bool = True,
) -> Union[BatchPublisher, DefaultPublisher]

publisher(
    topic: str,
    *,
    key: bytes | str | None = None,
    partition: int | None = None,
    headers: dict[str, str] | None = None,
    reply_to: str = "",
    batch: bool = False,
    title: str | None = None,
    description: str | None = None,
    schema: Any | None = None,
    include_in_schema: bool = True,
) -> Union[BatchPublisher, DefaultPublisher]

Publish messages to a Kafka topic.

PARAMETER	DESCRIPTION
`topic`	Topic where the message will be published. TYPE: `str`
`key`	A key to associate with the message. Can be used to determine which partition to send the message to. If partition is `None` (and producer's partitioner config is left as default), then messages with the same key will be delivered to the same partition (but if key is `None`, partition is chosen randomly). Must be type `bytes`, or be serializable to bytes via configured `key_serializer`. TYPE: `bytes \| str \| None` DEFAULT: `None`
`partition`	Specify a partition. If not set, the partition will be selected using the configured `partitioner`. TYPE: `int \| None` DEFAULT: `None`
`headers`	Message headers to store metainformation. content-type and correlation_id will be set automatically by framework anyway. Can be overridden by `publish.headers` if specified. TYPE: `dict[str, str] \| None` DEFAULT: `None`
`reply_to`	Topic name to send response. TYPE: `str` DEFAULT: `''`
`batch`	Whether to send messages in batches or not. TYPE: `bool` DEFAULT: `False`
`title`	Specification publisher object title. TYPE: `str \| None` DEFAULT: `None`
`description`	Specification publisher object description. TYPE: `str \| None` DEFAULT: `None`
`schema`	Specification publishing message type. Should be any python-native object annotation or `pydantic.BaseModel`. TYPE: `Any \| None` DEFAULT: `None`
`include_in_schema`	Whetever to include operation in Specification schema or not. TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`Union[BatchPublisher, DefaultPublisher]`	Union["BatchPublisher", "DefaultPublisher"]: The publisher instance.

Source code in faststream/confluent/fastapi/fastapi.py

@override
def publisher(
    self,
    topic: str,
    *,
    key: bytes | str | None = None,
    partition: int | None = None,
    headers: dict[str, str] | None = None,
    reply_to: str = "",
    batch: bool = False,
    # basic args
    # Specification args
    title: str | None = None,
    description: str | None = None,
    schema: Any | None = None,
    include_in_schema: bool = True,
) -> Union["BatchPublisher", "DefaultPublisher"]:
    """Publish messages to a Kafka topic.

    Args:
        topic: Topic where the message will be published.
        key: A key to associate with the message. Can be used to
            determine which partition to send the message to. If partition
            is `None` (and producer's partitioner config is left as default),
            then messages with the same key will be delivered to the same
            partition (but if key is `None`, partition is chosen randomly).
            Must be type `bytes`, or be serializable to bytes via configured
            `key_serializer`.
        partition: Specify a partition. If not set, the partition will be
            selected using the configured `partitioner`.
        headers: Message headers to store metainformation.
            **content-type** and **correlation_id** will be set automatically by framework anyway.
            Can be overridden by `publish.headers` if specified.
        reply_to: Topic name to send response.
        batch: Whether to send messages in batches or not.
        title: Specification publisher object title.
        description: Specification publisher object description.
        schema: Specification publishing message type.
            Should be any python-native object annotation or `pydantic.BaseModel`.
        include_in_schema: Whetever to include operation in Specification schema or not.

    Returns:
        Union["BatchPublisher", "DefaultPublisher"]: The publisher instance.
    """
    return self.broker.publisher(
        topic=topic,
        key=key,
        partition=partition,
        headers=headers,
        batch=batch,
        reply_to=reply_to,
        # Specification options
        title=title,
        description=description,
        schema=schema,
        include_in_schema=include_in_schema,
    )

KafkaRouter

faststream.confluent.fastapi.KafkaRouter #

context property #

routes instance-attribute #

redirect_slashes instance-attribute #

default instance-attribute #

lifespan_context instance-attribute #

middleware_stack instance-attribute #

on_startup instance-attribute #

on_shutdown instance-attribute #

prefix instance-attribute #

tags instance-attribute #

dependencies instance-attribute #

deprecated instance-attribute #

include_in_schema instance-attribute #

responses instance-attribute #

callbacks instance-attribute #

dependency_overrides_provider instance-attribute #

route_class instance-attribute #

default_response_class instance-attribute #

generate_unique_id_function instance-attribute #

strict_content_type instance-attribute #

docs_router instance-attribute #

title instance-attribute #

description instance-attribute #

version instance-attribute #

license instance-attribute #

contact instance-attribute #

setup_state instance-attribute #

fastapi_config instance-attribute #

broker_class class-attribute instance-attribute #

broker instance-attribute #

set_broker #

add_broker #

not_found async #

url_path_for #

lifespan async #

app async #

mount #

host #

add_route #

add_websocket_route #

frontend #

handle async #

matches #

route #

add_api_route #

api_route #

add_api_websocket_route #

websocket #

Example#

websocket_route #

include_router #

get #

Example#

put #

Example#

post #

Example#

delete #

Example#

options #

Example#

head #

Example#

patch #

Example#

trace #

Example#

add_event_handler #

on_event #

after_startup #

on_broker_shutdown #

subscriber #

publisher #

context `property` #

routes `instance-attribute` #

redirect_slashes `instance-attribute` #

default `instance-attribute` #

lifespan_context `instance-attribute` #

middleware_stack `instance-attribute` #

on_startup `instance-attribute` #

on_shutdown `instance-attribute` #

prefix `instance-attribute` #

tags `instance-attribute` #

dependencies `instance-attribute` #

deprecated `instance-attribute` #

include_in_schema `instance-attribute` #

responses `instance-attribute` #

callbacks `instance-attribute` #

dependency_overrides_provider `instance-attribute` #

route_class `instance-attribute` #

default_response_class `instance-attribute` #

generate_unique_id_function `instance-attribute` #

strict_content_type `instance-attribute` #

docs_router `instance-attribute` #

title `instance-attribute` #

description `instance-attribute` #

version `instance-attribute` #

license `instance-attribute` #

contact `instance-attribute` #

setup_state `instance-attribute` #

fastapi_config `instance-attribute` #

broker_class `class-attribute` `instance-attribute` #

broker `instance-attribute` #

not_found `async` #

lifespan `async` #

app `async` #

handle `async` #