fastapi_14448

{
 "command": "grep -rn \"is_async_gen\" fastapi | head -20"
}

{
 "command": "grep -rn \"is_async_callable\" fastapi | head -20"
}

{
 "command": "grep -rn \"is_async\" fastapi | head -20"
}

{
 "command": "sed -n '80,130p' fastapi/dependencies/models.py"
}

{
 "command": "sed -n '60,100p' fastapi/dependencies/models.py"
}

{
 "command": "grep -n \"is_async_callable\" fastapi/routing.py"
}

{
 "command": "sed -n '80,100p' fastapi/routing.py"
}

{
 "command": "cat /workspace/fastapi/dependencies/models.py"
}

{
 "command": "grep -rn \"is_async_callable\" ."
}

{
 "command": "cat /workspace/fastapi/routing.py"
}

{
 "command": "grep -rn \"is_async_callable\" /workspace/starlette | head -20"
}

{
 "command": "grep -rn \"is_async_callable\" /workspace/fastapi | head -20"
}

{
 "command": "python3 -c \"import starlette._utils; print(starlette._utils.is_async_callable)\""
}

{
 "command": "python3 -c \"import starlette._utils; print(starlette._utils.is_async_callable.__code__.co_name)\""
}

{
 "command": "python3 -c \"import starlette._utils; import inspect; print(inspect.getsource(starlette._utils.is_async_callable))\""
}

{
 "command": "python3 -c \"import functools; import asyncio; import inspect; \ndef sync_func(): pass\nasync def async_func(): pass\ndef wrapper(f):\n    @functools.wraps(f)\n    def wrapped(*args, **kwargs): return f(*args, **kwargs)\n    return wrapped\nasync def async_wrapper(f):\n    @functools.wraps(f)\n    async def wrapped(*args, **kwargs): return await f(*args, **kwargs)\n    return wrapped\n\nf1 = async_wrapper(sync_func)\nf2 = wrapper(async_func)\nf3 = functools.partial(async_wrapper(sync_func))\nf4 = functools.partial(wrapper(async_func))\n\nfrom starlette._utils import is_async_callable\nprint(f'f1: {is_async_callable(f1)}') # Expected: True\nprint(f'f2: {is_async_callable(f2)}') # Expected: True (if unwraps)\nprint(f'f3: {is_async_callable(f3)}') # Expected: True\nprint(f'f4: {is_async_callable(f4)}') # Expected: True\n\""
}

{
 "command": "python3 -c \"import functools; import asyncio; import inspect; \ndef sync_func(): pass\nasync def async_func(): pass\ndef wrapper(f):\n    @functools.wraps(f)\n    def wrapped(*args, **kwargs): return f(*args, **kwargs)\n    return wrapped\ndef async_wrapper(f):\n    @functools.wraps(f)\n    async def wrapped(*args, **kwargs): return await f(*args, **kwargs) if asyncio.iscoroutinefunction(f) else f(*args, **kwargs)\n    return wrapped\n\nf1 = async_wrapper(sync_func)\nf2 = wrapper(async_func)\nf3 = functools.partial(async_wrapper(sync_func)) # WRONG: async_wrapper returns a coroutine if it's async_def\n# Wait, I defined async_wrapper as a normal def that returns an async def.\n\n# Correction:\ndef sync_wrapper(f):\n    @functools.wraps(f)\n    def wrapped(*args, **kwargs): return f(*args, **kwargs)\n    return wrapped\ndef async_wrapper(f):\n    @functools.wraps(f)\n    async def wrapped(*args, **kwargs): return await f(*args, **kwargs) if asyncio.iscoroutinefunction(f) else f(*args, **kwargs)\n    return wrapped\n\nf1 = async_wrapper(sync_func)\nf2 = sync_wrapper(async_func)\nf3 = functools.partial(async_wrapper(sync_func)) # This is still wrong. async_wrapper(sync_func) returns the wrapped function.\nf3 = functools.partial(async_wrapper(sync_func)) # Correct\n\n# Actually, let's just use them:\nf1 = async_wrapper(sync_func)\nf2 = sync_wrapper(async_func)\nf3 = functools.partial(f1)\nf4 = functools.partial(f2)\n\nfrom starlette._utils import is_async_callable\nprint(f'f1: {is_async_callable(f1)}')\nprint(f'f2: {is_async_callable(f2)}')\nprint(f'f3: {is_async_callable(f3)}')\nprint(f'f4: {is_async_callable(f4)}')\n\""
}

