mirror of
https://github.com/simonw/datasette.git
synced 2025-12-10 16:51:24 +01:00
datasette.utils.check_callable(obj) - refs #2078
This commit is contained in:
parent
49184c569c
commit
2e43a14da1
2 changed files with 71 additions and 0 deletions
25
datasette/utils/callable.py
Normal file
25
datasette/utils/callable.py
Normal file
|
|
@ -0,0 +1,25 @@
|
||||||
|
import asyncio
|
||||||
|
import types
|
||||||
|
from typing import NamedTuple, Any
|
||||||
|
|
||||||
|
|
||||||
|
class CallableStatus(NamedTuple):
|
||||||
|
is_callable: bool
|
||||||
|
is_async_callable: bool
|
||||||
|
|
||||||
|
|
||||||
|
def check_callable(obj: Any) -> CallableStatus:
|
||||||
|
if not callable(obj):
|
||||||
|
return CallableStatus(False, False)
|
||||||
|
|
||||||
|
if isinstance(obj, type):
|
||||||
|
# It's a class
|
||||||
|
return CallableStatus(True, False)
|
||||||
|
|
||||||
|
if isinstance(obj, types.FunctionType):
|
||||||
|
return CallableStatus(True, asyncio.iscoroutinefunction(obj))
|
||||||
|
|
||||||
|
if hasattr(obj, "__call__"):
|
||||||
|
return CallableStatus(True, asyncio.iscoroutinefunction(obj.__call__))
|
||||||
|
|
||||||
|
assert False, "obj {} is somehow callable with no __call__ method".format(repr(obj))
|
||||||
46
tests/test_utils_callable.py
Normal file
46
tests/test_utils_callable.py
Normal file
|
|
@ -0,0 +1,46 @@
|
||||||
|
from datasette.utils.callable import check_callable
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
|
||||||
|
class AsyncClass:
|
||||||
|
async def __call__(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class NotAsyncClass:
|
||||||
|
def __call__(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class ClassNoCall:
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
async def async_func():
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def non_async_func():
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
"obj,expected_is_callable,expected_is_async_callable",
|
||||||
|
(
|
||||||
|
(async_func, True, True),
|
||||||
|
(non_async_func, True, False),
|
||||||
|
(AsyncClass(), True, True),
|
||||||
|
(NotAsyncClass(), True, False),
|
||||||
|
(ClassNoCall(), False, False),
|
||||||
|
(AsyncClass, True, False),
|
||||||
|
(NotAsyncClass, True, False),
|
||||||
|
(ClassNoCall, True, False),
|
||||||
|
("", False, False),
|
||||||
|
(1, False, False),
|
||||||
|
(str, True, False),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
def test_check_callable(obj, expected_is_callable, expected_is_async_callable):
|
||||||
|
status = check_callable(obj)
|
||||||
|
assert status.is_callable == expected_is_callable
|
||||||
|
assert status.is_async_callable == expected_is_async_callable
|
||||||
Loading…
Add table
Add a link
Reference in a new issue