from fastapi import Depends, FastAPI, Security
from fastapi.security.open_id_connect_url import OpenIdConnect
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()

oid = OpenIdConnect(
    openIdConnectUrl="/openid", description="OpenIdConnect security scheme"

            

Reported by Pylint.

              from fastapi import Depends, FastAPI, Security
from fastapi.security.open_id_connect_url import OpenIdConnect
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()

oid = OpenIdConnect(
    openIdConnectUrl="/openid", description="OpenIdConnect security scheme"

            

Reported by Pylint.

              )


class User(BaseModel):
    username: str


def get_current_user(oauth_header: str = Security(oid)):
    user = User(username=oauth_header)

            

Reported by Pylint.

              )


class User(BaseModel):
    username: str


def get_current_user(oauth_header: str = Security(oid)):
    user = User(username=oauth_header)

            

Reported by Pylint.

                  username: str


def get_current_user(oauth_header: str = Security(oid)):
    user = User(username=oauth_header)
    return user


@app.get("/users/me")

            

Reported by Pylint.

              

@app.get("/users/me")
def read_current_user(current_user: User = Depends(get_current_user)):
    return current_user


client = TestClient(app)


            

Reported by Pylint.

              }


def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema



            

Reported by Pylint.

              
def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


def test_security_oauth2():
    response = client.get("/users/me", headers={"Authorization": "Bearer footokenbar"})

            

Reported by Bandit.

              def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


def test_security_oauth2():
    response = client.get("/users/me", headers={"Authorization": "Bearer footokenbar"})
    assert response.status_code == 200, response.text

            

Reported by Bandit.

                  assert response.json() == openapi_schema


def test_security_oauth2():
    response = client.get("/users/me", headers={"Authorization": "Bearer footokenbar"})
    assert response.status_code == 200, response.text
    assert response.json() == {"username": "Bearer footokenbar"}



            

Reported by Pylint.

              from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.testclient import TestClient
from starlette.responses import JSONResponse


def http_exception_handler(request, exception):
    return JSONResponse({"exception": "http-exception"})


            

Reported by Pylint.

              from starlette.responses import JSONResponse


def http_exception_handler(request, exception):
    return JSONResponse({"exception": "http-exception"})


def request_validation_exception_handler(request, exception):
    return JSONResponse({"exception": "request-validation"})

            

Reported by Pylint.

              from starlette.responses import JSONResponse


def http_exception_handler(request, exception):
    return JSONResponse({"exception": "http-exception"})


def request_validation_exception_handler(request, exception):
    return JSONResponse({"exception": "request-validation"})

            

Reported by Pylint.

                  return JSONResponse({"exception": "http-exception"})


def request_validation_exception_handler(request, exception):
    return JSONResponse({"exception": "request-validation"})