{
 "command": "python3 -c \"import functools; import asyncio; import inspect; \ndef sync_func(): pass\nasync def async_func(): pass\ndef sync_wrapper(f):\n    @functools.wraps(f)\n    def wrapped(*args, **kwargs): return f(*args, **kwargs)\n    return wrapped\ndef async_wrapper(f):\n    @functools.wraps(f)\n    async def wrapped(*args, **kwargs): return await f(*args, **kwargs) if asyncio.iscoroutinefunction(f) else f(*args, **kwargs)\n    return wrapped\n\nf1 = async_wrapper(sync_func)\nf2 = sync_wrapper(async_func)\nf3 = functools.partial(f1)\nf4 = functools.partial(f2)\n\nfrom starlette._utils import is_async_callable\nprint(f'f1: {is_async_callable(f1)}') # True - because wrapped is async\nprint(f'f2: {is_async_callable(f2)}') # False - because wrapped is sync, and is_async_callable doesn't unwrap wraps\nprint(f'f3: {is_async_callable(f3)}') # True - because partial.func is async\nprint(f'f4: {is_async_callable(f4)}') # False - because partial.func is sync\n\""
}

{
 "command": "python3 -c \"import functools; import asyncio; import inspect; \ndef sync_func(): pass\nasync def async_func(): pass\ndef sync_wrapper(f):\n    @functools.wraps(f)\n    def wrapped(*args, **kwargs): return f(*args, **kwargs)\n    return wrapped\ndef async_wrapper(f):\n    @functools.wraps(f)\n    async def wrapped(*args, **kwargs): return await f(*args, **kwargs) if asyncio.iscoroutinefunction(f) else f(*args, **kwargs)\n    return wrapped\n\nf2 = sync_wrapper(async_func)\nprint(f'f2: {inspect.unwrap(f2)} is async: {asyncio.iscoroutinefunction(inspect.unwrap(f2))}')\n\""
}

{
 "command": "python3 -c \"import functools; import asyncio; import inspect; \ndef sync_func(): pass\nasync def async_func(): pass\ndef sync_wrapper(f):\n    @functools.wraps(f)\n    def wrapped(*args, **kwargs): return f(*args, **kwargs)\n    return wrapped\ndef async_wrapper(f):\n    @functools.wraps(f)\n    async def wrapped(*args, **kwargs): return await f(*args, **kwargs) if asyncio.iscoroutinefunction(f) else f(*args, **kwargs)\n    return wrapped\n\nf2 = sync_wrapper(async_func)\nprint(f'f2: {inspect.iscoroutinefunction(f2)}')\nprint(f'unwrap(f2): {inspect.iscoroutinefunction(inspect.unwrap(f2))}')\n\""
}

{
 "command": "cat /workspace/fastapi/dependencies/models.py"
}

{
 "command": "python3 -c \"import functools; import asyncio; import inspect; \ndef sync_func(): pass\nasync def async_func(): pass\ndef sync_wrapper(f):\n    @functools.wraps(f)\n    def wrapped(*args, **kwargs): return f(*args, **kwargs)\n    return wrapped\ndef async_wrapper(f):\n    @functools.wraps(f)\n    async def wrapped(*args, **kwargs): return await f(*args, **kwargs) if asyncio.iscoroutinefunction(f) else f(*args, **kwargs)\n    return wrapped\n\n# case 1: sync wrapper of async func\nf2 = sync_wrapper(async_func)\n# case 2: async wrapper of sync func\nf1 = async_wrapper(sync_func)\n\nfrom starlette._utils import is_async_callable\nprint(f'f1: {is_async_callable(f1)}') # True (wrapped is async)\nprint(f'f2: {is_async_callable(f2)}') # False (wrapped is sync, is_async_callable doesn't unwrap)\n\""
}

