owner_ids: list[int] | None = None


@app.get("/items/valid", response_model=Item)
def get_valid():
    return Item(aliased_name="valid", price=1.0)


@app.get("/items/coerce", response_model=Item)
def get_coerce():
    return Item(aliased_name="coerce", price="1.0")


@app.get("/items/validlist", response_model=list[Item])
def get_validlist():
    return [
        Item(aliased_name="foo"),

        yield i
        i += 1


async def _run_asgi_and_cancel(app: FastAPI, path: str, timeout: float) -> bool:
    """Call the ASGI app for *path* and cancel after *timeout* seconds.

    Returns `True` if the cancellation was delivered (i.e. it did not hang).
    """
    chunks: list[bytes] = []

    async def receive():  # type: ignore[no-untyped-def]
        # Simulate a client that never disconnects, rely on cancellation

# Without aliases


@app.get("/required-str")
async def read_required_str(p: Annotated[str, Header()]):
    return {"p": p}


class HeaderModelRequiredStr(BaseModel):
    p: str


@app.get("/model-required-str")
async def read_model_required_str(p: Annotated[HeaderModelRequiredStr, Header()]):
    return {"p": p.p}


@pytest.mark.parametrize(

class UserInputError(Exception):
  pass


def is_windows():
  return platform.system() == 'Windows'


def is_linux():
  return platform.system() == 'Linux'


def is_macos():
  return platform.system() == 'Darwin'


def is_ppc64le():
  return platform.machine() == 'ppc64le'


def is_s390x():
  return platform.machine() == 's390x'


def is_cygwin():

# Without aliases


@app.post("/required-str", operation_id="required_str")
async def read_required_str(p: Annotated[str, Body(embed=True)]):
    return {"p": p}


class BodyModelRequiredStr(BaseModel):
    p: str


@app.post("/model-required-str", operation_id="model_required_str")
async def read_model_required_str(p: BodyModelRequiredStr):
    return {"p": p.p}


@pytest.mark.parametrize(

# Without aliases


@app.post("/optional-list-bytes")
async def read_optional_list_bytes(p: Annotated[list[bytes] | None, File()] = None):
    return {"file_size": [len(file) for file in p] if p else None}


@app.post("/optional-list-uploadfile")
async def read_optional_list_uploadfile(
    p: Annotated[list[UploadFile] | None, File()] = None,
):

      max_content_size (optional): the maximum content size allowed in bytes, None for no limit
    """

    def __init__(self, app: APIRouter, max_content_size: int | None = None):
        self.app = app
        self.max_content_size = max_content_size

    def receive_wrapper(self, receive):
        received = 0

        async def inner():
            nonlocal received
            message = await receive()

else:  # pragma: no cover
    from asyncio import iscoroutinefunction


def noop_wrap(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)

    return wrapper


def noop_wrap_async(func):
    if inspect.isgeneratorfunction(func):

        @wraps(func)
        async def gen_wrapper(*args, **kwargs):

        }

        // workaround to IBM JDK bug
        def createDslDocModelClosure = this.&createDslDocModel.curry(classDocbookDirectory.get().asFile, mainDocbookTemplate, classRepository)

        def doc = mainDocbookTemplate
        use(BuildableDOMCategory) {
            DslDocModel model = createDslDocModelClosure(loadPluginsMetaData())
            def root = doc.documentElement

    Document document
    Node parent
    List elements = []

    def DomBuilder(Document document) {
        this.document = document
        this.parent = document
    }

    def DomBuilder(Node parent) {
        this.document = parent.ownerDocument
        this.parent = parent
    }

    def DomBuilder(Document document, Node parent) {
        this.document = document
        this.parent = parent