body - Code Search

docs_src/websockets/tutorial002_py310.py

    WebSocketException,
    status,
)
from fastapi.responses import HTMLResponse

app = FastAPI()

html = """
<!DOCTYPE html>
<html>
    <head>
        <title>Chat</title>
    </head>
    <body>
        <h1>WebSocket Chat</h1>
        <form action="" onsubmit="sendMessage(event)">
            <label>Item ID: <input type="text" id="itemId" autocomplete="off" value="foo"/></label>

Created: Sun Dec 28 07:19:09 GMT 2025

- Last Modified: Sat Mar 18 12:29:59 GMT 2023

- 2.7K bytes

- Click Count (0)

github.com/tiangolo/fastapi

tests/test_no_schema_split.py

        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": []},
    }

Created: Sun Dec 28 07:19:09 GMT 2025

- Last Modified: Sat Dec 20 15:55:38 GMT 2025

- 6.4K bytes

- Click Count (0)

github.com/tiangolo/fastapi

tests/test_request_params/test_body/test_optional_str.py

from typing import Annotated, Optional

import pytest
from fastapi import Body, FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field

from .utils import get_body_model_name

app = FastAPI()

# =====================================================================================
# Without aliases


@app.post("/optional-str", operation_id="optional_str")

Created: Sun Dec 28 07:19:09 GMT 2025

- Last Modified: Sat Dec 27 18:19:10 GMT 2025

- 11.6K bytes

- Click Count (0)

github.com/tiangolo/fastapi

docs_src/websockets/tutorial002_an_py39.py

    WebSocketException,
    status,
)
from fastapi.responses import HTMLResponse

app = FastAPI()

html = """
<!DOCTYPE html>
<html>
    <head>
        <title>Chat</title>
    </head>
    <body>
        <h1>WebSocket Chat</h1>
        <form action="" onsubmit="sendMessage(event)">
            <label>Item ID: <input type="text" id="itemId" autocomplete="off" value="foo"/></label>

Created: Sun Dec 28 07:19:09 GMT 2025

- Last Modified: Sat Mar 18 12:29:59 GMT 2023

- 2.8K bytes

- Click Count (0)

github.com/tiangolo/fastapi

tests/test_webhooks_security.py

bearer_scheme = HTTPBearer()


class Subscription(BaseModel):
    username: str
    monthly_fee: float
    start_date: datetime


@app.webhooks.post("new-subscription")
def new_subscription(
    body: Subscription, token: Annotated[str, Security(bearer_scheme)]
):
    """
    When a new user subscribes to your service we'll send you a POST request with this

Created: Sun Dec 28 07:19:09 GMT 2025

- Last Modified: Wed Dec 17 21:25:59 GMT 2025

- 4.5K bytes

- Click Count (0)

github.com/tiangolo/fastapi

tests/test_request_params/test_file/test_required.py

    response = client.post(path)
    assert response.status_code == 422
    assert response.json() == {
        "detail": [
            {
                "type": "missing",
                "loc": ["body", "p"],
                "msg": "Field required",
                "input": None,
            }
        ]
    }


@pytest.mark.parametrize(
    "path",
    [
        "/required-bytes",

Created: Sun Dec 28 07:19:09 GMT 2025

- Last Modified: Sat Dec 27 18:19:10 GMT 2025

- 10.8K bytes

- Click Count (0)

github.com/minio/minio

cmd/http-tracer.go

		// Setup a http request response recorder - this is needed for
		// http stats requests and audit if enabled.
		respRecorder := xhttp.NewResponseRecorder(w)

		// Setup a http request body recorder
		reqRecorder := &xhttp.RequestRecorder{Reader: r.Body}
		r.Body = reqRecorder

		// Create tracing data structure and associate it to the request context
		tc := mcontext.TraceCtxt{
			AmzReqID:         w.Header().Get(xhttp.AmzRequestID),

Created: Sun Dec 28 19:28:13 GMT 2025

- Last Modified: Tue Sep 24 17:13:00 GMT 2024

- 6K bytes

- Click Count (0)

github.com/minio/minio

internal/config/policy/plugin/config.go

func (o *AuthZPlugin) IsAllowed(args policy.Args) (bool, error) {
	if o == nil {
		return false, nil
	}

	// Access Management Plugin Input
	body := make(map[string]any)
	body["input"] = args

	inputBytes, err := json.Marshal(body)
	if err != nil {
		return false, err
	}

	req, err := http.NewRequest(http.MethodPost, o.args.URL.String(), bytes.NewReader(inputBytes))
	if err != nil {

Created: Sun Dec 28 19:28:13 GMT 2025

- Last Modified: Fri Aug 29 02:39:48 GMT 2025

- 5.8K bytes

- Click Count (0)

github.com/tiangolo/fastapi

tests/test_request_params/test_form/test_list.py

    response = client.post(path)
    assert response.status_code == 422
    assert response.json() == {
        "detail": [
            {
                "type": "missing",
                "loc": ["body", "p"],
                "msg": "Field required",
                "input": IsOneOf(None, {}),
            }
        ]
    }


@pytest.mark.parametrize(
    "path",

Created: Sun Dec 28 07:19:09 GMT 2025

- Last Modified: Sat Dec 27 18:31:34 GMT 2025

- 11.7K bytes

- Click Count (0)

github.com/minio/minio

cmd/bucket-policy-handlers_test.go

			t.Fatalf("Case %d: Expected the response status to be `%d`, but instead found `%d`", i+1, testCase.expectedRespStatus, recV4.Code)
		}
		// read the response body.
		bucketPolicyReadBuf, err := io.ReadAll(recV4.Body)
		if err != nil {
			t.Fatalf("Test %d: %s: Failed parsing response body: <ERROR> %v", i+1, instanceType, err)
		}

		if recV4.Code != testCase.expectedRespStatus {

Created: Sun Dec 28 19:28:13 GMT 2025

- Last Modified: Fri Aug 29 02:39:48 GMT 2025

- 32.9K bytes

- Click Count (0)

Search Options