fastapi_14448

{"status": "ok", "stdout": "fastapi/dependencies/models.py:4:from functools import cached_property, partial\nfastapi/_compat/main.py:2:from functools import lru_cache\nfastapi/routing.py:3:import functools\nfastapi/routing.py:92:        func if is_async_callable(func) else functools.partial(run_in_threadpool, func)  # type:ignore\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "fastapi/routing.py:65:from starlette._utils import is_async_callable\nfastapi/routing.py:92:        func if is_async_callable(func) else functools.partial(run_in_threadpool, func)  # type:ignore\n", "stderr": "", "exit_code": 0}

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

{"status": "ok", "stdout": "fastapi/dependencies/models.py:95:    def is_async_gen_callable(self) -> bool:\nfastapi/dependencies/models.py:114:        if self.is_gen_callable or self.is_async_gen_callable:\nfastapi/dependencies/utils.py:270:                (dependant.is_gen_callable or dependant.is_async_gen_callable)\nfastapi/dependencies/utils.py:553:    elif dependant.is_async_gen_callable:\nfastapi/dependencies/utils.py:636:            use_sub_dependant.is_gen_callable or use_sub_dependant.is_async_gen_callable\nfastapi/routing.py:65:from starlette._utils import is_async_callable\nfastapi/routing.py:92:        func if is_async_callable(func) else functools.partial(run_in_threadpool, func)  # type:ignore\n", "stderr": "", "exit_code": 0}

{"status": "error", "error_type": "FileReadError", "error_message": "'>' not supported between instances of 'int' and 'str'"}

{"status": "ok", "filepath": "fastapi/dependencies/models.py", "content": "import inspect\nimport sys\nfrom dataclasses import dataclass, field\nfrom functools import cached_property, partial\nfrom typing import Any, Callable, List, Optional, Sequence, Union\n\nfrom fastapi._compat import ModelField\nfrom fastapi.security.base import SecurityBase\nfrom fastapi.types import DependencyCacheKey\nfrom typing_extensions import Literal\n\nif sys.version_info >= (3, 13):  # pragma: no cover\n    from inspect import iscoroutinefunction\nelse:  # pragma: no cover\n    from asyncio import iscoroutinefunction\n\n\n@dataclass\nclass SecurityRequirement:\n    security_scheme: SecurityBase\n    scopes: Optional[Sequence[str]] = None\n\n\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    security_requirements: List[SecurityRequirement] = 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_scop

{"status": "ok", "stdout": "65:from starlette._utils import is_async_callable\n92:        func if is_async_callable(func) else functools.partial(run_in_threadpool, func)  # type:ignore\n", "stderr": "", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/routing.py", "content": "import dataclasses\nimport email.message\nimport functools\nimport inspect\nimport json\nfrom contextlib import AsyncExitStack, asynccontextmanager\nfrom enum import Enum, IntEnum\nfrom typing import (\n    Any,\n    AsyncIterator,\n    Awaitable,\n    Callable,\n    Collection,\n    Coroutine,\n    Dict,\n    List,\n    Mapping,\n    Optional,\n    Sequence,\n    Set,\n    Tuple,\n    Type,\n    Union,\n)\n\nfrom annotated_doc import Doc\nfrom fastapi import params, temp_pydantic_v1_params\nfrom fastapi._compat import (\n    ModelField,\n    Undefined,\n    _get_model_config,\n    _model_dump,\n    _normalize_errors,\n    lenient_issubclass,\n)\nfrom fastapi.datastructures import Default, DefaultPlaceholder\nfrom fastapi.dependencies.models import Dependant\nfrom fastapi.dependencies.utils import (\n    _should_embed_body_fields,\n    get_body_field,\n    get_dependant,\n    get_flat_dependant,\n    get_parameterless_sub_dependant,\n    get_typed_return_annotation,\n    solve_dependencies,\n)\nfrom fastapi.encoders import jsonable_encoder\nfrom fastapi.exceptions import (\n    FastAPIError,\n    RequestValidationError,\n    ResponseValidationError,\n    WebSocketRequestValidationError,\n)\nfrom fastapi.types import DecoratedCallable, IncEx\nfrom fastapi.utils import (\n    create_cloned_field,\n    create_model_field,\n    generate_unique_id,\n    get_value_or_default,\n    is_body_allowed_for_status_code,\n)\nfrom pydantic import BaseModel\nfrom starlette import routing\nfrom starlette._exception_handler import wrap_app_handling_exceptions\nfrom starlette._utils import is_async_callable\nfrom starlette.concurrency import run_in_threadpool\nfrom starlette.exceptions import HTTPException\nfrom starlette.requests import Request\nfrom starlette.responses import JSONResponse, Response\nfrom starlette.routing import (\n    BaseRoute,\n    Match,\n    compile_path,\n    get_name,\n)\nfrom starlette.routing im

{"status": "ok", "stdout": "65:from starlette._utils import is_async_callable\n92:        func if is_async_callable(func) else functools.partial(run_in_threadpool, func)  # type:ignore\n", "stderr": "", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/dependencies/models.py", "content": "import inspect\nimport sys\nfrom dataclasses import dataclass, field\nfrom functools import cached_property, partial\nfrom typing import Any, Callable, List, Optional, Sequence, Union\n\nfrom fastapi._compat import ModelField\nfrom fastapi.security.base import SecurityBase\nfrom fastapi.types import DependencyCacheKey\nfrom typing_extensions import Literal\n\nif sys.version_info >= (3, 13):  # pragma: no cover\n    from inspect import iscoroutinefunction\nelse:  # pragma: no cover\n    from asyncio import iscoroutinefunction\n\n\n@dataclass\nclass SecurityRequirement:\n    security_scheme: SecurityBase\n    scopes: Optional[Sequence[str]] = None\n\n\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    security_requirements: List[SecurityRequirement] = 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_scop

{"status": "ok", "stdout": "65:from starlette._utils import is_async_callable\n92:        func if is_async_callable(func) else functools.partial(run_in_threadpool, func)  # type:ignore\n", "stderr": "", "exit_code": 0}

{"status": "error", "error_type": "CommandError", "error_message": "grep: starlette/_utils.py: No such file or directory\n", "details": {"stdout": "", "stderr": "grep: starlette/_utils.py: No such file or directory\n", "exit_code": 2}}

{"status": "ok", "stdout": "/kaggle/tmp/envs/overlays/starlette-0.50.0-py3-none-any/starlette/_utils.py\n", "stderr": "", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/dependencies/models.py", "content": "import inspect\nimport sys\nfrom dataclasses import dataclass, field\nfrom functools import cached_property, partial\nfrom typing import Any, Callable, List, Optional, Sequence, Union\n\nfrom fastapi._compat import ModelField\nfrom fastapi.security.base import SecurityBase\nfrom fastapi.types import DependencyCacheKey\nfrom typing_extensions import Literal\n\nif sys.version_info >= (3, 13):  # pragma: no cover\n    from inspect import iscoroutinefunction\nelse:  # pragma: no cover\n    from asyncio import iscoroutinefunction\n\n\n@dataclass\nclass SecurityRequirement:\n    security_scheme: SecurityBase\n    scopes: Optional[Sequence[str]] = None\n\n\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    security_requirements: List[SecurityRequirement] = 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_scop

{"status": "ok", "stdout": "65:from starlette._utils import is_async_callable\n92:        func if is_async_callable(func) else functools.partial(run_in_threadpool, func)  # type:ignore\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "is_async_callable\n", "stderr": "", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/dependencies/models.py", "content": "import inspect\nimport sys\nfrom dataclasses import dataclass, field\nfrom functools import cached_property, partial\nfrom typing import Any, Callable, List, Optional, Sequence, Union\n\nfrom fastapi._compat import ModelField\nfrom fastapi.security.base import SecurityBase\nfrom fastapi.types import DependencyCacheKey\nfrom typing_extensions import Literal\n\nif sys.version_info >= (3, 13):  # pragma: no cover\n    from inspect import iscoroutinefunction\nelse:  # pragma: no cover\n    from asyncio import iscoroutinefunction\n\n\n@dataclass\nclass SecurityRequirement:\n    security_scheme: SecurityBase\n    scopes: Optional[Sequence[str]] = None\n\n\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    security_requirements: List[SecurityRequirement] = 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_scop

{"status": "ok", "stdout": "65:from starlette._utils import is_async_callable\n92:        func if is_async_callable(func) else functools.partial(run_in_threadpool, func)  # type:ignore\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "<function is_async_callable at 0x79d41b5a84a0>\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "def is_async_callable(obj: Any) -> Any:\n    while isinstance(obj, functools.partial):\n        obj = obj.func\n\n    return iscoroutinefunction(obj) or (callable(obj) and iscoroutinefunction(obj.__call__))\n\n", "stderr": "", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/dependencies/models.py", "content": "import inspect\nimport sys\nfrom dataclasses import dataclass, field\nfrom functools import cached_property, partial\nfrom typing import Any, Callable, List, Optional, Sequence, Union\n\nfrom fastapi._compat import ModelField\nfrom fastapi.security.base import SecurityBase\nfrom fastapi.types import DependencyCacheKey\nfrom typing_extensions import Literal\n\nif sys.version_info >= (3, 13):  # pragma: no cover\n    from inspect import iscoroutinefunction\nelse:  # pragma: no cover\n    from asyncio import iscoroutinefunction\n\n\n@dataclass\nclass SecurityRequirement:\n    security_scheme: SecurityBase\n    scopes: Optional[Sequence[str]] = None\n\n\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    security_requirements: List[SecurityRequirement] = 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_scop

{"status": "ok", "stdout": "65:from starlette._utils import is_async_callable\n92:        func if is_async_callable(func) else functools.partial(run_in_threadpool, func)  # type:ignore\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "def is_async_callable(obj: Any) -> Any:\n    while isinstance(obj, functools.partial):\n        obj = obj.func\n\n    return iscoroutinefunction(obj) or (callable(obj) and iscoroutinefunction(obj.__call__))\n\n", "stderr": "", "exit_code": 0}

{"status": "ok", "filepath": "fastapi/dependencies/models.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/dependencies/models.py\n+++ b/fastapi/dependencies/models.py\n@@ -79,9 +79,14 @@\n     def _unwrapped_call(self) -> Any:\n         if self.call is None:\n             return self.call  # pragma: no cover\n-        unwrapped = inspect.unwrap(self.call)\n-        if isinstance(unwrapped, partial):\n-            unwrapped = unwrapped.func\n+        unwrapped = self.call\n+        while True:\n+            if isinstance(unwrapped, partial):\n+                unwrapped = unwrapped.func\n+            elif hasattr(unwrapped, \"__wrapped__\"):\n+                unwrapped = unwrapped.__wrapped__\n+            else:\n+                break\n         return unwrapped\n \n     @cached_property\n", "is_truncated": false}

{"status": "ok", "stdout": "65:from starlette._utils import is_async_callable\n92:        func if is_async_callable(func) else functools.partial(run_in_threadpool, func)  # type:ignore\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "def is_async_callable(obj: Any) -> Any:\n    while isinstance(obj, functools.partial):\n        obj = obj.func\n\n    return iscoroutinefunction(obj) or (callable(obj) and iscoroutinefunction(obj.__call__))\n\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "65:from starlette._utils import is_async_callable\n92:        func if is_async_callable(func) else functools.partial(run_in_threadpool, func)  # type:ignore\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "def is_async_callable(obj: Any) -> Any:\n    while isinstance(obj, functools.partial):\n        obj = obj.func\n\n    return iscoroutinefunction(obj) or (callable(obj) and iscoroutinefunction(obj.__call__))\n\n", "stderr": "", "exit_code": 0}

{"status": "ok", "patch_size": 810, "files_changed": 1}