app = FastAPI(
    exception_handlers={

            

Reported by Pylint.

                  return JSONResponse({"exception": "http-exception"})


def request_validation_exception_handler(request, exception):
    return JSONResponse({"exception": "request-validation"})


app = FastAPI(
    exception_handlers={

            

Reported by Pylint.

              

@app.get("/request-validation/{param}/")
def route_with_request_validation_exception(param: int):
    pass  # pragma: no cover


def test_override_http_exception():
    response = client.get("/http-exception")

            

Reported by Pylint.

              from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.testclient import TestClient
from starlette.responses import JSONResponse


def http_exception_handler(request, exception):
    return JSONResponse({"exception": "http-exception"})


            

Reported by Pylint.

              from starlette.responses import JSONResponse


def http_exception_handler(request, exception):
    return JSONResponse({"exception": "http-exception"})


def request_validation_exception_handler(request, exception):
    return JSONResponse({"exception": "request-validation"})

            

Reported by Pylint.

                  return JSONResponse({"exception": "http-exception"})


def request_validation_exception_handler(request, exception):
    return JSONResponse({"exception": "request-validation"})


app = FastAPI(
    exception_handlers={

            

Reported by Pylint.

              

@app.get("/http-exception")
def route_with_http_exception():
    raise HTTPException(status_code=400)


@app.get("/request-validation/{param}/")
def route_with_request_validation_exception(param: int):

            

Reported by Pylint.

              
from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel, condecimal

app = FastAPI()


class Item(BaseModel):

            

Reported by Pylint.

              from decimal import Decimal
from typing import List

from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel, condecimal

app = FastAPI()


            

Reported by Pylint.

              app = FastAPI()


class Item(BaseModel):
    name: str
    age: condecimal(gt=Decimal(0.0))  # type: ignore


@app.post("/items/")

            

Reported by Pylint.

              app = FastAPI()


class Item(BaseModel):
    name: str
    age: condecimal(gt=Decimal(0.0))  # type: ignore


@app.post("/items/")

            

Reported by Pylint.

              

@app.post("/items/")
def save_item_no_body(item: List[Item]):
    return {"item": item}


client = TestClient(app)


            

Reported by Pylint.

              }


def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema



            

Reported by Pylint.

              
def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


def test_put_correct_body():
    response = client.post("/items/", json=[{"name": "Foo", "age": 5}])

            

Reported by Bandit.

              def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


def test_put_correct_body():
    response = client.post("/items/", json=[{"name": "Foo", "age": 5}])
    assert response.status_code == 200, response.text

            

Reported by Bandit.

                  assert response.json() == openapi_schema


def test_put_correct_body():
    response = client.post("/items/", json=[{"name": "Foo", "age": 5}])
    assert response.status_code == 200, response.text
    assert response.json() == {"item": [{"name": "Foo", "age": 5}]}



            

Reported by Pylint.

              
def test_put_correct_body():
    response = client.post("/items/", json=[{"name": "Foo", "age": 5}])
    assert response.status_code == 200, response.text
    assert response.json() == {"item": [{"name": "Foo", "age": 5}]}


def test_jsonable_encoder_requiring_error():
    response = client.post("/items/", json=[{"name": "Foo", "age": -1.0}])

            

Reported by Bandit.

              from types import GeneratorType
from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union

from pydantic import BaseModel
from pydantic.json import ENCODERS_BY_TYPE

SetIntStr = Set[Union[int, str]]
DictIntStrAny = Dict[Union[int, str], Any]


            

Reported by Pylint.

              from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union

from pydantic import BaseModel
from pydantic.json import ENCODERS_BY_TYPE

SetIntStr = Set[Union[int, str]]
DictIntStrAny = Dict[Union[int, str], Any]



            

Reported by Pylint.

              def generate_encoders_by_class_tuples(
    type_encoder_map: Dict[Any, Callable[[Any], Any]]
) -> Dict[Callable[[Any], Any], Tuple[Any, ...]]:
    encoders_by_class_tuples: Dict[Callable[[Any], Any], Tuple[Any, ...]] = defaultdict(
        tuple
    )
    for type_, encoder in type_encoder_map.items():
        encoders_by_class_tuples[encoder] += (type_,)
    return encoders_by_class_tuples

            

Reported by Pylint.

              encoders_by_class_tuples = generate_encoders_by_class_tuples(ENCODERS_BY_TYPE)


def jsonable_encoder(
    obj: Any,
    include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
    exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
    by_alias: bool = True,
    exclude_unset: bool = False,

            

Reported by Pylint.

                  errors: List[Exception] = []
    try:
        data = dict(obj)
    except Exception as e:
        errors.append(e)
        try:
            data = vars(obj)
        except Exception as e:
            errors.append(e)

            

Reported by Pylint.

                          data = vars(obj)
        except Exception as e:
            errors.append(e)
            raise ValueError(errors)
    return jsonable_encoder(
        data,
        by_alias=by_alias,
        exclude_unset=exclude_unset,
        exclude_defaults=exclude_defaults,

            

Reported by Pylint.

              import dataclasses
from collections import defaultdict
from enum import Enum
from pathlib import PurePath
from types import GeneratorType
from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union

from pydantic import BaseModel
from pydantic.json import ENCODERS_BY_TYPE

            

Reported by Pylint.

              DictIntStrAny = Dict[Union[int, str], Any]


def generate_encoders_by_class_tuples(
    type_encoder_map: Dict[Any, Callable[[Any], Any]]
) -> Dict[Callable[[Any], Any], Tuple[Any, ...]]:
    encoders_by_class_tuples: Dict[Callable[[Any], Any], Tuple[Any, ...]] = defaultdict(
        tuple
    )

            

Reported by Pylint.

              encoders_by_class_tuples = generate_encoders_by_class_tuples(ENCODERS_BY_TYPE)


def jsonable_encoder(
    obj: Any,
    include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
    exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
    by_alias: bool = True,
    exclude_unset: bool = False,

            

Reported by Pylint.

              encoders_by_class_tuples = generate_encoders_by_class_tuples(ENCODERS_BY_TYPE)


def jsonable_encoder(
    obj: Any,
    include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
    exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
    by_alias: bool = True,
    exclude_unset: bool = False,

            

Reported by Pylint.

              from fastapi import FastAPI, Path
from fastapi.testclient import TestClient

app = FastAPI()


@app.get("/int/{param:int}")
def int_convertor(param: int = Path(...)):
    return {"int": param}

            

Reported by Pylint.

              

@app.get("/int/{param:int}")
def int_convertor(param: int = Path(...)):
    return {"int": param}


@app.get("/float/{param:float}")
def float_convertor(param: float = Path(...)):

            

Reported by Pylint.

              

@app.get("/float/{param:float}")
def float_convertor(param: float = Path(...)):
    return {"float": param}


@app.get("/path/{param:path}")
def path_convertor(param: str = Path(...)):

            

Reported by Pylint.

              

@app.get("/path/{param:path}")
def path_convertor(param: str = Path(...)):
    return {"path": param}


client = TestClient(app)


            

Reported by Pylint.

              client = TestClient(app)


def test_route_converters_int():
    # Test integer conversion
    response = client.get("/int/5")
    assert response.status_code == 200, response.text
    assert response.json() == {"int": 5}
    assert app.url_path_for("int_convertor", param=5) == "/int/5"  # type: ignore

            

Reported by Pylint.

              def test_route_converters_int():
    # Test integer conversion
    response = client.get("/int/5")
    assert response.status_code == 200, response.text
    assert response.json() == {"int": 5}
    assert app.url_path_for("int_convertor", param=5) == "/int/5"  # type: ignore


def test_route_converters_float():

            

Reported by Bandit.

                  # Test integer conversion
    response = client.get("/int/5")
    assert response.status_code == 200, response.text
    assert response.json() == {"int": 5}
    assert app.url_path_for("int_convertor", param=5) == "/int/5"  # type: ignore


def test_route_converters_float():
    # Test float conversion

            

Reported by Bandit.

                  response = client.get("/int/5")
    assert response.status_code == 200, response.text
    assert response.json() == {"int": 5}
    assert app.url_path_for("int_convertor", param=5) == "/int/5"  # type: ignore


def test_route_converters_float():
    # Test float conversion
    response = client.get("/float/25.5")

            

Reported by Bandit.

                  assert app.url_path_for("int_convertor", param=5) == "/int/5"  # type: ignore


def test_route_converters_float():
    # Test float conversion
    response = client.get("/float/25.5")
    assert response.status_code == 200, response.text
    assert response.json() == {"float": 25.5}
    assert app.url_path_for("float_convertor", param=25.5) == "/float/25.5"  # type: ignore

            

Reported by Pylint.

              def test_route_converters_float():
    # Test float conversion
    response = client.get("/float/25.5")
    assert response.status_code == 200, response.text
    assert response.json() == {"float": 25.5}
    assert app.url_path_for("float_convertor", param=25.5) == "/float/25.5"  # type: ignore


def test_route_converters_path():

            

Reported by Bandit.

              from typing import Optional

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

fake_users_db = {
    "johndoe": {
        "username": "johndoe",

            

Reported by Pylint.

              from typing import Optional

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

fake_users_db = {
    "johndoe": {
        "username": "johndoe",

            

Reported by Pylint.

              
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",

            

Reported by Pylint.

              from typing import Optional

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

fake_users_db = {
    "johndoe": {
        "username": "johndoe",

            

Reported by Pylint.

              app = FastAPI()


def fake_hash_password(password: str):
    return "fakehashed" + password


oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


            

Reported by Pylint.

              oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None


            

Reported by Pylint.

              oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None


            

Reported by Pylint.

                  disabled: Optional[bool] = None


class UserInDB(User):
    hashed_password: str


def get_user(db, username: str):
    if username in db:

            

Reported by Pylint.

                  disabled: Optional[bool] = None


class UserInDB(User):
    hashed_password: str


def get_user(db, username: str):
    if username in db:

            

Reported by Pylint.

                  hashed_password: str


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)



            

Reported by Pylint.

              
from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel

from .utils import needs_py37

# In Python 3.6:
# u = Union[ExtendedItem, Item] == __main__.Item

            

Reported by Pylint.

              from fastapi.testclient import TestClient
from pydantic import BaseModel

from .utils import needs_py37

# In Python 3.6:
# u = Union[ExtendedItem, Item] == __main__.Item

# But in Python 3.7:

            

Reported by Pylint.

              from typing import Optional, Union

from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel

from .utils import needs_py37

# In Python 3.6:

            

Reported by Pylint.

              app = FastAPI()


class Item(BaseModel):
    name: Optional[str] = None


class ExtendedItem(Item):
    age: int

            

Reported by Pylint.

              app = FastAPI()


class Item(BaseModel):
    name: Optional[str] = None


class ExtendedItem(Item):
    age: int

            

Reported by Pylint.

                  name: Optional[str] = None


class ExtendedItem(Item):
    age: int


@app.post("/items/")
def save_union_different_body(item: Union[ExtendedItem, Item]):

            

Reported by Pylint.

                  name: Optional[str] = None


class ExtendedItem(Item):
    age: int


@app.post("/items/")
def save_union_different_body(item: Union[ExtendedItem, Item]):

            

Reported by Pylint.

              

@app.post("/items/")
def save_union_different_body(item: Union[ExtendedItem, Item]):
    return {"item": item}


client = TestClient(app)


            

Reported by Pylint.

              

@needs_py37
def test_inherited_item_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == inherited_item_openapi_schema



            

Reported by Pylint.

              @needs_py37
def test_inherited_item_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == inherited_item_openapi_schema


@needs_py37
def test_post_extended_item():

            

Reported by Bandit.

              from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()


class Model(BaseModel):
    pass

            

Reported by Pylint.

              from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()


class Model(BaseModel):
    pass

            

Reported by Pylint.

              app = FastAPI()


class Model(BaseModel):
    pass


class Model2(BaseModel):
    a: Model

            

Reported by Pylint.

              app = FastAPI()


class Model(BaseModel):
    pass


class Model2(BaseModel):
    a: Model

            

Reported by Pylint.

                  pass


class Model2(BaseModel):
    a: Model


class Model3(BaseModel):
    c: Model

            

Reported by Pylint.

                  pass


class Model2(BaseModel):
    a: Model


class Model3(BaseModel):
    c: Model

            

Reported by Pylint.

                  a: Model


class Model3(BaseModel):
    c: Model
    d: Model2


@app.get("/", response_model=Model3)

            

Reported by Pylint.

                  a: Model


class Model3(BaseModel):
    c: Model
    d: Model2


@app.get("/", response_model=Model3)

            

Reported by Pylint.

              

@app.get("/", response_model=Model3)
def f():
    return {"c": {}, "d": {"a": {}}}


openapi_schema = {
    "openapi": "3.0.2",

            

Reported by Pylint.

              

@app.get("/", response_model=Model3)
def f():
    return {"c": {}, "d": {"a": {}}}


openapi_schema = {
    "openapi": "3.0.2",

            

Reported by Pylint.

              from fastapi import Depends, FastAPI, Security
from fastapi.security import APIKeyCookie
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()

api_key = APIKeyCookie(name="key", auto_error=False)


            

Reported by Pylint.

              from typing import Optional

from fastapi import Depends, FastAPI, Security
from fastapi.security import APIKeyCookie
from fastapi.testclient import TestClient
from pydantic import BaseModel

app = FastAPI()


            

Reported by Pylint.

              api_key = APIKeyCookie(name="key", auto_error=False)


class User(BaseModel):
    username: str


def get_current_user(oauth_header: Optional[str] = Security(api_key)):
    if oauth_header is None:

            

Reported by Pylint.

              api_key = APIKeyCookie(name="key", auto_error=False)


class User(BaseModel):
    username: str


def get_current_user(oauth_header: Optional[str] = Security(api_key)):
    if oauth_header is None:

            

Reported by Pylint.

                  username: str


def get_current_user(oauth_header: Optional[str] = Security(api_key)):
    if oauth_header is None:
        return None
    user = User(username=oauth_header)
    return user


            

Reported by Pylint.

              

@app.get("/users/me")
def read_current_user(current_user: User = Depends(get_current_user)):
    if current_user is None:
        return {"msg": "Create an account first"}
    else:
        return current_user


            

Reported by Pylint.

              
@app.get("/users/me")
def read_current_user(current_user: User = Depends(get_current_user)):
    if current_user is None:
        return {"msg": "Create an account first"}
    else:
        return current_user



            

Reported by Pylint.

              }


def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema



            

Reported by Pylint.

              
def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


def test_security_api_key():
    response = client.get("/users/me", cookies={"key": "secret"})

            

Reported by Bandit.

              def test_openapi_schema():
    response = client.get("/openapi.json")
    assert response.status_code == 200, response.text
    assert response.json() == openapi_schema


def test_security_api_key():
    response = client.get("/users/me", cookies={"key": "secret"})
    assert response.status_code == 200, response.text

            

Reported by Bandit.

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

app = FastAPI()

media_type = "application/vnd.api+json"


            

Reported by Pylint.

              

@app.post("/products")
async def create_product(data: Product = Body(..., media_type=media_type, embed=True)):
    pass  # pragma: no cover


@app.post("/shops")
async def create_shop(

            

Reported by Pylint.

              
@app.post("/shops")
async def create_shop(
    data: Shop = Body(..., media_type=media_type),
    included: typing.List[Product] = Body([], media_type=media_type),
):
    pass  # pragma: no cover



            

Reported by Pylint.

              @app.post("/shops")
async def create_shop(
    data: Shop = Body(..., media_type=media_type),
    included: typing.List[Product] = Body([], media_type=media_type),
):
    pass  # pragma: no cover


create_product_request_body = {

            

Reported by Pylint.

              import typing

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

app = FastAPI()

media_type = "application/vnd.api+json"

            

Reported by Pylint.

              
app = FastAPI()

media_type = "application/vnd.api+json"


# NOTE: These are not valid JSON:API resources
# but they are fine for testing requestBody with custom media_type
class Product(BaseModel):

            

Reported by Pylint.

              
# NOTE: These are not valid JSON:API resources
# but they are fine for testing requestBody with custom media_type
class Product(BaseModel):
    name: str
    price: float


class Shop(BaseModel):

            

Reported by Pylint.

              
# NOTE: These are not valid JSON:API resources
# but they are fine for testing requestBody with custom media_type
class Product(BaseModel):
    name: str
    price: float


class Shop(BaseModel):

            

Reported by Pylint.

                  price: float


class Shop(BaseModel):
    name: str


@app.post("/products")
async def create_product(data: Product = Body(..., media_type=media_type, embed=True)):

            

Reported by Pylint.

                  price: float


class Shop(BaseModel):
    name: str


@app.post("/products")
async def create_product(data: Product = Body(..., media_type=media_type, embed=True)):

            

Reported by Pylint.