|
|
|
@ -3,92 +3,173 @@ from typing import Any, Dict, Union |
|
|
|
from fastapi import FastAPI |
|
|
|
from fastapi.testclient import TestClient |
|
|
|
from inline_snapshot import snapshot |
|
|
|
from pydantic import BaseModel, Field |
|
|
|
from pydantic import BaseModel, Field, Tag |
|
|
|
from typing_extensions import Annotated, Literal |
|
|
|
|
|
|
|
from .utils import needs_pydanticv1, needs_pydanticv2 |
|
|
|
from .utils import needs_pydanticv2 |
|
|
|
|
|
|
|
app = FastAPI() |
|
|
|
|
|
|
|
@needs_pydanticv2 |
|
|
|
def test_discriminator_pydantic_v2() -> None: |
|
|
|
app = FastAPI() |
|
|
|
|
|
|
|
class FirstItem(BaseModel): |
|
|
|
value: Literal["first"] |
|
|
|
price: int |
|
|
|
|
|
|
|
|
|
|
|
class OtherItem(BaseModel): |
|
|
|
value: Literal["other"] |
|
|
|
price: float |
|
|
|
|
|
|
|
|
|
|
|
Item = Annotated[ |
|
|
|
Union[FirstItem, OtherItem], |
|
|
|
Field(discriminator="value"), |
|
|
|
] |
|
|
|
|
|
|
|
class FirstItem(BaseModel): |
|
|
|
value: Literal["first"] |
|
|
|
price: int |
|
|
|
|
|
|
|
@app.post("/items/") |
|
|
|
def save_union_body_discriminator( |
|
|
|
item: Item, q: Annotated[str, Field(..., description="Query string")] |
|
|
|
) -> Dict[str, Any]: |
|
|
|
return {"item": item} |
|
|
|
class OtherItem(BaseModel): |
|
|
|
value: Literal["other"] |
|
|
|
price: float |
|
|
|
|
|
|
|
Item = Annotated[ |
|
|
|
Union[Annotated[FirstItem, Tag("first")], Annotated[OtherItem, Tag("other")]], |
|
|
|
Field(discriminator="value"), |
|
|
|
] |
|
|
|
|
|
|
|
client = TestClient(app) |
|
|
|
@app.post("/items/") |
|
|
|
def save_union_body_discriminator( |
|
|
|
item: Item, q: Annotated[str, Field(description="Query string")] |
|
|
|
) -> Dict[str, Any]: |
|
|
|
return {"item": item} |
|
|
|
|
|
|
|
client = TestClient(app) |
|
|
|
response = client.post("/items/?q=first", json={"value": "first", "price": 100}) |
|
|
|
assert response.status_code == 200, response.text |
|
|
|
assert response.json() == {"item": {"value": "first", "price": 100}} |
|
|
|
|
|
|
|
@needs_pydanticv1 |
|
|
|
def test_openapi_schema_pydantic_v1() -> None: |
|
|
|
openapi = app.openapi() |
|
|
|
response = client.post("/items/?q=other", json={"value": "other", "price": 100.5}) |
|
|
|
assert response.status_code == 200, response.text |
|
|
|
assert response.json() == {"item": {"value": "other", "price": 100.5}} |
|
|
|
|
|
|
|
assert openapi["paths"]["/items/"]["post"]["requestBody"]["content"] == snapshot( |
|
|
|
response = client.get("/openapi.json") |
|
|
|
assert response.status_code == 200, response.text |
|
|
|
assert response.json() == snapshot( |
|
|
|
{ |
|
|
|
"application/json": { |
|
|
|
"schema": { |
|
|
|
"anyOf": [ |
|
|
|
{"$ref": "#/components/schemas/FirstItem"}, |
|
|
|
{"$ref": "#/components/schemas/OtherItem"}, |
|
|
|
], |
|
|
|
"title": "Item", |
|
|
|
"openapi": "3.1.0", |
|
|
|
"info": {"title": "FastAPI", "version": "0.1.0"}, |
|
|
|
"paths": { |
|
|
|
"/items/": { |
|
|
|
"post": { |
|
|
|
"summary": "Save Union Body Discriminator", |
|
|
|
"operationId": "save_union_body_discriminator_items__post", |
|
|
|
"parameters": [ |
|
|
|
{ |
|
|
|
"name": "q", |
|
|
|
"in": "query", |
|
|
|
"required": True, |
|
|
|
"schema": { |
|
|
|
"type": "string", |
|
|
|
"description": "Query string", |
|
|
|
"title": "Q", |
|
|
|
}, |
|
|
|
} |
|
|
|
], |
|
|
|
"requestBody": { |
|
|
|
"required": True, |
|
|
|
"content": { |
|
|
|
"application/json": { |
|
|
|
"schema": { |
|
|
|
"oneOf": [ |
|
|
|
{"$ref": "#/components/schemas/FirstItem"}, |
|
|
|
{"$ref": "#/components/schemas/OtherItem"}, |
|
|
|
], |
|
|
|
"discriminator": { |
|
|
|
"propertyName": "value", |
|
|
|
"mapping": { |
|
|
|
"first": "#/components/schemas/FirstItem", |
|
|
|
"other": "#/components/schemas/OtherItem", |
|
|
|
}, |
|
|
|
}, |
|
|
|
"title": "Item", |
|
|
|
} |
|
|
|
} |
|
|
|
}, |
|
|
|
}, |
|
|
|
"responses": { |
|
|
|
"200": { |
|
|
|
"description": "Successful Response", |
|
|
|
"content": { |
|
|
|
"application/json": { |
|
|
|
"schema": { |
|
|
|
"type": "object", |
|
|
|
"additionalProperties": True, |
|
|
|
"title": "Response Save Union Body Discriminator Items Post", |
|
|
|
} |
|
|
|
} |
|
|
|
}, |
|
|
|
}, |
|
|
|
"422": { |
|
|
|
"description": "Validation Error", |
|
|
|
"content": { |
|
|
|
"application/json": { |
|
|
|
"schema": { |
|
|
|
"$ref": "#/components/schemas/HTTPValidationError" |
|
|
|
} |
|
|
|
} |
|
|
|
}, |
|
|
|
}, |
|
|
|
}, |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
) |
|
|
|
|
|
|
|
|
|
|
|
@needs_pydanticv2 |
|
|
|
def test_openapi_schema() -> None: |
|
|
|
openapi = app.openapi() |
|
|
|
|
|
|
|
assert openapi["paths"]["/items/"]["post"]["requestBody"]["content"] == snapshot( |
|
|
|
{ |
|
|
|
"application/json": { |
|
|
|
"schema": { |
|
|
|
"oneOf": [ |
|
|
|
{"$ref": "#/components/schemas/FirstItem"}, |
|
|
|
{"$ref": "#/components/schemas/OtherItem"}, |
|
|
|
], |
|
|
|
"discriminator": { |
|
|
|
"propertyName": "value", |
|
|
|
"mapping": { |
|
|
|
"first": "#/components/schemas/FirstItem", |
|
|
|
"other": "#/components/schemas/OtherItem", |
|
|
|
}, |
|
|
|
"components": { |
|
|
|
"schemas": { |
|
|
|
"FirstItem": { |
|
|
|
"properties": { |
|
|
|
"value": { |
|
|
|
"type": "string", |
|
|
|
"const": "first", |
|
|
|
"title": "Value", |
|
|
|
}, |
|
|
|
"price": {"type": "integer", "title": "Price"}, |
|
|
|
}, |
|
|
|
"type": "object", |
|
|
|
"required": ["value", "price"], |
|
|
|
"title": "FirstItem", |
|
|
|
}, |
|
|
|
"HTTPValidationError": { |
|
|
|
"properties": { |
|
|
|
"detail": { |
|
|
|
"items": { |
|
|
|
"$ref": "#/components/schemas/ValidationError" |
|
|
|
}, |
|
|
|
"type": "array", |
|
|
|
"title": "Detail", |
|
|
|
} |
|
|
|
}, |
|
|
|
"type": "object", |
|
|
|
"title": "HTTPValidationError", |
|
|
|
}, |
|
|
|
"OtherItem": { |
|
|
|
"properties": { |
|
|
|
"value": { |
|
|
|
"type": "string", |
|
|
|
"const": "other", |
|
|
|
"title": "Value", |
|
|
|
}, |
|
|
|
"price": {"type": "number", "title": "Price"}, |
|
|
|
}, |
|
|
|
"type": "object", |
|
|
|
"required": ["value", "price"], |
|
|
|
"title": "OtherItem", |
|
|
|
}, |
|
|
|
"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", |
|
|
|
}, |
|
|
|
"title": "Item", |
|
|
|
} |
|
|
|
} |
|
|
|
}, |
|
|
|
} |
|
|
|
) |
|
|
|
|
|
|
|
|
|
|
|
def test_post_item() -> None: |
|
|
|
response = client.post("/items/?q=first", json={"value": "first", "price": 100}) |
|
|
|
assert response.status_code == 200, response.text |
|
|
|
assert response.json() == {"item": {"value": "first", "price": 100}} |
|
|
|
|
|
|
|
|
|
|
|
def test_post_other_item() -> None: |
|
|
|
response = client.post("/items/?q=other", json={"value": "other", "price": 100.5}) |
|
|
|
assert response.status_code == 200, response.text |
|
|
|
assert response.json() == {"item": {"value": "other", "price": 100.5}} |
|
|
|
|