String result;
        Path src = sourcePathProvider.get();
        if (src == null) {
            Files.deleteIfExists(target);
            result = null;
        } else if (!Files.exists(src)) {
            Files.deleteIfExists(target);
            result = "";
        } else {
            String current = sha1(src);
            String existing = sourceState.get();

        "name": "Foo",
        "price": 42,
        "description": None,
        "tax": None,
        "tags": [],
    }


@needs_pydanticv2
def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == {
        "openapi": "3.1.0",
        "info": {"title": "FastAPI", "version": "0.1.0"},
        "paths": {

        final String pLC = pageCacheMap.get(pathLC);
        if (pLC != null) {
            return pLC;
        }
        if (existsPage(pathLC)) {
            pageCacheMap.put(pathLC, pathLC);
            return pathLC;
        }

        final String pathL = getLocalizedPagePath(page, lang, null);
        final String pL = pageCacheMap.get(pathL);
        if (pL != null) {
            return pL;

    @Override
    public SortedSet<V> get() {
      return new TreeSet<>(comparator);
    }
  }

  private static final class EnumSetSupplier<V extends Enum<V>>
      implements Supplier<Set<V>>, Serializable {
    private final Class<V> clazz;

    EnumSetSupplier(Class<V> clazz) {
      this.clazz = checkNotNull(clazz);
    }

    @Override
    public Set<V> get() {
      return EnumSet.noneOf(clazz);

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
async def root():

from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: str | None = Query(default=None, min_length=3, max_length=50)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})

from fastapi import FastAPI, Query
from typing_extensions import Annotated

app = FastAPI()


@app.get("/items/")
async def read_items(q: Annotated[str, Query(min_length=3)] = ...):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})

import java.io.InputStream;

/**
 * Provides access to the contents of a source independently of the backing store (e.g. file system, database, memory).
 *
 */
public interface Source {

    /**
     * Gets a byte stream to the source contents. Closing the returned stream is the responsibility of the caller.
     *
     * @return A byte stream to the source contents, never {@code null}.
     * @throws IOException in case of IO issue

from pydantic import BaseModel

app = FastAPI()


class Cookies(BaseModel):
    session_id: str
    fatebook_tracker: Union[str, None] = None
    googall_tracker: Union[str, None] = None


@app.get("/items/")
async def read_items(cookies: Cookies = Cookie()):

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

some_file_path = "large-video-file.mp4"
app = FastAPI()


@app.get("/")
def main():
    def iterfile():  # (1)
        with open(some_file_path, mode="rb") as file_like:  # (2)
            yield from file_like  # (3)