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": "fastapi/dependencies/utils.py:131:    return get_dependant(\nfastapi/dependencies/utils.py:231:def get_dependant(\nfastapi/dependencies/utils.py:283:            sub_dependant = get_dependant(\nfastapi/dependencies/utils.py:610:            use_sub_dependant = get_dependant(\nfastapi/routing.py:41:    get_dependant,\nfastapi/routing.py:477:        self.dependant = get_dependant(\nfastapi/routing.py:630:        self.dependant = get_dependant(\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/utils.py", "content": "import dataclasses\nimport inspect\nfrom contextlib import AsyncExitStack, contextmanager\nfrom copy import copy, deepcopy\nfrom dataclasses import dataclass\nfrom typing import (\n    Any,\n    Callable,\n    Coroutine,\n    Dict,\n    ForwardRef,\n    List,\n    Mapping,\n    Optional,\n    Sequence,\n    Tuple,\n    Type,\n    Union,\n    cast,\n)\n\nimport anyio\nfrom fastapi import params\nfrom fastapi._compat import (\n    PYDANTIC_V2,\n    ModelField,\n    RequiredParam,\n    Undefined,\n    _is_error_wrapper,\n    _is_model_class,\n    copy_field_info,\n    create_body_model,\n    evaluate_forwardref,\n    field_annotation_is_scalar,\n    get_annotation_from_field_info,\n    get_cached_model_fields,\n    get_missing_field_error,\n    is_bytes_field,\n    is_bytes_sequence_field,\n    is_scalar_field,\n    is_scalar_sequence_field,\n    is_sequence_field,\n    is_uploadfile_or_nonable_uploadfile_annotation,\n    is_uploadfile_sequence_annotation,\n    lenient_issubclass,\n    may_v1,\n    sequence_types,\n    serialize_sequence_value,\n    value_is_sequence,\n)\nfrom fastapi._compat.shared import annotation_is_pydantic_v1\nfrom fastapi.background import BackgroundTasks\nfrom fastapi.concurrency import (\n    asynccontextmanager,\n    contextmanager_in_threadpool,\n)\nfrom fastapi.dependencies.models import Dependant, SecurityRequirement\nfrom fastapi.exceptions import DependencyScopeError\nfrom fastapi.logger import logger\nfrom fastapi.security.base import SecurityBase\nfrom fastapi.security.oauth2 import SecurityScopes\nfrom fastapi.types import DependencyCacheKey\nfrom fastapi.utils import create_model_field, get_path_param_names\nfrom pydantic import BaseModel\nfrom pydantic.fields import FieldInfo\nfrom starlette.background import BackgroundTasks as StarletteBackgroundTasks\nfrom starlette.concurrency import run_in_threadpool\nfrom starlette.datastructures import (\n    FormData,\n

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

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

{"status": "ok", "stdout": "231:def get_dependant(\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "def get_dependant(\n    *,\n    path: str,\n    call: Callable[..., Any],\n    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    scope: Union[Literal[\"function\", \"request\"], None] = None,\n) -> Dependant:\n    dependant = Dependant(\n        call=call,\n        name=name,\n        path=path,\n        use_cache=use_cache,\n        scope=scope,\n        own_oauth_scopes=own_oauth_scopes,\n        parent_oauth_scopes=parent_oauth_scopes,\n    )\n    current_scopes = (parent_oauth_scopes or []) + (own_oauth_scopes or [])\n    path_param_names = get_path_param_names(path)\n    endpoint_signature = get_typed_signature(call)\n    signature_params = endpoint_signature.parameters\n    if isinstance(call, SecurityBase):\n        security_requirement = SecurityRequirement(\n            security_scheme=call, scopes=current_scopes\n        )\n        dependant.security_requirements.append(security_requirement)\n    for param_name, param in signature_params.items():\n        is_path_param = param_name in path_param_names\n        param_details = analyze_param(\n            param_name=param_name,\n            annotation=param.annotation,\n            value=param.default,\n            is_path_param=is_path_param,\n        )\n        if param_details.depends is not None:\n            assert param_details.depends.dependency\n            if (\n                (dependant.is_gen_callable or dependant.is_async_gen_callable)\n                and dependant.computed_scope == \"request\"\n                and param_details.depends.scope == \"function\"\n            ):\n                assert dependant.call\n                raise DependencyScopeError(\n                    f'The dependency \"{dependant.call.__name__}\" has a scope of '\n                    '\"request\", it cannot depend on dependencies with scope \"function\".'\n                )\n

{"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": "fastapi/dependencies/models.py:102:    def is_coroutine_callable(self) -> bool:\nfastapi/dependencies/utils.py:646:        elif use_sub_dependant.is_coroutine_callable:\nfastapi/routing.py:305:    is_coroutine = dependant.is_coroutine_callable\n", "stderr": "", "exit_code": 0}

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

{"status": "ok", "stdout": "        return unwrapped\n\n    @cached_property\n    def is_gen_callable(self) -> bool:\n        if inspect.isgeneratorfunction(self._unwrapped_call):\n            return True\n        dunder_call = getattr(self._unwrapped_call, \"__call__\", None)  # noqa: B004\n        return inspect.isgeneratorfunction(dunder_call)\n\n    @cached_property\n    def is_async_gen_callable(self) -> bool:\n        if inspect.isasyncgenfunction(self._unwrapped_call):\n            return True\n        dunder_call = getattr(self._unwrapped_call, \"__call__\", None)  # noqa: B004\n        return inspect.isasyncgenfunction(dunder_call)\n\n    @cached_property\n    def is_coroutine_callable(self) -> bool:\n        if inspect.isroutine(self._unwrapped_call):\n            return iscoroutinefunction(self._unwrapped_call)\n        if inspect.isclass(self._unwrapped_call):\n            return False\n        dunder_call = getattr(self._unwrapped_call, \"__call__\", None)  # noqa: B004\n        return iscoroutinefunction(dunder_call)\n\n    @cached_property\n    def computed_scope(self) -> Union[str, None]:\n        if self.scope:\n            return self.scope\n        if self.is_gen_callable or self.is_async_gen_callable:\n            return \"request\"\n        return None\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "                return True\n        return False\n\n    @cached_property\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", "stderr": "", "exit_code": 0}

{"status": "error", "error_type": "CommandError", "error_message": "Traceback (most recent call last):\n  File \"/tmp/repro.py\", line 49, in <module>\n    test_callable(async_wrapped_sync, True)\n    ~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/tmp/repro.py\", line 44, in test_callable\n    assert actual_async == expected_async\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\nAssertionError\n", "details": {"stdout": "Call: <function async_func at 0x7b91a9275620>, Expected Async: True, Actual: True\nCall: <function sync_func at 0x7b91a8f4b6a0>, Expected Async: False, Actual: False\nCall: <function async_func at 0x7b91a7bd9800>, Expected Async: True, Actual: True\nCall: <function sync_func at 0x7b91a7bd9760>, Expected Async: True, Actual: False\n", "stderr": "Traceback (most recent call last):\n  File \"/tmp/repro.py\", line 49, in <module>\n    test_callable(async_wrapped_sync, True)\n    ~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/tmp/repro.py\", line 44, in test_callable\n    assert actual_async == expected_async\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\nAssertionError\n", "exit_code": 1}}

{"status": "error", "error_type": "CommandError", "error_message": "Traceback (most recent call last):\n  File \"/tmp/repro.py\", line 48, in <module>\n    test_callable(async_wrapped_sync, True)\n    ~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/tmp/repro.py\", line 42, in test_callable\n    assert actual_async == expected_async\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\nAssertionError\n", "details": {"stdout": "Testing...\nCall: <function async_func at 0x7cd5f0ec37e0>, Expected Async: True, Actual: True\nCall: <function sync_func at 0x7cd5f0d3b560>, Expected Async: False, Actual: False\nCall: <function async_func at 0x7cd5f093c4a0>, Expected Async: True, Actual: True\nCall: <function sync_func at 0x7cd5f093ca40>, Expected Async: True, Actual: False\n", "stderr": "Traceback (most recent call last):\n  File \"/tmp/repro.py\", line 48, in <module>\n    test_callable(async_wrapped_sync, True)\n    ~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/tmp/repro.py\", line 42, in test_callable\n    assert actual_async == expected_async\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\nAssertionError\n", "exit_code": 1}}

{"status": "error", "error_type": "CommandError", "error_message": "Traceback (most recent call last):\n  File \"/tmp/repro.py\", line 58, in <module>\n    raise e\n  File \"/tmp/repro.py\", line 49, in <module>\n    test_callable(async_wrapped_sync, True)\n    ~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/tmp/repro.py\", line 42, in test_callable\n    assert actual_async == expected_async\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\nAssertionError\n", "details": {"stdout": "Testing...\nCall: <function async_func at 0x7b6f44faf7e0>, Expected Async: True, Actual: True\nCall: <function sync_func at 0x7b6f44e27560>, Expected Async: False, Actual: False\nCall: <function async_func at 0x7b6f449ec4a0>, Expected Async: True, Actual: True\nCall: <function sync_func at 0x7b6f449eca40>, Expected Async: True, Actual: False\nAssertion failed!\n", "stderr": "Traceback (most recent call last):\n  File \"/tmp/repro.py\", line 58, in <module>\n    raise e\n  File \"/tmp/repro.py\", line 49, in <module>\n    test_callable(async_wrapped_sync, True)\n    ~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/tmp/repro.py\", line 42, in test_callable\n    assert actual_async == expected_async\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\nAssertionError\n", "exit_code": 1}}

{"status": "ok", "stdout": "Testing...\nCall: <function async_func at 0x7d3d26fcf7e0>, Expected Async: True, Actual: True\nCall: <function sync_func at 0x7d3d26e47560>, Expected Async: False, Actual: False\nCall: <function async_func at 0x7d3d26a004a0>, Expected Async: True, Actual: True\nAll tests passed!\n", "stderr": "", "exit_code": 0}

{"status": "ok", "stdout": "                return True\n        return False\n\n    @cached_property\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        return unwrapped\n\n    @cached_property\n    def is_gen_callable(self) -> bool:\n        if inspect.isgeneratorfunction(self._unwrapped_call):\n            return True\n        dunder_call = getattr(self._unwrapped_call, \"__call__\", None)  # noqa: B004\n        return inspect.isgeneratorfunction(dunder_call)\n\n    @cached_property\n    def is_async_gen_callable(self) -> bool:\n        if inspect.isasyncgenfunction(self._unwrapped_call):\n            return True\n        dunder_call = getattr(self._unwrapped_call, \"__call__\", None)  # noqa: B004\n        return inspect.isasyncgenfunction(dunder_call)\n\n    @cached_property\n    def is_coroutine_callable(self) -> bool:\n        if inspect.isroutine(self._unwrapped_call):\n            return iscoroutinefunction(self._unwrapped_call)\n        if inspect.isclass(self._unwrapped_call):\n            return False\n        dunder_call = getattr(self._unwrapped_call, \"__call__\", None)  # noqa: B004\n        return iscoroutinefunction(dunder_call)\n\n    @cached_property\n    def computed_scope(self) -> Union[str, None]:\n        if self.scope:\n            return self.scope\n        if self.is_gen_callable or self.is_async_gen_callable:\n            return \"request\"\n        return None\n", "stderr": "", "exit_code": 0}

{"status": "error", "error_type": "CommandError", "error_message": "Traceback (most recent call last):\n  File \"/tmp/repro.py\", line 58, in <module>\n    raise e\n  File \"/tmp/repro.py\", line 49, in <module>\n    test_callable(async_wrapped_sync, True)\n    ~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/tmp/repro.py\", line 42, in test_callable\n    assert actual_async == expected_async\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\nAssertionError\n", "details": {"stdout": "Testing...\nCall: <function async_func at 0x7c5b28bab7e0>, Expected Async: True, Actual: True\nCall: <function sync_func at 0x7c5b28a23560>, Expected Async: False, Actual: False\nCall: <function async_func at 0x7c5b285e84a0>, Expected Async: True, Actual: True\nCall: <function sync_func at 0x7c5b285e8a40>, Expected Async: True, Actual: False\nAssertion failed!\n", "stderr": "Traceback (most recent call last):\n  File \"/tmp/repro.py\", line 58, in <module>\n    raise e\n  File \"/tmp/repro.py\", line 49, in <module>\n    test_callable(async_wrapped_sync, True)\n    ~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/tmp/repro.py\", line 42, in test_callable\n    assert actual_async == expected_async\n           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\nAssertionError\n", "exit_code": 1}}

{"status": "ok", "stdout": "Testing...\nCall: <function async_func at 0x7d33bd0d37e0>, Expected Async: True, Actual: True\nCall: <function sync_func at 0x7d33bcf4b560>, Expected Async: False, Actual: False\nCall: <function async_func at 0x7d33bcb044a0>, Expected Async: True, Actual: True\nAll tests passed!\n", "stderr": "", "exit_code": 0}