This commit is contained in:
Motov Yurii 2026-03-16 09:14:15 +01:00 committed by GitHub
commit c6f9396d2a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 348 additions and 230 deletions

View File

@ -45,14 +45,25 @@ class GenerateJsonSchema(_GenerateJsonSchema):
# TODO: remove when this is merged (or equivalent): https://github.com/pydantic/pydantic/pull/12841
# and dropping support for any version of Pydantic before that one (so, in a very long time)
def bytes_schema(self, schema: CoreSchema) -> JsonSchemaValue:
json_schema = {"type": "string", "contentMediaType": "application/octet-stream"}
bytes_mode = (
self._config.ser_json_bytes
if self.mode == "serialization"
else self._config.val_json_bytes
)
if bytes_mode == "base64":
json_schema["contentEncoding"] = "base64"
is_file_upload = schema.get("metadata", {}).get("fastapi_file_upload", False)
if is_file_upload:
json_schema: JsonSchemaValue = {
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
}
else:
json_schema = {
"type": "string",
"contentMediaType": "application/octet-stream",
}
bytes_mode = (
self._config.ser_json_bytes
if self.mode == "serialization"
else self._config.val_json_bytes
)
if bytes_mode == "base64":
json_schema["contentEncoding"] = "base64"
self.update_with_validations(json_schema, schema, self.ValidationsMapping.bytes)
return json_schema

View File

@ -139,7 +139,11 @@ class UploadFile(StarletteUploadFile):
def __get_pydantic_json_schema__(
cls, core_schema: Mapping[str, Any], handler: GetJsonSchemaHandler
) -> dict[str, Any]:
return {"type": "string", "contentMediaType": "application/octet-stream"}
return {
"type": "string",
"format": "binary", # For compatibility with OAS 3.0
"contentMediaType": "application/octet-stream",
}
@classmethod
def __get_pydantic_core_schema__(

View File

@ -2,7 +2,7 @@ import warnings
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from enum import Enum
from typing import Annotated, Any, Literal
from typing import Annotated, Any, Literal, cast
from fastapi.exceptions import FastAPIDeprecationWarning
from fastapi.openapi.models import Example
@ -660,6 +660,32 @@ class Form(Body): # type: ignore[misc] # ty: ignore[unused-ignore-comment]
)
class _FileUploadMarker:
"Pydantic metadata marker to tag bytes CoreSchemas as file uploads."
@classmethod
def __get_pydantic_core_schema__(
cls, source: type[Any], handler: Any
) -> dict[str, Any]:
schema = cast(dict[str, Any], handler(source))
# Find the inner type schema (if nullable or list)
inner_type_schema = schema
if inner_type_schema.get("type") != "bytes":
if inner_type_schema.get("type") == "list":
inner_type_schema = inner_type_schema["items_schema"]
elif "schema" in inner_type_schema:
inner_type_schema = inner_type_schema["schema"]
if inner_type_schema.get("type") == "list":
inner_type_schema = inner_type_schema["items_schema"]
# If the inner type is bytes, add the file upload marker metadata
if inner_type_schema.get("type") == "bytes":
metadata: dict[str, Any] = inner_type_schema.setdefault("metadata", {})
metadata["fastapi_file_upload"] = True
return schema
class File(Form): # type: ignore[misc] # ty: ignore[unused-ignore-comment]
def __init__(
self,
@ -740,6 +766,7 @@ class File(Form): # type: ignore[misc] # ty: ignore[unused-ignore-comment]
json_schema_extra=json_schema_extra,
**extra,
)
self.metadata.append(_FileUploadMarker())
@dataclass(frozen=True)

View File

