Used by Starlette:
* <a href="https://requests.readthedocs.io" target="_blank"><code>requests</code></a> - Required if you want to use the `TestClient`.
-* <a href="https://github.com/Tinche/aiofiles" target="_blank"><code>aiofiles</code></a> - Required if you want to use `FileResponse` or `StaticFiles`.
* <a href="https://jinja.palletsprojects.com" target="_blank"><code>jinja2</code></a> - Required if you want to use the default template configuration.
* <a href="https://andrew-d.github.io/python-multipart/" target="_blank"><code>python-multipart</code></a> - Required if you want to support form <abbr title="converting the string that comes from an HTTP request into Python data">"parsing"</abbr>, with `request.form()`.
* <a href="https://pythonhosted.org/itsdangerous/" target="_blank"><code>itsdangerous</code></a> - Required for `SessionMiddleware` support.
Let's look at how we can make that work.
-## pytest-asyncio
+## pytest.mark.anyio
-If we want to call asynchronous functions in our tests, our test functions have to be asynchronous. Pytest provides a neat library for this, called `pytest-asyncio`, that allows us to specify that some test functions are to be called asynchronously.
-
-You can install it via:
-
-<div class="termy">
-
-```console
-$ pip install pytest-asyncio
-
----> 100%
-```
-
-</div>
+If we want to call asynchronous functions in our tests, our test functions have to be asynchronous. Anyio provides a neat plugin for this, that allows us to specify that some test functions are to be called asynchronously.
## HTTPX
## In Detail
-The marker `@pytest.mark.asyncio` tells pytest that this test function should be called asynchronously:
+The marker `@pytest.mark.anyio` tells pytest that this test function should be called asynchronously:
```Python hl_lines="7"
{!../../../docs_src/async_tests/test_main.py!}
As the testing function is now asynchronous, you can now also call (and `await`) other `async` functions apart from sending requests to your FastAPI application in your tests, exactly as you would call them anywhere else in your code.
!!! tip
- If you encounter a `RuntimeError: Task attached to a different loop` when integrating asynchronous function calls in your tests (e.g. when using <a href="https://stackoverflow.com/questions/41584243/runtimeerror-task-attached-to-a-different-loop" class="external-link" target="_blank">MongoDB's MotorClient</a>) check out <a href="https://github.com/pytest-dev/pytest-asyncio/issues/38#issuecomment-264418154" class="external-link" target="_blank">this issue</a> in the pytest-asyncio repository.
+ If you encounter a `RuntimeError: Task attached to a different loop` when integrating asynchronous function calls in your tests (e.g. when using <a href="https://stackoverflow.com/questions/41584243/runtimeerror-task-attached-to-a-different-loop" class="external-link" target="_blank">MongoDB's MotorClient</a>) Remember to instantiate objects that need an event loop only within async functions, e.g. an `'@app.on_event("startup")` callback.
└── swagger-ui.css
```
-### Install `aiofiles`
-
-Now you need to install `aiofiles`:
-
-
-<div class="termy">
-
-```console
-$ pip install aiofiles
-
----> 100%
-```
-
-</div>
-
### Serve the static files
* Import `StaticFiles`.
</div>
-If you need to also serve static files (as in this example), install `aiofiles`:
-
-<div class="termy">
-
-```console
-$ pip install aiofiles
-
----> 100%
-```
-
-</div>
-
## Using `Jinja2Templates`
* Import `Jinja2Templates`.
Used by Starlette:
* <a href="https://requests.readthedocs.io" target="_blank"><code>requests</code></a> - Required if you want to use the `TestClient`.
-* <a href="https://github.com/Tinche/aiofiles" target="_blank"><code>aiofiles</code></a> - Required if you want to use `FileResponse` or `StaticFiles`.
* <a href="https://jinja.palletsprojects.com" target="_blank"><code>jinja2</code></a> - Required if you want to use the default template configuration.
* <a href="https://andrew-d.github.io/python-multipart/" target="_blank"><code>python-multipart</code></a> - Required if you want to support form <abbr title="converting the string that comes from an HTTP request into Python data">"parsing"</abbr>, with `request.form()`.
* <a href="https://pythonhosted.org/itsdangerous/" target="_blank"><code>itsdangerous</code></a> - Required for `SessionMiddleware` support.
!!! tip
Make sure to use `yield` one single time.
-!!! info
- For this to work, you need to use **Python 3.7** or above, or in **Python 3.6**, install the "backports":
-
- ```
- pip install async-exit-stack async-generator
- ```
-
- This installs <a href="https://github.com/sorcio/async_exit_stack" class="external-link" target="_blank">async-exit-stack</a> and <a href="https://github.com/python-trio/async_generator" class="external-link" target="_blank">async-generator</a>.
-
!!! note "Technical Details"
Any function that is valid to use with:
### Create a dependency
-!!! info
- For this to work, you need to use **Python 3.7** or above, or in **Python 3.6**, install the "backports":
-
- ```console
- $ pip install async-exit-stack async-generator
- ```
-
- This installs <a href="https://github.com/sorcio/async_exit_stack" class="external-link" target="_blank">async-exit-stack</a> and <a href="https://github.com/python-trio/async_generator" class="external-link" target="_blank">async-generator</a>.
-
- You can also use the alternative method with a "middleware" explained at the end.
-
Now use the `SessionLocal` class we created in the `sql_app/databases.py` file to create a dependency.
We need to have an independent database session/connection (`SessionLocal`) per request, use the same session through all the request and then close it after the request is finished.
You can serve static files automatically from a directory using `StaticFiles`.
-## Install `aiofiles`
-
-First you need to install `aiofiles`:
-
-<div class="termy">
-
-```console
-$ pip install aiofiles
-
----> 100%
-```
-
-</div>
-
## Use `StaticFiles`
* Import `StaticFiles`.
from .main import app
-@pytest.mark.asyncio
+@pytest.mark.anyio
async def test_root():
async with AsyncClient(app=app, base_url="http://test") as ac:
response = await ac.get("/")
-from typing import Any, Callable
+import sys
+from typing import AsyncGenerator, ContextManager, TypeVar
from starlette.concurrency import iterate_in_threadpool as iterate_in_threadpool # noqa
from starlette.concurrency import run_in_threadpool as run_in_threadpool # noqa
run_until_first_complete as run_until_first_complete,
)
-asynccontextmanager_error_message = """
-FastAPI's contextmanager_in_threadpool require Python 3.7 or above,
-or the backport for Python 3.6, installed with:
- pip install async-generator
-"""
+if sys.version_info >= (3, 7):
+ from contextlib import AsyncExitStack as AsyncExitStack
+ from contextlib import asynccontextmanager as asynccontextmanager
+else:
+ from contextlib2 import AsyncExitStack as AsyncExitStack # noqa
+ from contextlib2 import asynccontextmanager as asynccontextmanager # noqa
-def _fake_asynccontextmanager(func: Callable[..., Any]) -> Callable[..., Any]:
- def raiser(*args: Any, **kwargs: Any) -> Any:
- raise RuntimeError(asynccontextmanager_error_message)
+_T = TypeVar("_T")
- return raiser
-
-try:
- from contextlib import asynccontextmanager as asynccontextmanager # type: ignore
-except ImportError:
- try:
- from async_generator import ( # type: ignore # isort: skip
- asynccontextmanager as asynccontextmanager,
- )
- except ImportError: # pragma: no cover
- asynccontextmanager = _fake_asynccontextmanager
-
-try:
- from contextlib import AsyncExitStack as AsyncExitStack # type: ignore
-except ImportError:
- try:
- from async_exit_stack import AsyncExitStack as AsyncExitStack # type: ignore
- except ImportError: # pragma: no cover
- AsyncExitStack = None # type: ignore
-
-
-@asynccontextmanager # type: ignore
-async def contextmanager_in_threadpool(cm: Any) -> Any:
+@asynccontextmanager
+async def contextmanager_in_threadpool(
+ cm: ContextManager[_T],
+) -> AsyncGenerator[_T, None]:
try:
yield await run_in_threadpool(cm.__enter__)
except Exception as e:
-import asyncio
import dataclasses
import inspect
from contextlib import contextmanager
from typing import (
Any,
Callable,
+ Coroutine,
Dict,
List,
Mapping,
cast,
)
+import anyio
from fastapi import params
from fastapi.concurrency import (
AsyncExitStack,
- _fake_asynccontextmanager,
asynccontextmanager,
contextmanager_in_threadpool,
)
return annotation
-async_contextmanager_dependencies_error = """
-FastAPI dependencies with yield require Python 3.7 or above,
-or the backports for Python 3.6, installed with:
- pip install async-exit-stack async-generator
-"""
-
-
-def check_dependency_contextmanagers() -> None:
- if AsyncExitStack is None or asynccontextmanager == _fake_asynccontextmanager:
- raise RuntimeError(async_contextmanager_dependencies_error) # pragma: no cover
-
-
def get_dependant(
*,
path: str,
path_param_names = get_path_param_names(path)
endpoint_signature = get_typed_signature(call)
signature_params = endpoint_signature.parameters
- if is_gen_callable(call) or is_async_gen_callable(call):
- check_dependency_contextmanagers()
dependant = Dependant(call=call, name=name, path=path, use_cache=use_cache)
for param_name, param in signature_params.items():
if isinstance(param.default, params.Depends):
if is_gen_callable(call):
cm = contextmanager_in_threadpool(contextmanager(call)(**sub_values))
elif is_async_gen_callable(call):
- if not inspect.isasyncgenfunction(call):
- # asynccontextmanager from the async_generator backfill pre python3.7
- # does not support callables that are not functions or methods.
- # See https://github.com/python-trio/async_generator/issues/32
- #
- # Expand the callable class into its __call__ method before decorating it.
- # This approach will work on newer python versions as well.
- call = getattr(call, "__call__", None)
cm = asynccontextmanager(call)(**sub_values)
return await stack.enter_async_context(cm)
solved = dependency_cache[sub_dependant.cache_key]
elif is_gen_callable(call) or is_async_gen_callable(call):
stack = request.scope.get("fastapi_astack")
- if stack is None:
- raise RuntimeError(
- async_contextmanager_dependencies_error
- ) # pragma: no cover
+ assert isinstance(stack, AsyncExitStack)
solved = await solve_generator(
call=call, stack=stack, sub_values=sub_values
)
and lenient_issubclass(field.type_, bytes)
and isinstance(value, sequence_types)
):
- awaitables = [sub_value.read() for sub_value in value]
- contents = await asyncio.gather(*awaitables)
- value = sequence_shape_to_type[field.shape](contents)
+ results: List[Union[bytes, str]] = []
+
+ async def process_fn(
+ fn: Callable[[], Coroutine[Any, Any, Any]]
+ ) -> None:
+ result = await fn()
+ results.append(result)
+
+ async with anyio.create_task_group() as tg:
+ for sub_value in value:
+ tg.start_soon(process_fn, sub_value.read)
+ value = sequence_shape_to_type[field.shape](results)
v_, errors_ = field.validate(value, values, loc=loc)
"Topic :: Internet :: WWW/HTTP",
]
requires = [
- "starlette ==0.14.2",
- "pydantic >=1.6.2,!=1.7,!=1.7.1,!=1.7.2,!=1.7.3,!=1.8,!=1.8.1,<2.0.0"
+ "starlette ==0.15.0",
+ "pydantic >=1.6.2,!=1.7,!=1.7.1,!=1.7.2,!=1.7.3,!=1.8,!=1.8.1,<2.0.0",
+ # TODO: remove contextlib2 as a direct dependency after upgrading Starlette
+ "contextlib2 >= 21.6.0; python_version < '3.7'",
]
description-file = "README.md"
requires-python = ">=3.6.1"
test = [
"pytest >=6.2.4,<7.0.0",
"pytest-cov >=2.12.0,<4.0.0",
- "pytest-asyncio >=0.14.0,<0.16.0",
"mypy ==0.910",
"flake8 >=3.8.3,<4.0.0",
"black ==21.9b0",
"orjson >=3.2.1,<4.0.0",
"ujson >=4.0.1,<5.0.0",
"python-multipart >=0.0.5,<0.0.6",
- "aiofiles >=0.5.0,<0.8.0",
# TODO: try to upgrade after upgrading Starlette
"flask >=1.1.2,<2.0.0",
- "async_exit_stack >=1.0.1,<2.0.0; python_version < '3.7'",
- "async_generator >=1.10,<2.0.0; python_version < '3.7'",
+ "anyio[trio] >=3.2.1,<4.0.0",
# types
"types-ujson ==0.1.1",
]
all = [
"requests >=2.24.0,<3.0.0",
- "aiofiles >=0.5.0,<0.8.0",
# TODO: try to upgrade after upgrading Starlette
"jinja2 >=2.11.2,<3.0.0",
"python-multipart >=0.0.5,<0.0.6",
"orjson >=3.2.1,<4.0.0",
"email_validator >=1.1.1,<2.0.0",
"uvicorn[standard] >=0.12.0,<0.16.0",
- "async_exit_stack >=1.0.1,<2.0.0; python_version < '3.7'",
- "async_generator >=1.10,<2.0.0; python_version < '3.7'",
]
[tool.isort]
filterwarnings = [
"error",
'ignore:"@coroutine" decorator is deprecated since Python 3\.8, use "async def" instead:DeprecationWarning',
+ # TODO: needed by AnyIO in Python 3.9, try to remove after an AnyIO upgrade
+ 'ignore:The loop argument is deprecated since Python 3\.8, and scheduled for removal in Python 3\.10:DeprecationWarning',
# TODO: if these ignores are needed, enable them, otherwise remove them
# 'ignore:The explicit passing of coroutine objects to asyncio\.wait\(\) is deprecated since Python 3\.8:DeprecationWarning',
# 'ignore:Exception ignored in. <socket\.socket fd=-1:pytest.PytestUnraisableExceptionWarning',
+++ /dev/null
-import pytest
-from fastapi.concurrency import _fake_asynccontextmanager
-
-
-@_fake_asynccontextmanager
-def never_run():
- pass # pragma: no cover
-
-
-def test_fake_async():
- with pytest.raises(RuntimeError):
- never_run()
from docs_src.async_tests.test_main import test_root
-@pytest.mark.asyncio
+@pytest.mark.anyio
async def test_async_testing():
await test_root()
def test_websocket_no_credentials():
with pytest.raises(WebSocketDisconnect):
- client.websocket_connect("/items/foo/ws")
+ with client.websocket_connect("/items/foo/ws"):
+ pytest.fail(
+ "did not raise WebSocketDisconnect on __enter__"
+ ) # pragma: no cover
def test_websocket_invalid_data():
with pytest.raises(WebSocketDisconnect):
- client.websocket_connect("/items/foo/ws?q=bar&token=some-token")
+ with client.websocket_connect("/items/foo/ws?q=bar&token=some-token"):
+ pytest.fail(
+ "did not raise WebSocketDisconnect on __enter__"
+ ) # pragma: no cover
projects / thirdparty / fastapi / fastapi.git / commitdiff