{
 "command": "python3 -c \"import functools; import asyncio; import inspect; \ndef sync_func(): pass\nasync def async_func(): pass\ndef sync_wrapper(f):\n    @functools.wraps(f)\n    def wrapped(*args, **kwargs): return f(*args, **kwargs)\n    return wrapped\ndef async_wrapper(f):\n    @functools.wraps(f)\n    async def wrapped(*args, **kwargs): return await f(*args, **kwargs) if asyncio.iscoroutinefunction(f) else f(*args, **kwargs)\n    return wrapped\n\n# case 1: sync wrapper of async func\nf2 = sync_wrapper(async_func)\n# case 2: async wrapper of sync func\nf1 = async_wrapper(sync_func)\n\ndef improved_is_async_callable(obj):\n    while isinstance(obj, functools.partial):\n        obj = obj.func\n    \n    if asyncio.iscoroutinefunction(obj) or (callable(obj) and asyncio.iscoroutinefunction(getattr(obj, '__call__', None))):\n        return True\n    \n    unwrapped = inspect.unwrap(obj)\n    if asyncio.iscoroutinefunction(unwrapped) or (callable(unwrapped) and asyncio.iscoroutinefunction(getattr(unwrapped, '__call__', None))):\n        return True\n        \n    return False\n\nprint(f'f1: {improved_is_async_callable(f1)}') # True\nprint(f'f2: {improved_is_async_callable(f2)}') # True\n\""
}

{
 "command": "python3 -c \"import functools; import asyncio; import inspect; \ndef sync_func(): pass\nasync def async_func(): pass\ndef sync_wrapper(f):\n    @functools.wraps(f)\n    def wrapped(*args, **kwargs): return f(*args, **kwargs)\n    return wrapped\ndef async_wrapper(f):\n    @functools.wraps(f)\n    async def wrapped(*args, **kwargs): return await f(*args, **kwargs) if asyncio.iscoroutinefunction(f) else f(*args, **kwargs)\n    return wrapped\n\n# f3 = functools.partial(async_wrapper(sync_func))\nf3 = functools.partial(async_wrapper(sync_func))\n# f4 = functools.partial(sync_wrapper(async_func))\nf4 = functools.partial(sync_wrapper(async_func))\n\ndef improved_is_async_callable(obj):\n    while isinstance(obj, functools.partial):\n        obj = obj.func\n    \n    if asyncio.iscoroutinefunction(obj) or (callable(obj) and asyncio.iscoroutinefunction(getattr(obj, '__call__', None))):\n        return True\n    \n    unwrapped = inspect.unwrap(obj)\n    if asyncio.iscoroutinefunction(unwrapped) or (callable(unwrapped) and asyncio.iscoroutinefunction(getattr(unwrapped, '__call__', None))):\n        return True\n        \n    return False\n\nprint(f'f3: {improved_is_async_callable(f3)}') # True\nprint(f'f4: {improved_is_async_callable(f4)}') # True\n\""
}

