field_mapping: dict[
        tuple[ModelField, Literal["validation", "serialization"]], dict[str, Any]
    ],
    separate_input_output_schemas: bool = True,
) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
    path = {}
    security_schemes: dict[str, Any] = {}
    definitions: dict[str, Any] = {}
    assert route.methods is not None, "Methods must be a list"

    set: list,
    UUID: str,
    Url: str,
    AnyUrl: str,
}


def generate_encoders_by_class_tuples(
    type_encoder_map: dict[Any, Callable[[Any], Any]],
) -> dict[Callable[[Any], Any], tuple[Any, ...]]:
    encoders_by_class_tuples: dict[Callable[[Any], Any], tuple[Any, ...]] = defaultdict(
        tuple
    )
    for type_, encoder in type_encoder_map.items():
        encoders_by_class_tuples[encoder] += (type_,)

docs:
  - all:
    - changed-files:
      - any-glob-to-any-file:
        - docs/en/docs/**
        - docs_src/**
      - all-globs-to-all-files:
        - '!fastapi/**'
        - '!pyproject.toml'
        - '!docs/en/data/sponsors.yml'
        - '!docs/en/overrides/main.html'

lang-all:
  - all:
    - changed-files:
      - any-glob-to-any-file:
        - docs/*/docs/**
      - all-globs-to-all-files:

        when(mockSearchBuilder.setScroll(any(Scroll.class))).thenReturn(mockSearchBuilder);
        when(mockSearchBuilder.setSize(10)).thenReturn(mockSearchBuilder);
        when(mockSearchBuilder.setQuery(any())).thenReturn(mockSearchBuilder);
        when(mockSearchBuilder.execute()).thenReturn(mockSearchFuture);
        when(mockSearchFuture.actionGet(anyLong(), any(TimeUnit.class))).thenReturn(mockSearchResponse);

@app.get("/data")
def get_data(session: SessionDep) -> Any:
    data = list(session)
    return data


@app.get("/stream-simple")
def get_stream_simple(session: SessionDep) -> Any:
    def iter_data():
        yield from ["x", "y", "z"]

    return StreamingResponse(iter_data())


@app.get("/stream-session")
def get_stream_session(session: SessionDep) -> Any:
    def iter_data():
        yield from session

        self,
        errors: Sequence[Any],
        *,
        body: Any = None,
        endpoint_ctx: EndpointContext | None = None,
    ) -> None:
        super().__init__(errors, endpoint_ctx=endpoint_ctx)
        self.body = body


class WebSocketRequestValidationError(ValidationException):
    def __init__(
        self,
        errors: Sequence[Any],
        *,

* don't have a `Content-Type` header (e.g. using `fetch()` with a `Blob` body)
* and don't send any authentication credentials.

This type of attack is mainly relevant when:

* the application is running locally (e.g. on `localhost`) or in an internal network

    tuple: tuple,
    set: set,
    frozenset: frozenset,
    deque: deque,
}

sequence_types: tuple[type[Any], ...] = tuple(sequence_annotation_to_type.keys())


# Copy of Pydantic: pydantic/_internal/_utils.py with added TypeGuard
def lenient_issubclass(
    cls: Any, class_or_tuple: type[_T] | tuple[type[_T], ...] | None
) -> TypeGuard[type[_T]]:
    try:

  private val forbiddenLocks = mutableSetOf<Any>()

  /** The timestamp of the last taken event, used to measure elapsed time between events. */
  private var lastTimestampNs: Long? = null

  /** Confirm that the thread does not hold a lock on `lock` during the callback. */
  fun forbidLock(lock: Any) {
    forbiddenLocks.add(lock)
  }

  /**

You can return a `Response` or any sub-class of it.

/// info

`JSONResponse` itself is a sub-class of `Response`.

///

And when you return a `Response`, **FastAPI** will pass it directly.

It won't do any data conversion with Pydantic models, it won't convert the contents to any type, etc.

This gives you a lot of **flexibility**. You can return any data type, override any data declaration or validation, etc.