mirror
/
tiangolo.fastapi
mirror of https://github.com/tiangolo/fastapi
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

🔊 Add deprecation warnings when using `pydantic.v1` (#14583)

pull/14586/head
Sebastián Ramírez 6 months ago
committed by GitHub
parent
6513d4daa1
commit
6e42bcd8ce
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
20 changed files with 802 additions and 602 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 8
      fastapi/dependencies/utils.py
  2. 16
      fastapi/routing.py
  3. 188
      tests/benchmarks/test_general_performance.py
  4. 187
      tests/test_compat_params_v1.py
  5. 10
      tests/test_datetime_custom_encoder.py
  6. 20
      tests/test_filter_pydantic_sub_model/app_pv1.py
  7. 30
      tests/test_get_model_definitions_formfeed_escape.py
  8. 98
      tests/test_pydantic_v1_deprecation_warnings.py
  9. 48
      tests/test_pydantic_v1_v2_01.py
  10. 75
      tests/test_pydantic_v1_v2_list.py
  11. 315
      tests/test_pydantic_v1_v2_mixed.py
  12. 217
      tests/test_pydantic_v1_v2_multifile/main.py
  13. 123
      tests/test_pydantic_v1_v2_noneable.py
  14. 12
      tests/test_read_with_orm_mode.py
  15. 12
      tests/test_response_model_as_return_annotation.py
  16. 9
      tests/test_tutorial/test_pydantic_v1_in_v2/test_tutorial002.py
  17. 9
      tests/test_tutorial/test_pydantic_v1_in_v2/test_tutorial003.py
  18. 9
      tests/test_tutorial/test_pydantic_v1_in_v2/test_tutorial004.py
  19. 9
      tests/test_tutorial/test_request_form_models/test_tutorial002_pv1.py
  20. 9
      tests/test_tutorial/test_schema_extra_example/test_tutorial001_pv1.py

8
fastapi/dependencies/utils.py
View File

@ -1,6 +1,7 @@
import dataclasses import dataclasses
import inspect import inspect
import sys import sys
import warnings
from collections.abc import Coroutine, Mapping, Sequence from collections.abc import Coroutine, Mapping, Sequence
from contextlib import AsyncExitStack, contextmanager from contextlib import AsyncExitStack, contextmanager
from copy import copy, deepcopy from copy import copy, deepcopy
@ -322,6 +323,13 @@ def get_dependant(
) )
continue continue
assert param_details.field is not None assert param_details.field is not None
if isinstance(param_details.field, may_v1.ModelField):
warnings.warn(
"pydantic.v1 is deprecated and will soon stop being supported by FastAPI."
f" Please update the param {param_name}: {param_details.type_annotation!r}.",
category=DeprecationWarning,
stacklevel=5,
)
if isinstance( if isinstance(
param_details.field.field_info, (params.Body, temp_pydantic_v1_params.Body) param_details.field.field_info, (params.Body, temp_pydantic_v1_params.Body)
): ):

16
fastapi/routing.py
View File

@ -2,6 +2,7 @@ import email.message
import functools import functools
import inspect import inspect
import json import json
import warnings
from collections.abc import ( from collections.abc import (
AsyncIterator, AsyncIterator,
Awaitable, Awaitable,
@ -28,6 +29,7 @@ from fastapi._compat import (
_get_model_config, _get_model_config,
_model_dump, _model_dump,
_normalize_errors, _normalize_errors,
annotation_is_pydantic_v1,
lenient_issubclass, lenient_issubclass,
may_v1, may_v1,
) )
@ -634,6 +636,13 @@ class APIRoute(routing.Route):
f"Status code {status_code} must not have a response body" f"Status code {status_code} must not have a response body"
) )
response_name = "Response_" + self.unique_id response_name = "Response_" + self.unique_id
if annotation_is_pydantic_v1(self.response_model):
warnings.warn(
"pydantic.v1 is deprecated and will soon stop being supported by FastAPI."
f" Please update the response model {self.response_model!r}.",
category=DeprecationWarning,
stacklevel=4,
)
self.response_field = create_model_field( self.response_field = create_model_field(
name=response_name, name=response_name,
type_=self.response_model, type_=self.response_model,
@ -667,6 +676,13 @@ class APIRoute(routing.Route):
f"Status code {additional_status_code} must not have a response body" f"Status code {additional_status_code} must not have a response body"
) )
response_name = f"Response_{additional_status_code}_{self.unique_id}" response_name = f"Response_{additional_status_code}_{self.unique_id}"
if annotation_is_pydantic_v1(model):
warnings.warn(
"pydantic.v1 is deprecated and will soon stop being supported by FastAPI."
f" In responses={{}}, please update {model}.",
category=DeprecationWarning,
stacklevel=4,
)
response_field = create_model_field( response_field = create_model_field(
name=response_name, type_=model, mode="serialization" name=response_name, type_=model, mode="serialization"
) )

188
tests/benchmarks/test_general_performance.py
View File

