from typing import Optional

from fastapi import FastAPI
from fastapi.responses import FileResponse
from pydantic import BaseModel


class Item(BaseModel):
    id: str

            

Reported by Pylint.

              from typing import Optional

from fastapi import FastAPI
from fastapi.responses import FileResponse
from pydantic import BaseModel


class Item(BaseModel):
    id: str

            

Reported by Pylint.

              
from fastapi import FastAPI
from fastapi.responses import FileResponse
from pydantic import BaseModel


class Item(BaseModel):
    id: str
    value: str

            

Reported by Pylint.

                      }
    },
)
async def read_item(item_id: str, img: Optional[bool] = None):
    if img:
        return FileResponse("image.png", media_type="image/png")
    else:
        return {"id": "foo", "value": "there goes my hero"}

            

Reported by Pylint.

              from typing import Optional

from fastapi import FastAPI
from fastapi.responses import FileResponse
from pydantic import BaseModel


class Item(BaseModel):
    id: str

            

Reported by Pylint.

              from pydantic import BaseModel


class Item(BaseModel):
    id: str
    value: str


app = FastAPI()

            

Reported by Pylint.

              from pydantic import BaseModel


class Item(BaseModel):
    id: str
    value: str


app = FastAPI()

            

Reported by Pylint.

                      200: {
            "content": {"image/png": {}},
            "description": "Return the JSON item or an image.",
        }
    },
)
async def read_item(item_id: str, img: Optional[bool] = None):
    if img:
        return FileResponse("image.png", media_type="image/png")

            

Reported by Pylint.

                  },
)
async def read_item(item_id: str, img: Optional[bool] = None):
    if img:
        return FileResponse("image.png", media_type="image/png")
    else:
        return {"id": "foo", "value": "there goes my hero"}

            

Reported by Pylint.

              from fastapi.testclient import TestClient

from docs_src.body_updates.tutorial001 import app

client = TestClient(app)

