|
|
|
@ -1,7 +1,9 @@ |
|
|
|
import json |
|
|
|
from typing import Any |
|
|
|
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 |
|
|
|
@ -20,8 +22,47 @@ def dep_session() -> Any: |
|
|
|
|
|
|
|
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() |
|
|
|
|
|
|
|
|
|
|
|
@ -53,6 +94,36 @@ def get_stream_session( |
|
|
|
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) |
|
|
|
|
|
|
|
|
|
|
|
@ -76,3 +147,38 @@ def test_two_scopes() -> None: |
|
|
|
data = json.loads(response.content) |
|
|
|
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 = json.loads(response.content) |
|
|
|
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 = json.loads(response.content) |
|
|
|
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 = json.loads(response.content) |
|
|
|
assert data["named_session_open"] is True |
|
|
|
assert data["session_open"] is False |
|
|
|
|