@ -3,6 +3,7 @@ from typing import Annotated
import pytest
from fastapi import FastAPI, File, UploadFile
from fastapi.testclient import TestClient
from inline_snapshot import Is, snapshot
from .utils import get_body_model_name
@ -33,21 +34,24 @@ def test_list_schema(path: str):
openapi = app.openapi()
body_model_name = get_body_model_name(openapi, path)
assert app.openapi()["components"]["schemas"][body_model_name] == {
"properties": {
"p": {
"type": "array",
"items": {
"type": "string",
"contentMediaType": "application/octet-stream",
assert app.openapi()["components"]["schemas"][body_model_name] == snapshot(
{
"properties": {
"p": {
"type": "array",
"items": {
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
"title": "P",
},
"title": "P",
},
},
"required": ["p"],
"title": body_model_name,
"type": "object",
}
"required": ["p"],
"title": Is(body_model_name),
"type": "object",
}
)
@pytest.mark.parametrize(
@ -114,21 +118,24 @@ def test_list_alias_schema(path: str):
openapi = app.openapi()
body_model_name = get_body_model_name(openapi, path)
assert app.openapi()["components"]["schemas"][body_model_name] == {
"properties": {
"p_alias": {
"type": "array",
"items": {
"type": "string",
"contentMediaType": "application/octet-stream",
assert app.openapi()["components"]["schemas"][body_model_name] == snapshot(
{
"properties": {
"p_alias": {
"type": "array",
"items": {
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
"title": "P Alias",
},
"title": "P Alias",
},
},
"required": ["p_alias"],
"title": body_model_name,
"type": "object",
}
"required": ["p_alias"],
"title": Is(body_model_name),
"type": "object",
}
)
@pytest.mark.parametrize(
@ -223,21 +230,24 @@ def test_list_validation_alias_schema(path: str):
openapi = app.openapi()
body_model_name = get_body_model_name(openapi, path)
assert app.openapi()["components"]["schemas"][body_model_name] == {
"properties": {
"p_val_alias": {
"type": "array",
"items": {
"type": "string",
"contentMediaType": "application/octet-stream",
assert app.openapi()["components"]["schemas"][body_model_name] == snapshot(
{
"properties": {
"p_val_alias": {
"type": "array",
"items": {
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
"title": "P Val Alias",
},
"title": "P Val Alias",
},
},
"required": ["p_val_alias"],
"title": body_model_name,
"type": "object",
}
"required": ["p_val_alias"],
"title": Is(body_model_name),
"type": "object",
}
)
@pytest.mark.parametrize(
@ -343,21 +353,24 @@ def test_list_alias_and_validation_alias_schema(path: str):
openapi = app.openapi()
body_model_name = get_body_model_name(openapi, path)
assert app.openapi()["components"]["schemas"][body_model_name] == {
"properties": {
"p_val_alias": {
"type": "array",
"items": {
"type": "string",
"contentMediaType": "application/octet-stream",
assert app.openapi()["components"]["schemas"][body_model_name] == snapshot(
{
"properties": {
"p_val_alias": {
"type": "array",
"items": {
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
"title": "P Val Alias",
},
"title": "P Val Alias",
},
},
"required": ["p_val_alias"],
"title": body_model_name,
"type": "object",
}
"required": ["p_val_alias"],
"title": Is(body_model_name),
"type": "object",
}
)
@pytest.mark.parametrize(

View File

@ -3,6 +3,7 @@ from typing import Annotated
import pytest
from fastapi import FastAPI, File, UploadFile
from fastapi.testclient import TestClient
from inline_snapshot import Is, snapshot
from .utils import get_body_model_name
@ -33,19 +34,25 @@ def test_optional_schema(path: str):
openapi = app.openapi()
body_model_name = get_body_model_name(openapi, path)
assert app.openapi()["components"]["schemas"][body_model_name] == {
"properties": {
"p": {
"anyOf": [
{"type": "string", "contentMediaType": "application/octet-stream"},
{"type": "null"},
],
"title": "P",
}
},
"title": body_model_name,
"type": "object",
}
assert app.openapi()["components"]["schemas"][body_model_name] == snapshot(
{
"properties": {
"p": {
"anyOf": [
{
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
{"type": "null"},
],
"title": "P",
}
},
"title": Is(body_model_name),
"type": "object",
}
)
@pytest.mark.parametrize(
@ -105,19 +112,25 @@ def test_optional_alias_schema(path: str):
openapi = app.openapi()
body_model_name = get_body_model_name(openapi, path)
assert app.openapi()["components"]["schemas"][body_model_name] == {
"properties": {
"p_alias": {
"anyOf": [
{"type": "string", "contentMediaType": "application/octet-stream"},
{"type": "null"},
],
"title": "P Alias",
}
},
"title": body_model_name,
"type": "object",
}
assert app.openapi()["components"]["schemas"][body_model_name] == snapshot(
{
"properties": {
"p_alias": {
"anyOf": [
{
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
{"type": "null"},
],
"title": "P Alias",
}
},
"title": Is(body_model_name),
"type": "object",
}
)
@pytest.mark.parametrize(
@ -196,19 +209,25 @@ def test_optional_validation_alias_schema(path: str):
openapi = app.openapi()
body_model_name = get_body_model_name(openapi, path)
assert app.openapi()["components"]["schemas"][body_model_name] == {
"properties": {
"p_val_alias": {
"anyOf": [
{"type": "string", "contentMediaType": "application/octet-stream"},
{"type": "null"},
],
"title": "P Val Alias",
}
},
"title": body_model_name,
"type": "object",
}
assert app.openapi()["components"]["schemas"][body_model_name] == snapshot(
{
"properties": {
"p_val_alias": {
"anyOf": [
{
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
{"type": "null"},
],
"title": "P Val Alias",
}
},
"title": Is(body_model_name),
"type": "object",
}
)
@pytest.mark.parametrize(
@ -292,19 +311,25 @@ def test_optional_alias_and_validation_alias_schema(path: str):
openapi = app.openapi()
body_model_name = get_body_model_name(openapi, path)
assert app.openapi()["components"]["schemas"][body_model_name] == {
"properties": {
"p_val_alias": {
"anyOf": [
{"type": "string", "contentMediaType": "application/octet-stream"},
{"type": "null"},
],
"title": "P Val Alias",
}
},
"title": body_model_name,
"type": "object",
}
assert app.openapi()["components"]["schemas"][body_model_name] == snapshot(
{
"properties": {
"p_val_alias": {
"anyOf": [
{
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
{"type": "null"},
],
"title": "P Val Alias",
}
},
"title": Is(body_model_name),
"type": "object",
}
)
@pytest.mark.parametrize(

View File

@ -3,6 +3,7 @@ from typing import Annotated
import pytest
from fastapi import FastAPI, File, UploadFile
from fastapi.testclient import TestClient
from inline_snapshot import Is, snapshot
from .utils import get_body_model_name
@ -35,25 +36,28 @@ def test_optional_list_schema(path: str):
openapi = app.openapi()
body_model_name = get_body_model_name(openapi, path)
assert app.openapi()["components"]["schemas"][body_model_name] == {
"properties": {
"p": {
"anyOf": [
{
"type": "array",
"items": {
"type": "string",
"contentMediaType": "application/octet-stream",
assert app.openapi()["components"]["schemas"][body_model_name] == snapshot(
{
"properties": {
"p": {
"anyOf": [
{
"type": "array",
"items": {
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
},
},
{"type": "null"},
],
"title": "P",
}
},
"title": body_model_name,
"type": "object",
}
{"type": "null"},
],
"title": "P",
}
},
"title": Is(body_model_name),
"type": "object",
}
)
@pytest.mark.parametrize(
@ -113,25 +117,28 @@ def test_optional_list_alias_schema(path: str):
openapi = app.openapi()
body_model_name = get_body_model_name(openapi, path)
assert app.openapi()["components"]["schemas"][body_model_name] == {
"properties": {
"p_alias": {
"anyOf": [
{
"type": "array",
"items": {
"type": "string",
"contentMediaType": "application/octet-stream",
assert app.openapi()["components"]["schemas"][body_model_name] == snapshot(
{
"properties": {
"p_alias": {
"anyOf": [
{
"type": "array",
"items": {
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
},
},
{"type": "null"},
],
"title": "P Alias",
}
},
"title": body_model_name,
"type": "object",
}
{"type": "null"},
],
"title": "P Alias",
}
},
"title": Is(body_model_name),
"type": "object",
}
)
@pytest.mark.parametrize(
@ -205,25 +212,28 @@ def test_optional_validation_alias_schema(path: str):
openapi = app.openapi()
body_model_name = get_body_model_name(openapi, path)
assert app.openapi()["components"]["schemas"][body_model_name] == {
"properties": {
"p_val_alias": {
"anyOf": [
{
"type": "array",
"items": {
"type": "string",
"contentMediaType": "application/octet-stream",
assert app.openapi()["components"]["schemas"][body_model_name] == snapshot(
{
"properties": {
"p_val_alias": {
"anyOf": [
{
"type": "array",
"items": {
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
},
},
{"type": "null"},
],
"title": "P Val Alias",
}
},
"title": body_model_name,
"type": "object",
}
{"type": "null"},
],
"title": "P Val Alias",
}
},
"title": Is(body_model_name),
"type": "object",
}
)
@pytest.mark.parametrize(
@ -304,25 +314,28 @@ def test_optional_list_alias_and_validation_alias_schema(path: str):
openapi = app.openapi()
body_model_name = get_body_model_name(openapi, path)
assert app.openapi()["components"]["schemas"][body_model_name] == {
"properties": {
"p_val_alias": {
"anyOf": [
{
"type": "array",
"items": {
"type": "string",
"contentMediaType": "application/octet-stream",
assert app.openapi()["components"]["schemas"][body_model_name] == snapshot(
{
"properties": {
"p_val_alias": {
"anyOf": [
{
"type": "array",
"items": {
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
},
},
{"type": "null"},
],
"title": "P Val Alias",
}
},
"title": body_model_name,
"type": "object",
}
{"type": "null"},
],
"title": "P Val Alias",
}
},
"title": Is(body_model_name),
"type": "object",
}
)
@pytest.mark.parametrize(

View File

@ -3,6 +3,7 @@ from typing import Annotated
import pytest
from fastapi import FastAPI, File, UploadFile
from fastapi.testclient import TestClient
from inline_snapshot import Is, snapshot
from .utils import get_body_model_name
@ -33,18 +34,21 @@ def test_required_schema(path: str):
openapi = app.openapi()
body_model_name = get_body_model_name(openapi, path)
assert app.openapi()["components"]["schemas"][body_model_name] == {
"properties": {
"p": {
"title": "P",
"type": "string",
"contentMediaType": "application/octet-stream",
assert app.openapi()["components"]["schemas"][body_model_name] == snapshot(
{
"properties": {
"p": {
"title": "P",
"format": "binary",
"type": "string",
"contentMediaType": "application/octet-stream",
},
},
},
"required": ["p"],
"title": body_model_name,
"type": "object",
}
"required": ["p"],
"title": Is(body_model_name),
"type": "object",
}
)
@pytest.mark.parametrize(
@ -111,18 +115,21 @@ def test_required_alias_schema(path: str):
openapi = app.openapi()
body_model_name = get_body_model_name(openapi, path)
assert app.openapi()["components"]["schemas"][body_model_name] == {
"properties": {
"p_alias": {
"title": "P Alias",
"type": "string",
"contentMediaType": "application/octet-stream",
assert app.openapi()["components"]["schemas"][body_model_name] == snapshot(
{
"properties": {
"p_alias": {
"title": "P Alias",
"format": "binary",
"type": "string",
"contentMediaType": "application/octet-stream",
},
},
},
"required": ["p_alias"],
"title": body_model_name,
"type": "object",
}
"required": ["p_alias"],
"title": Is(body_model_name),
"type": "object",
}
)
@pytest.mark.parametrize(
@ -219,18 +226,21 @@ def test_required_validation_alias_schema(path: str):
openapi = app.openapi()
body_model_name = get_body_model_name(openapi, path)
assert app.openapi()["components"]["schemas"][body_model_name] == {
"properties": {
"p_val_alias": {
"title": "P Val Alias",
"type": "string",
"contentMediaType": "application/octet-stream",
assert app.openapi()["components"]["schemas"][body_model_name] == snapshot(
{
"properties": {
"p_val_alias": {
"title": "P Val Alias",
"format": "binary",
"type": "string",
"contentMediaType": "application/octet-stream",
},
},
},
"required": ["p_val_alias"],
"title": body_model_name,
"type": "object",
}
"required": ["p_val_alias"],
"title": Is(body_model_name),
"type": "object",
}
)
@pytest.mark.parametrize(
@ -332,18 +342,21 @@ def test_required_alias_and_validation_alias_schema(path: str):
openapi = app.openapi()
body_model_name = get_body_model_name(openapi, path)
assert app.openapi()["components"]["schemas"][body_model_name] == {
"properties": {
"p_val_alias": {
"title": "P Val Alias",
"type": "string",
"contentMediaType": "application/octet-stream",
assert app.openapi()["components"]["schemas"][body_model_name] == snapshot(
{
"properties": {
"p_val_alias": {
"title": "P Val Alias",
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
},
},
"required": ["p_val_alias"],
"title": body_model_name,
"type": "object",
}
"required": ["p_val_alias"],
"title": Is(body_model_name),
"type": "object",
}
)
@pytest.mark.parametrize(

View File

@ -162,6 +162,7 @@ def test_openapi_schema(client: TestClient):
"properties": {
"file": {
"title": "File",
"format": "binary",
"contentMediaType": "application/octet-stream",
"type": "string",
}
@ -174,6 +175,7 @@ def test_openapi_schema(client: TestClient):
"properties": {
"file": {
"title": "File",
"format": "binary",
"type": "string",
"contentMediaType": "application/octet-stream",
}

View File

@ -136,6 +136,7 @@ def test_openapi_schema(client: TestClient):
"anyOf": [
{
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
{"type": "null"},
@ -152,6 +153,7 @@ def test_openapi_schema(client: TestClient):
"anyOf": [
{
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
{"type": "null"},

View File

@ -121,6 +121,7 @@ def test_openapi_schema(client: TestClient):
"properties": {
"file": {
"title": "File",
"format": "binary",
"type": "string",
"description": "A file read as bytes",
"contentMediaType": "application/octet-stream",
@ -134,6 +135,7 @@ def test_openapi_schema(client: TestClient):
"properties": {
"file": {
"title": "File",
"format": "binary",
"contentMediaType": "application/octet-stream",
"type": "string",
"description": "A file read as UploadFile",

View File

@ -197,6 +197,7 @@ def test_openapi_schema(client: TestClient):
"type": "array",
"items": {
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
}
@ -212,6 +213,7 @@ def test_openapi_schema(client: TestClient):
"type": "array",
"items": {
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
}

View File

@ -167,6 +167,7 @@ def test_openapi_schema(client: TestClient):
"type": "array",
"items": {
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
"description": "Multiple files as bytes",
@ -183,6 +184,7 @@ def test_openapi_schema(client: TestClient):
"type": "array",
"items": {
"type": "string",
"format": "binary",
"contentMediaType": "application/octet-stream",
},
"description": "Multiple files as UploadFile",

View File

@ -197,11 +197,13 @@ def test_openapi_schema(client: TestClient):
"properties": {
"file": {
"title": "File",
"format": "binary",
"type": "string",
"contentMediaType": "application/octet-stream",
},
"fileb": {
"title": "Fileb",
"format": "binary",
"contentMediaType": "application/octet-stream",
"type": "string",
},