Browse Source
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>pull/14287/head
Sebastián Ramírez
7 months ago
committed by
GitHub
GitHub
14 changed files with 653 additions and 70 deletions
@ -0,0 +1,15 @@ |
|||
from fastapi import Depends, FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
def get_username(): |
|||
try: |
|||
yield "Rick" |
|||
finally: |
|||
print("Cleanup up before response is sent") |
|||
|
|||
|
|||
@app.get("/users/me") |
|||
def get_user_me(username: str = Depends(get_username, scope="function")): |
|||
return username |
|||
@ -0,0 +1,16 @@ |
|||
from fastapi import Depends, FastAPI |
|||
from typing_extensions import Annotated |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
def get_username(): |
|||
try: |
|||
yield "Rick" |
|||
finally: |
|||
print("Cleanup up before response is sent") |
|||
|
|||
|
|||
@app.get("/users/me") |
|||
def get_user_me(username: Annotated[str, Depends(get_username, scope="function")]): |
|||
return username |
|||
@ -0,0 +1,17 @@ |
|||
from typing import Annotated |
|||
|
|||
from fastapi import Depends, FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
def get_username(): |
|||
try: |
|||
yield "Rick" |
|||
finally: |
|||
print("Cleanup up before response is sent") |
|||
|
|||
|
|||
@app.get("/users/me") |
|||
def get_user_me(username: Annotated[str, Depends(get_username, scope="function")]): |
|||
return username |
|||
@ -0,0 +1,184 @@ |
|||
import json |
|||
from typing import Any, Tuple |
|||
|
|||
import pytest |
|||
from fastapi import Depends, FastAPI |
|||
from fastapi.exceptions import FastAPIError |
|||
from fastapi.responses import StreamingResponse |
|||
from fastapi.testclient import TestClient |
|||
from typing_extensions import Annotated |
|||
|
|||
|
|||
class Session: |
|||
def __init__(self) -> None: |
|||
self.open = True |
|||
|
|||
|
|||
def dep_session() -> Any: |
|||
s = Session() |
|||
yield s |
|||
s.open = False |
|||
|
|||
|
|||
SessionFuncDep = Annotated[Session, Depends(dep_session, scope="function")] |
|||
SessionRequestDep = Annotated[Session, Depends(dep_session, scope="request")] |
|||
SessionDefaultDep = Annotated[Session, Depends(dep_session)] |
|||
|
|||
|
|||
class NamedSession: |
|||
def __init__(self, name: str = "default") -> None: |
|||
self.name = name |
|||
self.open = True |
|||
|
|||
|
|||
def get_named_session(session: SessionRequestDep, session_b: SessionDefaultDep) -> Any: |
|||
assert session is session_b |
|||
named_session = NamedSession(name="named") |
|||
yield named_session, session_b |
|||
named_session.open = False |
|||
|
|||
|
|||
NamedSessionsDep = Annotated[Tuple[NamedSession, Session], Depends(get_named_session)] |
|||
|
|||
|
|||
def get_named_func_session(session: SessionFuncDep) -> Any: |
|||
named_session = NamedSession(name="named") |
|||
yield named_session, session |
|||
named_session.open = False |
|||
|
|||
|
|||
def get_named_regular_func_session(session: SessionFuncDep) -> Any: |
|||
named_session = NamedSession(name="named") |
|||
return named_session, session |
|||
|
|||
|
|||
BrokenSessionsDep = Annotated[ |
|||
Tuple[NamedSession, Session], Depends(get_named_func_session) |
|||
] |
|||
NamedSessionsFuncDep = Annotated[ |
|||
Tuple[NamedSession, Session], Depends(get_named_func_session, scope="function") |
|||
] |
|||
|
|||
RegularSessionsDep = Annotated[ |
|||
Tuple[NamedSession, Session], Depends(get_named_regular_func_session) |
|||
] |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/function-scope") |
|||
def function_scope(session: SessionFuncDep) -> Any: |
|||
def iter_data(): |
|||
yield json.dumps({"is_open": session.open}) |
|||
|
|||
return StreamingResponse(iter_data()) |
|||
|
|||
|
|||
@app.get("/request-scope") |
|||
def request_scope(session: SessionRequestDep) -> Any: |
|||
def iter_data(): |
|||
yield json.dumps({"is_open": session.open}) |
|||
|
|||
return StreamingResponse(iter_data()) |
|||
|
|||
|
|||
@app.get("/two-scopes") |
|||
def get_stream_session( |
|||
function_session: SessionFuncDep, request_session: SessionRequestDep |
|||
) -> Any: |
|||
def iter_data(): |
|||
yield json.dumps( |
|||
{"func_is_open": function_session.open, "req_is_open": request_session.open} |
|||
) |
|||
|
|||
return StreamingResponse(iter_data()) |
|||
|
|||
|
|||
@app.get("/sub") |
|||
def get_sub(sessions: NamedSessionsDep) -> Any: |
|||
def iter_data(): |
|||
yield json.dumps( |
|||
{"named_session_open": sessions[0].open, "session_open": sessions[1].open} |
|||
) |
|||
|
|||
return StreamingResponse(iter_data()) |
|||
|
|||
|
|||
@app.get("/named-function-scope") |
|||
def get_named_function_scope(sessions: NamedSessionsFuncDep) -> Any: |
|||
def iter_data(): |
|||
yield json.dumps( |
|||
{"named_session_open": sessions[0].open, "session_open": sessions[1].open} |
|||
) |
|||
|
|||
return StreamingResponse(iter_data()) |
|||
|
|||
|
|||
@app.get("/regular-function-scope") |
|||
def get_regular_function_scope(sessions: RegularSessionsDep) -> Any: |
|||
def iter_data(): |
|||
yield json.dumps( |
|||
{"named_session_open": sessions[0].open, "session_open": sessions[1].open} |
|||
) |
|||
|
|||
return StreamingResponse(iter_data()) |
|||
|
|||
|
|||
client = TestClient(app) |
|||
|
|||
|
|||
def test_function_scope() -> None: |
|||
response = client.get("/function-scope") |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
assert data["is_open"] is False |
|||
|
|||
|
|||
def test_request_scope() -> None: |
|||
response = client.get("/request-scope") |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
assert data["is_open"] is True |
|||
|
|||
|
|||
def test_two_scopes() -> None: |
|||
response = client.get("/two-scopes") |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
assert data["func_is_open"] is False |
|||
assert data["req_is_open"] is True |
|||
|
|||
|
|||
def test_sub() -> None: |
|||
response = client.get("/sub") |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
assert data["named_session_open"] is True |
|||
assert data["session_open"] is True |
|||
|
|||
|
|||
def test_broken_scope() -> None: |
|||
with pytest.raises( |
|||
FastAPIError, |
|||
match='The dependency "get_named_func_session" has a scope of "request", it cannot depend on dependencies with scope "function"', |
|||
): |
|||
|
|||
@app.get("/broken-scope") |
|||
def get_broken(sessions: BrokenSessionsDep) -> Any: # pragma: no cover |
|||
pass |
|||
|
|||
|
|||
def test_named_function_scope() -> None: |
|||
response = client.get("/named-function-scope") |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
assert data["named_session_open"] is False |
|||
assert data["session_open"] is False |
|||
|
|||
|
|||
def test_regular_function_scope() -> None: |
|||
response = client.get("/regular-function-scope") |
|||
assert response.status_code == 200 |
|||
data = response.json() |
|||
assert data["named_session_open"] is True |
|||
assert data["session_open"] is False |
|||
@ -0,0 +1,201 @@ |
|||
from contextvars import ContextVar |
|||
from typing import Any, Dict, Tuple |
|||
|
|||
import pytest |
|||
from fastapi import Depends, FastAPI, WebSocket |
|||
from fastapi.exceptions import FastAPIError |
|||
from fastapi.testclient import TestClient |
|||
from typing_extensions import Annotated |
|||
|
|||
global_context: ContextVar[Dict[str, Any]] = ContextVar("global_context", default={}) # noqa: B039 |
|||
|
|||
|
|||
class Session: |
|||
def __init__(self) -> None: |
|||
self.open = True |
|||
|
|||
|
|||
async def dep_session() -> Any: |
|||
s = Session() |
|||
yield s |
|||
s.open = False |
|||
global_state = global_context.get() |
|||
global_state["session_closed"] = True |
|||
|
|||
|
|||
SessionFuncDep = Annotated[Session, Depends(dep_session, scope="function")] |
|||
SessionRequestDep = Annotated[Session, Depends(dep_session, scope="request")] |
|||
SessionDefaultDep = Annotated[Session, Depends(dep_session)] |
|||
|
|||
|
|||
class NamedSession: |
|||
def __init__(self, name: str = "default") -> None: |
|||
self.name = name |
|||
self.open = True |
|||
|
|||
|
|||
def get_named_session(session: SessionRequestDep, session_b: SessionDefaultDep) -> Any: |
|||
assert session is session_b |
|||
named_session = NamedSession(name="named") |
|||
yield named_session, session_b |
|||
named_session.open = False |
|||
global_state = global_context.get() |
|||
global_state["named_session_closed"] = True |
|||
|
|||
|
|||
NamedSessionsDep = Annotated[Tuple[NamedSession, Session], Depends(get_named_session)] |
|||
|
|||
|
|||
def get_named_func_session(session: SessionFuncDep) -> Any: |
|||
named_session = NamedSession(name="named") |
|||
yield named_session, session |
|||
named_session.open = False |
|||
global_state = global_context.get() |
|||
global_state["named_func_session_closed"] = True |
|||
|
|||
|
|||
def get_named_regular_func_session(session: SessionFuncDep) -> Any: |
|||
named_session = NamedSession(name="named") |
|||
return named_session, session |
|||
|
|||
|
|||
BrokenSessionsDep = Annotated[ |
|||
Tuple[NamedSession, Session], Depends(get_named_func_session) |
|||
] |
|||
NamedSessionsFuncDep = Annotated[ |
|||
Tuple[NamedSession, Session], Depends(get_named_func_session, scope="function") |
|||
] |
|||
|
|||
RegularSessionsDep = Annotated[ |
|||
Tuple[NamedSession, Session], Depends(get_named_regular_func_session) |
|||
] |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.websocket("/function-scope") |
|||
async def function_scope(websocket: WebSocket, session: SessionFuncDep) -> Any: |
|||
await websocket.accept() |
|||
await websocket.send_json({"is_open": session.open}) |
|||
|
|||
|
|||
@app.websocket("/request-scope") |
|||
async def request_scope(websocket: WebSocket, session: SessionRequestDep) -> Any: |
|||
await websocket.accept() |
|||
await websocket.send_json({"is_open": session.open}) |
|||
|
|||
|
|||
@app.websocket("/two-scopes") |
|||
async def get_stream_session( |
|||
websocket: WebSocket, |
|||
function_session: SessionFuncDep, |
|||
request_session: SessionRequestDep, |
|||
) -> Any: |
|||
await websocket.accept() |
|||
await websocket.send_json( |
|||
{"func_is_open": function_session.open, "req_is_open": request_session.open} |
|||
) |
|||
|
|||
|
|||
@app.websocket("/sub") |
|||
async def get_sub(websocket: WebSocket, sessions: NamedSessionsDep) -> Any: |
|||
await websocket.accept() |
|||
await websocket.send_json( |
|||
{"named_session_open": sessions[0].open, "session_open": sessions[1].open} |
|||
) |
|||
|
|||
|
|||
@app.websocket("/named-function-scope") |
|||
async def get_named_function_scope( |
|||
websocket: WebSocket, sessions: NamedSessionsFuncDep |
|||
) -> Any: |
|||
await websocket.accept() |
|||
await websocket.send_json( |
|||
{"named_session_open": sessions[0].open, "session_open": sessions[1].open} |
|||
) |
|||
|
|||
|
|||
@app.websocket("/regular-function-scope") |
|||
async def get_regular_function_scope( |
|||
websocket: WebSocket, sessions: RegularSessionsDep |
|||
) -> Any: |
|||
await websocket.accept() |
|||
await websocket.send_json( |
|||
{"named_session_open": sessions[0].open, "session_open": sessions[1].open} |
|||
) |
|||
|
|||
|
|||
client = TestClient(app) |
|||
|
|||
|
|||
def test_function_scope() -> None: |
|||
global_context.set({}) |
|||
global_state = global_context.get() |
|||
with client.websocket_connect("/function-scope") as websocket: |
|||
data = websocket.receive_json() |
|||
assert data["is_open"] is True |
|||
assert global_state["session_closed"] is True |
|||
|
|||
|
|||
def test_request_scope() -> None: |
|||
global_context.set({}) |
|||
global_state = global_context.get() |
|||
with client.websocket_connect("/request-scope") as websocket: |
|||
data = websocket.receive_json() |
|||
assert data["is_open"] is True |
|||
assert global_state["session_closed"] is True |
|||
|
|||
|
|||
def test_two_scopes() -> None: |
|||
global_context.set({}) |
|||
global_state = global_context.get() |
|||
with client.websocket_connect("/two-scopes") as websocket: |
|||
data = websocket.receive_json() |
|||
assert data["func_is_open"] is True |
|||
assert data["req_is_open"] is True |
|||
assert global_state["session_closed"] is True |
|||
|
|||
|
|||
def test_sub() -> None: |
|||
global_context.set({}) |
|||
global_state = global_context.get() |
|||
with client.websocket_connect("/sub") as websocket: |
|||
data = websocket.receive_json() |
|||
assert data["named_session_open"] is True |
|||
assert data["session_open"] is True |
|||
assert global_state["session_closed"] is True |
|||
assert global_state["named_session_closed"] is True |
|||
|
|||
|
|||
def test_broken_scope() -> None: |
|||
with pytest.raises( |
|||
FastAPIError, |
|||
match='The dependency "get_named_func_session" has a scope of "request", it cannot depend on dependencies with scope "function"', |
|||
): |
|||
|
|||
@app.websocket("/broken-scope") |
|||
async def get_broken( |
|||
websocket: WebSocket, sessions: BrokenSessionsDep |
|||
) -> Any: # pragma: no cover |
|||
pass |
|||
|
|||
|
|||
def test_named_function_scope() -> None: |
|||
global_context.set({}) |
|||
global_state = global_context.get() |
|||
with client.websocket_connect("/named-function-scope") as websocket: |
|||
data = websocket.receive_json() |
|||
assert data["named_session_open"] is True |
|||
assert data["session_open"] is True |
|||
assert global_state["session_closed"] is True |
|||
assert global_state["named_func_session_closed"] is True |
|||
|
|||
|
|||
def test_regular_function_scope() -> None: |
|||
global_context.set({}) |
|||
global_state = global_context.get() |
|||
with client.websocket_connect("/regular-function-scope") as websocket: |
|||
data = websocket.receive_json() |
|||
assert data["named_session_open"] is True |
|||
assert data["session_open"] is True |
|||
assert global_state["session_closed"] is True |
|||
@ -0,0 +1,27 @@ |
|||
import importlib |
|||
|
|||
import pytest |
|||
from fastapi.testclient import TestClient |
|||
|
|||
from ...utils import needs_py39 |
|||
|
|||
|
|||
@pytest.fixture( |
|||
name="client", |
|||
params=[ |
|||
"tutorial008e", |
|||
"tutorial008e_an", |
|||
pytest.param("tutorial008e_an_py39", marks=needs_py39), |
|||
], |
|||
) |
|||
def get_client(request: pytest.FixtureRequest): |
|||
mod = importlib.import_module(f"docs_src.dependencies.{request.param}") |
|||
|
|||
client = TestClient(mod.app) |
|||
return client |
|||
|
|||
|
|||
def test_get_users_me(client: TestClient): |
|||
response = client.get("/users/me") |
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == "Rick" |
|||
Loading…
Reference in new issue