{
 "command": "python3 -c \"import functools; import asyncio; import inspect; \ndef sync_func(): pass\nasync def async_func(): pass\ndef sync_wrapper(f):\n    @functools.wraps(f)\n    def wrapped(*args, **kwargs): return f(*args, **kwargs)\n    return wrapped\ndef async_wrapper(f):\n    @functools.wraps(f)\n    async def wrapped(*args, **kwargs): return await f(*args, **kwargs) if asyncio.iscoroutinefunction(f) else f(*args, **kwargs)\n    return wrapped\n\n# test combined\nf5 = functools.partial(sync_wrapper(async_func))\nf6 = functools.partial(async_wrapper(sync_func))\n\ndef improved_is_async_callable(obj):\n    while isinstance(obj, functools.partial):\n        obj = obj.func\n    \n    if asyncio.iscoroutinefunction(obj) or (callable(obj) and asyncio.iscoroutinefunction(getattr(obj, '__call__', None))):\n        return True\n    \n    unwrapped = inspect.unwrap(obj)\n    if asyncio.iscoroutinefunction(unwrapped) or (callable(unwrapped) and asyncio.iscoroutinefunction(getattr(unwrapped, '__call__', None))):\n        return True\n        \n    return False\n\nprint(f'f5: {improved_is_async_callable(f5)}') # True\nprint(f'f6: {improved_is_async_callable(f6)}') # True\n\""
}

{
 "command": "python3 -c \"import functools; import asyncio; import inspect; \ndef sync_func(): pass\nasync def async_func(): pass\ndef sync_wrapper(f):\n    @functools.wraps(f)\n    def wrapped(*args, **kwargs): return f(*args, **kwargs)\n    return wrapped\ndef async_wrapper(f):\n    @functools.wraps(f)\n    async def wrapped(*args, **kwargs): return await f(*args, **kwargs) if asyncio.iscoroutinefunction(f) else f(*args, **kwargs)\n    return wrapped\n\n# combined wraps and partials\nf7 = functools.partial(sync_wrapper(async_func))\nf8 = sync_wrapper(functools.partial(async_func))\n\ndef improved_is_async_callable(obj):\n    # This is tricky. We might need to repeatedly unwrap and check for partials.\n    current = obj\n    while True:\n        if asyncio.iscoroutinefunction(current) or (callable(current) and asyncio.iscoroutinefunction(getattr(current, '__call__', None))):\n            return True\n        \n        if isinstance(current, functools.partial):\n            current = current.func\n            continue\n        \n        unwrapped = inspect.unwrap(current)\n        if unwrapped is current:\n            break\n        current = unwrapped\n        \n    return False\n\nprint(f'f7: {improved_is_async_callable(f7)}') # True\nprint(f'f8: {improved_is_async_callable(f8)}') # True\n\""
}

{
 "command": "python3 -c \"import functools; import asyncio; import inspect; \ndef sync_func(): pass\nasync def async_func(): pass\ndef sync_wrapper(f):\n    @functools.wraps(f)\n    def wrapped(*args, **kwargs): return f(*args, **kwargs)\n    return wrapped\ndef async_wrapper(f):\n    @functools.wraps(f)\n    async def wrapped(*args, **kwargs): return await f(*args, **kwargs) if asyncio.iscoroutinefunction(f) else f(*args, **kwargs)\n    return wrapped\n\n# Combined wraps and partials\nf7 = functools.partial(sync_wrapper(async_func))\nf8 = sync_wrapper(functools.partial(async_func))\n\ndef improved_is_async_callable(obj):\n    current = obj\n    while True:\n        if asyncio.iscoroutinefunction(current) or (callable(current) and asyncio.iscoroutinefunction(getattr(current, '__call__', None))):\n            return True\n        \n        if isinstance(current, functools.partial):\n            current = current.func\n            continue\n        \n        unwrapped = inspect.unwrap(current)\n        if unwrapped is current:\n            break\n        current = unwrapped\n        \n    return False\n\nprint(f'f7: {improved_is_async_callable(f7)}')\nprint(f'f8: {improved_is_async_callable(f8)}')\n\""
}

