fastapi_14306

{"status": "ok", "stdout": "165:class RequestValidationError(ValidationException):\n171:class WebSocketRequestValidationError(ValidationException):\n", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/exceptions.py", "content": "class DependencyScopeError(FastAPIError):\n    \"\"\"\n    A dependency declared that it depends on another dependency with an invalid\n    (narrower) scope.\n    \"\"\"\n\n\nclass ValidationException(Exception):\n    def __init__(self, errors: Sequence[Any]) -> None:\n        self._errors = errors\n\n    def errors(self) -> Sequence[Any]:\n        return self._errors\n\n\nclass RequestValidationError(ValidationException):\n    def __init__(self, errors: Sequence[Any], *, body: Any = None) -> None:\n        super().__init__(errors)\n        self.body = body\n\n\nclass WebSocketRequestValidationError(ValidationException):\n    pass\n\n\nclass ResponseValidationError(ValidationException):\n    def __init__(self, errors: Sequence[Any], *, body: Any = None) -> None:\n        super().__init__(errors)\n        self.body = body\n\n    def __str__(self) -> str:\n        message = f\"{len(self._errors)} validation errors:\\n\"\n        for err in self._errors:\n            message += f\"  {err}\\n\"\n        return message\n", "start_line": 150, "end_line": 184, "total_lines": 184, "is_truncated": false}

