from starlette.websockets import WebSocket

app = FastAPI()
app.state.value = 42


async def extract_value_from_http_connection(conn: HTTPConnection):
    return conn.app.state.value


@app.get("/http")
async def get_value_by_http(value: int = Depends(extract_value_from_http_connection)):
    return value


@app.websocket("/ws")
async def get_value_by_ws(
    websocket: WebSocket, value: int = Depends(extract_value_from_http_connection)

oid = OpenIdConnect(openIdConnectUrl="/openid", auto_error=False)


class User(BaseModel):
    username: str


def get_current_user(oauth_header: str | None = Security(oid)):
    if oauth_header is None:
        return None
    user = User(username=oauth_header)
    return user


@app.get("/users/me")
def read_current_user(current_user: User | None = Depends(get_current_user)):
    if current_user is None:

        ],
    )


def get_en_config() -> dict[str, Any]:
    return mkdocs.utils.yaml_load(en_config_path.read_text(encoding="utf-8"))


def get_lang_paths() -> list[Path]:
    return sorted(docs_path.iterdir())


def lang_callback(lang: str | None) -> str | None:
    if lang is None:
        return None
    lang = lang.lower()
    return lang


def complete_existing_lang(incomplete: str):

app = FastAPI()

security = HTTPBase(scheme="Other", description="Other Security Scheme")


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


client = TestClient(app)


def test_security_http_base():
    response = client.get("/users/me", headers={"Authorization": "Other foobar"})

state = initial_state.copy()

app = FastAPI()


async def get_database():
    temp_database = fake_database.copy()
    try:
        yield temp_database
        fake_database.update(temp_database)
    except HTTPException:
        state["except"] = True
        raise
    finally:
        state["finally"] = True


@app.put("/invalid-user/{user_id}")
def put_invalid_user(

app = FastAPI()


def get_user(required_scopes: SecurityScopes):
    return "john", required_scopes.scopes


def get_user_override(required_scopes: SecurityScopes):
    return "alice", required_scopes.scopes


def get_data():
    return [1, 2, 3]


def get_data_override():
    return [3, 4, 5]


@app.get("/user")
def read_user(

@app.get("/default")
async def default(foo: Annotated[str, Query()] = "foo"):
    return {"foo": foo}


@app.get("/required")
async def required(foo: Annotated[str, Query(min_length=1)]):
    return {"foo": foo}


@app.get("/multiple")
async def multiple(foo: Annotated[str, object(), Query(min_length=1)]):
    return {"foo": foo}


@app.get("/unrelated")
async def unrelated(foo: Annotated[str, object()]):

class Session:
    def __init__(self) -> None:
        self.data = ["foo", "bar", "baz"]
        self.open = True

    def __iter__(self) -> Generator[str, None, None]:
        for item in self.data:
            if self.open:
                yield item
            else:
                raise ValueError("Session closed")


@contextmanager
def acquire_session() -> Generator[Session, None, None]:

app = FastAPI()

security = HTTPBase(scheme="Other")


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


client = TestClient(app)


def test_security_http_base():
    response = client.get("/users/me", headers={"Authorization": "Other foobar"})

     * @param key the property key to look up
     * @param def the default value to return if key is not found or cannot be parsed
     * @return the integer value of the property or the default value
     */
    public static int getInt(final Properties props, final String key, int def) {
        final String s = props.getProperty(key);
        if (s != null) {
            try {
                def = Integer.parseInt(s);