Motov Yurii
9 months ago
committed by
GitHub
GitHub
53 changed files with 1668 additions and 529 deletions
@ -0,0 +1,133 @@ |
|||
# Von Pydantic v1 zu Pydantic v2 migrieren { #migrate-from-pydantic-v1-to-pydantic-v2 } |
|||
|
|||
Wenn Sie eine ältere FastAPI-App haben, nutzen Sie möglicherweise Pydantic Version 1. |
|||
|
|||
FastAPI unterstützt seit Version 0.100.0 sowohl Pydantic v1 als auch v2. |
|||
|
|||
Wenn Sie Pydantic v2 installiert hatten, wurde dieses verwendet. Wenn stattdessen Pydantic v1 installiert war, wurde jenes verwendet. |
|||
|
|||
Pydantic v1 ist jetzt deprecatet und die Unterstützung dafür wird in den nächsten Versionen von FastAPI entfernt, Sie sollten also zu **Pydantic v2 migrieren**. Auf diese Weise erhalten Sie die neuesten Features, Verbesserungen und Fixes. |
|||
|
|||
/// warning | Achtung |
|||
|
|||
Außerdem hat das Pydantic-Team die Unterstützung für Pydantic v1 in den neuesten Python-Versionen eingestellt, beginnend mit **Python 3.14**. |
|||
|
|||
Wenn Sie die neuesten Features von Python nutzen möchten, müssen Sie sicherstellen, dass Sie Pydantic v2 verwenden. |
|||
|
|||
/// |
|||
|
|||
Wenn Sie eine ältere FastAPI-App mit Pydantic v1 haben, zeige ich Ihnen hier, wie Sie sie zu Pydantic v2 migrieren, und die **neuen Features in FastAPI 0.119.0**, die Ihnen bei einer schrittweisen Migration helfen. |
|||
|
|||
## Offizieller Leitfaden { #official-guide } |
|||
|
|||
Pydantic hat einen offiziellen <a href="https://docs.pydantic.dev/latest/migration/" class="external-link" target="_blank">Migrationsleitfaden</a> von v1 zu v2. |
|||
|
|||
Er enthält auch, was sich geändert hat, wie Validierungen nun korrekter und strikter sind, mögliche Stolpersteine, usw. |
|||
|
|||
Sie können ihn lesen, um besser zu verstehen, was sich geändert hat. |
|||
|
|||
## Tests { #tests } |
|||
|
|||
Stellen Sie sicher, dass Sie [Tests](../tutorial/testing.md){.internal-link target=_blank} für Ihre App haben und diese in Continuous Integration (CI) ausführen. |
|||
|
|||
Auf diese Weise können Sie das Update durchführen und sicherstellen, dass weiterhin alles wie erwartet funktioniert. |
|||
|
|||
## `bump-pydantic` { #bump-pydantic } |
|||
|
|||
In vielen Fällen, wenn Sie reguläre Pydantic-Modelle ohne Anpassungen verwenden, können Sie den Großteil des Prozesses der Migration von Pydantic v1 auf Pydantic v2 automatisieren. |
|||
|
|||
Sie können <a href="https://github.com/pydantic/bump-pydantic" class="external-link" target="_blank">`bump-pydantic`</a> vom selben Pydantic-Team verwenden. |
|||
|
|||
Dieses Tool hilft Ihnen, den Großteil des zu ändernden Codes automatisch anzupassen. |
|||
|
|||
Danach können Sie die Tests ausführen und prüfen, ob alles funktioniert. Falls ja, sind Sie fertig. 😎 |
|||
|
|||
## Pydantic v1 in v2 { #pydantic-v1-in-v2 } |
|||
|
|||
Pydantic v2 enthält alles aus Pydantic v1 als Untermodul `pydantic.v1`. |
|||
|
|||
Das bedeutet, Sie können die neueste Version von Pydantic v2 installieren und die alten Pydantic‑v1‑Komponenten aus diesem Untermodul importieren und verwenden, als hätten Sie das alte Pydantic v1 installiert. |
|||
|
|||
{* ../../docs_src/pydantic_v1_in_v2/tutorial001_an_py310.py hl[1,4] *} |
|||
|
|||
### FastAPI-Unterstützung für Pydantic v1 in v2 { #fastapi-support-for-pydantic-v1-in-v2 } |
|||
|
|||
Seit FastAPI 0.119.0 gibt es außerdem eine teilweise Unterstützung für Pydantic v1 innerhalb von Pydantic v2, um die Migration auf v2 zu erleichtern. |
|||
|
|||
Sie könnten also Pydantic auf die neueste Version 2 aktualisieren und die Importe so ändern, dass das Untermodul `pydantic.v1` verwendet wird, und in vielen Fällen würde es einfach funktionieren. |
|||
|
|||
{* ../../docs_src/pydantic_v1_in_v2/tutorial002_an_py310.py hl[2,5,15] *} |
|||
|
|||
/// warning | Achtung |
|||
|
|||
Beachten Sie, dass, da das Pydantic‑Team Pydantic v1 in neueren Python‑Versionen nicht mehr unterstützt, beginnend mit Python 3.14, auch die Verwendung von `pydantic.v1` unter Python 3.14 und höher nicht unterstützt wird. |
|||
|
|||
/// |
|||
|
|||
### Pydantic v1 und v2 in derselben App { #pydantic-v1-and-v2-on-the-same-app } |
|||
|
|||
Es wird von Pydantic **nicht unterstützt**, dass ein Pydantic‑v2‑Modell Felder hat, die als Pydantic‑v1‑Modelle definiert sind, und umgekehrt. |
|||
|
|||
```mermaid |
|||
graph TB |
|||
subgraph "❌ Nicht unterstützt" |
|||
direction TB |
|||
subgraph V2["Pydantic-v2-Modell"] |
|||
V1Field["Pydantic-v1-Modell"] |
|||
end |
|||
subgraph V1["Pydantic-v1-Modell"] |
|||
V2Field["Pydantic-v2-Modell"] |
|||
end |
|||
end |
|||
|
|||
style V2 fill:#f9fff3 |
|||
style V1 fill:#fff6f0 |
|||
style V1Field fill:#fff6f0 |
|||
style V2Field fill:#f9fff3 |
|||
``` |
|||
|
|||
... aber Sie können getrennte Modelle, die Pydantic v1 bzw. v2 nutzen, in derselben App verwenden. |
|||
|
|||
```mermaid |
|||
graph TB |
|||
subgraph "✅ Unterstützt" |
|||
direction TB |
|||
subgraph V2["Pydantic-v2-Modell"] |
|||
V2Field["Pydantic-v2-Modell"] |
|||
end |
|||
subgraph V1["Pydantic-v1-Modell"] |
|||
V1Field["Pydantic-v1-Modell"] |
|||
end |
|||
end |
|||
|
|||
style V2 fill:#f9fff3 |
|||
style V1 fill:#fff6f0 |
|||
style V1Field fill:#fff6f0 |
|||
style V2Field fill:#f9fff3 |
|||
``` |
|||
|
|||
In einigen Fällen ist es sogar möglich, sowohl Pydantic‑v1‑ als auch Pydantic‑v2‑Modelle in derselben **Pfadoperation** Ihrer FastAPI‑App zu verwenden: |
|||
|
|||
{* ../../docs_src/pydantic_v1_in_v2/tutorial003_an_py310.py hl[2:3,6,12,21:22] *} |
|||
|
|||
Im obigen Beispiel ist das Eingabemodell ein Pydantic‑v1‑Modell, und das Ausgabemodell (definiert in `response_model=ItemV2`) ist ein Pydantic‑v2‑Modell. |
|||
|
|||
### Pydantic v1 Parameter { #pydantic-v1-parameters } |
|||
|
|||
Wenn Sie einige der FastAPI-spezifischen Tools für Parameter wie `Body`, `Query`, `Form`, usw. zusammen mit Pydantic‑v1‑Modellen verwenden müssen, können Sie die aus `fastapi.temp_pydantic_v1_params` importieren, während Sie die Migration zu Pydantic v2 abschließen: |
|||
|
|||
{* ../../docs_src/pydantic_v1_in_v2/tutorial004_an_py310.py hl[4,18] *} |
|||
|
|||
### In Schritten migrieren { #migrate-in-steps } |
|||
|
|||
/// tip | Tipp |
|||
|
|||
Probieren Sie zuerst `bump-pydantic` aus. Wenn Ihre Tests erfolgreich sind und das funktioniert, sind Sie mit einem einzigen Befehl fertig. ✨ |
|||
|
|||
/// |
|||
|
|||
Wenn `bump-pydantic` für Ihren Anwendungsfall nicht funktioniert, können Sie die Unterstützung für Pydantic‑v1‑ und Pydantic‑v2‑Modelle in derselben App nutzen, um die Migration zu Pydantic v2 schrittweise durchzuführen. |
|||
|
|||
Sie könnten zuerst Pydantic auf die neueste Version 2 aktualisieren und die Importe so ändern, dass für all Ihre Modelle `pydantic.v1` verwendet wird. |
|||
|
|||
Anschließend können Sie beginnen, Ihre Modelle gruppenweise von Pydantic v1 auf v2 zu migrieren – in kleinen, schrittweisen Etappen. 🚶 |
|||
|
After Width: | Height: | Size: 6.3 KiB |
|
After Width: | Height: | Size: 10 KiB |
@ -1,7 +1,9 @@ |
|||
/// warning |
|||
|
|||
The current page still doesn't have a translation for this language. |
|||
This page hasn’t been translated into your language yet. 🌍 |
|||
|
|||
But you can help translating it: [Contributing](https://fastapi.tiangolo.com/contributing/){.internal-link target=_blank}. |
|||
We’re currently switching to an automated translation system 🤖, which will help keep all translations complete and up to date. |
|||
|
|||
Learn more: [Contributing – Translations](https://fastapi.tiangolo.com/contributing/#translations){.internal-link target=_blank} |
|||
|
|||
/// |
|||
|
|||
@ -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 |
|||
@ -7,6 +7,8 @@ name = "fastapi" |
|||
dynamic = ["version"] |
|||
description = "FastAPI framework, high performance, easy to learn, fast to code, ready for production" |
|||
readme = "README.md" |
|||
license = "MIT" |
|||
license-files = ["LICENSE"] |
|||
requires-python = ">=3.8" |
|||
authors = [ |
|||
{ name = "Sebastián Ramírez", email = "[email protected]" }, |
|||
@ -31,7 +33,6 @@ classifiers = [ |
|||
"Framework :: Pydantic :: 1", |
|||
"Framework :: Pydantic :: 2", |
|||
"Intended Audience :: Developers", |
|||
"License :: OSI Approved :: MIT License", |
|||
"Programming Language :: Python :: 3 :: Only", |
|||
"Programming Language :: Python :: 3.8", |
|||
"Programming Language :: Python :: 3.9", |
|||
@ -44,9 +45,10 @@ classifiers = [ |
|||
"Topic :: Internet :: WWW/HTTP", |
|||
] |
|||
dependencies = [ |
|||
"starlette>=0.40.0,<0.49.0", |
|||
"starlette>=0.40.0,<0.50.0", |
|||
"pydantic>=1.7.4,!=1.8,!=1.8.1,!=2.0.0,!=2.0.1,!=2.1.0,<3.0.0", |
|||
"typing-extensions>=4.8.0", |
|||
"annotated-doc>=0.0.2", |
|||
] |
|||
|
|||
[project.urls] |
|||
|
|||
@ -0,0 +1,78 @@ |
|||
from typing import Union |
|||
|
|||
from fastapi import FastAPI, HTTPException, Security |
|||
from fastapi.security import ( |
|||
OAuth2PasswordBearer, |
|||
SecurityScopes, |
|||
) |
|||
from fastapi.testclient import TestClient |
|||
from typing_extensions import Annotated |
|||
|
|||
app = FastAPI() |
|||
|
|||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") |
|||
|
|||
|
|||
def process_auth( |
|||
credentials: Annotated[Union[str, None], Security(oauth2_scheme)], |
|||
security_scopes: SecurityScopes, |
|||
): |
|||
# This is an incorrect way of using it, this is not checking if the scopes are |
|||
# provided by the token, only if the endpoint is requesting them, but the test |
|||
# here is just to check if FastAPI is indeed registering and passing the scopes |
|||
# correctly when using Security with parameterless dependencies. |
|||
if "a" not in security_scopes.scopes or "b" not in security_scopes.scopes: |
|||
raise HTTPException(detail="a or b not in scopes", status_code=401) |
|||
return {"token": credentials, "scopes": security_scopes.scopes} |
|||
|
|||
|
|||
@app.get("/get-credentials") |
|||
def get_credentials( |
|||
credentials: Annotated[dict, Security(process_auth, scopes=["a", "b"])], |
|||
): |
|||
return credentials |
|||
|
|||
|
|||
@app.get( |
|||
"/parameterless-with-scopes", |
|||
dependencies=[Security(process_auth, scopes=["a", "b"])], |
|||
) |
|||
def get_parameterless_with_scopes(): |
|||
return {"status": "ok"} |
|||
|
|||
|
|||
@app.get( |
|||
"/parameterless-without-scopes", |
|||
dependencies=[Security(process_auth)], |
|||
) |
|||
def get_parameterless_without_scopes(): |
|||
return {"status": "ok"} |
|||
|
|||
|
|||
client = TestClient(app) |
|||
|
|||
|
|||
def test_get_credentials(): |
|||
response = client.get("/get-credentials", headers={"authorization": "Bearer token"}) |
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == {"token": "token", "scopes": ["a", "b"]} |
|||
|
|||
|
|||
def test_parameterless_with_scopes(): |
|||
response = client.get( |
|||
"/parameterless-with-scopes", headers={"authorization": "Bearer token"} |
|||
) |
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == {"status": "ok"} |
|||
|
|||
|
|||
def test_parameterless_without_scopes(): |
|||
response = client.get( |
|||
"/parameterless-without-scopes", headers={"authorization": "Bearer token"} |
|||
) |
|||
assert response.status_code == 401, response.text |
|||
assert response.json() == {"detail": "a or b not in scopes"} |
|||
|
|||
|
|||
def test_call_get_parameterless_without_scopes_for_coverage(): |
|||
assert get_parameterless_without_scopes() == {"status": "ok"} |
|||
@ -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,203 @@ |
|||
# Test with parts from, and to verify the report in: |
|||
# https://github.com/fastapi/fastapi/discussions/14177 |
|||
# Made an issue in: |
|||
# https://github.com/fastapi/fastapi/issues/14247 |
|||
from enum import Enum |
|||
from typing import List |
|||
|
|||
from fastapi import FastAPI |
|||
from fastapi.testclient import TestClient |
|||
from inline_snapshot import snapshot |
|||
from pydantic import BaseModel, Field |
|||
|
|||
from tests.utils import pydantic_snapshot |
|||
|
|||
|
|||
class MessageEventType(str, Enum): |
|||
alpha = "alpha" |
|||
beta = "beta" |
|||
|
|||
|
|||
class MessageEvent(BaseModel): |
|||
event_type: MessageEventType = Field(default=MessageEventType.alpha) |
|||
output: str |
|||
|
|||
|
|||
class MessageOutput(BaseModel): |
|||
body: str = "" |
|||
events: List[MessageEvent] = [] |
|||
|
|||
|
|||
class Message(BaseModel): |
|||
input: str |
|||
output: MessageOutput |
|||
|
|||
|
|||
app = FastAPI(title="Minimal FastAPI App", version="1.0.0") |
|||
|
|||
|
|||
@app.post("/messages", response_model=Message) |
|||
async def create_message(input_message: str) -> Message: |
|||
return Message( |
|||
input=input_message, |
|||
output=MessageOutput(body=f"Processed: {input_message}"), |
|||
) |
|||
|
|||
|
|||
client = TestClient(app) |
|||
|
|||
|
|||
def test_create_message(): |
|||
response = client.post("/messages", params={"input_message": "Hello"}) |
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == { |
|||
"input": "Hello", |
|||
"output": {"body": "Processed: Hello", "events": []}, |
|||
} |
|||
|
|||
|
|||
def test_openapi_schema(): |
|||
response = client.get("/openapi.json") |
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == snapshot( |
|||
{ |
|||
"openapi": "3.1.0", |
|||
"info": {"title": "Minimal FastAPI App", "version": "1.0.0"}, |
|||
"paths": { |
|||
"/messages": { |
|||
"post": { |
|||
"summary": "Create Message", |
|||
"operationId": "create_message_messages_post", |
|||
"parameters": [ |
|||
{ |
|||
"name": "input_message", |
|||
"in": "query", |
|||
"required": True, |
|||
"schema": {"type": "string", "title": "Input Message"}, |
|||
} |
|||
], |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/Message" |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
"422": { |
|||
"description": "Validation Error", |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/HTTPValidationError" |
|||
} |
|||
} |
|||
}, |
|||
}, |
|||
}, |
|||
} |
|||
} |
|||
}, |
|||
"components": { |
|||
"schemas": { |
|||
"HTTPValidationError": { |
|||
"properties": { |
|||
"detail": { |
|||
"items": { |
|||
"$ref": "#/components/schemas/ValidationError" |
|||
}, |
|||
"type": "array", |
|||
"title": "Detail", |
|||
} |
|||
}, |
|||
"type": "object", |
|||
"title": "HTTPValidationError", |
|||
}, |
|||
"Message": { |
|||
"properties": { |
|||
"input": {"type": "string", "title": "Input"}, |
|||
"output": {"$ref": "#/components/schemas/MessageOutput"}, |
|||
}, |
|||
"type": "object", |
|||
"required": ["input", "output"], |
|||
"title": "Message", |
|||
}, |
|||
"MessageEvent": { |
|||
"properties": { |
|||
"event_type": pydantic_snapshot( |
|||
v2=snapshot( |
|||
{ |
|||
"$ref": "#/components/schemas/MessageEventType", |
|||
"default": "alpha", |
|||
} |
|||
), |
|||
v1=snapshot( |
|||
{ |
|||
"allOf": [ |
|||
{ |
|||
"$ref": "#/components/schemas/MessageEventType" |
|||
} |
|||
], |
|||
"default": "alpha", |
|||
} |
|||
), |
|||
), |
|||
"output": {"type": "string", "title": "Output"}, |
|||
}, |
|||
"type": "object", |
|||
"required": ["output"], |
|||
"title": "MessageEvent", |
|||
}, |
|||
"MessageEventType": pydantic_snapshot( |
|||
v2=snapshot( |
|||
{ |
|||
"type": "string", |
|||
"enum": ["alpha", "beta"], |
|||
"title": "MessageEventType", |
|||
} |
|||
), |
|||
v1=snapshot( |
|||
{ |
|||
"type": "string", |
|||
"enum": ["alpha", "beta"], |
|||
"title": "MessageEventType", |
|||
"description": "An enumeration.", |
|||
} |
|||
), |
|||
), |
|||
"MessageOutput": { |
|||
"properties": { |
|||
"body": {"type": "string", "title": "Body", "default": ""}, |
|||
"events": { |
|||
"items": {"$ref": "#/components/schemas/MessageEvent"}, |
|||
"type": "array", |
|||
"title": "Events", |
|||
"default": [], |
|||
}, |
|||
}, |
|||
"type": "object", |
|||
"title": "MessageOutput", |
|||
}, |
|||
"ValidationError": { |
|||
"properties": { |
|||
"loc": { |
|||
"items": { |
|||
"anyOf": [{"type": "string"}, {"type": "integer"}] |
|||
}, |
|||
"type": "array", |
|||
"title": "Location", |
|||
}, |
|||
"msg": {"type": "string", "title": "Message"}, |
|||
"type": {"type": "string", "title": "Error Type"}, |
|||
}, |
|||
"type": "object", |
|||
"required": ["loc", "msg", "type"], |
|||
"title": "ValidationError", |
|||
}, |
|||
} |
|||
}, |
|||
} |
|||
) |
|||
@ -0,0 +1,60 @@ |
|||
# Test security scheme at the top level, including OpenAPI |
|||
# Ref: https://github.com/fastapi/fastapi/discussions/14263 |
|||
# Ref: https://github.com/fastapi/fastapi/issues/14271 |
|||
from fastapi import Depends, FastAPI |
|||
from fastapi.security import HTTPBearer |
|||
from fastapi.testclient import TestClient |
|||
from inline_snapshot import snapshot |
|||
|
|||
app = FastAPI() |
|||
|
|||
bearer_scheme = HTTPBearer() |
|||
|
|||
|
|||
@app.get("/", dependencies=[Depends(bearer_scheme)]) |
|||
async def get_root(): |
|||
return {"message": "Hello, World!"} |
|||
|
|||
|
|||
client = TestClient(app) |
|||
|
|||
|
|||
def test_get_root(): |
|||
response = client.get("/", headers={"Authorization": "Bearer token"}) |
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == {"message": "Hello, World!"} |
|||
|
|||
|
|||
def test_get_root_no_token(): |
|||
response = client.get("/") |
|||
assert response.status_code == 403, response.text |
|||
assert response.json() == {"detail": "Not authenticated"} |
|||
|
|||
|
|||
def test_openapi_schema(): |
|||
response = client.get("/openapi.json") |
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == snapshot( |
|||
{ |
|||
"openapi": "3.1.0", |
|||
"info": {"title": "FastAPI", "version": "0.1.0"}, |
|||
"paths": { |
|||
"/": { |
|||
"get": { |
|||
"summary": "Get Root", |
|||
"operationId": "get_root__get", |
|||
"responses": { |
|||
"200": { |
|||
"description": "Successful Response", |
|||
"content": {"application/json": {"schema": {}}}, |
|||
} |
|||
}, |
|||
"security": [{"HTTPBearer": []}], |
|||
} |
|||
} |
|||
}, |
|||
"components": { |
|||
"securitySchemes": {"HTTPBearer": {"type": "http", "scheme": "bearer"}} |
|||
}, |
|||
} |
|||
) |
|||
@ -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