openapi_schema = {
    "openapi": "3.0.2",
    "info": {"title": "FastAPI", "version": "0.1.0"},

            

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_get():
    response = client.get("/items/baz")

            

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_get():
    response = client.get("/items/baz")
    assert response.status_code == 200, response.text

            

Reported by Bandit.

                  assert response.json() == openapi_schema


def test_get():
    response = client.get("/items/baz")
    assert response.status_code == 200, response.text
    assert response.json() == {
        "name": "Baz",
        "description": None,

            

Reported by Pylint.

              
def test_get():
    response = client.get("/items/baz")
    assert response.status_code == 200, response.text
    assert response.json() == {
        "name": "Baz",
        "description": None,
        "price": 50.2,
        "tax": 10.5,

            

Reported by Bandit.

              def test_get():
    response = client.get("/items/baz")
    assert response.status_code == 200, response.text
    assert response.json() == {
        "name": "Baz",
        "description": None,
        "price": 50.2,
        "tax": 10.5,
        "tags": [],

            

Reported by Bandit.

                  }


def test_put():
    response = client.put(
        "/items/bar", json={"name": "Barz", "price": 3, "description": None}
    )
    assert response.json() == {
        "name": "Barz",

            

Reported by Pylint.

                  response = client.put(
        "/items/bar", json={"name": "Barz", "price": 3, "description": None}
    )
    assert response.json() == {
        "name": "Barz",
        "description": None,
        "price": 3,
        "tax": 10.5,
        "tags": [],

            

Reported by Bandit.

              from typing import List

import yaml
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, ValidationError

app = FastAPI()



            

Reported by Pylint.

              
import yaml
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, ValidationError

app = FastAPI()


class Item(BaseModel):

            

Reported by Pylint.

                  try:
        data = yaml.safe_load(raw_body)
    except yaml.YAMLError:
        raise HTTPException(status_code=422, detail="Invalid YAML")
    try:
        item = Item.parse_obj(data)
    except ValidationError as e:
        raise HTTPException(status_code=422, detail=e.errors())
    return item

            

Reported by Pylint.

                  try:
        item = Item.parse_obj(data)
    except ValidationError as e:
        raise HTTPException(status_code=422, detail=e.errors())
    return item

            

Reported by Pylint.

              from typing import List

import yaml
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, ValidationError

app = FastAPI()



            

Reported by Pylint.

              app = FastAPI()


class Item(BaseModel):
    name: str
    tags: List[str]


@app.post(

            

Reported by Pylint.

              app = FastAPI()


class Item(BaseModel):
    name: str
    tags: List[str]


@app.post(

            

Reported by Pylint.

                      "requestBody": {
            "content": {"application/x-yaml": {"schema": Item.schema()}},
            "required": True,
        },
    },
)
async def create_item(request: Request):
    raw_body = await request.body()
    try:

            

Reported by Pylint.

                      raise HTTPException(status_code=422, detail="Invalid YAML")
    try:
        item = Item.parse_obj(data)
    except ValidationError as e:
        raise HTTPException(status_code=422, detail=e.errors())
    return item

            

Reported by Pylint.

              import pytest
from fastapi.testclient import TestClient

from docs_src.body_fields.tutorial001 import app

client = TestClient(app)


openapi_schema = {

            

Reported by Pylint.

              import pytest
from fastapi.testclient import TestClient

from docs_src.body_fields.tutorial001 import app

client = TestClient(app)


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



            

Reported by Pylint.

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


price_not_greater = {
    "detail": [

            

Reported by Bandit.

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


price_not_greater = {
    "detail": [
        {

            

Reported by Bandit.

                          },
        ),
        ("/items/5", {"item": {"name": "Foo", "price": -3.0}}, 422, price_not_greater),
    ],
)
def test(path, body, expected_status, expected_response):
    response = client.put(path, json=body)
    assert response.status_code == expected_status
    assert response.json() == expected_response

            

Reported by Pylint.

              )
def test(path, body, expected_status, expected_response):
    response = client.put(path, json=body)
    assert response.status_code == expected_status
    assert response.json() == expected_response

            

Reported by Bandit.

              def test(path, body, expected_status, expected_response):
    response = client.put(path, json=body)
    assert response.status_code == expected_status
    assert response.json() == expected_response

            

Reported by Bandit.

              import time
from typing import Callable

from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute


class TimedRoute(APIRoute):
    def get_route_handler(self) -> Callable:

            

Reported by Pylint.

              from typing import Callable

from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute


class TimedRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

            

Reported by Pylint.

              import time
from typing import Callable

from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute


class TimedRoute(APIRoute):
    def get_route_handler(self) -> Callable:

            

Reported by Pylint.

              from fastapi.routing import APIRoute


class TimedRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            before = time.time()

            

Reported by Pylint.

              from fastapi.routing import APIRoute


class TimedRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            before = time.time()

            

Reported by Pylint.

              

class TimedRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            before = time.time()
            response: Response = await original_route_handler(request)

            

Reported by Pylint.

              

@app.get("/")
async def not_timed():
    return {"message": "Not timed"}


@router.get("/timed")
async def timed():

            

Reported by Pylint.

              

@router.get("/timed")
async def timed():
    return {"message": "It's the time of my life"}


app.include_router(router)

            

Reported by Pylint.

              
from fastapi.openapi.models import OpenIdConnect as OpenIdConnectModel
from fastapi.security.base import SecurityBase
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.status import HTTP_403_FORBIDDEN


class OpenIdConnect(SecurityBase):

            

Reported by Pylint.

              from fastapi.openapi.models import OpenIdConnect as OpenIdConnectModel
from fastapi.security.base import SecurityBase
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.status import HTTP_403_FORBIDDEN


class OpenIdConnect(SecurityBase):
    def __init__(

            

Reported by Pylint.

              from fastapi.security.base import SecurityBase
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.status import HTTP_403_FORBIDDEN


class OpenIdConnect(SecurityBase):
    def __init__(
        self,

            

Reported by Pylint.

              from typing import Optional

from fastapi.openapi.models import OpenIdConnect as OpenIdConnectModel
from fastapi.security.base import SecurityBase
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.status import HTTP_403_FORBIDDEN



            

Reported by Pylint.

              from starlette.status import HTTP_403_FORBIDDEN


class OpenIdConnect(SecurityBase):
    def __init__(
        self,
        *,
        openIdConnectUrl: str,
        scheme_name: Optional[str] = None,

            

Reported by Pylint.

              from starlette.status import HTTP_403_FORBIDDEN


class OpenIdConnect(SecurityBase):
    def __init__(
        self,
        *,
        openIdConnectUrl: str,
        scheme_name: Optional[str] = None,

            

Reported by Pylint.

                  def __init__(
        self,
        *,
        openIdConnectUrl: str,
        scheme_name: Optional[str] = None,
        description: Optional[str] = None,
        auto_error: bool = True
    ):
        self.model = OpenIdConnectModel(

            

Reported by Pylint.

                  async def __call__(self, request: Request) -> Optional[str]:
        authorization: str = request.headers.get("Authorization")
        if not authorization:
            if self.auto_error:
                raise HTTPException(
                    status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
                )
            else:
                return None

            

Reported by Pylint.

              import pytest
from fastapi.testclient import TestClient

from docs_src.body_multiple_params.tutorial001 import app

client = TestClient(app)

openapi_schema = {
    "openapi": "3.0.2",

            

Reported by Pylint.

              import pytest
from fastapi.testclient import TestClient

from docs_src.body_multiple_params.tutorial001 import app

client = TestClient(app)

openapi_schema = {
    "openapi": "3.0.2",

            

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


item_id_not_int = {
    "detail": [

            

Reported by Bandit.

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


item_id_not_int = {
    "detail": [
        {

            

Reported by Bandit.

                      ("/items/5?q=bar", None, 200, {"item_id": 5, "q": "bar"}),
        ("/items/5", None, 200, {"item_id": 5}),
        ("/items/foo", None, 422, item_id_not_int),
    ],
)
def test_post_body(path, body, expected_status, expected_response):
    response = client.put(path, json=body)
    assert response.status_code == expected_status
    assert response.json() == expected_response

            

Reported by Pylint.

              )
def test_post_body(path, body, expected_status, expected_response):
    response = client.put(path, json=body)
    assert response.status_code == expected_status
    assert response.json() == expected_response

            

Reported by Bandit.

              def test_post_body(path, body, expected_status, expected_response):
    response = client.put(path, json=body)
    assert response.status_code == expected_status
    assert response.json() == expected_response

            

Reported by Bandit.

              from fastapi import FastAPI, HTTPException
from fastapi.exception_handlers import (
    http_exception_handler,
    request_validation_exception_handler,
)
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

            

Reported by Pylint.

              from fastapi import FastAPI, HTTPException
from fastapi.exception_handlers import (
    http_exception_handler,
    request_validation_exception_handler,
)
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

            

Reported by Pylint.

                  http_exception_handler,
    request_validation_exception_handler,
)
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()



            

Reported by Pylint.

                  request_validation_exception_handler,
)
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()


@app.exception_handler(StarletteHTTPException)

            

Reported by Pylint.

              from fastapi import FastAPI, HTTPException
from fastapi.exception_handlers import (
    http_exception_handler,
    request_validation_exception_handler,
)
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

            

Reported by Pylint.

              

@app.exception_handler(StarletteHTTPException)
async def custom_http_exception_handler(request, exc):
    print(f"OMG! An HTTP error!: {repr(exc)}")
    return await http_exception_handler(request, exc)


@app.exception_handler(RequestValidationError)

            

Reported by Pylint.

              

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    print(f"OMG! The client sent invalid data!: {exc}")
    return await request_validation_exception_handler(request, exc)


@app.get("/items/{item_id}")

            

Reported by Pylint.

              

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}

            

Reported by Pylint.

              from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name



            

Reported by Pylint.

              from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name



            

Reported by Pylint.

              

class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name


app = FastAPI()


            

Reported by Pylint.

              

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
    )


            

Reported by Pylint.

              from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse


class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name



            

Reported by Pylint.

              from fastapi.responses import JSONResponse


class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name


app = FastAPI()

            

Reported by Pylint.

              

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
    )


            

Reported by Pylint.

              

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

            

Reported by Pylint.

              from typing import List, Optional

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()



            

Reported by Pylint.

              from typing import List, Optional

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()



            

Reported by Pylint.

              
from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):

            

Reported by Pylint.

              from typing import List, Optional

from fastapi import FastAPI
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

app = FastAPI()



            

Reported by Pylint.

              app = FastAPI()


class Item(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    tax: float = 10.5
    tags: List[str] = []

            

Reported by Pylint.

              app = FastAPI()


class Item(BaseModel):
    name: Optional[str] = None
    description: Optional[str] = None
    price: Optional[float] = None
    tax: float = 10.5
    tags: List[str] = []

            

Reported by Pylint.

              

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: str):
    return items[item_id]


@app.patch("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):

            

Reported by Pylint.

              

@app.patch("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
    stored_item_data = items[item_id]
    stored_item_model = Item(**stored_item_data)
    update_data = item.dict(exclude_unset=True)
    updated_item = stored_item_model.copy(update=update_data)
    items[item_id] = jsonable_encoder(updated_item)

            

Reported by Pylint.