Browse Source
* ➕ 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 <[email protected]>pull/613/head
Sebastián Ramírez
6 years ago
committed by
GitHub
GitHub
19 changed files with 1238 additions and 88 deletions
@ -1,21 +1,6 @@ |
|||
from fastapi import Depends, FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class FixedContentQueryChecker: |
|||
def __init__(self, fixed_content: str): |
|||
self.fixed_content = fixed_content |
|||
|
|||
def __call__(self, q: str = ""): |
|||
if q: |
|||
return self.fixed_content in q |
|||
return False |
|||
|
|||
|
|||
checker = FixedContentQueryChecker("bar") |
|||
|
|||
|
|||
@app.get("/query-checker/") |
|||
async def read_query_check(fixed_content_included: bool = Depends(checker)): |
|||
return {"fixed_content_in_query": fixed_content_included} |
|||
async def get_db(): |
|||
db = DBSession() |
|||
try: |
|||
yield db |
|||
finally: |
|||
db.close() |
|||
|
|||
@ -0,0 +1,25 @@ |
|||
from fastapi import Depends |
|||
|
|||
|
|||
async def dependency_a(): |
|||
dep_a = generate_dep_a() |
|||
try: |
|||
yield dep_a |
|||
finally: |
|||
dep_a.close() |
|||
|
|||
|
|||
async def dependency_b(dep_a=Depends(dependency_a)): |
|||
dep_b = generate_dep_b() |
|||
try: |
|||
yield dep_b |
|||
finally: |
|||
dep_b.close(dep_a) |
|||
|
|||
|
|||
async def dependency_c(dep_b=Depends(dependency_b)): |
|||
dep_c = generate_dep_c() |
|||
try: |
|||
yield dep_c |
|||
finally: |
|||
dep_c.close(dep_b) |
|||
@ -0,0 +1,25 @@ |
|||
from fastapi import Depends |
|||
|
|||
|
|||
async def dependency_a(): |
|||
dep_a = generate_dep_a() |
|||
try: |
|||
yield dep_a |
|||
finally: |
|||
dep_a.close() |
|||
|
|||
|
|||
async def dependency_b(dep_a=Depends(dependency_a)): |
|||
dep_b = generate_dep_b() |
|||
try: |
|||
yield dep_b |
|||
finally: |
|||
dep_b.close(dep_a) |
|||
|
|||
|
|||
async def dependency_c(dep_b=Depends(dependency_b)): |
|||
dep_c = generate_dep_c() |
|||
try: |
|||
yield dep_c |
|||
finally: |
|||
dep_c.close(dep_b) |
|||
@ -0,0 +1,14 @@ |
|||
class MySuperContextManager: |
|||
def __init__(self): |
|||
self.db = DBSession() |
|||
|
|||
def __enter__(self): |
|||
return self.db |
|||
|
|||
def __exit__(self, exc_type, exc_value, traceback): |
|||
self.db.close() |
|||
|
|||
|
|||
async def get_db(): |
|||
with MySuperContextManager() as db: |
|||
yield db |
|||
@ -0,0 +1,21 @@ |
|||
from fastapi import Depends, FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class FixedContentQueryChecker: |
|||
def __init__(self, fixed_content: str): |
|||
self.fixed_content = fixed_content |
|||
|
|||
def __call__(self, q: str = ""): |
|||
if q: |
|||
return self.fixed_content in q |
|||
return False |
|||
|
|||
|
|||
checker = FixedContentQueryChecker("bar") |
|||
|
|||
|
|||
@app.get("/query-checker/") |
|||
async def read_query_check(fixed_content_included: bool = Depends(checker)): |
|||
return {"fixed_content_in_query": fixed_content_included} |
|||
@ -0,0 +1,64 @@ |
|||
from typing import List |
|||
|
|||
from fastapi import Depends, FastAPI, HTTPException |
|||
from sqlalchemy.orm import Session |
|||
from starlette.requests import Request |
|||
from starlette.responses import Response |
|||
|
|||
from . import crud, models, schemas |
|||
from .database import SessionLocal, engine |
|||
|
|||
models.Base.metadata.create_all(bind=engine) |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.middleware("http") |
|||
async def db_session_middleware(request: Request, call_next): |
|||
response = Response("Internal server error", status_code=500) |
|||
try: |
|||
request.state.db = SessionLocal() |
|||
response = await call_next(request) |
|||
finally: |
|||
request.state.db.close() |
|||
return response |
|||
|
|||
|
|||
# Dependency |
|||
def get_db(request: Request): |
|||
return request.state.db |
|||
|
|||
|
|||
@app.post("/users/", response_model=schemas.User) |
|||
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)): |
|||
db_user = crud.get_user_by_email(db, email=user.email) |
|||
if db_user: |
|||
raise HTTPException(status_code=400, detail="Email already registered") |
|||
return crud.create_user(db=db, user=user) |
|||
|
|||
|
|||
@app.get("/users/", response_model=List[schemas.User]) |
|||
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): |
|||
users = crud.get_users(db, skip=skip, limit=limit) |
|||
return users |
|||
|
|||
|
|||
@app.get("/users/{user_id}", response_model=schemas.User) |
|||
def read_user(user_id: int, db: Session = Depends(get_db)): |
|||
db_user = crud.get_user(db, user_id=user_id) |
|||
if db_user is None: |
|||
raise HTTPException(status_code=404, detail="User not found") |
|||
return db_user |
|||
|
|||
|
|||
@app.post("/users/{user_id}/items/", response_model=schemas.Item) |
|||
def create_item_for_user( |
|||
user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db) |
|||
): |
|||
return crud.create_user_item(db=db, item=item, user_id=user_id) |
|||
|
|||
|
|||
@app.get("/items/", response_model=List[schemas.Item]) |
|||
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): |
|||
items = crud.get_items(db, skip=skip, limit=limit) |
|||
return items |
|||
@ -0,0 +1,153 @@ |
|||
# Dependencies with `yield` |
|||
|
|||
FastAPI supports dependencies that do some <abbr title='sometimes also called "exit", "cleanup", "teardown", "close", "context managers", ...'>extra steps after finishing</abbr>. |
|||
|
|||
To do this, use `yield` instead of `return`, and write the extra steps after. |
|||
|
|||
!!! 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": |
|||
|
|||
```bash |
|||
pip install async-exit-stack async-generator |
|||
``` |
|||
|
|||
This installs <a href="https://github.com/sorcio/async_exit_stack" target="_blank">async-exit-stack</a> and <a href="https://github.com/python-trio/async_generator" target="_blank">async-generator</a>. |
|||
|
|||
!!! note "Technical Details" |
|||
Any function that is valid to use with: |
|||
|
|||
* <a href="https://docs.python.org/3/library/contextlib.html#contextlib.contextmanager" target="_blank">`@contextlib.contextmanager`</a> or |
|||
* <a href="https://docs.python.org/3/library/contextlib.html#contextlib.asynccontextmanager" target="_blank">`@contextlib.asynccontextmanager`</a> |
|||
|
|||
would be valid to use as a **FastAPI** dependency. |
|||
|
|||
In fact, FastAPI uses those two decorators internally. |
|||
|
|||
## A database dependency with `yield` |
|||
|
|||
For example, you could use this to create a database session and close it after finishing. |
|||
|
|||
Only the code prior to and including the `yield` statement is executed before sending a response: |
|||
|
|||
```Python hl_lines="2 3 4" |
|||
{!./src/dependencies/tutorial007.py!} |
|||
``` |
|||
|
|||
The yielded value is what is injected into *path operations* and other dependencies: |
|||
|
|||
```Python hl_lines="4" |
|||
{!./src/dependencies/tutorial007.py!} |
|||
``` |
|||
|
|||
The code following the `yield` statement is executed after the response has been delivered: |
|||
|
|||
```Python hl_lines="5 6" |
|||
{!./src/dependencies/tutorial007.py!} |
|||
``` |
|||
|
|||
!!! tip |
|||
You can use `async` or normal functions. |
|||
|
|||
**FastAPI** will do the right thing with each, the same as with normal dependencies. |
|||
|
|||
## A dependency with `yield` and `try` |
|||
|
|||
If you use a `try` block in a dependency with `yield`, you'll receive any exception that was thrown when using the dependency. |
|||
|
|||
For example, if some code at some point in the middle, in another dependency or in a *path operation*, made a database transaction "rollback" or create any other error, you will receive the exception in your dependency. |
|||
|
|||
So, you can look for that specific exception inside the dependency with `except SomeException`. |
|||
|
|||
In the same way, you can use `finally` to make sure the exit steps are executed, no matter if there was an exception or not. |
|||
|
|||
```Python hl_lines="3 5" |
|||
{!./src/dependencies/tutorial007.py!} |
|||
``` |
|||
|
|||
## Sub-dependencies with `yield` |
|||
|
|||
You can have sub-dependencies and "trees" of sub-dependencies of any size and shape, and any or all of them can use `yield`. |
|||
|
|||
**FastAPI** will make sure that the "exit code" in each dependency with `yield` is run in the correct order. |
|||
|
|||
For example, `dependency_c` can have a dependency on `dependency_b`, and `dependency_b` on `dependency_a`: |
|||
|
|||
```Python hl_lines="4 12 20" |
|||
{!./src/dependencies/tutorial008.py!} |
|||
``` |
|||
|
|||
And all of them can use `yield`. |
|||
|
|||
In this case `dependency_c`, to execute its exit code, needs the value from `dependency_b` (here named `dep_b`) to still be available. |
|||
|
|||
And, in turn, `dependency_b` needs the value from `dependency_a` (here named `dep_a`) to be available for its exit code. |
|||
|
|||
```Python hl_lines="16 17 24 25" |
|||
{!./src/dependencies/tutorial008.py!} |
|||
``` |
|||
|
|||
The same way, you could have dependencies with `yield` and `return` mixed. |
|||
|
|||
And you could have a single dependency that requires several other dependencies with `yield`, etc. |
|||
|
|||
You can have any combinations of dependencies that you want. |
|||
|
|||
**FastAPI** will make sure everything is run in the correct order. |
|||
|
|||
!!! note "Technical Details" |
|||
This works thanks to Python's <a href="https://docs.python.org/3/library/contextlib.html" target="_blank">Context Managers</a>. |
|||
|
|||
**FastAPI** uses them internally to achieve this. |
|||
|
|||
## Context Managers |
|||
|
|||
### What are "Context Managers" |
|||
|
|||
"Context Managers" are any of those Python objects that you can use in a `with` statement. |
|||
|
|||
For example, <a href="https://docs.python.org/3/tutorial/inputoutput.html#reading-and-writing-files" target="_blank">you can use `with` to read a file</a>: |
|||
|
|||
```Python |
|||
with open("./somefile.txt") as f: |
|||
contents = f.read() |
|||
print(contents) |
|||
``` |
|||
|
|||
Underneath, the `open("./somefile.txt")` returns an object that is a called a "Context Manager". |
|||
|
|||
When the `with` block finishes, it makes sure to close the file, even if there were exceptions. |
|||
|
|||
When you create a dependency with `yield`, **FastAPI** will internally convert it to a context manager, and combine it with some other related tools. |
|||
|
|||
### Using context managers in dependencies with `yield` |
|||
|
|||
!!! warning |
|||
This is, more or less, an "advanced" idea. |
|||
|
|||
If you are just starting with **FastAPI** you might want to skip it for now. |
|||
|
|||
In Python, you can create context managers by <a href="https://docs.python.org/3/reference/datamodel.html#context-managers" target="_blank">creating a class with two methods: `__enter__()` and `__exit__()`</a>. |
|||
|
|||
You can also use them with **FastAPI** dependencies with `yield` by using |
|||
`with` or `async with` statements inside of the dependency function: |
|||
|
|||
```Python hl_lines="1 2 3 4 5 6 7 8 9 13" |
|||
{!./src/dependencies/tutorial010.py!} |
|||
``` |
|||
|
|||
!!! tip |
|||
Another way to create a context manager is with: |
|||
|
|||
* <a href="https://docs.python.org/3/library/contextlib.html#contextlib.contextmanager" target="_blank">`@contextlib.contextmanager`</a> or |
|||
* <a href="https://docs.python.org/3/library/contextlib.html#contextlib.asynccontextmanager" target="_blank">`@contextlib.asynccontextmanager`</a> |
|||
|
|||
using them to decorate a function with a single `yield`. |
|||
|
|||
That's what **FastAPI** uses internally for dependencies with `yield`. |
|||
|
|||
But you don't have to use the decorators for FastAPI dependencies (and you shouldn't). |
|||
|
|||
FastAPI will do it for you internally. |
|||
@ -0,0 +1,45 @@ |
|||
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) |
|||
@ -0,0 +1,349 @@ |
|||
from typing import Dict |
|||
|
|||
import pytest |
|||
from fastapi import BackgroundTasks, Depends, FastAPI |
|||
from starlette.testclient import TestClient |
|||
|
|||
app = FastAPI() |
|||
state = { |
|||
"/async": "asyncgen not started", |
|||
"/sync": "generator not started", |
|||
"/async_raise": "asyncgen raise not started", |
|||
"/sync_raise": "generator raise not started", |
|||
"context_a": "not started a", |
|||
"context_b": "not started b", |
|||
"bg": "not set", |
|||
"sync_bg": "not set", |
|||
} |
|||
|
|||
errors = [] |
|||
|
|||
|
|||
async def get_state(): |
|||
return state |
|||
|
|||
|
|||
class AsyncDependencyError(Exception): |
|||
pass |
|||
|
|||
|
|||
class SyncDependencyError(Exception): |
|||
pass |
|||
|
|||
|
|||
class OtherDependencyError(Exception): |
|||
pass |
|||
|
|||
|
|||
async def asyncgen_state(state: Dict[str, str] = Depends(get_state)): |
|||
state["/async"] = "asyncgen started" |
|||
yield state["/async"] |
|||
state["/async"] = "asyncgen completed" |
|||
|
|||
|
|||
def generator_state(state: Dict[str, str] = Depends(get_state)): |
|||
state["/sync"] = "generator started" |
|||
yield state["/sync"] |
|||
state["/sync"] = "generator completed" |
|||
|
|||
|
|||
async def asyncgen_state_try(state: Dict[str, str] = Depends(get_state)): |
|||
state["/async_raise"] = "asyncgen raise started" |
|||
try: |
|||
yield state["/async_raise"] |
|||
except AsyncDependencyError: |
|||
errors.append("/async_raise") |
|||
finally: |
|||
state["/async_raise"] = "asyncgen raise finalized" |
|||
|
|||
|
|||
def generator_state_try(state: Dict[str, str] = Depends(get_state)): |
|||
state["/sync_raise"] = "generator raise started" |
|||
try: |
|||
yield state["/sync_raise"] |
|||
except SyncDependencyError: |
|||
errors.append("/sync_raise") |
|||
finally: |
|||
state["/sync_raise"] = "generator raise finalized" |
|||
|
|||
|
|||
async def context_a(state: dict = Depends(get_state)): |
|||
state["context_a"] = "started a" |
|||
try: |
|||
yield state |
|||
finally: |
|||
state["context_a"] = "finished a" |
|||
|
|||
|
|||
async def context_b(state: dict = Depends(context_a)): |
|||
state["context_b"] = "started b" |
|||
try: |
|||
yield state |
|||
finally: |
|||
state["context_b"] = f"finished b with a: {state['context_a']}" |
|||
|
|||
|
|||
@app.get("/async") |
|||
async def get_async(state: str = Depends(asyncgen_state)): |
|||
return state |
|||
|
|||
|
|||
@app.get("/sync") |
|||
async def get_sync(state: str = Depends(generator_state)): |
|||
return state |
|||
|
|||
|
|||
@app.get("/async_raise") |
|||
async def get_async_raise(state: str = Depends(asyncgen_state_try)): |
|||
assert state == "asyncgen raise started" |
|||
raise AsyncDependencyError() |
|||
|
|||
|
|||
@app.get("/sync_raise") |
|||
async def get_sync_raise(state: str = Depends(generator_state_try)): |
|||
assert state == "generator raise started" |
|||
raise SyncDependencyError() |
|||
|
|||
|
|||
@app.get("/async_raise_other") |
|||
async def get_async_raise_other(state: str = Depends(asyncgen_state_try)): |
|||
assert state == "asyncgen raise started" |
|||
raise OtherDependencyError() |
|||
|
|||
|
|||
@app.get("/sync_raise_other") |
|||
async def get_sync_raise_other(state: str = Depends(generator_state_try)): |
|||
assert state == "generator raise started" |
|||
raise OtherDependencyError() |
|||
|
|||
|
|||
@app.get("/context_b") |
|||
async def get_context_b(state: dict = Depends(context_b)): |
|||
return state |
|||
|
|||
|
|||
@app.get("/context_b_raise") |
|||
async def get_context_b_raise(state: dict = Depends(context_b)): |
|||
assert state["context_b"] == "started b" |
|||
assert state["context_a"] == "started a" |
|||
raise OtherDependencyError() |
|||
|
|||
|
|||
@app.get("/context_b_bg") |
|||
async def get_context_b_bg(tasks: BackgroundTasks, state: dict = Depends(context_b)): |
|||
async def bg(state: dict): |
|||
state["bg"] = f"bg set - b: {state['context_b']} - a: {state['context_a']}" |
|||
|
|||
tasks.add_task(bg, state) |
|||
return state |
|||
|
|||
|
|||
# Sync versions |
|||
|
|||
|
|||
@app.get("/sync_async") |
|||
def get_sync_async(state: str = Depends(asyncgen_state)): |
|||
return state |
|||
|
|||
|
|||
@app.get("/sync_sync") |
|||
def get_sync_sync(state: str = Depends(generator_state)): |
|||
return state |
|||
|
|||
|
|||
@app.get("/sync_async_raise") |
|||
def get_sync_async_raise(state: str = Depends(asyncgen_state_try)): |
|||
assert state == "asyncgen raise started" |
|||
raise AsyncDependencyError() |
|||
|
|||
|
|||
@app.get("/sync_sync_raise") |
|||
def get_sync_sync_raise(state: str = Depends(generator_state_try)): |
|||
assert state == "generator raise started" |
|||
raise SyncDependencyError() |
|||
|
|||
|
|||
@app.get("/sync_async_raise_other") |
|||
def get_sync_async_raise_other(state: str = Depends(asyncgen_state_try)): |
|||
assert state == "asyncgen raise started" |
|||
raise OtherDependencyError() |
|||
|
|||
|
|||
@app.get("/sync_sync_raise_other") |
|||
def get_sync_sync_raise_other(state: str = Depends(generator_state_try)): |
|||
assert state == "generator raise started" |
|||
raise OtherDependencyError() |
|||
|
|||
|
|||
@app.get("/sync_context_b") |
|||
def get_sync_context_b(state: dict = Depends(context_b)): |
|||
return state |
|||
|
|||
|
|||
@app.get("/sync_context_b_raise") |
|||
def get_sync_context_b_raise(state: dict = Depends(context_b)): |
|||
assert state["context_b"] == "started b" |
|||
assert state["context_a"] == "started a" |
|||
raise OtherDependencyError() |
|||
|
|||
|
|||
@app.get("/sync_context_b_bg") |
|||
async def get_sync_context_b_bg( |
|||
tasks: BackgroundTasks, state: dict = Depends(context_b) |
|||
): |
|||
async def bg(state: dict): |
|||
state[ |
|||
"sync_bg" |
|||
] = f"sync_bg set - b: {state['context_b']} - a: {state['context_a']}" |
|||
|
|||
tasks.add_task(bg, state) |
|||
return state |
|||
|
|||
|
|||
client = TestClient(app) |
|||
|
|||
|
|||
def test_async_state(): |
|||
assert state["/async"] == f"asyncgen not started" |
|||
response = client.get("/async") |
|||
assert response.status_code == 200 |
|||
assert response.json() == f"asyncgen started" |
|||
assert state["/async"] == f"asyncgen completed" |
|||
|
|||
|
|||
def test_sync_state(): |
|||
assert state["/sync"] == f"generator not started" |
|||
response = client.get("/sync") |
|||
assert response.status_code == 200 |
|||
assert response.json() == f"generator started" |
|||
assert state["/sync"] == f"generator completed" |
|||
|
|||
|
|||
def test_async_raise_other(): |
|||
assert state["/async_raise"] == "asyncgen raise not started" |
|||
with pytest.raises(OtherDependencyError): |
|||
client.get("/async_raise_other") |
|||
assert state["/async_raise"] == "asyncgen raise finalized" |
|||
assert "/async_raise" not in errors |
|||
|
|||
|
|||
def test_sync_raise_other(): |
|||
assert state["/sync_raise"] == "generator raise not started" |
|||
with pytest.raises(OtherDependencyError): |
|||
client.get("/sync_raise_other") |
|||
assert state["/sync_raise"] == "generator raise finalized" |
|||
assert "/sync_raise" not in errors |
|||
|
|||
|
|||
def test_async_raise(): |
|||
response = client.get("/async_raise") |
|||
assert response.status_code == 500 |
|||
assert state["/async_raise"] == "asyncgen raise finalized" |
|||
assert "/async_raise" in errors |
|||
errors.clear() |
|||
|
|||
|
|||
def test_context_b(): |
|||
response = client.get("/context_b") |
|||
data = response.json() |
|||
assert data["context_b"] == "started b" |
|||
assert data["context_a"] == "started a" |
|||
assert state["context_b"] == "finished b with a: started a" |
|||
assert state["context_a"] == "finished a" |
|||
|
|||
|
|||
def test_context_b_raise(): |
|||
with pytest.raises(OtherDependencyError): |
|||
client.get("/context_b_raise") |
|||
assert state["context_b"] == "finished b with a: started a" |
|||
assert state["context_a"] == "finished a" |
|||
|
|||
|
|||
def test_background_tasks(): |
|||
response = client.get("/context_b_bg") |
|||
data = response.json() |
|||
assert data["context_b"] == "started b" |
|||
assert data["context_a"] == "started a" |
|||
assert data["bg"] == "not set" |
|||
assert state["context_b"] == "finished b with a: started a" |
|||
assert state["context_a"] == "finished a" |
|||
assert state["bg"] == "bg set - b: started b - a: started a" |
|||
|
|||
|
|||
def test_sync_raise(): |
|||
response = client.get("/sync_raise") |
|||
assert response.status_code == 500 |
|||
assert state["/sync_raise"] == "generator raise finalized" |
|||
assert "/sync_raise" in errors |
|||
errors.clear() |
|||
|
|||
|
|||
def test_sync_async_state(): |
|||
response = client.get("/sync_async") |
|||
assert response.status_code == 200 |
|||
assert response.json() == f"asyncgen started" |
|||
assert state["/async"] == f"asyncgen completed" |
|||
|
|||
|
|||
def test_sync_sync_state(): |
|||
response = client.get("/sync_sync") |
|||
assert response.status_code == 200 |
|||
assert response.json() == f"generator started" |
|||
assert state["/sync"] == f"generator completed" |
|||
|
|||
|
|||
def test_sync_async_raise_other(): |
|||
with pytest.raises(OtherDependencyError): |
|||
client.get("/sync_async_raise_other") |
|||
assert state["/async_raise"] == "asyncgen raise finalized" |
|||
assert "/async_raise" not in errors |
|||
|
|||
|
|||
def test_sync_sync_raise_other(): |
|||
with pytest.raises(OtherDependencyError): |
|||
client.get("/sync_sync_raise_other") |
|||
assert state["/sync_raise"] == "generator raise finalized" |
|||
assert "/sync_raise" not in errors |
|||
|
|||
|
|||
def test_sync_async_raise(): |
|||
response = client.get("/sync_async_raise") |
|||
assert response.status_code == 500 |
|||
assert state["/async_raise"] == "asyncgen raise finalized" |
|||
assert "/async_raise" in errors |
|||
errors.clear() |
|||
|
|||
|
|||
def test_sync_sync_raise(): |
|||
response = client.get("/sync_sync_raise") |
|||
assert response.status_code == 500 |
|||
assert state["/sync_raise"] == "generator raise finalized" |
|||
assert "/sync_raise" in errors |
|||
errors.clear() |
|||
|
|||
|
|||
def test_sync_context_b(): |
|||
response = client.get("/sync_context_b") |
|||
data = response.json() |
|||
assert data["context_b"] == "started b" |
|||
assert data["context_a"] == "started a" |
|||
assert state["context_b"] == "finished b with a: started a" |
|||
assert state["context_a"] == "finished a" |
|||
|
|||
|
|||
def test_sync_context_b_raise(): |
|||
with pytest.raises(OtherDependencyError): |
|||
client.get("/sync_context_b_raise") |
|||
assert state["context_b"] == "finished b with a: started a" |
|||
assert state["context_a"] == "finished a" |
|||
|
|||
|
|||
def test_sync_background_tasks(): |
|||
response = client.get("/sync_context_b_bg") |
|||
data = response.json() |
|||
assert data["context_b"] == "started b" |
|||
assert data["context_a"] == "started a" |
|||
assert data["sync_bg"] == "not set" |
|||
assert state["context_b"] == "finished b with a: started a" |
|||
assert state["context_a"] == "finished a" |
|||
assert state["sync_bg"] == "sync_bg set - b: started b - a: started a" |
|||
@ -0,0 +1,12 @@ |
|||
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() |
|||
@ -1,8 +1,7 @@ |
|||
from starlette.testclient import TestClient |
|||
|
|||
from sql_databases.sql_app.main import app |
|||
from pathlib import Path |
|||
|
|||
client = TestClient(app) |
|||
import pytest |
|||
from starlette.testclient import TestClient |
|||
|
|||
openapi_schema = { |
|||
"openapi": "3.0.2", |
|||
@ -282,13 +281,24 @@ openapi_schema = { |
|||
} |
|||
|
|||
|
|||
def test_openapi_schema(): |
|||
@pytest.fixture(scope="module") |
|||
def client(): |
|||
# Import while creating the client to create the DB after starting the test session |
|||
from sql_databases.sql_app.main import app |
|||
|
|||
test_db = Path("./test.db") |
|||
with TestClient(app) as c: |
|||
yield c |
|||
test_db.unlink() |
|||
|
|||
|
|||
def test_openapi_schema(client): |
|||
response = client.get("/openapi.json") |
|||
assert response.status_code == 200 |
|||
assert response.json() == openapi_schema |
|||
|
|||
|
|||
def test_create_user(): |
|||
def test_create_user(client): |
|||
test_user = {"email": "[email protected]", "password": "secret"} |
|||
response = client.post("/users/", json=test_user) |
|||
assert response.status_code == 200 |
|||
@ -299,7 +309,7 @@ def test_create_user(): |
|||
assert response.status_code == 400 |
|||
|
|||
|
|||
def test_get_user(): |
|||
def test_get_user(client): |
|||
response = client.get("/users/1") |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
@ -307,12 +317,12 @@ def test_get_user(): |
|||
assert "id" in data |
|||
|
|||
|
|||
def test_inexistent_user(): |
|||
def test_inexistent_user(client): |
|||
response = client.get("/users/999") |
|||
assert response.status_code == 404 |
|||
|
|||
|
|||
def test_get_users(): |
|||
def test_get_users(client): |
|||
response = client.get("/users/") |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
@ -320,7 +330,7 @@ def test_get_users(): |
|||
assert "id" in data[0] |
|||
|
|||
|
|||
def test_create_item(): |
|||
def test_create_item(client): |
|||
item = {"title": "Foo", "description": "Something that fights"} |
|||
response = client.post("/users/1/items/", json=item) |
|||
assert response.status_code == 200 |
|||
@ -343,7 +353,7 @@ def test_create_item(): |
|||
assert item_to_check["description"] == item["description"] |
|||
|
|||
|
|||
def test_read_items(): |
|||
def test_read_items(client): |
|||
response = client.get("/items/") |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
|
|||
@ -0,0 +1,363 @@ |
|||
from pathlib import Path |
|||
|
|||
import pytest |
|||
from starlette.testclient import TestClient |
|||
|
|||
openapi_schema = { |
|||
"openapi": "3.0.2", |
|||
"info": {"title": "Fast API", "version": "0.1.0"}, |
|||
"paths": { |
|||
"/users/": { |
|||
"get": { |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"title": "Response_Read_Users_Users__Get", |
|||
"type": "array", |
|||
"items": {"$ref": "#/components/schemas/User"}, |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
"422": { |
|||
"description": "Validation Error", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/HTTPValidationError" |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
}, |
|||
"summary": "Read Users", |
|||
"operationId": "read_users_users__get", |
|||
"parameters": [ |
|||
{ |
|||
"required": False, |
|||
"schema": {"title": "Skip", "type": "integer", "default": 0}, |
|||
"name": "skip", |
|||
"in": "query", |
|||
}, |
|||
{ |
|||
"required": False, |
|||
"schema": {"title": "Limit", "type": "integer", "default": 100}, |
|||
"name": "limit", |
|||
"in": "query", |
|||
}, |
|||
], |
|||
}, |
|||
"post": { |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": {"$ref": "#/components/schemas/User"} |
|||
} |
|||
}, |
|||
}, |
|||
"422": { |
|||
"description": "Validation Error", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/HTTPValidationError" |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
}, |
|||
"summary": "Create User", |
|||
"operationId": "create_user_users__post", |
|||
"requestBody": { |
|||
"content": { |
|||
"application/json": { |
|||
"schema": {"$ref": "#/components/schemas/UserCreate"} |
|||
} |
|||
}, |
|||
"required": True, |
|||
}, |
|||
}, |
|||
}, |
|||
"/users/{user_id}": { |
|||
"get": { |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": {"$ref": "#/components/schemas/User"} |
|||
} |
|||
}, |
|||
}, |
|||
"422": { |
|||
"description": "Validation Error", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/HTTPValidationError" |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
}, |
|||
"summary": "Read User", |
|||
"operationId": "read_user_users__user_id__get", |
|||
"parameters": [ |
|||
{ |
|||
"required": True, |
|||
"schema": {"title": "User_Id", "type": "integer"}, |
|||
"name": "user_id", |
|||
"in": "path", |
|||
} |
|||
], |
|||
} |
|||
}, |
|||
"/users/{user_id}/items/": { |
|||
"post": { |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": {"$ref": "#/components/schemas/Item"} |
|||
} |
|||
}, |
|||
}, |
|||
"422": { |
|||
"description": "Validation Error", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/HTTPValidationError" |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
}, |
|||
"summary": "Create Item For User", |
|||
"operationId": "create_item_for_user_users__user_id__items__post", |
|||
"parameters": [ |
|||
{ |
|||
"required": True, |
|||
"schema": {"title": "User_Id", "type": "integer"}, |
|||
"name": "user_id", |
|||
"in": "path", |
|||
} |
|||
], |
|||
"requestBody": { |
|||
"content": { |
|||
"application/json": { |
|||
"schema": {"$ref": "#/components/schemas/ItemCreate"} |
|||
} |
|||
}, |
|||
"required": True, |
|||
}, |
|||
} |
|||
}, |
|||
"/items/": { |
|||
"get": { |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"title": "Response_Read_Items_Items__Get", |
|||
"type": "array", |
|||
"items": {"$ref": "#/components/schemas/Item"}, |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
"422": { |
|||
"description": "Validation Error", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/HTTPValidationError" |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
}, |
|||
"summary": "Read Items", |
|||
"operationId": "read_items_items__get", |
|||
"parameters": [ |
|||
{ |
|||
"required": False, |
|||
"schema": {"title": "Skip", "type": "integer", "default": 0}, |
|||
"name": "skip", |
|||
"in": "query", |
|||
}, |
|||
{ |
|||
"required": False, |
|||
"schema": {"title": "Limit", "type": "integer", "default": 100}, |
|||
"name": "limit", |
|||
"in": "query", |
|||
}, |
|||
], |
|||
} |
|||
}, |
|||
}, |
|||
"components": { |
|||
"schemas": { |
|||
"ItemCreate": { |
|||
"title": "ItemCreate", |
|||
"required": ["title"], |
|||
"type": "object", |
|||
"properties": { |
|||
"title": {"title": "Title", "type": "string"}, |
|||
"description": {"title": "Description", "type": "string"}, |
|||
}, |
|||
}, |
|||
"Item": { |
|||
"title": "Item", |
|||
"required": ["title", "id", "owner_id"], |
|||
"type": "object", |
|||
"properties": { |
|||
"title": {"title": "Title", "type": "string"}, |
|||
"description": {"title": "Description", "type": "string"}, |
|||
"id": {"title": "Id", "type": "integer"}, |
|||
"owner_id": {"title": "Owner_Id", "type": "integer"}, |
|||
}, |
|||
}, |
|||
"User": { |
|||
"title": "User", |
|||
"required": ["email", "id", "is_active"], |
|||
"type": "object", |
|||
"properties": { |
|||
"email": {"title": "Email", "type": "string"}, |
|||
"id": {"title": "Id", "type": "integer"}, |
|||
"is_active": {"title": "Is_Active", "type": "boolean"}, |
|||
"items": { |
|||
"title": "Items", |
|||
"type": "array", |
|||
"items": {"$ref": "#/components/schemas/Item"}, |
|||
"default": [], |
|||
}, |
|||
}, |
|||
}, |
|||
"UserCreate": { |
|||
"title": "UserCreate", |
|||
"required": ["email", "password"], |
|||
"type": "object", |
|||
"properties": { |
|||
"email": {"title": "Email", "type": "string"}, |
|||
"password": {"title": "Password", "type": "string"}, |
|||
}, |
|||
}, |
|||
"ValidationError": { |
|||
"title": "ValidationError", |
|||
"required": ["loc", "msg", "type"], |
|||
"type": "object", |
|||
"properties": { |
|||
"loc": { |
|||
"title": "Location", |
|||
"type": "array", |
|||
"items": {"type": "string"}, |
|||
}, |
|||
"msg": {"title": "Message", "type": "string"}, |
|||
"type": {"title": "Error Type", "type": "string"}, |
|||
}, |
|||
}, |
|||
"HTTPValidationError": { |
|||
"title": "HTTPValidationError", |
|||
"type": "object", |
|||
"properties": { |
|||
"detail": { |
|||
"title": "Detail", |
|||
"type": "array", |
|||
"items": {"$ref": "#/components/schemas/ValidationError"}, |
|||
} |
|||
}, |
|||
}, |
|||
} |
|||
}, |
|||
} |
|||
|
|||
|
|||
@pytest.fixture(scope="module") |
|||
def client(): |
|||
# Import while creating the client to create the DB after starting the test session |
|||
from sql_databases.sql_app.alt_main import app |
|||
|
|||
test_db = Path("./test.db") |
|||
with TestClient(app) as c: |
|||
yield c |
|||
test_db.unlink() |
|||
|
|||
|
|||
def test_openapi_schema(client): |
|||
response = client.get("/openapi.json") |
|||
assert response.status_code == 200 |
|||
assert response.json() == openapi_schema |
|||
|
|||
|
|||
def test_create_user(client): |
|||
test_user = {"email": "[email protected]", "password": "secret"} |
|||
response = client.post("/users/", json=test_user) |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
assert test_user["email"] == data["email"] |
|||
assert "id" in data |
|||
response = client.post("/users/", json=test_user) |
|||
assert response.status_code == 400 |
|||
|
|||
|
|||
def test_get_user(client): |
|||
response = client.get("/users/1") |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
assert "email" in data |
|||
assert "id" in data |
|||
|
|||
|
|||
def test_inexistent_user(client): |
|||
response = client.get("/users/999") |
|||
assert response.status_code == 404 |
|||
|
|||
|
|||
def test_get_users(client): |
|||
response = client.get("/users/") |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
assert "email" in data[0] |
|||
assert "id" in data[0] |
|||
|
|||
|
|||
def test_create_item(client): |
|||
item = {"title": "Foo", "description": "Something that fights"} |
|||
response = client.post("/users/1/items/", json=item) |
|||
assert response.status_code == 200 |
|||
item_data = response.json() |
|||
assert item["title"] == item_data["title"] |
|||
assert item["description"] == item_data["description"] |
|||
assert "id" in item_data |
|||
assert "owner_id" in item_data |
|||
response = client.get("/users/1") |
|||
assert response.status_code == 200 |
|||
user_data = response.json() |
|||
item_to_check = [it for it in user_data["items"] if it["id"] == item_data["id"]][0] |
|||
assert item_to_check["title"] == item["title"] |
|||
assert item_to_check["description"] == item["description"] |
|||
response = client.get("/users/1") |
|||
assert response.status_code == 200 |
|||
user_data = response.json() |
|||
item_to_check = [it for it in user_data["items"] if it["id"] == item_data["id"]][0] |
|||
assert item_to_check["title"] == item["title"] |
|||
assert item_to_check["description"] == item["description"] |
|||
|
|||
|
|||
def test_read_items(client): |
|||
response = client.get("/items/") |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
assert data |
|||
first_item = data[0] |
|||
assert "title" in first_item |
|||
assert "description" in first_item |
|||
Loading…
Reference in new issue