{
 "command": "python3 -c \"import functools; import asyncio; import inspect; \ndef sync_func(): pass\nasync def async_func(): pass\ndef sync_wrapper(f):\n    @functools.wraps(f)\n    def wrapped(*args, **kwargs): return f(*args, **kwargs)\n    return wrapped\ndef async_wrapper(f):\n    @functools.wraps(f)\n    async def wrapped(*args, **kwargs): return await f(*args, **kwargs) if asyncio.iscoroutinefunction(f) else f(*args, **kwargs)\n    return wrapped\n\n# Test a sync one\nf9 = sync_wrapper(sync_func)\nf10 = functools.partial(f9)\n\ndef improved_is_async_callable(obj):\n    current = obj\n    while True:\n        if asyncio.iscoroutinefunction(current) or (callable(current) and asyncio.iscoroutinefunction(getattr(current, '__call__', None))):\n            return True\n        \n        if isinstance(current, functools.partial):\n            current = current.func\n            continue\n        \n        unwrapped = inspect.unwrap(current)\n        if unwrapped is current:\n            break\n        current = unwrapped\n        \n    return False\n\nprint(f'f9: {improved_is_async_callable(f9)}') # False\nprint(f'f10: {improved_is_async_callable(f10)}') # False\n\""
}

{
 "command": "python3 -c \"import functools; import asyncio; import inspect; \ndef sync_func(): pass\nasync def async_func(): pass\ndef sync_wrapper(f):\n    @functools.wraps(f)\n    def wrapped(*args, **kwargs): return f(*args, **kwargs)\n    return wrapped\ndef async_wrapper(f):\n    @functools.wraps(f)\n    async def wrapped(*args, **kwargs): return await f(*args, **kwargs) if asyncio.iscoroutinefunction(f) else f(*args, **kwargs)\n    return wrapped\n\nclass SyncCallable:\n    def __call__(self): pass\nclass AsyncCallable:\n    async def __call__(self): pass\n\nc1 = SyncCallable()\nc2 = AsyncCallable()\nc3 = functools.partial(c1)\nc4 = functools.partial(c2)\n\ndef improved_is_async_callable(obj):\n    current = obj\n    while True:\n        if asyncio.iscoroutinefunction(current) or (callable(current) and asyncio.iscoroutinefunction(getattr(current, '__call__', None))):\n            return True\n        \n        if isinstance(current, functools.partial):\n            current = current.func\n            continue\n        \n        unwrapped = inspect.unwrap(current)\n        if unwrapped is current:\n            break\n        current = unwrapped\n        \n    return False\n\nprint(f'c1: {improved_is_async_callable(c1)}') # False\nprint(f'c2: {improved_is_async_callable(c2)}') # True\nprint(f'c3: {improved_is_async_callable(c3)}') # False\nprint(f'c4: {improved_is_async_callable(c4)}') # True\n\""
}

{
 "command": "cat /workspace/fastapi/dependencies/models.py"
}

{
 "command": "python3 - <<'PY'\nfrom pathlib import Path\np = Path(\"fastapi/dependencies/models.py\")\ns = p.read_text()\nold = \"\"\"    @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\"\"\"\nnew = \"\"\"    @cached_property\n    def _unwrapped_call(self) -> Any:\n        if self.call is None:\n            return self.call  # pragma: no cover\n        current = self.call\n        while True:\n            if isinstance(current, partial):\n                current = current.func\n                continue\n            unwrapped = inspect.unwrap(current)\n            if unwrapped is current:\n                break\n            current = unwrapped\n        return current\"\"\"\nassert s.count(old) == 1, (\"old text found this many times:\", s.count(old))\nt = s.replace(old, new)\np.write_text(t)\nPY\n"
}

{
 "command": "python3 -m py_compile fastapi/dependencies/models.py"
}

{
 "command": "python3 -c \"import fastapi.dependencies.models; print('Import successful')\""
}