@ -1,5 +1,6 @@
import json import json
import sys import sys
import warnings
from collections.abc import Iterator from collections.abc import Iterator
from typing import Annotated, Any from typing import Annotated, Any
@ -84,96 +85,103 @@ def app(basemodel_class: type[Any]) -> FastAPI:
app = FastAPI() app = FastAPI()
@app.post("/sync/validated", response_model=ItemOut) with warnings.catch_warnings(record=True):
def sync_validated(item: ItemIn, dep: Annotated[int, Depends(dep_b)]): warnings.filterwarnings(
return ItemOut(name=item.name, value=item.value, dep=dep) "ignore",
message=r"pydantic\.v1 is deprecated and will soon stop being supported by FastAPI\..*",
@app.get("/sync/dict-no-response-model") category=DeprecationWarning,
def sync_dict_no_response_model(): )
return {"name": "foo", "value": 123}
@app.post("/sync/validated", response_model=ItemOut)
@app.get("/sync/dict-with-response-model", response_model=ItemOut) def sync_validated(item: ItemIn, dep: Annotated[int, Depends(dep_b)]):
def sync_dict_with_response_model( return ItemOut(name=item.name, value=item.value, dep=dep)
dep: Annotated[int, Depends(dep_b)],
): @app.get("/sync/dict-no-response-model")
return {"name": "foo", "value": 123, "dep": dep} def sync_dict_no_response_model():
return {"name": "foo", "value": 123}
@app.get("/sync/model-no-response-model")
def sync_model_no_response_model(dep: Annotated[int, Depends(dep_b)]): @app.get("/sync/dict-with-response-model", response_model=ItemOut)
return ItemOut(name="foo", value=123, dep=dep) def sync_dict_with_response_model(
dep: Annotated[int, Depends(dep_b)],
@app.get("/sync/model-with-response-model", response_model=ItemOut) ):
def sync_model_with_response_model(dep: Annotated[int, Depends(dep_b)]): return {"name": "foo", "value": 123, "dep": dep}
return ItemOut(name="foo", value=123, dep=dep)
@app.get("/sync/model-no-response-model")
@app.post("/async/validated", response_model=ItemOut) def sync_model_no_response_model(dep: Annotated[int, Depends(dep_b)]):
async def async_validated( return ItemOut(name="foo", value=123, dep=dep)
item: ItemIn,
dep: Annotated[int, Depends(dep_b)], @app.get("/sync/model-with-response-model", response_model=ItemOut)
): def sync_model_with_response_model(dep: Annotated[int, Depends(dep_b)]):
return ItemOut(name=item.name, value=item.value, dep=dep) return ItemOut(name="foo", value=123, dep=dep)
@app.post("/sync/large-receive") @app.post("/async/validated", response_model=ItemOut)
def sync_large_receive(payload: LargeIn): async def async_validated(
return {"received": len(payload.items)} item: ItemIn,
dep: Annotated[int, Depends(dep_b)],
@app.post("/async/large-receive") ):
async def async_large_receive(payload: LargeIn): return ItemOut(name=item.name, value=item.value, dep=dep)
return {"received": len(payload.items)}
@app.post("/sync/large-receive")
@app.get("/sync/large-dict-no-response-model") def sync_large_receive(payload: LargeIn):
def sync_large_dict_no_response_model(): return {"received": len(payload.items)}
return LARGE_PAYLOAD
@app.post("/async/large-receive")
@app.get("/sync/large-dict-with-response-model", response_model=LargeOut) async def async_large_receive(payload: LargeIn):
def sync_large_dict_with_response_model(): return {"received": len(payload.items)}
return LARGE_PAYLOAD
@app.get("/sync/large-dict-no-response-model")
@app.get("/sync/large-model-no-response-model") def sync_large_dict_no_response_model():
def sync_large_model_no_response_model(): return LARGE_PAYLOAD
return LargeOut(items=LARGE_ITEMS, metadata=LARGE_METADATA)
@app.get("/sync/large-dict-with-response-model", response_model=LargeOut)
@app.get("/sync/large-model-with-response-model", response_model=LargeOut) def sync_large_dict_with_response_model():
def sync_large_model_with_response_model(): return LARGE_PAYLOAD
return LargeOut(items=LARGE_ITEMS, metadata=LARGE_METADATA)
@app.get("/sync/large-model-no-response-model")
@app.get("/async/large-dict-no-response-model") def sync_large_model_no_response_model():
async def async_large_dict_no_response_model(): return LargeOut(items=LARGE_ITEMS, metadata=LARGE_METADATA)
return LARGE_PAYLOAD
@app.get("/sync/large-model-with-response-model", response_model=LargeOut)
@app.get("/async/large-dict-with-response-model", response_model=LargeOut) def sync_large_model_with_response_model():
async def async_large_dict_with_response_model(): return LargeOut(items=LARGE_ITEMS, metadata=LARGE_METADATA)
return LARGE_PAYLOAD
@app.get("/async/large-dict-no-response-model")
@app.get("/async/large-model-no-response-model") async def async_large_dict_no_response_model():
async def async_large_model_no_response_model(): return LARGE_PAYLOAD
return LargeOut(items=LARGE_ITEMS, metadata=LARGE_METADATA)
@app.get("/async/large-dict-with-response-model", response_model=LargeOut)
@app.get("/async/large-model-with-response-model", response_model=LargeOut) async def async_large_dict_with_response_model():
async def async_large_model_with_response_model(): return LARGE_PAYLOAD
return LargeOut(items=LARGE_ITEMS, metadata=LARGE_METADATA)
@app.get("/async/large-model-no-response-model")
@app.get("/async/dict-no-response-model") async def async_large_model_no_response_model():
async def async_dict_no_response_model(): return LargeOut(items=LARGE_ITEMS, metadata=LARGE_METADATA)
return {"name": "foo", "value": 123}
@app.get("/async/large-model-with-response-model", response_model=LargeOut)
@app.get("/async/dict-with-response-model", response_model=ItemOut) async def async_large_model_with_response_model():
async def async_dict_with_response_model( return LargeOut(items=LARGE_ITEMS, metadata=LARGE_METADATA)
dep: Annotated[int, Depends(dep_b)],
): @app.get("/async/dict-no-response-model")
return {"name": "foo", "value": 123, "dep": dep} async def async_dict_no_response_model():
return {"name": "foo", "value": 123}
@app.get("/async/model-no-response-model")
async def async_model_no_response_model( @app.get("/async/dict-with-response-model", response_model=ItemOut)
dep: Annotated[int, Depends(dep_b)], async def async_dict_with_response_model(
): dep: Annotated[int, Depends(dep_b)],
return ItemOut(name="foo", value=123, dep=dep) ):
return {"name": "foo", "value": 123, "dep": dep}
@app.get("/async/model-with-response-model", response_model=ItemOut)
async def async_model_with_response_model( @app.get("/async/model-no-response-model")
dep: Annotated[int, Depends(dep_b)], async def async_model_no_response_model(
): dep: Annotated[int, Depends(dep_b)],
return ItemOut(name="foo", value=123, dep=dep) ):
return ItemOut(name="foo", value=123, dep=dep)
@app.get("/async/model-with-response-model", response_model=ItemOut)
async def async_model_with_response_model(
dep: Annotated[int, Depends(dep_b)],
):
return ItemOut(name="foo", value=123, dep=dep)
return app return app

187
tests/test_compat_params_v1.py
View File

