Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion litestar/_openapi/path_item.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,12 +62,17 @@ def create_operation_for_handler_method(
signature_fields = route_handler.parsed_fn_signature.parameters

request_body = None
request_body_field = None
if data_field := signature_fields.get("data"):
request_body = create_request_body(
self.context, route_handler.handler_id, route_handler.data_dto, data_field
)
request_body_field = data_field
elif body_field := signature_fields.get("body"):
request_body = create_request_body(self.context, route_handler.handler_id, None, body_field)
request_body_field = body_field

raises_validation_error = bool(data_field or self._path_item.parameters or parameters)
raises_validation_error = bool(request_body_field or self._path_item.parameters or parameters)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't think we should set raises_validation_error=True in this case, as the body parameter is only allowed to be bytes and will always receive the raw request body

responses = create_responses_for_handler(
self.context, route_handler, raises_validation_error=raises_validation_error
)
Expand Down
4 changes: 3 additions & 1 deletion litestar/_openapi/request_body.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,9 @@ def create_request_body(
Returns:
A RequestBody instance.
"""
media_type: RequestEncodingType | str = RequestEncodingType.JSON
media_type: RequestEncodingType | str = (
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is incorrect; body: bytes is just the raw request body, it does not signify a binary payload. That would be data: bytes. So no changes should be made here

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm new to the framework and not fully confident about the correct place to enforce the expected behavior.
Could you outline where in the existing code (e.g. specific modules or functions) the fix should be implemented, so I can update the PR accordingly?

Copy link
Member

@provinzkraut provinzkraut Nov 23, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think there might be a slight misunderstanding!

There's no need to change anything here, you just need to revert the changes you've made in this section.

Currently, (with your changes implemented - this line here) the proper way to achieve what you want would be to do something like:

@post("/")
async def handler(body: bytes = Body(content_encoding="application/octet-stream")) -> None:
    pass

Am I correct in assuming that what you're asking is to be able to omit the Body param, and default to application/octet-stream for a bare body: bytes?

"application/octet-stream" if data_field.is_subclass_of(bytes) else RequestEncodingType.JSON
)
schema_creator = SchemaCreator.from_openapi_context(context, prefer_alias=True)
if isinstance(data_field.kwarg_definition, BodyKwarg) and data_field.kwarg_definition.media_type:
media_type = data_field.kwarg_definition.media_type
Expand Down
66 changes: 66 additions & 0 deletions tests/unit/test_openapi/test_request_body.py
Original file line number Diff line number Diff line change
Expand Up @@ -217,3 +217,69 @@ async def handler(
}
}
}

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think these tests would be more readable if you used pytest.mark.parametrize


def test_body_parameter_binary_request() -> None:
@post("/upload/")
async def handle_binary_upload(body: bytes) -> None:
return None

app = Litestar([handle_binary_upload])
schema = app.openapi_schema.to_schema()

assert "requestBody" in schema["paths"]["/upload"]["post"]
assert schema["paths"]["/upload"]["post"]["requestBody"] == {
"required": True,
"content": {"application/octet-stream": {"schema": {"type": "string"}}},
}


def test_body_parameter_with_body_annotation() -> None:
@post("/upload/")
async def handle_binary_upload(
body: Annotated[bytes, Body(media_type="application/octet-stream", title="Binary Data")],
) -> None:
return None

app = Litestar([handle_binary_upload])
schema = app.openapi_schema.to_schema()

assert "requestBody" in schema["paths"]["/upload"]["post"]
assert schema["paths"]["/upload"]["post"]["requestBody"] == {
"required": True,
"content": {"application/octet-stream": {"schema": {"type": "string", "title": "Binary Data"}}},
}


def test_body_parameter_with_default_value() -> None:
@post("/upload/")
async def handle_binary_upload(
body: bytes = Body(media_type="application/octet-stream", title="Binary Data"),
) -> None:
return None

app = Litestar([handle_binary_upload])
schema = app.openapi_schema.to_schema()

assert "requestBody" in schema["paths"]["/upload"]["post"]
assert schema["paths"]["/upload"]["post"]["requestBody"] == {
"required": True,
"content": {"application/octet-stream": {"schema": {"type": "string", "title": "Binary Data"}}},
}


def test_body_parameter_with_custom_media_type() -> None:
@post("/upload/")
async def handle_binary_upload(
body: Annotated[bytes, Body(media_type="application/x-custom-binary")],
) -> None:
return None

app = Litestar([handle_binary_upload])
schema = app.openapi_schema.to_schema()

assert "requestBody" in schema["paths"]["/upload"]["post"]
assert schema["paths"]["/upload"]["post"]["requestBody"] == {
"required": True,
"content": {"application/x-custom-binary": {"schema": {"type": "string"}}},
}
Loading