{
 "command": "python3 - <<'PY'\nimport functools\nimport asyncio\nimport inspect\nfrom fastapi.dependencies.models import Dependant\nfrom functools import partial\n\ndef sync_func(): pass\nasync def async_func(): pass\ndef sync_wrapper(f):\n    @functools.wraps(f)\n    def wrapped(*args, **kwargs): return f(*args, **kwargs)\n    return wrapped\ndef async_wrapper(f):\n    @functools.wraps(f)\n    async def wrapped(*args, **kwargs): return await f(*args, **kwargs) if asyncio.iscoroutinefunction(f) else f(*args, **kwargs)\n    return wrapped\n\n# Test cases\ncases = [\n    (\"sync_func\", sync_func),\n    (\"async_func\", async_func),\n    (\"sync_wrapped_sync\", sync_wrapper(sync_func)),\n    (\"async_wrapped_sync\", async_wrapper(sync_func)),\n    (\"sync_wrapped_async\", sync_wrapper(async_func)),\n    (\"async_wrapped_async\", async_wrapper(async_func)),\n    (\"partial_sync\", partial(sync_func)),\n    (\"partial_async\", partial(async_func)),\n    (\"partial_sync_wrapped_sync\", partial(sync_wrapper(sync_func))),\n    (\"partial_async_wrapped_sync\", partial(async_wrapper(sync_func))),\n    (\"partial_sync_wrapped_async\", partial(sync_wrapper(async_func))),\n    (\"partial_async_wrapped_async\", partial(async_wrapper(async_func))),\n    (\"sync_wrapped_partial_async\", sync_wrapper(partial(async_func))),\n]\n\nfor name, func in cases:\n    dep = Dependant(call=func)\n    is_coro = dep.is_coroutine_callable\n    # Determine expected\n    # It's a coroutine if any part of the chain is async\n    \n    # Manual check for expected\n    expected = False\n    current = func\n    while True:\n        if asyncio.iscoroutinefunction(current) or (callable(current) and asyncio.iscoroutinefunction(getattr(current, '__call__', None))):\n            expected = True\n            break\n        if isinstance(current, partial):\n            current = current.func\n            continue\n        unwrapped = inspect.unwrap(current)\n        if unwrapped is current:\n            break\n        current = unwrapped\n    \n    print(f\"{name:30} | Actual: {str(is_coro):5} | Expected: {str(expected):5} | {'PASS' if is_coro == expected else 'FAIL'}\")\nPY\n"
}

{
 "command": "python3 - <<'PY'\nimport functools\nimport asyncio\nimport inspect\nfrom fastapi.dependencies.models import Dependant\nfrom functools import partial\n\ndef sync_func(): pass\nasync def async_func(): pass\ndef sync_wrapper(f):\n    @functools.wraps(f)\n    def wrapped(*args, **kwargs): return f(*args, **kwargs)\n    return wrapped\ndef async_wrapper(f):\n    @functools.wraps(f)\n    async def wrapped(*args, **kwargs): return await f(*args, **kwargs) if asyncio.iscoroutinefunction(f) else f(*args, **kwargs)\n    return wrapped\n\nf1 = async_wrapper(sync_func)\ndep = Dependant(call=f1)\nprint(f\"f1: {dep._unwrapped_call}\")\nprint(f\"f1 is coro: {dep.is_coroutine_callable}\")\nPY\n"
}

{
 "command": "python3 - <<'PY'\nimport functools\nimport asyncio\nimport inspect\nfrom fastapi.dependencies.models import Dependant\nfrom functools import partial\n\ndef sync_func(): pass\nasync def async_func(): pass\ndef sync_wrapper(f):\n    @functools.wraps(f)\n    def wrapped(*args, **kwargs): return f(*args, **kwargs)\n    return wrapped\ndef async_wrapper(f):\n    @functools.wraps(f)\n    async def wrapped(*args, **kwargs): return await f(*args, **kwargs) if asyncio.iscoroutinefunction(f) else f(*args, **kwargs)\n    return wrapped\n\nf1 = async_wrapper(sync_func)\n# f1 is the wrapped function, which is async.\nprint(f\"f1 is async: {asyncio.iscoroutinefunction(f1)}\")\nPY\n"
}