@ -1,4 +1,5 @@
import sys import sys
import warnings
from typing import Optional from typing import Optional
import pytest import pytest
@ -33,94 +34,90 @@ class Item(BaseModel):
app = FastAPI() app = FastAPI()
with warnings.catch_warnings(record=True):
@app.get("/items/{item_id}") warnings.simplefilter("always")
def get_item_with_path(
item_id: Annotated[int, Path(title="The ID of the item", ge=1, le=1000)], @app.get("/items/{item_id}")
): def get_item_with_path(
return {"item_id": item_id} item_id: Annotated[int, Path(title="The ID of the item", ge=1, le=1000)],
):
return {"item_id": item_id}
@app.get("/items/")
def get_items_with_query( @app.get("/items/")
q: Annotated[ def get_items_with_query(
Optional[str], Query(min_length=3, max_length=50, pattern="^[a-zA-Z0-9 ]+$") q: Annotated[
] = None, Optional[str],
skip: Annotated[int, Query(ge=0)] = 0, Query(min_length=3, max_length=50, pattern="^[a-zA-Z0-9 ]+$"),
limit: Annotated[int, Query(ge=1, le=100, examples=[5])] = 10, ] = None,
): skip: Annotated[int, Query(ge=0)] = 0,
return {"q": q, "skip": skip, "limit": limit} limit: Annotated[int, Query(ge=1, le=100, examples=[5])] = 10,
):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/users/")
def get_user_with_header( @app.get("/users/")
x_custom: Annotated[Optional[str], Header()] = None, def get_user_with_header(
x_token: Annotated[Optional[str], Header(convert_underscores=True)] = None, x_custom: Annotated[Optional[str], Header()] = None,
): x_token: Annotated[Optional[str], Header(convert_underscores=True)] = None,
return {"x_custom": x_custom, "x_token": x_token} ):
return {"x_custom": x_custom, "x_token": x_token}
@app.get("/cookies/") @app.get("/cookies/")
def get_cookies( def get_cookies(
session_id: Annotated[Optional[str], Cookie()] = None, session_id: Annotated[Optional[str], Cookie()] = None,
tracking_id: Annotated[Optional[str], Cookie(min_length=10)] = None, tracking_id: Annotated[Optional[str], Cookie(min_length=10)] = None,
): ):
return {"session_id": session_id, "tracking_id": tracking_id} return {"session_id": session_id, "tracking_id": tracking_id}
@app.post("/items/")
@app.post("/items/") def create_item(
def create_item( item: Annotated[
item: Annotated[ Item,
Item, Body(
Body(examples=[{"name": "Foo", "price": 35.4, "description": "The Foo item"}]), examples=[{"name": "Foo", "price": 35.4, "description": "The Foo item"}]
], ),
): ],
return {"item": item} ):
return {"item": item}
@app.post("/items-embed/") @app.post("/items-embed/")
def create_item_embed( def create_item_embed(
item: Annotated[Item, Body(embed=True)], item: Annotated[Item, Body(embed=True)],
): ):
return {"item": item} return {"item": item}
@app.put("/items/{item_id}")
@app.put("/items/{item_id}") def update_item(
def update_item( item_id: Annotated[int, Path(ge=1)],
item_id: Annotated[int, Path(ge=1)], item: Annotated[Item, Body()],
item: Annotated[Item, Body()], importance: Annotated[int, Body(gt=0, le=10)],
importance: Annotated[int, Body(gt=0, le=10)], ):
): return {"item": item, "importance": importance}
return {"item": item, "importance": importance}
@app.post("/form-data/")
def submit_form(
@app.post("/form-data/") username: Annotated[str, Form(min_length=3, max_length=50)],
def submit_form( password: Annotated[str, Form(min_length=8)],
username: Annotated[str, Form(min_length=3, max_length=50)], email: Annotated[Optional[str], Form()] = None,
password: Annotated[str, Form(min_length=8)], ):
email: Annotated[Optional[str], Form()] = None, return {"username": username, "password": password, "email": email}
):
return {"username": username, "password": password, "email": email} @app.post("/upload/")
def upload_file(
file: Annotated[bytes, File()],
@app.post("/upload/") description: Annotated[Optional[str], Form()] = None,
def upload_file( ):
file: Annotated[bytes, File()], return {"file_size": len(file), "description": description}
description: Annotated[Optional[str], Form()] = None,
): @app.post("/upload-multiple/")
return {"file_size": len(file), "description": description} def upload_multiple_files(
files: Annotated[list[bytes], File()],
note: Annotated[str, Form()] = "",
@app.post("/upload-multiple/") ):
def upload_multiple_files( return {
files: Annotated[list[bytes], File()], "file_count": len(files),
note: Annotated[str, Form()] = "", "total_size": sum(len(f) for f in files),
): "note": note,
return { }
"file_count": len(files),
"total_size": sum(len(f) for f in files),
"note": note,
}
client = TestClient(app) client = TestClient(app)
@ -211,10 +208,10 @@ def test_header_params_none():
# Cookie parameter tests # Cookie parameter tests
def test_cookie_params(): def test_cookie_params():
with TestClient(app) as client: with TestClient(app) as test_client:
client.cookies.set("session_id", "abc123") test_client.cookies.set("session_id", "abc123")
client.cookies.set("tracking_id", "1234567890abcdef") test_client.cookies.set("tracking_id", "1234567890abcdef")
response = client.get("/cookies/") response = test_client.get("/cookies/")
assert response.status_code == 200 assert response.status_code == 200
assert response.json() == { assert response.json() == {
"session_id": "abc123", "session_id": "abc123",
@ -223,9 +220,9 @@ def test_cookie_params():
def test_cookie_tracking_id_too_short(): def test_cookie_tracking_id_too_short():
with TestClient(app) as client: with TestClient(app) as test_client:
client.cookies.set("tracking_id", "short") test_client.cookies.set("tracking_id", "short")
response = client.get("/cookies/") response = test_client.get("/cookies/")
assert response.status_code == 422 assert response.status_code == 422
assert response.json() == snapshot( assert response.json() == snapshot(
{ {

10
tests/test_datetime_custom_encoder.py
View File

@ -1,3 +1,4 @@
import warnings
from datetime import datetime, timezone from datetime import datetime, timezone
from fastapi import FastAPI from fastapi import FastAPI
@ -48,9 +49,12 @@ def test_pydanticv1():
app = FastAPI() app = FastAPI()
model = ModelWithDatetimeField(dt_field=datetime(2019, 1, 1, 8)) model = ModelWithDatetimeField(dt_field=datetime(2019, 1, 1, 8))
@app.get("/model", response_model=ModelWithDatetimeField) with warnings.catch_warnings(record=True):
def get_model(): warnings.simplefilter("always")
return model
@app.get("/model", response_model=ModelWithDatetimeField)
def get_model():
return model
client = TestClient(app) client = TestClient(app)
with client: with client:

20
tests/test_filter_pydantic_sub_model/app_pv1.py
View File

@ -1,3 +1,4 @@
import warnings
from typing import Optional from typing import Optional
from fastapi import Depends, FastAPI from fastapi import Depends, FastAPI
@ -31,11 +32,14 @@ async def get_model_c() -> ModelC:
return ModelC(username="test-user", password="test-password") return ModelC(username="test-user", password="test-password")
@app.get("/model/{name}", response_model=ModelA) with warnings.catch_warnings(record=True):
async def get_model_a(name: str, model_c=Depends(get_model_c)): warnings.simplefilter("always")
return {
"name": name, @app.get("/model/{name}", response_model=ModelA)
"description": "model-a-desc", async def get_model_a(name: str, model_c=Depends(get_model_c)):
"model_b": model_c, return {
"tags": {"key1": "value1", "key2": "value2"}, "name": name,
} "description": "model-a-desc",
"model_b": model_c,
"tags": {"key1": "value1", "key2": "value2"},
}

30
tests/test_get_model_definitions_formfeed_escape.py
View File

@ -1,3 +1,5 @@
import warnings
import pytest import pytest
from fastapi import FastAPI from fastapi import FastAPI
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
@ -36,12 +38,28 @@ def client_fixture(request: pytest.FixtureRequest) -> TestClient:
app = FastAPI() app = FastAPI()
@app.get("/facilities/{facility_id}") if request.param == "pydantic-v1":
def get_facility(facility_id: str) -> Facility: with warnings.catch_warnings(record=True):
return Facility( warnings.simplefilter("always")
id=facility_id,
address=Address(line_1="123 Main St", city="Anytown", state_province="CA"), @app.get("/facilities/{facility_id}")
) def get_facility(facility_id: str) -> Facility:
return Facility(
id=facility_id,
address=Address(
line_1="123 Main St", city="Anytown", state_province="CA"
),
)
else:
@app.get("/facilities/{facility_id}")
def get_facility(facility_id: str) -> Facility:
return Facility(
id=facility_id,
address=Address(
line_1="123 Main St", city="Anytown", state_province="CA"
),
)
client = TestClient(app) client = TestClient(app)
return client return client

98
tests/test_pydantic_v1_deprecation_warnings.py
View File

@ -0,0 +1,98 @@
import sys
import pytest
from tests.utils import skip_module_if_py_gte_314
if sys.version_info >= (3, 14):
skip_module_if_py_gte_314()
from fastapi import FastAPI
from fastapi._compat.v1 import BaseModel
from fastapi.testclient import TestClient
def test_warns_pydantic_v1_model_in_endpoint_param() -> None:
class ParamModelV1(BaseModel):
name: str
app = FastAPI()
with pytest.warns(
DeprecationWarning,
match=r"pydantic\.v1 is deprecated.*Please update the param data:",
):
@app.post("/param")
def endpoint(data: ParamModelV1):
return data
client = TestClient(app)
response = client.post("/param", json={"name": "test"})
assert response.status_code == 200, response.text
assert response.json() == {"name": "test"}
def test_warns_pydantic_v1_model_in_return_type() -> None:
class ReturnModelV1(BaseModel):
name: str
app = FastAPI()
with pytest.warns(
DeprecationWarning,
match=r"pydantic\.v1 is deprecated.*Please update the response model",
):
@app.get("/return")
def endpoint() -> ReturnModelV1:
return ReturnModelV1(name="test")
client = TestClient(app)
response = client.get("/return")
assert response.status_code == 200, response.text
assert response.json() == {"name": "test"}
def test_warns_pydantic_v1_model_in_response_model() -> None:
class ResponseModelV1(BaseModel):
name: str
app = FastAPI()
with pytest.warns(
DeprecationWarning,
match=r"pydantic\.v1 is deprecated.*Please update the response model",
):
@app.get("/response-model", response_model=ResponseModelV1)
def endpoint():
return {"name": "test"}
client = TestClient(app)
response = client.get("/response-model")
assert response.status_code == 200, response.text
assert response.json() == {"name": "test"}
def test_warns_pydantic_v1_model_in_additional_responses_model() -> None:
class ErrorModelV1(BaseModel):
detail: str
app = FastAPI()
with pytest.warns(
DeprecationWarning,
match=r"pydantic\.v1 is deprecated.*In responses=\{\}, please update",
):
@app.get(
"/responses", response_model=None, responses={400: {"model": ErrorModelV1}}
)
def endpoint():
return {"ok": True}
client = TestClient(app)
response = client.get("/responses")
assert response.status_code == 200, response.text
assert response.json() == {"ok": True}

48
tests/test_pydantic_v1_v2_01.py
View File

@ -1,4 +1,5 @@
import sys import sys
import warnings
from typing import Any, Union from typing import Any, Union
from tests.utils import skip_module_if_py_gte_314 from tests.utils import skip_module_if_py_gte_314
@ -26,30 +27,29 @@ class Item(BaseModel):
app = FastAPI() app = FastAPI()
with warnings.catch_warnings(record=True):
@app.post("/simple-model") warnings.simplefilter("always")
def handle_simple_model(data: SubItem) -> SubItem:
return data @app.post("/simple-model")
def handle_simple_model(data: SubItem) -> SubItem:
return data
@app.post("/simple-model-filter", response_model=SubItem)
def handle_simple_model_filter(data: SubItem) -> Any: @app.post("/simple-model-filter", response_model=SubItem)
extended_data = data.dict() def handle_simple_model_filter(data: SubItem) -> Any:
extended_data.update({"secret_price": 42}) extended_data = data.dict()
return extended_data extended_data.update({"secret_price": 42})
return extended_data
@app.post("/item") @app.post("/item")
def handle_item(data: Item) -> Item: def handle_item(data: Item) -> Item:
return data return data
@app.post("/item-filter", response_model=Item)
@app.post("/item-filter", response_model=Item) def handle_item_filter(data: Item) -> Any:
def handle_item_filter(data: Item) -> Any: extended_data = data.dict()
extended_data = data.dict() extended_data.update({"secret_data": "classified", "internal_id": 12345})
extended_data.update({"secret_data": "classified", "internal_id": 12345}) extended_data["sub"].update({"internal_id": 67890})
extended_data["sub"].update({"internal_id": 67890}) return extended_data
return extended_data
client = TestClient(app) client = TestClient(app)

75
tests/test_pydantic_v1_v2_list.py
View File

@ -1,4 +1,5 @@
import sys import sys
import warnings
from typing import Any, Union from typing import Any, Union
from tests.utils import skip_module_if_py_gte_314 from tests.utils import skip_module_if_py_gte_314
@ -27,49 +28,47 @@ class Item(BaseModel):
app = FastAPI() app = FastAPI()
@app.post("/item") with warnings.catch_warnings(record=True):
def handle_item(data: Item) -> list[Item]: warnings.simplefilter("always")
return [data, data]
@app.post("/item")
def handle_item(data: Item) -> list[Item]:
return [data, data]
@app.post("/item-filter", response_model=list[Item]) @app.post("/item-filter", response_model=list[Item])
def handle_item_filter(data: Item) -> Any: def handle_item_filter(data: Item) -> Any:
extended_data = data.dict() extended_data = data.dict()
extended_data.update({"secret_data": "classified", "internal_id": 12345})
extended_data["sub"].update({"internal_id": 67890})
return [extended_data, extended_data]
@app.post("/item-list")
def handle_item_list(data: list[Item]) -> Item:
if data:
return data[0]
return Item(title="", size=0, sub=SubItem(name=""))
@app.post("/item-list-filter", response_model=Item)
def handle_item_list_filter(data: list[Item]) -> Any:
if data:
extended_data = data[0].dict()
extended_data.update({"secret_data": "classified", "internal_id": 12345})
extended_data["sub"].update({"internal_id": 67890})
return extended_data
return Item(title="", size=0, sub=SubItem(name=""))
@app.post("/item-list-to-list")
def handle_item_list_to_list(data: list[Item]) -> list[Item]:
return data
@app.post("/item-list-to-list-filter", response_model=list[Item])
def handle_item_list_to_list_filter(data: list[Item]) -> Any:
if data:
extended_data = data[0].dict()
extended_data.update({"secret_data": "classified", "internal_id": 12345}) extended_data.update({"secret_data": "classified", "internal_id": 12345})
extended_data["sub"].update({"internal_id": 67890}) extended_data["sub"].update({"internal_id": 67890})
return [extended_data, extended_data] return [extended_data, extended_data]
return []
@app.post("/item-list")
def handle_item_list(data: list[Item]) -> Item:
if data:
return data[0]
return Item(title="", size=0, sub=SubItem(name=""))
@app.post("/item-list-filter", response_model=Item)
def handle_item_list_filter(data: list[Item]) -> Any:
if data:
extended_data = data[0].dict()
extended_data.update({"secret_data": "classified", "internal_id": 12345})
extended_data["sub"].update({"internal_id": 67890})
return extended_data
return Item(title="", size=0, sub=SubItem(name=""))
@app.post("/item-list-to-list")
def handle_item_list_to_list(data: list[Item]) -> list[Item]:
return data
@app.post("/item-list-to-list-filter", response_model=list[Item])
def handle_item_list_to_list_filter(data: list[Item]) -> Any:
if data:
extended_data = data[0].dict()
extended_data.update({"secret_data": "classified", "internal_id": 12345})
extended_data["sub"].update({"internal_id": 67890})
return [extended_data, extended_data]
return []
client = TestClient(app) client = TestClient(app)

315
tests/test_pydantic_v1_v2_mixed.py
View File

@ -1,4 +1,5 @@
import sys import sys
import warnings
from typing import Any, Union from typing import Any, Union
from tests.utils import skip_module_if_py_gte_314 from tests.utils import skip_module_if_py_gte_314
@ -39,179 +40,181 @@ class NewItem(NewBaseModel):
app = FastAPI() app = FastAPI()
with warnings.catch_warnings(record=True):
warnings.simplefilter("always")
@app.post("/v1-to-v2/item") @app.post("/v1-to-v2/item")
def handle_v1_item_to_v2(data: Item) -> NewItem: def handle_v1_item_to_v2(data: Item) -> NewItem:
return NewItem( return NewItem(
new_title=data.title, new_title=data.title,
new_size=data.size, new_size=data.size,
new_description=data.description, new_description=data.description,
new_sub=NewSubItem(new_sub_name=data.sub.name), new_sub=NewSubItem(new_sub_name=data.sub.name),
new_multi=[NewSubItem(new_sub_name=s.name) for s in data.multi], new_multi=[NewSubItem(new_sub_name=s.name) for s in data.multi],
) )
@app.post("/v1-to-v2/item-filter", response_model=NewItem)
def handle_v1_item_to_v2_filter(data: Item) -> Any:
result = {
"new_title": data.title,
"new_size": data.size,
"new_description": data.description,
"new_sub": {"new_sub_name": data.sub.name, "new_sub_secret": "sub_hidden"},
"new_multi": [
{"new_sub_name": s.name, "new_sub_secret": "sub_hidden"} for s in data.multi
],
"secret": "hidden_v1_to_v2",
}
return result
@app.post("/v2-to-v1/item")
def handle_v2_item_to_v1(data: NewItem) -> Item:
return Item(
title=data.new_title,
size=data.new_size,
description=data.new_description,
sub=SubItem(name=data.new_sub.new_sub_name),
multi=[SubItem(name=s.new_sub_name) for s in data.new_multi],
)
@app.post("/v2-to-v1/item-filter", response_model=Item)
def handle_v2_item_to_v1_filter(data: NewItem) -> Any:
result = {
"title": data.new_title,
"size": data.new_size,
"description": data.new_description,
"sub": {"name": data.new_sub.new_sub_name, "sub_secret": "sub_hidden"},
"multi": [
{"name": s.new_sub_name, "sub_secret": "sub_hidden"} for s in data.new_multi
],
"secret": "hidden_v2_to_v1",
}
return result
@app.post("/v1-to-v2/item-to-list") @app.post("/v1-to-v2/item-filter", response_model=NewItem)
def handle_v1_item_to_v2_list(data: Item) -> list[NewItem]: def handle_v1_item_to_v2_filter(data: Item) -> Any:
converted = NewItem( result = {
new_title=data.title, "new_title": data.title,
new_size=data.size, "new_size": data.size,
new_description=data.description, "new_description": data.description,
new_sub=NewSubItem(new_sub_name=data.sub.name), "new_sub": {
new_multi=[NewSubItem(new_sub_name=s.name) for s in data.multi], "new_sub_name": data.sub.name,
) "new_sub_secret": "sub_hidden",
return [converted, converted] },
"new_multi": [
{"new_sub_name": s.name, "new_sub_secret": "sub_hidden"}
for s in data.multi
],
"secret": "hidden_v1_to_v2",
}
return result
@app.post("/v2-to-v1/item")
def handle_v2_item_to_v1(data: NewItem) -> Item:
return Item(
title=data.new_title,
size=data.new_size,
description=data.new_description,
sub=SubItem(name=data.new_sub.new_sub_name),
multi=[SubItem(name=s.new_sub_name) for s in data.new_multi],
)
@app.post("/v1-to-v2/list-to-list") @app.post("/v2-to-v1/item-filter", response_model=Item)
def handle_v1_list_to_v2_list(data: list[Item]) -> list[NewItem]: def handle_v2_item_to_v1_filter(data: NewItem) -> Any:
result = [] result = {
for item in data: "title": data.new_title,
result.append( "size": data.new_size,
NewItem( "description": data.new_description,
"sub": {"name": data.new_sub.new_sub_name, "sub_secret": "sub_hidden"},
"multi": [
{"name": s.new_sub_name, "sub_secret": "sub_hidden"}
for s in data.new_multi
],
"secret": "hidden_v2_to_v1",
}
return result
@app.post("/v1-to-v2/item-to-list")
def handle_v1_item_to_v2_list(data: Item) -> list[NewItem]:
converted = NewItem(
new_title=data.title,
new_size=data.size,
new_description=data.description,
new_sub=NewSubItem(new_sub_name=data.sub.name),
new_multi=[NewSubItem(new_sub_name=s.name) for s in data.multi],
)
return [converted, converted]
@app.post("/v1-to-v2/list-to-list")
def handle_v1_list_to_v2_list(data: list[Item]) -> list[NewItem]:
result = []
for item in data:
result.append(
NewItem(
new_title=item.title,
new_size=item.size,
new_description=item.description,
new_sub=NewSubItem(new_sub_name=item.sub.name),
new_multi=[NewSubItem(new_sub_name=s.name) for s in item.multi],
)
)
return result
@app.post("/v1-to-v2/list-to-list-filter", response_model=list[NewItem])
def handle_v1_list_to_v2_list_filter(data: list[Item]) -> Any:
result = []
for item in data:
converted = {
"new_title": item.title,
"new_size": item.size,
"new_description": item.description,
"new_sub": {
"new_sub_name": item.sub.name,
"new_sub_secret": "sub_hidden",
},
"new_multi": [
{"new_sub_name": s.name, "new_sub_secret": "sub_hidden"}
for s in item.multi
],
"secret": "hidden_v2_to_v1",
}
result.append(converted)
return result
@app.post("/v1-to-v2/list-to-item")
def handle_v1_list_to_v2_item(data: list[Item]) -> NewItem:
if data:
item = data[0]
return NewItem(
new_title=item.title, new_title=item.title,
new_size=item.size, new_size=item.size,
new_description=item.description, new_description=item.description,
new_sub=NewSubItem(new_sub_name=item.sub.name), new_sub=NewSubItem(new_sub_name=item.sub.name),
new_multi=[NewSubItem(new_sub_name=s.name) for s in item.multi], new_multi=[NewSubItem(new_sub_name=s.name) for s in item.multi],
) )
return NewItem(new_title="", new_size=0, new_sub=NewSubItem(new_sub_name=""))
@app.post("/v2-to-v1/item-to-list")
def handle_v2_item_to_v1_list(data: NewItem) -> list[Item]:
converted = Item(
title=data.new_title,
size=data.new_size,
description=data.new_description,
sub=SubItem(name=data.new_sub.new_sub_name),
multi=[SubItem(name=s.new_sub_name) for s in data.new_multi],
) )
return result return [converted, converted]
@app.post("/v2-to-v1/list-to-list")
@app.post("/v1-to-v2/list-to-list-filter", response_model=list[NewItem]) def handle_v2_list_to_v1_list(data: list[NewItem]) -> list[Item]:
def handle_v1_list_to_v2_list_filter(data: list[Item]) -> Any: result = []
result = [] for item in data:
for item in data: result.append(
converted = { Item(
"new_title": item.title, title=item.new_title,
"new_size": item.size, size=item.new_size,
"new_description": item.description, description=item.new_description,
"new_sub": {"new_sub_name": item.sub.name, "new_sub_secret": "sub_hidden"}, sub=SubItem(name=item.new_sub.new_sub_name),
"new_multi": [ multi=[SubItem(name=s.new_sub_name) for s in item.new_multi],
{"new_sub_name": s.name, "new_sub_secret": "sub_hidden"} )
for s in item.multi )
], return result
"secret": "hidden_v2_to_v1",
} @app.post("/v2-to-v1/list-to-list-filter", response_model=list[Item])
result.append(converted) def handle_v2_list_to_v1_list_filter(data: list[NewItem]) -> Any:
return result result = []
for item in data:
converted = {
@app.post("/v1-to-v2/list-to-item") "title": item.new_title,
def handle_v1_list_to_v2_item(data: list[Item]) -> NewItem: "size": item.new_size,
if data: "description": item.new_description,
item = data[0] "sub": {
return NewItem( "name": item.new_sub.new_sub_name,
new_title=item.title, "sub_secret": "sub_hidden",
new_size=item.size, },
new_description=item.description, "multi": [
new_sub=NewSubItem(new_sub_name=item.sub.name), {"name": s.new_sub_name, "sub_secret": "sub_hidden"}
new_multi=[NewSubItem(new_sub_name=s.name) for s in item.multi], for s in item.new_multi
) ],
return NewItem(new_title="", new_size=0, new_sub=NewSubItem(new_sub_name="")) "secret": "hidden_v2_to_v1",
}
result.append(converted)
@app.post("/v2-to-v1/item-to-list") return result
def handle_v2_item_to_v1_list(data: NewItem) -> list[Item]:
converted = Item( @app.post("/v2-to-v1/list-to-item")
title=data.new_title, def handle_v2_list_to_v1_item(data: list[NewItem]) -> Item:
size=data.new_size, if data:
description=data.new_description, item = data[0]
sub=SubItem(name=data.new_sub.new_sub_name), return Item(
multi=[SubItem(name=s.new_sub_name) for s in data.new_multi],
)
return [converted, converted]
@app.post("/v2-to-v1/list-to-list")
def handle_v2_list_to_v1_list(data: list[NewItem]) -> list[Item]:
result = []
for item in data:
result.append(
Item(
title=item.new_title, title=item.new_title,
size=item.new_size, size=item.new_size,
description=item.new_description, description=item.new_description,
sub=SubItem(name=item.new_sub.new_sub_name), sub=SubItem(name=item.new_sub.new_sub_name),
multi=[SubItem(name=s.new_sub_name) for s in item.new_multi], multi=[SubItem(name=s.new_sub_name) for s in item.new_multi],
) )
) return Item(title="", size=0, sub=SubItem(name=""))
return result
@app.post("/v2-to-v1/list-to-list-filter", response_model=list[Item])
def handle_v2_list_to_v1_list_filter(data: list[NewItem]) -> Any:
result = []
for item in data:
converted = {
"title": item.new_title,
"size": item.new_size,
"description": item.new_description,
"sub": {"name": item.new_sub.new_sub_name, "sub_secret": "sub_hidden"},
"multi": [
{"name": s.new_sub_name, "sub_secret": "sub_hidden"}
for s in item.new_multi
],
"secret": "hidden_v2_to_v1",
}
result.append(converted)
return result
@app.post("/v2-to-v1/list-to-item")
def handle_v2_list_to_v1_item(data: list[NewItem]) -> Item:
if data:
item = data[0]
return Item(
title=item.new_title,
size=item.new_size,
description=item.new_description,
sub=SubItem(name=item.new_sub.new_sub_name),
multi=[SubItem(name=s.new_sub_name) for s in item.new_multi],
)
return Item(title="", size=0, sub=SubItem(name=""))
client = TestClient(app) client = TestClient(app)

217
tests/test_pydantic_v1_v2_multifile/main.py
View File

@ -1,140 +1,137 @@
import warnings
from fastapi import FastAPI from fastapi import FastAPI
from . import modelsv1, modelsv2, modelsv2b from . import modelsv1, modelsv2, modelsv2b
app = FastAPI() app = FastAPI()
with warnings.catch_warnings(record=True):
warnings.simplefilter("always")
@app.post("/v1-to-v2/item") @app.post("/v1-to-v2/item")
def handle_v1_item_to_v2(data: modelsv1.Item) -> modelsv2.Item: def handle_v1_item_to_v2(data: modelsv1.Item) -> modelsv2.Item:
return modelsv2.Item( return modelsv2.Item(
new_title=data.title, new_title=data.title,
new_size=data.size, new_size=data.size,
new_description=data.description, new_description=data.description,
new_sub=modelsv2.SubItem(new_sub_name=data.sub.name), new_sub=modelsv2.SubItem(new_sub_name=data.sub.name),
new_multi=[modelsv2.SubItem(new_sub_name=s.name) for s in data.multi], new_multi=[modelsv2.SubItem(new_sub_name=s.name) for s in data.multi],
) )
@app.post("/v2-to-v1/item")
def handle_v2_item_to_v1(data: modelsv2.Item) -> modelsv1.Item:
return modelsv1.Item(
title=data.new_title,
size=data.new_size,
description=data.new_description,
sub=modelsv1.SubItem(name=data.new_sub.new_sub_name),
multi=[modelsv1.SubItem(name=s.new_sub_name) for s in data.new_multi],
)
@app.post("/v1-to-v2/item-to-list") @app.post("/v2-to-v1/item")
def handle_v1_item_to_v2_list(data: modelsv1.Item) -> list[modelsv2.Item]: def handle_v2_item_to_v1(data: modelsv2.Item) -> modelsv1.Item:
converted = modelsv2.Item( return modelsv1.Item(
new_title=data.title, title=data.new_title,
new_size=data.size, size=data.new_size,
new_description=data.description, description=data.new_description,
new_sub=modelsv2.SubItem(new_sub_name=data.sub.name), sub=modelsv1.SubItem(name=data.new_sub.new_sub_name),
new_multi=[modelsv2.SubItem(new_sub_name=s.name) for s in data.multi], multi=[modelsv1.SubItem(name=s.new_sub_name) for s in data.new_multi],
) )
return [converted, converted]
@app.post("/v1-to-v2/item-to-list")
def handle_v1_item_to_v2_list(data: modelsv1.Item) -> list[modelsv2.Item]:
converted = modelsv2.Item(
new_title=data.title,
new_size=data.size,
new_description=data.description,
new_sub=modelsv2.SubItem(new_sub_name=data.sub.name),
new_multi=[modelsv2.SubItem(new_sub_name=s.name) for s in data.multi],
)
return [converted, converted]
@app.post("/v1-to-v2/list-to-list")
def handle_v1_list_to_v2_list(data: list[modelsv1.Item]) -> list[modelsv2.Item]:
result = []
for item in data:
result.append(
modelsv2.Item(
new_title=item.title,
new_size=item.size,
new_description=item.description,
new_sub=modelsv2.SubItem(new_sub_name=item.sub.name),
new_multi=[
modelsv2.SubItem(new_sub_name=s.name) for s in item.multi
],
)
)
return result
@app.post("/v1-to-v2/list-to-list") @app.post("/v1-to-v2/list-to-item")
def handle_v1_list_to_v2_list(data: list[modelsv1.Item]) -> list[modelsv2.Item]: def handle_v1_list_to_v2_item(data: list[modelsv1.Item]) -> modelsv2.Item:
result = [] if data:
for item in data: item = data[0]
result.append( return modelsv2.Item(
modelsv2.Item(
new_title=item.title, new_title=item.title,
new_size=item.size, new_size=item.size,
new_description=item.description, new_description=item.description,
new_sub=modelsv2.SubItem(new_sub_name=item.sub.name), new_sub=modelsv2.SubItem(new_sub_name=item.sub.name),
new_multi=[modelsv2.SubItem(new_sub_name=s.name) for s in item.multi], new_multi=[modelsv2.SubItem(new_sub_name=s.name) for s in item.multi],
) )
)
return result
@app.post("/v1-to-v2/list-to-item")
def handle_v1_list_to_v2_item(data: list[modelsv1.Item]) -> modelsv2.Item:
if data:
item = data[0]
return modelsv2.Item( return modelsv2.Item(
new_title=item.title, new_title="", new_size=0, new_sub=modelsv2.SubItem(new_sub_name="")
new_size=item.size,
new_description=item.description,
new_sub=modelsv2.SubItem(new_sub_name=item.sub.name),
new_multi=[modelsv2.SubItem(new_sub_name=s.name) for s in item.multi],
) )
return modelsv2.Item(
new_title="", new_size=0, new_sub=modelsv2.SubItem(new_sub_name="")
)
@app.post("/v2-to-v1/item-to-list")
def handle_v2_item_to_v1_list(data: modelsv2.Item) -> list[modelsv1.Item]:
converted = modelsv1.Item(
title=data.new_title,
size=data.new_size,
description=data.new_description,
sub=modelsv1.SubItem(name=data.new_sub.new_sub_name),
multi=[modelsv1.SubItem(name=s.new_sub_name) for s in data.new_multi],
)
return [converted, converted]
@app.post("/v2-to-v1/item-to-list")
def handle_v2_item_to_v1_list(data: modelsv2.Item) -> list[modelsv1.Item]:
converted = modelsv1.Item(
title=data.new_title,
size=data.new_size,
description=data.new_description,
sub=modelsv1.SubItem(name=data.new_sub.new_sub_name),
multi=[modelsv1.SubItem(name=s.new_sub_name) for s in data.new_multi],
)
return [converted, converted]
@app.post("/v2-to-v1/list-to-list")
def handle_v2_list_to_v1_list(data: list[modelsv2.Item]) -> list[modelsv1.Item]:
result = []
for item in data:
result.append(
modelsv1.Item(
title=item.new_title,
size=item.new_size,
description=item.new_description,
sub=modelsv1.SubItem(name=item.new_sub.new_sub_name),
multi=[
modelsv1.SubItem(name=s.new_sub_name) for s in item.new_multi
],
)
)
return result
@app.post("/v2-to-v1/list-to-list") @app.post("/v2-to-v1/list-to-item")
def handle_v2_list_to_v1_list(data: list[modelsv2.Item]) -> list[modelsv1.Item]: def handle_v2_list_to_v1_item(data: list[modelsv2.Item]) -> modelsv1.Item:
result = [] if data:
for item in data: item = data[0]
result.append( return modelsv1.Item(
modelsv1.Item(
title=item.new_title, title=item.new_title,
size=item.new_size, size=item.new_size,
description=item.new_description, description=item.new_description,
sub=modelsv1.SubItem(name=item.new_sub.new_sub_name), sub=modelsv1.SubItem(name=item.new_sub.new_sub_name),
multi=[modelsv1.SubItem(name=s.new_sub_name) for s in item.new_multi], multi=[modelsv1.SubItem(name=s.new_sub_name) for s in item.new_multi],
) )
) return modelsv1.Item(title="", size=0, sub=modelsv1.SubItem(name=""))
return result
@app.post("/v2-to-v1/list-to-item") @app.post("/v2-to-v1/same-name")
def handle_v2_list_to_v1_item(data: list[modelsv2.Item]) -> modelsv1.Item: def handle_v2_same_name_to_v1(
if data: item1: modelsv2.Item, item2: modelsv2b.Item
item = data[0] ) -> modelsv1.Item:
return modelsv1.Item( return modelsv1.Item(
title=item.new_title, title=item1.new_title,
size=item.new_size, size=item2.dup_size,
description=item.new_description, description=item1.new_description,
sub=modelsv1.SubItem(name=item.new_sub.new_sub_name), sub=modelsv1.SubItem(name=item1.new_sub.new_sub_name),
multi=[modelsv1.SubItem(name=s.new_sub_name) for s in item.new_multi], multi=[modelsv1.SubItem(name=s.dup_sub_name) for s in item2.dup_multi],
) )
return modelsv1.Item(title="", size=0, sub=modelsv1.SubItem(name=""))
@app.post("/v2-to-v1/same-name")
def handle_v2_same_name_to_v1(
item1: modelsv2.Item, item2: modelsv2b.Item
) -> modelsv1.Item:
return modelsv1.Item(
title=item1.new_title,
size=item2.dup_size,
description=item1.new_description,
sub=modelsv1.SubItem(name=item1.new_sub.new_sub_name),
multi=[modelsv1.SubItem(name=s.dup_sub_name) for s in item2.dup_multi],
)
@app.post("/v2-to-v1/list-of-items-to-list-of-items") @app.post("/v2-to-v1/list-of-items-to-list-of-items")
def handle_v2_items_in_list_to_v1_item_in_list( def handle_v2_items_in_list_to_v1_item_in_list(
data1: list[modelsv2.ItemInList], data2: list[modelsv2b.ItemInList] data1: list[modelsv2.ItemInList], data2: list[modelsv2b.ItemInList]
) -> list[modelsv1.ItemInList]: ) -> list[modelsv1.ItemInList]:
result = [] item1 = data1[0]
item1 = data1[0] item2 = data2[0]
item2 = data2[0] return [
result = [ modelsv1.ItemInList(name1=item1.name2),
modelsv1.ItemInList(name1=item1.name2), modelsv1.ItemInList(name1=item2.dup_name2),
modelsv1.ItemInList(name1=item2.dup_name2), ]
]
return result

123
tests/test_pydantic_v1_v2_noneable.py
View File

@ -1,4 +1,5 @@
import sys import sys
import warnings
from typing import Any, Union from typing import Any, Union
from tests.utils import skip_module_if_py_gte_314 from tests.utils import skip_module_if_py_gte_314
@ -39,65 +40,69 @@ class NewItem(NewBaseModel):
app = FastAPI() app = FastAPI()
with warnings.catch_warnings(record=True):
@app.post("/v1-to-v2/") warnings.simplefilter("always")
def handle_v1_item_to_v2(data: Item) -> Union[NewItem, None]:
if data.size < 0: @app.post("/v1-to-v2/")
return None def handle_v1_item_to_v2(data: Item) -> Union[NewItem, None]:
return NewItem( if data.size < 0:
new_title=data.title, return None
new_size=data.size, return NewItem(
new_description=data.description, new_title=data.title,
new_sub=NewSubItem(new_sub_name=data.sub.name), new_size=data.size,
new_multi=[NewSubItem(new_sub_name=s.name) for s in data.multi], new_description=data.description,
) new_sub=NewSubItem(new_sub_name=data.sub.name),
new_multi=[NewSubItem(new_sub_name=s.name) for s in data.multi],
)
@app.post("/v1-to-v2/item-filter", response_model=Union[NewItem, None])
def handle_v1_item_to_v2_filter(data: Item) -> Any: @app.post("/v1-to-v2/item-filter", response_model=Union[NewItem, None])
if data.size < 0: def handle_v1_item_to_v2_filter(data: Item) -> Any:
return None if data.size < 0:
result = { return None
"new_title": data.title, result = {
"new_size": data.size, "new_title": data.title,
"new_description": data.description, "new_size": data.size,
"new_sub": {"new_sub_name": data.sub.name, "new_sub_secret": "sub_hidden"}, "new_description": data.description,
"new_multi": [ "new_sub": {
{"new_sub_name": s.name, "new_sub_secret": "sub_hidden"} for s in data.multi "new_sub_name": data.sub.name,
], "new_sub_secret": "sub_hidden",
"secret": "hidden_v1_to_v2", },
} "new_multi": [
return result {"new_sub_name": s.name, "new_sub_secret": "sub_hidden"}
for s in data.multi
],
@app.post("/v2-to-v1/item") "secret": "hidden_v1_to_v2",
def handle_v2_item_to_v1(data: NewItem) -> Union[Item, None]: }
if data.new_size < 0: return result
return None
return Item( @app.post("/v2-to-v1/item")
title=data.new_title, def handle_v2_item_to_v1(data: NewItem) -> Union[Item, None]:
size=data.new_size, if data.new_size < 0:
description=data.new_description, return None
sub=SubItem(name=data.new_sub.new_sub_name), return Item(
multi=[SubItem(name=s.new_sub_name) for s in data.new_multi], title=data.new_title,
) size=data.new_size,
description=data.new_description,
sub=SubItem(name=data.new_sub.new_sub_name),
@app.post("/v2-to-v1/item-filter", response_model=Union[Item, None]) multi=[SubItem(name=s.new_sub_name) for s in data.new_multi],
def handle_v2_item_to_v1_filter(data: NewItem) -> Any: )
if data.new_size < 0:
return None @app.post("/v2-to-v1/item-filter", response_model=Union[Item, None])
result = { def handle_v2_item_to_v1_filter(data: NewItem) -> Any:
"title": data.new_title, if data.new_size < 0:
"size": data.new_size, return None
"description": data.new_description, result = {
"sub": {"name": data.new_sub.new_sub_name, "sub_secret": "sub_hidden"}, "title": data.new_title,
"multi": [ "size": data.new_size,
{"name": s.new_sub_name, "sub_secret": "sub_hidden"} for s in data.new_multi "description": data.new_description,
], "sub": {"name": data.new_sub.new_sub_name, "sub_secret": "sub_hidden"},
"secret": "hidden_v2_to_v1", "multi": [
} {"name": s.new_sub_name, "sub_secret": "sub_hidden"}
return result for s in data.new_multi
],
"secret": "hidden_v2_to_v1",
}
return result
client = TestClient(app) client = TestClient(app)

12
tests/test_read_with_orm_mode.py
View File

@ -1,3 +1,4 @@
import warnings
from typing import Any from typing import Any
from fastapi import FastAPI from fastapi import FastAPI
@ -73,10 +74,13 @@ def test_read_with_orm_mode_pv1() -> None:
app = FastAPI() app = FastAPI()
@app.post("/people/", response_model=PersonRead) with warnings.catch_warnings(record=True):
def create_person(person: PersonCreate) -> Any: warnings.simplefilter("always")
db_person = Person.from_orm(person)
return db_person @app.post("/people/", response_model=PersonRead)
def create_person(person: PersonCreate) -> Any:
db_person = Person.from_orm(person)
return db_person
client = TestClient(app) client = TestClient(app)

12
tests/test_response_model_as_return_annotation.py
View File

@ -1,3 +1,4 @@
import warnings
from typing import Union from typing import Union
import pytest import pytest
@ -521,11 +522,14 @@ def test_invalid_response_model_field_pv1():
class Model(v1.BaseModel): class Model(v1.BaseModel):
foo: str foo: str
with pytest.raises(FastAPIError) as e: with warnings.catch_warnings(record=True):
warnings.simplefilter("always")
@app.get("/") with pytest.raises(FastAPIError) as e:
def read_root() -> Union[Response, Model, None]:
return Response(content="Foo") # pragma: no cover @app.get("/")
def read_root() -> Union[Response, Model, None]:
return Response(content="Foo") # pragma: no cover
assert "valid Pydantic field type" in e.value.args[0] assert "valid Pydantic field type" in e.value.args[0]
assert "parameter response_model=None" in e.value.args[0] assert "parameter response_model=None" in e.value.args[0]

9
tests/test_tutorial/test_pydantic_v1_in_v2/test_tutorial002.py
View File

@ -1,4 +1,5 @@
import sys import sys
import warnings
import pytest import pytest
from inline_snapshot import snapshot from inline_snapshot import snapshot
@ -24,7 +25,13 @@ from ...utils import needs_py310
], ],
) )
def get_client(request: pytest.FixtureRequest): def get_client(request: pytest.FixtureRequest):
mod = importlib.import_module(f"docs_src.pydantic_v1_in_v2.{request.param}") with warnings.catch_warnings(record=True):
warnings.filterwarnings(
"ignore",
message=r"pydantic\.v1 is deprecated and will soon stop being supported by FastAPI\..*",
category=DeprecationWarning,
)
mod = importlib.import_module(f"docs_src.pydantic_v1_in_v2.{request.param}")
c = TestClient(mod.app) c = TestClient(mod.app)
return c return c

9
tests/test_tutorial/test_pydantic_v1_in_v2/test_tutorial003.py
View File

@ -1,4 +1,5 @@
import sys import sys
import warnings
import pytest import pytest
from inline_snapshot import snapshot from inline_snapshot import snapshot
@ -24,7 +25,13 @@ from ...utils import needs_py310
], ],
) )
def get_client(request: pytest.FixtureRequest): def get_client(request: pytest.FixtureRequest):
mod = importlib.import_module(f"docs_src.pydantic_v1_in_v2.{request.param}") with warnings.catch_warnings(record=True):
warnings.filterwarnings(
"ignore",
message=r"pydantic\.v1 is deprecated and will soon stop being supported by FastAPI\..*",
category=DeprecationWarning,
)
mod = importlib.import_module(f"docs_src.pydantic_v1_in_v2.{request.param}")
c = TestClient(mod.app) c = TestClient(mod.app)
return c return c

