from typing import Annotated

from fastapi import FastAPI, File, UploadFile
from fastapi.responses import HTMLResponse

app = FastAPI()


@app.post("/files/")
async def create_files(
    files: Annotated[list[bytes], File(description="Multiple files as bytes")],
):
    return {"file_sizes": [len(file) for file in files]}


@app.post("/uploadfiles/")
async def create_upload_files(
    files: Annotated[

1a) Send encrypted data key and master key ID to KMS.
1b) Receive decrypted data key.
2)  Decrypt encrypted object key with the KEK derived from the data key.
3a) Receive new plain data key from the KMS using the master key ID of the server config.
3b) Receive encrypted form of the data key from the KMS.
4)  Derive a new KEK from the new data key and re-encrypt the OEK with it.
5)  Store the encrypted OEK encrypted data key and master key ID in object metadata.

import pytest
from dirty_equals import IsDict, IsOneOf
from fastapi.testclient import TestClient

from ...utils import needs_py310


@pytest.fixture(name="client")
def get_client():
    from docs_src.response_model.tutorial004_py310 import app

    client = TestClient(app)
    return client


@needs_py310
@pytest.mark.parametrize(
    "url,data",
    [
        ("/items/foo", {"name": "Foo", "price": 50.2}),
        (

from fastapi import FastAPI

from . import a, b

app = FastAPI()

app.include_router(a.router, prefix="/a")

- Upon receiving a LIST request with an expired continue token, the apiserver now returns a continue token together with the 410 "the from parameter is too old" error. If the client does not care about getting a list from a consistent snapshot, the client can use this token to continue listing from the next key, but the returned chunk will be from the latest snapshot. ([#67284](https://github.com/kubernetes/kubernetes/pull/67284), [@caesarxuchao](https://github.com/caesarxuchao))

   */
  void putCoalescing(Range<K> range, V value);

  /** Puts all the associations from {@code rangeMap} into this range map (optional operation). */
  void putAll(RangeMap<K, ? extends V> rangeMap);

  /** Removes all associations from this range map (optional operation). */
  void clear();

  /**
   * Removes all associations from this range map in the specified range (optional operation).
   *

from typing import Dict, List, Tuple

import pytest
from fastapi import FastAPI
from pydantic import BaseModel


def test_invalid_sequence():
    with pytest.raises(AssertionError):
        app = FastAPI()

        class Item(BaseModel):
            title: str

        @app.get("/items/{id}")
        def read_items(id: List[Item]):
            pass  # pragma: no cover


def test_invalid_tuple():

from fastapi import FastAPI, Security
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from fastapi.testclient import TestClient

app = FastAPI()

security = HTTPBearer(description="HTTP Bearer token scheme")


@app.get("/users/me")
def read_current_user(credentials: HTTPAuthorizationCredentials = Security(security)):
    return {"scheme": credentials.scheme, "credentials": credentials.credentials}

import pytest
from fastapi.testclient import TestClient

from ...utils import needs_py310


@pytest.fixture(name="client")
def get_client():
    from docs_src.query_params_str_validations.tutorial014_py310 import app

    client = TestClient(app)
    return client


@needs_py310
def test_hidden_query(client: TestClient):
    response = client.get("/items?hidden_query=somevalue")
    assert response.status_code == 200, response.text

from typing import Dict

from fastapi import FastAPI

app = FastAPI()


@app.post("/index-weights/")
async def create_index_weights(weights: Dict[int, float]):