mirror of
https://github.com/fastapi/fastapi.git
synced 2026-09-14 13:36:21 +08:00
Sebastián Ramírez
3f9f4a0f8f
* ➕ Add development/testing dependencies for Python 3.6 * ✨ Add concurrency submodule with contextmanager_in_threadpool * ✨ Add AsyncExitStack to ASGI scope in FastAPI app call * ✨ Use async stack for contextmanager-able dependencies including running in threadpool sync dependencies * ✅ Add tests for contextmanager dependencies including internal raise checks when exceptions should be handled and when not * ✅ Add test for fake asynccontextmanager raiser * 🐛 Fix mypy errors and coverage * 🔇 Remove development logs and prints * ✅ Add tests for sub-contextmanagers, background tasks, and sync functions * 🐛 Fix mypy errors for Python 3.7 * 💬 Fix error texts for clarity * 📝 Add docs for dependencies with yield * ✨ Update SQL with SQLAlchemy tutorial to use dependencies with yield and add an alternative with a middleware (from the old tutorial) * ✅ Update SQL tests to remove DB file during the same tests * ✅ Add tests for example with middleware as a copy from the tests with dependencies with yield, removing the DB in the tests * ✏️ Fix typos with suggestions from code review Co-Authored-By: dmontagu <35119617+dmontagu@users.noreply.github.com>
46 lines
1.3 KiB
Python
46 lines
1.3 KiB
Python
from typing import Any, Callable
|
|
|
|
from starlette.concurrency import iterate_in_threadpool, run_in_threadpool # noqa
|
|
|
|
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
|
|
"""
|
|
|
|
|
|
def _fake_asynccontextmanager(func: Callable) -> Callable:
|
|
def raiser(*args: Any, **kwargs: Any) -> Any:
|
|
raise RuntimeError(asynccontextmanager_error_message)
|
|
|
|
return raiser
|
|
|
|
|
|
try:
|
|
from contextlib import asynccontextmanager # type: ignore
|
|
except ImportError:
|
|
try:
|
|
from async_generator import asynccontextmanager # type: ignore
|
|
except ImportError: # pragma: no cover
|
|
asynccontextmanager = _fake_asynccontextmanager
|
|
|
|
try:
|
|
from contextlib import AsyncExitStack # type: ignore
|
|
except ImportError:
|
|
try:
|
|
from async_exit_stack import AsyncExitStack # type: ignore
|
|
except ImportError: # pragma: no cover
|
|
AsyncExitStack = None # type: ignore
|
|
|
|
|
|
@asynccontextmanager
|
|
async def contextmanager_in_threadpool(cm: Any) -> Any:
|
|
try:
|
|
yield await run_in_threadpool(cm.__enter__)
|
|
except Exception as e:
|
|
ok = await run_in_threadpool(cm.__exit__, type(e), e, None)
|
|
if not ok:
|
|
raise e
|
|
else:
|
|
await run_in_threadpool(cm.__exit__, None, None, None)
|