9
tests/test_tutorial/test_pydantic_v1_in_v2/test_tutorial004.py
View File

@ -1,4 +1,5 @@
import sys import sys
import warnings
import pytest import pytest
from inline_snapshot import snapshot from inline_snapshot import snapshot
@ -24,7 +25,13 @@ from ...utils import needs_py310
], ],
) )
def get_client(request: pytest.FixtureRequest): def get_client(request: pytest.FixtureRequest):
mod = importlib.import_module(f"docs_src.pydantic_v1_in_v2.{request.param}") with warnings.catch_warnings(record=True):
warnings.filterwarnings(
"ignore",
message=r"pydantic\.v1 is deprecated and will soon stop being supported by FastAPI\..*",
category=DeprecationWarning,
)
mod = importlib.import_module(f"docs_src.pydantic_v1_in_v2.{request.param}")
c = TestClient(mod.app) c = TestClient(mod.app)
return c return c

9
tests/test_tutorial/test_request_form_models/test_tutorial002_pv1.py
View File

@ -1,4 +1,5 @@
import importlib import importlib
import warnings
import pytest import pytest
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
@ -14,7 +15,13 @@ from ...utils import needs_pydanticv1
], ],
) )
def get_client(request: pytest.FixtureRequest): def get_client(request: pytest.FixtureRequest):
mod = importlib.import_module(f"docs_src.request_form_models.{request.param}") with warnings.catch_warnings(record=True):
warnings.filterwarnings(
"ignore",
message=r"pydantic\.v1 is deprecated and will soon stop being supported by FastAPI\..*",
category=DeprecationWarning,
)
mod = importlib.import_module(f"docs_src.request_form_models.{request.param}")
client = TestClient(mod.app) client = TestClient(mod.app)
return client return client

9
tests/test_tutorial/test_schema_extra_example/test_tutorial001_pv1.py
View File

@ -1,4 +1,5 @@
import importlib import importlib
import warnings
import pytest import pytest
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
@ -15,7 +16,13 @@ from ...utils import needs_py310, needs_pydanticv1
], ],
) )
def get_client(request: pytest.FixtureRequest): def get_client(request: pytest.FixtureRequest):
mod = importlib.import_module(f"docs_src.schema_extra_example.{request.param}") with warnings.catch_warnings(record=True):
warnings.filterwarnings(
"ignore",
message=r"pydantic\.v1 is deprecated and will soon stop being supported by FastAPI\..*",
category=DeprecationWarning,
)
mod = importlib.import_module(f"docs_src.schema_extra_example.{request.param}")
client = TestClient(mod.app) client = TestClient(mod.app)
return client return client

Write Preview
Loading…
Cancel
Save