{"status": "ok", "stdout": "fastapi/exception_handlers.py:2:from fastapi.exceptions import RequestValidationError, WebSocketRequestValidationError\nfastapi/exception_handlers.py:21:    request: Request, exc: RequestValidationError\nfastapi/exception_handlers.py:30:    websocket: WebSocket, exc: WebSocketRequestValidationError\nfastapi/applications.py:24:from fastapi.exceptions import RequestValidationError, WebSocketRequestValidationError\nfastapi/applications.py:988:            RequestValidationError, request_validation_exception_handler\nfastapi/applications.py:991:            WebSocketRequestValidationError,\nfastapi/routing.py:50:    RequestValidationError,\nfastapi/routing.py:52:    WebSocketRequestValidationError,\nfastapi/routing.py:347:            validation_error = RequestValidationError(\nfastapi/routing.py:423:            validation_error = RequestValidationError(\nfastapi/routing.py:453:            raise WebSocketRequestValidationError(\n", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/routing.py", "content": "                    if body_bytes:\n                        json_body: Any = Undefined\n                        content_type_value = request.headers.get(\"content-type\")\n                        if not content_type_value:\n                            json_body = await request.json()\n                        else:\n                            message = email.message.Message()\n                            message[\"content-type\"] = content_type_value\n                            if message.get_content_maintype() == \"application\":\n                                subtype = message.get_content_subtype()\n                                if subtype == \"json\" or subtype.endswith(\"+json\"):\n                                    json_body = await request.json()\n                        if json_body != Undefined:\n                            body = json_body\n                        else:\n                            body = body_bytes\n        except json.JSONDecodeError as e:\n            validation_error = RequestValidationError(\n                [\n                    {\n                        \"type\": \"json_invalid\",\n                        \"loc\": (\"body\", e.pos),\n                        \"msg\": \"JSON decode error\",\n                        \"input\": {},\n                        \"ctx\": {\"error\": e.msg},\n                    }\n                ],\n                body=e.doc,\n            )\n            raise validation_error from e\n        except HTTPException:\n            # If a middleware raises an HTTPException, it should be raised again\n            raise\n        except Exception as e:\n            http_error = HTTPException(\n                status_code=400, detail=\"There was an error parsing the body\"\n            )\n            raise http_error from e\n\n        # Solve dependencies and run path operation function, auto-closing dependencies\n        errors: List[Any] = []\n  

{"status": "ok", "stdout": "51:    ResponseValidationError,\n248:            raise ResponseValidationError(\n", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/routing.py", "content": "            # pydantic v1\n            response_content = _prepare_response_content(\n                response_content,\n                exclude_unset=exclude_unset,\n                exclude_defaults=exclude_defaults,\n                exclude_none=exclude_none,\n            )\n        if is_coroutine:\n            value, errors_ = field.validate(response_content, {}, loc=(\"response\",))\n        else:\n            value, errors_ = await run_in_threadpool(\n                field.validate, response_content, {}, loc=(\"response\",)\n            )\n        if isinstance(errors_, list):\n            errors.extend(errors_)\n        elif errors_:\n            errors.append(errors_)\n        if errors:\n            raise ResponseValidationError(\n                errors=_normalize_errors(errors), body=response_content\n            )\n\n        if hasattr(field, \"serialize\"):\n            return field.serialize(\n                value,\n                include=include,\n                exclude=exclude,\n                by_alias=by_alias,\n                exclude_unset=exclude_unset,\n                exclude_defaults=exclude_defaults,\n                exclude_none=exclude_none,\n", "start_line": 230, "end_line": 260, "total_lines": 4517, "is_truncated": false}

{"status": "ok", "filepath": "fastapi/exceptions.py", "content": "from typing import Any, Dict, Optional, Sequence, Type, Union\n\nfrom annotated_doc import Doc\nfrom pydantic import BaseModel, create_model\nfrom starlette.exceptions import HTTPException as StarletteHTTPException\nfrom starlette.exceptions import WebSocketException as StarletteWebSocketException\nfrom typing_extensions import Annotated\n\n\nclass HTTPException(StarletteHTTPException):\n    \"\"\"\n    An HTTP exception you can raise in your own code to show errors to the client.\n\n    This is for client errors, invalid authentication, invalid data, etc. Not for server\n    errors in your code.\n\n    Read more about it in the\n    [FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/).\n\n    ## Example\n\n    ```python\n    from fastapi import FastAPI, HTTPException\n\n    app = FastAPI()\n\n    items = {\"foo\": \"The Foo Wrestlers\"}\n\n\n    @app.get(\"/items/{item_id}\")\n    async def read_item(item_id: str):\n        if item_id not in items:\n            raise HTTPException(status_code=404, detail=\"Item not found\")\n        return {\"item\": items[item_id]}\n    ```\n    \"\"\"\n\n    def __init__(\n        self,\n        status_code: Annotated[\n            int,\n            Doc(\n                \"\"\"\n                HTTP status code to send to the client.\n                \"\"\"\n            ),\n        ],\n        detail: Annotated[\n            Any,\n            Doc(\n                \"\"\"\n                Any data to be sent to the client in the `detail` key of the JSON\n                response.\n                \"\"\"\n            ),\n        ] = None,\n        headers: Annotated[\n            Optional[Dict[str, str]],\n            Doc(\n                \"\"\"\n                Any headers to send to the client in the response.\n                \"\"\"\n            ),\n        ] = None,\n    ) -> None:\n        super().__init__(status_code=statu

{"status": "ok", "filepath": "fastapi/exceptions.py", "content": "class DependencyScopeError(FastAPIError):\n    \"\"\"\n    A dependency declared that it depends on another dependency with an invalid\n    (narrower) scope.\n    \"\"\"\n\n\nclass ValidationException(Exception):\n    def __init__(self, errors: Sequence[Any]) -> None:\n        self._errors = errors\n\n    def errors(self) -> Sequence[Any]:\n        return self._errors\n\n\nclass RequestValidationError(ValidationException):\n    def __init__(self, errors: Sequence[Any], *, body: Any = None) -> None:\n        super().__init__(errors)\n        self.body = body\n\n\nclass WebSocketRequestValidationError(ValidationException):\n    pass\n\n\nclass ResponseValidationError(ValidationException):\n    def __init__(self, errors: Sequence[Any], *, body: Any = None) -> None:\n        super().__init__(errors)\n        self.body = body\n\n    def __str__(self) -> str:\n        message = f\"{len(self._errors)} validation errors:\\n\"\n        for err in self._errors:\n            message += f\"  {err}\\n\"\n        return message\n", "start_line": 150, "end_line": 184, "total_lines": 184, "is_truncated": false}

{"status": "ok", "stdout": "fastapi/dependencies/models.py:32:class Dependant:\n", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/dependencies/models.py", "content": "\n@dataclass\nclass Dependant:\n    path_params: List[ModelField] = field(default_factory=list)\n    query_params: List[ModelField] = field(default_factory=list)\n    header_params: List[ModelField] = field(default_factory=list)\n    cookie_params: List[ModelField] = field(default_factory=list)\n    body_params: List[ModelField] = field(default_factory=list)\n    dependencies: List[\"Dependant\"] = field(default_factory=list)\n    name: Optional[str] = None\n    call: Optional[Callable[..., Any]] = None\n    request_param_name: Optional[str] = None\n    websocket_param_name: Optional[str] = None\n    http_connection_param_name: Optional[str] = None\n    response_param_name: Optional[str] = None\n    background_tasks_param_name: Optional[str] = None\n    security_scopes_param_name: Optional[str] = None\n    own_oauth_scopes: Optional[List[str]] = None\n    parent_oauth_scopes: Optional[List[str]] = None\n    use_cache: bool = True\n    path: Optional[str] = None\n    scope: Union[Literal[\"function\", \"request\"], None] = None\n\n    @cached_property\n    def oauth_scopes(self) -> List[str]:\n        scopes = self.parent_oauth_scopes.copy() if self.parent_oauth_scopes else []\n        # This doesn't use a set to preserve order, just in case\n        for scope in self.own_oauth_scopes or []:\n            if scope not in scopes:\n                scopes.append(scope)\n        return scopes\n\n    @cached_property\n    def cache_key(self) -> DependencyCacheKey:\n        scopes_for_cache = (\n            tuple(sorted(set(self.oauth_scopes or []))) if self._uses_scopes else ()\n        )\n        return (\n            self.call,\n            scopes_for_cache,\n            self.computed_scope or \"\",\n        )\n\n    @cached_property\n    def _uses_scopes(self) -> bool:\n        if self.own_oauth_scopes:\n            return True\n        if self.security_scopes_param_name is not None:\n           

{"status": "ok", "filepath": "fastapi/routing.py", "content": "            response_awaited = False\n            async with AsyncExitStack() as request_stack:\n                scope[\"fastapi_inner_astack\"] = request_stack\n                async with AsyncExitStack() as function_stack:\n                    scope[\"fastapi_function_astack\"] = function_stack\n                    response = await f(request)\n                await response(scope, receive, send)\n                # Continues customization\n                response_awaited = True\n            if not response_awaited:\n                raise FastAPIError(\n                    \"Response not awaited. There's a high chance that the \"\n                    \"application code is raising an exception and a dependency with yield \"\n                    \"has a block with a bare except, or a block with except Exception, \"\n                    \"and is not raising the exception again. Read more about it in the \"\n                    \"docs: https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/#dependencies-with-yield-and-except\"\n                )\n\n        # Same as in Starlette\n        await wrap_app_handling_exceptions(app, request)(scope, receive, send)\n\n    return app\n\n\n# Copy of starlette.routing.websocket_session modified to include the\n# dependencies' AsyncExitStack\ndef websocket_session(\n    func: Callable[[WebSocket], Awaitable[None]],\n) -> ASGIApp:\n    \"\"\"\n    Takes a coroutine `func(session)`, and returns an ASGI application.\n    \"\"\"\n    # assert asyncio.iscoroutinefunction(func), \"WebSocket endpoints must be async\"\n\n    async def app(scope: Scope, receive: Receive, send: Send) -> None:\n        session = WebSocket(scope, receive=receive, send=send)\n\n        async def app(scope: Scope, receive: Receive, send: Send) -> None:\n            async with AsyncExitStack() as request_stack:\n                scope[\"fastapi_inner_astack\"] = request_stack\n    

{"status": "ok", "filepath": "fastapi/routing.py", "content": "                It will be added to the generated OpenAPI (e.g. visible at `/docs`).\n\n                Read more about it in the\n                [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/).\n                \"\"\"\n            ),\n        ] = None,\n        openapi_extra: Annotated[\n            Optional[Dict[str, Any]],\n            Doc(\n                \"\"\"\n                Extra metadata to be included in the OpenAPI schema for this *path\n                operation*.\n\n                Read more about it in the\n                [FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema).\n                \"\"\"\n            ),\n        ] = None,\n        generate_unique_id_function: Annotated[\n            Callable[[APIRoute], str],\n            Doc(\n                \"\"\"\n                Customize the function used to generate unique IDs for the *path\n                operations* shown in the generated OpenAPI.\n\n                This is particularly useful when automatically generating clients or\n                SDKs for your API.\n\n                Read more about it in the\n                [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).\n                \"\"\"\n            ),\n        ] = Default(generate_unique_id),\n    ) -> Callable[[DecoratedCallable], DecoratedCallable]:\n        \"\"\"\n        Add a *path operation* using an HTTP TRACE operation.\n\n        ## Example\n\n        ```python\n        from fastapi import APIRouter, FastAPI\n        from pydantic import BaseModel\n\n        class Item(BaseModel):\n            name: str\n            description: str | None = None\n\n        app = FastAPI()\n        router = APIRouter()\n\n 

{"status": "ok", "stdout": "504:class APIRoute(routing.Route):\n673:class APIRouter(routing.Router):\n", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/routing.py", "content": "            child_scope[\"route\"] = self\n        return match, child_scope\n\n\nclass APIRoute(routing.Route):\n    def __init__(\n        self,\n        path: str,\n        endpoint: Callable[..., Any],\n        *,\n        response_model: Any = Default(None),\n        status_code: Optional[int] = None,\n        tags: Optional[List[Union[str, Enum]]] = None,\n        dependencies: Optional[Sequence[params.Depends]] = None,\n        summary: Optional[str] = None,\n        description: Optional[str] = None,\n        response_description: str = \"Successful Response\",\n        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,\n        deprecated: Optional[bool] = None,\n        name: Optional[str] = None,\n        methods: Optional[Union[Set[str], List[str]]] = None,\n        operation_id: Optional[str] = None,\n        response_model_include: Optional[IncEx] = None,\n        response_model_exclude: Optional[IncEx] = None,\n        response_model_by_alias: bool = True,\n        response_model_exclude_unset: bool = False,\n        response_model_exclude_defaults: bool = False,\n        response_model_exclude_none: bool = False,\n        include_in_schema: bool = True,\n        response_class: Union[Type[Response], DefaultPlaceholder] = Default(\n            JSONResponse\n        ),\n        dependency_overrides_provider: Optional[Any] = None,\n        callbacks: Optional[List[BaseRoute]] = None,\n        openapi_extra: Optional[Dict[str, Any]] = None,\n        generate_unique_id_function: Union[\n            Callable[[\"APIRoute\"], str], DefaultPlaceholder\n        ] = Default(generate_unique_id),\n    ) -> None:\n        self.path = path\n        self.endpoint = endpoint\n        if isinstance(response_model, DefaultPlaceholder):\n            return_annotation = get_typed_return_annotation(endpoint)\n            if lenient_issubclass(return_annotation, Response):\n                respo

{"status": "ok", "filepath": "fastapi/routing.py", "content": "                create_cloned_field(self.response_field)\n            )\n        else:\n            self.response_field = None  # type: ignore\n            self.secure_cloned_response_field = None\n        self.dependencies = list(dependencies or [])\n        self.description = description or inspect.cleandoc(self.endpoint.__doc__ or \"\")\n        # if a \"form feed\" character (page break) is found in the description text,\n        # truncate description text to the content preceding the first \"form feed\"\n        self.description = self.description.split(\"\\f\")[0].strip()\n        response_fields = {}\n        for additional_status_code, response in self.responses.items():\n            assert isinstance(response, dict), \"An additional response must be a dict\"\n            model = response.get(\"model\")\n            if model:\n                assert is_body_allowed_for_status_code(additional_status_code), (\n                    f\"Status code {additional_status_code} must not have a response body\"\n                )\n                response_name = f\"Response_{additional_status_code}_{self.unique_id}\"\n                response_field = create_model_field(\n                    name=response_name, type_=model, mode=\"serialization\"\n                )\n                response_fields[additional_status_code] = response_field\n        if response_fields:\n            self.response_fields: Dict[Union[int, str], ModelField] = response_fields\n        else:\n            self.response_fields = {}\n\n        assert callable(endpoint), \"An endpoint must be a callable\"\n        self.dependant = get_dependant(\n            path=self.path_format, call=self.endpoint, scope=\"function\"\n        )\n        for depends in self.dependencies[::-1]:\n            self.dependant.dependencies.insert(\n                0,\n                get_parameterless_sub_dependant(depends=depends, path=self.path_format),\n   

{"status": "ok", "filepath": "fastapi/routing.py", "content": "            dependant=self.dependant,\n            body_field=self.body_field,\n            status_code=self.status_code,\n            response_class=self.response_class,\n            response_field=self.secure_cloned_response_field,\n            response_model_include=self.response_model_include,\n            response_model_exclude=self.response_model_exclude,\n            response_model_by_alias=self.response_model_by_alias,\n            response_model_exclude_unset=self.response_model_exclude_unset,\n            response_model_exclude_defaults=self.response_model_exclude_defaults,\n            response_model_exclude_none=self.response_model_exclude_none,\n            dependency_overrides_provider=self.dependency_overrides_provider,\n            embed_body_fields=self._embed_body_fields,\n        )\n\n    def matches(self, scope: Scope) -> Tuple[Match, Scope]:\n        match, child_scope = super().matches(scope)\n        if match != Match.NONE:\n            child_scope[\"route\"] = self\n        return match, child_scope\n\n\nclass APIRouter(routing.Router):\n    \"\"\"\n    `APIRouter` class, used to group *path operations*, for example to structure\n    an app in multiple files. It would then be included in the `FastAPI` app, or\n    in another `APIRouter` (ultimately included in the app).\n\n    Read more about it in the\n    [FastAPI docs for Bigger Applications - Multiple Files](https://fastapi.tiangolo.com/tutorial/bigger-applications/).\n\n    ## Example\n\n    ```python\n    from fastapi import APIRouter, FastAPI\n\n    app = FastAPI()\n    router = APIRouter()\n\n\n    @router.get(\"/users/\", tags=[\"users\"])\n    async def read_users():\n        return [{\"username\": \"Rick\"}, {\"username\": \"Morty\"}]\n\n\n    app.include_router(router)\n    ```\n    \"\"\"\n\n    def __init__(\n        self,\n        *,\n        prefix: Annotated[str, Doc(\"An optional path prefix for the router.\")] =

{"status": "ok", "stdout": "289:def get_request_handler(\n", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/routing.py", "content": "def get_request_handler(\n    dependant: Dependant,\n    body_field: Optional[ModelField] = None,\n    status_code: Optional[int] = None,\n    response_class: Union[Type[Response], DefaultPlaceholder] = Default(JSONResponse),\n    response_field: Optional[ModelField] = None,\n    response_model_include: Optional[IncEx] = None,\n    response_model_exclude: Optional[IncEx] = None,\n    response_model_by_alias: bool = True,\n    response_model_exclude_unset: bool = False,\n    response_model_exclude_defaults: bool = False,\n    response_model_exclude_none: bool = False,\n    dependency_overrides_provider: Optional[Any] = None,\n    embed_body_fields: bool = False,\n) -> Callable[[Request], Coroutine[Any, Any, Response]]:\n    assert dependant.call is not None, \"dependant.call must be a function\"\n    is_coroutine = dependant.is_coroutine_callable\n    is_body_form = body_field and isinstance(\n        body_field.field_info, (params.Form, temp_pydantic_v1_params.Form)\n    )\n    if isinstance(response_class, DefaultPlaceholder):\n        actual_response_class: Type[Response] = response_class.value\n    else:\n        actual_response_class = response_class\n\n    async def app(request: Request) -> Response:\n        response: Union[Response, None] = None\n        file_stack = request.scope.get(\"fastapi_middleware_astack\")\n        assert isinstance(file_stack, AsyncExitStack), (\n            \"fastapi_middleware_astack not found in request scope\"\n        )\n\n        # Read body and auto-close files\n        try:\n            body: Any = None\n            if body_field:\n                if is_body_form:\n                    body = await request.form()\n                    file_stack.push_async_callback(body.close)\n                else:\n                    body_bytes = await request.body()\n                    if body_bytes:\n                        json_body: Any = Undefined\n                        

{"status": "ok", "filepath": "fastapi/routing.py", "content": "                    status_code if status_code else solved_result.response.status_code\n                )\n                if current_status_code is not None:\n                    response_args[\"status_code\"] = current_status_code\n                if solved_result.response.status_code:\n                    response_args[\"status_code\"] = solved_result.response.status_code\n                content = await serialize_response(\n                    field=response_field,\n                    response_content=raw_response,\n                    include=response_model_include,\n                    exclude=response_model_exclude,\n                    by_alias=response_model_by_alias,\n                    exclude_unset=response_model_exclude_unset,\n                    exclude_defaults=response_model_exclude_defaults,\n                    exclude_none=response_model_exclude_none,\n                    is_coroutine=is_coroutine,\n                )\n                response = actual_response_class(content, **response_args)\n                if not is_body_allowed_for_status_code(response.status_code):\n                    response.body = b\"\"\n                response.headers.raw.extend(solved_result.response.headers.raw)\n        if errors:\n            validation_error = RequestValidationError(\n                _normalize_errors(errors), body=body\n            )\n            raise validation_error\n\n        # Return response\n        assert response\n        return response\n\n    return app\n\n\ndef get_websocket_app(\n    dependant: Dependant,\n    dependency_overrides_provider: Optional[Any] = None,\n    embed_body_fields: bool = False,\n) -> Callable[[WebSocket], Coroutine[Any, Any, Any]]:\n    async def app(websocket: WebSocket) -> None:\n        async_exit_stack = websocket.scope.get(\"fastapi_inner_astack\")\n        assert isinstance(async_exit_stack, AsyncExitStack), (\n            \"fastapi_inner_asta

{"status": "ok", "filepath": "fastapi/routing.py", "content": "            # pydantic v1\n            response_content = _prepare_response_content(\n                response_content,\n                exclude_unset=exclude_unset,\n                exclude_defaults=exclude_defaults,\n                exclude_none=exclude_none,\n            )\n        if is_coroutine:\n            value, errors_ = field.validate(response_content, {}, loc=(\"response\",))\n        else:\n            value, errors_ = await run_in_threadpool(\n                field.validate, response_content, {}, loc=(\"response\",)\n            )\n        if isinstance(errors_, list):\n            errors.extend(errors_)\n        elif errors_:\n            errors.append(errors_)\n        if errors:\n            raise ResponseValidationError(\n                errors=_normalize_errors(errors), body=response_content\n            )\n\n        if hasattr(field, \"serialize\"):\n            return field.serialize(\n                value,\n                include=include,\n                exclude=exclude,\n                by_alias=by_alias,\n                exclude_unset=exclude_unset,\n                exclude_defaults=exclude_defaults,\n                exclude_none=exclude_none,\n", "start_line": 230, "end_line": 260, "total_lines": 4517, "is_truncated": false}

{"status": "ok", "filepath": "fastapi/routing.py", "content": ") -> Lifespan[Any]:\n    @asynccontextmanager\n    async def merged_lifespan(\n        app: AppType,\n    ) -> AsyncIterator[Optional[Mapping[str, Any]]]:\n        async with original_context(app) as maybe_original_state:\n            async with nested_context(app) as maybe_nested_state:\n                if maybe_nested_state is None and maybe_original_state is None:\n                    yield None  # old ASGI compatibility\n                else:\n                    yield {**(maybe_nested_state or {}), **(maybe_original_state or {})}\n\n    return merged_lifespan  # type: ignore[return-value]\n\n\nasync def serialize_response(\n    *,\n    field: Optional[ModelField] = None,\n    response_content: Any,\n    include: Optional[IncEx] = None,\n    exclude: Optional[IncEx] = None,\n    by_alias: bool = True,\n    exclude_unset: bool = False,\n    exclude_defaults: bool = False,\n    exclude_none: bool = False,\n    is_coroutine: bool = True,\n) -> Any:\n    if field:\n        errors = []\n        if not hasattr(field, \"serialize\"):\n            # pydantic v1\n            response_content = _prepare_response_content(\n                response_content,\n                exclude_unset=exclude_unset,\n                exclude_defaults=exclude_defaults,\n                exclude_none=exclude_none,\n            )\n        if is_coroutine:\n            value, errors_ = field.validate(response_content, {}, loc=(\"response\",))\n        else:\n            value, errors_ = await run_in_threadpool(\n                field.validate, response_content, {}, loc=(\"response\",)\n            )\n        if isinstance(errors_, list):\n            errors.extend(errors_)\n        elif errors_:\n            errors.append(errors_)\n        if errors:\n            raise ResponseValidationError(\n                errors=_normalize_errors(errors), body=response_content\n            )\n\n        if hasattr(field, \"serialize\"):\n      

{"status": "ok", "stdout": "4:import inspect\n", "exit_code": 0}

{"status": "error", "error_type": "CommandError", "error_message": "Command failed with exit code 1", "details": {"stdout": "", "stderr": "", "exit_code": 1}}

{"status": "ok", "stdout": "1:import dataclasses\n2:import email.message\n3:import functools\n4:import inspect\n5:import json\n6:from contextlib import AsyncExitStack, asynccontextmanager\n7:from enum import Enum, IntEnum\n8:from typing import (\n26:from annotated_doc import Doc\n27:from fastapi import params, temp_pydantic_v1_params\n28:from fastapi._compat import (\n36:from fastapi.datastructures import Default, DefaultPlaceholder\n37:from fastapi.dependencies.models import Dependant\n38:from fastapi.dependencies.utils import (\n47:from fastapi.encoders import jsonable_encoder\n48:from fastapi.exceptions import (\n54:from fastapi.types import DecoratedCallable, IncEx\n55:from fastapi.utils import (\n62:from pydantic import BaseModel\n63:from starlette import routing\n", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/exceptions.py", "content": "class DependencyScopeError(FastAPIError):\n    \"\"\"\n    A dependency declared that it depends on another dependency with an invalid\n    (narrower) scope.\n    \"\"\"\n\n\nclass ValidationException(Exception):\n    def __init__(self, errors: Sequence[Any]) -> None:\n        self._errors = errors\n\n    def errors(self) -> Sequence[Any]:\n        return self._errors\n\n\nclass RequestValidationError(ValidationException):\n    def __init__(self, errors: Sequence[Any], *, body: Any = None) -> None:\n        super().__init__(errors)\n        self.body = body\n\n\nclass WebSocketRequestValidationError(ValidationException):\n    pass\n\n\nclass ResponseValidationError(ValidationException):\n    def __init__(self, errors: Sequence[Any], *, body: Any = None) -> None:\n        super().__init__(errors)\n        self.body = body\n\n    def __str__(self) -> str:\n        message = f\"{len(self._errors)} validation errors:\\n\"\n        for err in self._errors:\n            message += f\"  {err}\\n\"\n        return message\n", "start_line": 150, "end_line": 184, "total_lines": 184, "is_truncated": false}

{"status": "ok", "filepath": "fastapi/exceptions.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/exceptions.py\n+++ b/fastapi/exceptions.py\n@@ -157,9 +157,19 @@\n class ValidationException(Exception):\n     def __init__(self, errors: Sequence[Any]) -> None:\n         self._errors = errors\n+        self.endpoint_context: Optional[str] = None\n \n     def errors(self) -> Sequence[Any]:\n         return self._errors\n+\n+    def __str__(self) -> str:\n+        message = f\"{len(self._errors)} validation errors:\\n\"\n+        for err in self._errors:\n+            message += f\"  {err}\\n\"\n+        if self.endpoint_context:\n+            message += f\"\\n{self.endpoint_context}\"\n+        return message\n+\n \n \n class RequestValidationError(ValidationException):\n@@ -168,17 +178,13 @@\n         self.body = body\n \n \n+\n class WebSocketRequestValidationError(ValidationException):\n     pass\n+\n \n \n class ResponseValidationError(ValidationException):\n     def __init__(self, errors: Sequence[Any], *, body: Any = None) -> None:\n         super().__init__(errors)\n         self.body = body\n-\n-    def __str__(self) -> str:\n-        message = f\"{len(self._errors)} validation errors:\\n\"\n-        for err in self._errors:\n-            message += f\"  {err}\\n\"\n-        return message\n", "is_truncated": false}

{"status": "ok", "stdout": "4:import inspect\n", "exit_code": 0}

{"status": "error", "error_type": "CommandError", "error_message": "Command failed with exit code 1", "details": {"stdout": "", "stderr": "", "exit_code": 1}}

{"status": "error", "error_type": "CommandError", "error_message": "Command failed with exit code 1", "details": {"stdout": "", "stderr": "", "exit_code": 1}}

{"status": "ok", "stdout": "", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/routing.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/routing.py\n+++ b/fastapi/routing.py\n@@ -3,6 +3,7 @@\n import functools\n import inspect\n import json\n+import os\n from contextlib import AsyncExitStack, asynccontextmanager\n from enum import Enum, IntEnum\n from typing import (\n", "is_truncated": false}

{"status": "ok", "filepath": "fastapi/routing.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/routing.py\n+++ b/fastapi/routing.py\n@@ -662,6 +662,7 @@\n             response_model_exclude_none=self.response_model_exclude_none,\n             dependency_overrides_provider=self.dependency_overrides_provider,\n             embed_body_fields=self._embed_body_fields,\n+            route=self,\n         )\n \n     def matches(self, scope: Scope) -> Tuple[Match, Scope]:\n", "is_truncated": false}

{"status": "ok", "filepath": "fastapi/routing.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/routing.py\n+++ b/fastapi/routing.py\n@@ -301,6 +301,7 @@\n     response_model_exclude_none: bool = False,\n     dependency_overrides_provider: Optional[Any] = None,\n     embed_body_fields: bool = False,\n+    route: Optional[Any] = None,\n ) -> Callable[[Request], Coroutine[Any, Any, Response]]:\n     assert dependant.call is not None, \"dependant.call must be a function\"\n     is_coroutine = dependant.is_coroutine_callable\n", "is_truncated": false}

{"status": "ok", "filepath": "fastapi/routing.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/routing.py\n+++ b/fastapi/routing.py\n@@ -319,6 +319,35 @@\n         assert isinstance(file_stack, AsyncExitStack), (\n             \"fastapi_middleware_astack not found in request scope\"\n         )\n+\n+        # Cache endpoint context\n+        endpoint_context: Optional[str] = None\n+        if route:\n+            if not hasattr(route, \"_endpoint_context\"):\n+                endpoint = route.endpoint\n+                try:\n+                    # Get the file path and line number of the endpoint function\n+                    source_file = inspect.getsourcefile(endpoint)\n+                    lines, start_line = inspect.getsourcelines(endpoint)\n+                    # Get the function name\n+                    func_name = endpoint.__name__\n+                    # Get the HTTP method and route path\n+                    method = \"UNKNOWN\"\n+                    if hasattr(route, \"methods\"):\n+                        method = list(route.methods)[0] if route.methods else \"UNKNOWN\"\n+                    path = route.path\n+                    \n+                    endpoint_context = (\n+                        f\"File \\\"{source_file}\\\", line {start_line}, in {func_name}\\n\"\n+                        f\"    {method} {path}\"\n+                    )\n+                except Exception:\n+                    endpoint_context = None\n+            endpoint_context = getattr(route, \"_endpoint_context\", None)\n+            # Wait, I should actually cache it on the route object\n+            if not hasattr(route, \"_endpoint_context\"):\n+                # (The logic above should be here)\n+                pass\n \n         # Read body and auto-close files\n         try:\n", "is_truncated": false}

{"status": "ok", "filepath": "fastapi/routing.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/routing.py\n+++ b/fastapi/routing.py\n@@ -326,28 +326,20 @@\n             if not hasattr(route, \"_endpoint_context\"):\n                 endpoint = route.endpoint\n                 try:\n-                    # Get the file path and line number of the endpoint function\n                     source_file = inspect.getsourcefile(endpoint)\n-                    lines, start_line = inspect.getsourcelines(endpoint)\n-                    # Get the function name\n+                    _, start_line = inspect.getsourcelines(endpoint)\n                     func_name = endpoint.__name__\n-                    # Get the HTTP method and route path\n                     method = \"UNKNOWN\"\n                     if hasattr(route, \"methods\"):\n                         method = list(route.methods)[0] if route.methods else \"UNKNOWN\"\n                     path = route.path\n-                    \n-                    endpoint_context = (\n+                    route._endpoint_context = (\n                         f\"File \\\"{source_file}\\\", line {start_line}, in {func_name}\\n\"\n                         f\"    {method} {path}\"\n                     )\n                 except Exception:\n-                    endpoint_context = None\n+                    route._endpoint_context = None\n             endpoint_context = getattr(route, \"_endpoint_context\", None)\n-            # Wait, I should actually cache it on the route object\n-            if not hasattr(route, \"_endpoint_context\"):\n-                # (The logic above should be here)\n-                pass\n \n         # Read body and auto-close files\n         try:\n", "is_truncated": false}