📝 Update all docs to use Annotated as the main recommendation, with new examples and tests (#9268)

* 🍱 Add new source examples with Annotated for Query Params and String Validations

* 📝 Add new docs with Annotated for Query Params and String Validations

* 🚚 Rename incorrectly named tests for Query Params and str validations

*  Add new tests with Annotated for Query Params and Sring Validations examples

* 🍱 Add new examples with Annotated for Intro to Python Types

* 📝 Update Python Types Intro, include Annotated

* 🎨 Fix formatting in Query params and string validation, and highlight

* 🍱 Add new Annotated source examples for Path Params and Numeric Validations

* 📝 Update docs for Path Params and Numeric Validations with Annotated

* 🍱 Add new source examples with Annotated for Body - Multiple Params

* 📝 Update docs with Annotated for Body - Multiple Parameters

*  Add test for new Annotated examples in Body - Multiple Parameters

* 🍱 Add new Annotated source examples for Body Fields

* 📝 Update docs for Body Fields with new Annotated examples

*  Add new tests for new Annotated examples for Body Fields

* 🍱 Add new Annotated source examples for Schema Extra (Example Data)

* 📝 Update docs for Schema Extra with Annotated

*  Add tests for new Annotated examples for Schema Extra

* 🍱 Add new Annnotated source examples for Extra Data Types

* 📝 Update docs with Annotated for Extra Data Types

*  Add tests for new Annotated examples for Extra Data Types

* 🍱 Add new Annotated source examples for Cookie Parameters

* 📝 Update docs for Cookie Parameters with Annotated examples

*  Add tests for new Annotated source examples in Cookie Parameters

* 🍱 Add new Annotated examples for Header Params

* 📝 Update docs with Annotated examples for Header Parameters

*  Add tests for new Annotated examples for Header Params

* 🍱 Add new Annotated examples for Form Data

* 📝 Update Annotated docs for Form Data

*  Add tests for new Annotated examples in Form Data

* 🍱 Add new Annotated source examples for Request Files

* 📝 Update Annotated docs for Request Files

*  Test new Annotated examples for Request Files

* 🍱 Add new Annotated source examples for Request Forms and Files

*  Add tests for new Anotated examples for Request Forms and Files

* 🍱 Add new Annotated source examples for Dependencies and Advanced Dependencies

*  Add tests for new Annotated dependencies

* 📝 Add new docs for using Annotated with dependencies including type aliases

* 📝 Update docs for Classes as Dependencies with Annotated

* 📝 Update docs for Sub-dependencies with Annotated

* 📝 Update docs for Dependencies in path operation decorators with Annotated

* 📝 Update docs for Global Dependencies with Annotated

* 📝 Update docs for Dependencies with yield with Annotated

* 🎨 Update format in example for dependencies with Annotated

* 🍱 Add source examples with Annotated for Security

*  Add tests for new Annotated examples for security

* 📝 Update docs for Security - First Steps with Annotated

* 📝 Update docs for Security: Get Current User with Annotated

* 📝 Update docs for Simple OAuth2 with Password and Bearer with Annotated

* 📝 Update docs for OAuth2 with Password (and hashing), Bearer with JWT tokens with Annotated

* 📝 Update docs for Request Forms and Files with Annotated

* 🍱 Add new source examples for Bigger Applications with Annotated

*  Add new tests for Bigger Applications with Annotated

* 📝 Update docs for Bigger Applications - Multiple Files with Annotated

* 🍱 Add source examples for background tasks with Annotated

* 📝 Update docs for Background Tasks with Annotated

*  Add test for Background Tasks with Anotated

* 🍱 Add new source examples for docs for Testing with Annotated

* 📝 Update docs for Testing with Annotated

*  Add tests for Annotated examples for Testing

* 🍱 Add new source examples for Additional Status Codes with Annotated

*  Add tests for new Annotated examples for Additional Status Codes

* 📝 Update docs for Additional Status Codes with Annotated

* 📝 Update docs for Advanced Dependencies with Annotated

* 📝 Update docs for OAuth2 scopes with Annotated

* 📝 Update docs for HTTP Basic Auth with Annotated

* 🍱 Add source examples with Annotated for WebSockets

*  Add tests for new Annotated examples for WebSockets

* 📝 Update docs for WebSockets with new Annotated examples

* 🍱 Add source examples with Annotated for Settings and Environment Variables

* 📝 Update docs for Settings and Environment Variables with Annotated

* 🍱 Add new source examples for testing dependencies with Annotated

*  Add tests for new examples for testing dependencies

* 📝 Update docs for testing dependencies with new Annotated examples

*  Update and fix marker for Python 3.9 test

* 🔧 Update Ruff ignores for source examples in docs

*  Fix some tests in the grid for Python 3.9 (incorrectly testing 3.10)

* 🔥 Remove source examples and tests for (non existent) docs section about Annotated, as it's covered in all the rest of the docs
This commit is contained in:
Sebastián Ramírez
2023-03-18 13:29:59 +01:00
committed by GitHub
parent f63b3ad53e
commit 9eaed2eb37
347 changed files with 21815 additions and 589 deletions

View File

@@ -0,0 +1,26 @@
from typing import Union
from fastapi import Body, FastAPI, status
from fastapi.responses import JSONResponse
from typing_extensions import Annotated
app = FastAPI()
items = {"foo": {"name": "Fighters", "size": 6}, "bar": {"name": "Tenders", "size": 3}}
@app.put("/items/{item_id}")
async def upsert_item(
item_id: str,
name: Annotated[Union[str, None], Body()] = None,
size: Annotated[Union[int, None], Body()] = None,
):
if item_id in items:
item = items[item_id]
item["name"] = name
item["size"] = size
return item
else:
item = {"name": name, "size": size}
items[item_id] = item
return JSONResponse(status_code=status.HTTP_201_CREATED, content=item)

View File

@@ -0,0 +1,25 @@
from typing import Annotated
from fastapi import Body, FastAPI, status
from fastapi.responses import JSONResponse
app = FastAPI()
items = {"foo": {"name": "Fighters", "size": 6}, "bar": {"name": "Tenders", "size": 3}}
@app.put("/items/{item_id}")
async def upsert_item(
item_id: str,
name: Annotated[str | None, Body()] = None,
size: Annotated[int | None, Body()] = None,
):
if item_id in items:
item = items[item_id]
item["name"] = name
item["size"] = size
return item
else:
item = {"name": name, "size": size}
items[item_id] = item
return JSONResponse(status_code=status.HTTP_201_CREATED, content=item)

View File

@@ -0,0 +1,25 @@
from typing import Annotated, Union
from fastapi import Body, FastAPI, status
from fastapi.responses import JSONResponse
app = FastAPI()
items = {"foo": {"name": "Fighters", "size": 6}, "bar": {"name": "Tenders", "size": 3}}
@app.put("/items/{item_id}")
async def upsert_item(
item_id: str,
name: Annotated[Union[str, None], Body()] = None,
size: Annotated[Union[int, None], Body()] = None,
):
if item_id in items:
item = items[item_id]
item["name"] = name
item["size"] = size
return item
else:
item = {"name": name, "size": size}
items[item_id] = item
return JSONResponse(status_code=status.HTTP_201_CREATED, content=item)

View File

@@ -0,0 +1,23 @@
from fastapi import Body, FastAPI, status
from fastapi.responses import JSONResponse
app = FastAPI()
items = {"foo": {"name": "Fighters", "size": 6}, "bar": {"name": "Tenders", "size": 3}}
@app.put("/items/{item_id}")
async def upsert_item(
item_id: str,
name: str | None = Body(default=None),
size: int | None = Body(default=None),
):
if item_id in items:
item = items[item_id]
item["name"] = name
item["size"] = size
return item
else:
item = {"name": name, "size": size}
items[item_id] = item
return JSONResponse(status_code=status.HTTP_201_CREATED, content=item)

View File

@@ -1,18 +0,0 @@
from typing import Optional
from fastapi import Depends, FastAPI
from typing_extensions import Annotated
app = FastAPI()
async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
CommonParamsDepends = Annotated[dict, Depends(common_parameters)]
@app.get("/items/")
async def read_items(commons: CommonParamsDepends):
return commons

View File

@@ -1,17 +0,0 @@
from typing import Annotated, Optional
from fastapi import Depends, FastAPI
app = FastAPI()
async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
CommonParamsDepends = Annotated[dict, Depends(common_parameters)]
@app.get("/items/")
async def read_items(commons: CommonParamsDepends):
return commons

View File

@@ -1,21 +0,0 @@
from typing import Optional
from fastapi import Depends, FastAPI
from typing_extensions import Annotated
app = FastAPI()
class CommonQueryParams:
def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
CommonQueryParamsDepends = Annotated[CommonQueryParams, Depends()]
@app.get("/items/")
async def read_items(commons: CommonQueryParamsDepends):
return commons

View File

@@ -1,20 +0,0 @@
from typing import Annotated, Optional
from fastapi import Depends, FastAPI
app = FastAPI()
class CommonQueryParams:
def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
CommonQueryParamsDepends = Annotated[CommonQueryParams, Depends()]
@app.get("/items/")
async def read_items(commons: CommonQueryParamsDepends):
return commons

View File

@@ -1,15 +0,0 @@
from fastapi import FastAPI, Path
from fastapi.param_functions import Query
from typing_extensions import Annotated
app = FastAPI()
@app.get("/items/{item_id}")
async def read_items(item_id: Annotated[int, Path(gt=0)]):
return {"item_id": item_id}
@app.get("/users")
async def read_users(user_id: Annotated[str, Query(min_length=1)] = "me"):
return {"user_id": user_id}

View File

@@ -1,16 +0,0 @@
from typing import Annotated
from fastapi import FastAPI, Path
from fastapi.param_functions import Query
app = FastAPI()
@app.get("/items/{item_id}")
async def read_items(item_id: Annotated[int, Path(gt=0)]):
return {"item_id": item_id}
@app.get("/users")
async def read_users(user_id: Annotated[str, Query(min_length=1)] = "me"):
return {"user_id": user_id}

View File

View File

@@ -0,0 +1,39 @@
from typing import Union
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel
from typing_extensions import Annotated
fake_secret_token = "coneofsilence"
fake_db = {
"foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
"bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}
app = FastAPI()
class Item(BaseModel):
id: str
title: str
description: Union[str, None] = None
@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: Annotated[str, Header()]):
if x_token != fake_secret_token:
raise HTTPException(status_code=400, detail="Invalid X-Token header")
if item_id not in fake_db:
raise HTTPException(status_code=404, detail="Item not found")
return fake_db[item_id]
@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: Annotated[str, Header()]):
if x_token != fake_secret_token:
raise HTTPException(status_code=400, detail="Invalid X-Token header")
if item.id in fake_db:
raise HTTPException(status_code=400, detail="Item already exists")
fake_db[item.id] = item
return item

View File

@@ -0,0 +1,65 @@
from fastapi.testclient import TestClient
from .main import app
client = TestClient(app)
def test_read_item():
response = client.get("/items/foo", headers={"X-Token": "coneofsilence"})
assert response.status_code == 200
assert response.json() == {
"id": "foo",
"title": "Foo",
"description": "There goes my hero",
}
def test_read_item_bad_token():
response = client.get("/items/foo", headers={"X-Token": "hailhydra"})
assert response.status_code == 400
assert response.json() == {"detail": "Invalid X-Token header"}
def test_read_inexistent_item():
response = client.get("/items/baz", headers={"X-Token": "coneofsilence"})
assert response.status_code == 404
assert response.json() == {"detail": "Item not found"}
def test_create_item():
response = client.post(
"/items/",
headers={"X-Token": "coneofsilence"},
json={"id": "foobar", "title": "Foo Bar", "description": "The Foo Barters"},
)
assert response.status_code == 200
assert response.json() == {
"id": "foobar",
"title": "Foo Bar",
"description": "The Foo Barters",
}
def test_create_item_bad_token():
response = client.post(
"/items/",
headers={"X-Token": "hailhydra"},
json={"id": "bazz", "title": "Bazz", "description": "Drop the bazz"},
)
assert response.status_code == 400
assert response.json() == {"detail": "Invalid X-Token header"}
def test_create_existing_item():
response = client.post(
"/items/",
headers={"X-Token": "coneofsilence"},
json={
"id": "foo",
"title": "The Foo ID Stealers",
"description": "There goes my stealer",
},
)
assert response.status_code == 400
assert response.json() == {"detail": "Item already exists"}

View File

View File

@@ -0,0 +1,38 @@
from typing import Annotated
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel
fake_secret_token = "coneofsilence"
fake_db = {
"foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
"bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}
app = FastAPI()
class Item(BaseModel):
id: str
title: str
description: str | None = None
@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: Annotated[str, Header()]):
if x_token != fake_secret_token:
raise HTTPException(status_code=400, detail="Invalid X-Token header")
if item_id not in fake_db:
raise HTTPException(status_code=404, detail="Item not found")
return fake_db[item_id]
@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: Annotated[str, Header()]):
if x_token != fake_secret_token:
raise HTTPException(status_code=400, detail="Invalid X-Token header")
if item.id in fake_db:
raise HTTPException(status_code=400, detail="Item already exists")
fake_db[item.id] = item
return item

View File

@@ -0,0 +1,65 @@
from fastapi.testclient import TestClient
from .main import app
client = TestClient(app)
def test_read_item():
response = client.get("/items/foo", headers={"X-Token": "coneofsilence"})
assert response.status_code == 200
assert response.json() == {
"id": "foo",
"title": "Foo",
"description": "There goes my hero",
}
def test_read_item_bad_token():
response = client.get("/items/foo", headers={"X-Token": "hailhydra"})
assert response.status_code == 400
assert response.json() == {"detail": "Invalid X-Token header"}
def test_read_inexistent_item():
response = client.get("/items/baz", headers={"X-Token": "coneofsilence"})
assert response.status_code == 404
assert response.json() == {"detail": "Item not found"}
def test_create_item():
response = client.post(
"/items/",
headers={"X-Token": "coneofsilence"},
json={"id": "foobar", "title": "Foo Bar", "description": "The Foo Barters"},
)
assert response.status_code == 200
assert response.json() == {
"id": "foobar",
"title": "Foo Bar",
"description": "The Foo Barters",
}
def test_create_item_bad_token():
response = client.post(
"/items/",
headers={"X-Token": "hailhydra"},
json={"id": "bazz", "title": "Bazz", "description": "Drop the bazz"},
)
assert response.status_code == 400
assert response.json() == {"detail": "Invalid X-Token header"}
def test_create_existing_item():
response = client.post(
"/items/",
headers={"X-Token": "coneofsilence"},
json={
"id": "foo",
"title": "The Foo ID Stealers",
"description": "There goes my stealer",
},
)
assert response.status_code == 400
assert response.json() == {"detail": "Item already exists"}

View File

View File

@@ -0,0 +1,38 @@
from typing import Annotated, Union
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel
fake_secret_token = "coneofsilence"
fake_db = {
"foo": {"id": "foo", "title": "Foo", "description": "There goes my hero"},
"bar": {"id": "bar", "title": "Bar", "description": "The bartenders"},
}
app = FastAPI()
class Item(BaseModel):
id: str
title: str
description: Union[str, None] = None
@app.get("/items/{item_id}", response_model=Item)
async def read_main(item_id: str, x_token: Annotated[str, Header()]):
if x_token != fake_secret_token:
raise HTTPException(status_code=400, detail="Invalid X-Token header")
if item_id not in fake_db:
raise HTTPException(status_code=404, detail="Item not found")
return fake_db[item_id]
@app.post("/items/", response_model=Item)
async def create_item(item: Item, x_token: Annotated[str, Header()]):
if x_token != fake_secret_token:
raise HTTPException(status_code=400, detail="Invalid X-Token header")
if item.id in fake_db:
raise HTTPException(status_code=400, detail="Item already exists")
fake_db[item.id] = item
return item

View File

@@ -0,0 +1,65 @@
from fastapi.testclient import TestClient
from .main import app
client = TestClient(app)
def test_read_item():
response = client.get("/items/foo", headers={"X-Token": "coneofsilence"})
assert response.status_code == 200
assert response.json() == {
"id": "foo",
"title": "Foo",
"description": "There goes my hero",
}
def test_read_item_bad_token():
response = client.get("/items/foo", headers={"X-Token": "hailhydra"})
assert response.status_code == 400
assert response.json() == {"detail": "Invalid X-Token header"}
def test_read_inexistent_item():
response = client.get("/items/baz", headers={"X-Token": "coneofsilence"})
assert response.status_code == 404
assert response.json() == {"detail": "Item not found"}
def test_create_item():
response = client.post(
"/items/",
headers={"X-Token": "coneofsilence"},
json={"id": "foobar", "title": "Foo Bar", "description": "The Foo Barters"},
)
assert response.status_code == 200
assert response.json() == {
"id": "foobar",
"title": "Foo Bar",
"description": "The Foo Barters",
}
def test_create_item_bad_token():
response = client.post(
"/items/",
headers={"X-Token": "hailhydra"},
json={"id": "bazz", "title": "Bazz", "description": "Drop the bazz"},
)
assert response.status_code == 400
assert response.json() == {"detail": "Invalid X-Token header"}
def test_create_existing_item():
response = client.post(
"/items/",
headers={"X-Token": "coneofsilence"},
json={
"id": "foo",
"title": "The Foo ID Stealers",
"description": "There goes my stealer",
},
)
assert response.status_code == 400
assert response.json() == {"detail": "Item already exists"}

View File

@@ -0,0 +1,27 @@
from typing import Union
from fastapi import BackgroundTasks, Depends, FastAPI
from typing_extensions import Annotated
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
def get_query(background_tasks: BackgroundTasks, q: Union[str, None] = None):
if q:
message = f"found query: {q}\n"
background_tasks.add_task(write_log, message)
return q
@app.post("/send-notification/{email}")
async def send_notification(
email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):
message = f"message to {email}\n"
background_tasks.add_task(write_log, message)
return {"message": "Message sent"}

View File

@@ -0,0 +1,26 @@
from typing import Annotated
from fastapi import BackgroundTasks, Depends, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
def get_query(background_tasks: BackgroundTasks, q: str | None = None):
if q:
message = f"found query: {q}\n"
background_tasks.add_task(write_log, message)
return q
@app.post("/send-notification/{email}")
async def send_notification(
email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):
message = f"message to {email}\n"
background_tasks.add_task(write_log, message)
return {"message": "Message sent"}

View File

@@ -0,0 +1,26 @@
from typing import Annotated, Union
from fastapi import BackgroundTasks, Depends, FastAPI
app = FastAPI()
def write_log(message: str):
with open("log.txt", mode="a") as log:
log.write(message)
def get_query(background_tasks: BackgroundTasks, q: Union[str, None] = None):
if q:
message = f"found query: {q}\n"
background_tasks.add_task(write_log, message)
return q
@app.post("/send-notification/{email}")
async def send_notification(
email: str, background_tasks: BackgroundTasks, q: Annotated[str, Depends(get_query)]
):
message = f"message to {email}\n"
background_tasks.add_task(write_log, message)
return {"message": "Message sent"}

View File

View File

@@ -0,0 +1,12 @@
from fastapi import Header, HTTPException
from typing_extensions import Annotated
async def get_token_header(x_token: Annotated[str, Header()]):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="X-Token header invalid")
async def get_query_token(token: str):
if token != "jessica":
raise HTTPException(status_code=400, detail="No Jessica token provided")

View File

View File

@@ -0,0 +1,8 @@
from fastapi import APIRouter
router = APIRouter()
@router.post("/")
async def update_admin():
return {"message": "Admin getting schwifty"}

View File

@@ -0,0 +1,23 @@
from fastapi import Depends, FastAPI
from .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, users
app = FastAPI(dependencies=[Depends(get_query_token)])
app.include_router(users.router)
app.include_router(items.router)
app.include_router(
admin.router,
prefix="/admin",
tags=["admin"],
dependencies=[Depends(get_token_header)],
responses={418: {"description": "I'm a teapot"}},
)
@app.get("/")
async def root():
return {"message": "Hello Bigger Applications!"}

View File

View File

@@ -0,0 +1,38 @@
from fastapi import APIRouter, Depends, HTTPException
from ..dependencies import get_token_header
router = APIRouter(
prefix="/items",
tags=["items"],
dependencies=[Depends(get_token_header)],
responses={404: {"description": "Not found"}},
)
fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}
@router.get("/")
async def read_items():
return fake_items_db
@router.get("/{item_id}")
async def read_item(item_id: str):
if item_id not in fake_items_db:
raise HTTPException(status_code=404, detail="Item not found")
return {"name": fake_items_db[item_id]["name"], "item_id": item_id}
@router.put(
"/{item_id}",
tags=["custom"],
responses={403: {"description": "Operation forbidden"}},
)
async def update_item(item_id: str):
if item_id != "plumbus":
raise HTTPException(
status_code=403, detail="You can only update the item: plumbus"
)
return {"item_id": item_id, "name": "The great Plumbus"}

View File

@@ -0,0 +1,18 @@
from fastapi import APIRouter
router = APIRouter()
@router.get("/users/", tags=["users"])
async def read_users():
return [{"username": "Rick"}, {"username": "Morty"}]
@router.get("/users/me", tags=["users"])
async def read_user_me():
return {"username": "fakecurrentuser"}
@router.get("/users/{username}", tags=["users"])
async def read_user(username: str):
return {"username": username}

View File

View File

@@ -0,0 +1,13 @@
from typing import Annotated
from fastapi import Header, HTTPException
async def get_token_header(x_token: Annotated[str, Header()]):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="X-Token header invalid")
async def get_query_token(token: str):
if token != "jessica":
raise HTTPException(status_code=400, detail="No Jessica token provided")

View File

View File

@@ -0,0 +1,8 @@
from fastapi import APIRouter
router = APIRouter()
@router.post("/")
async def update_admin():
return {"message": "Admin getting schwifty"}

View File

@@ -0,0 +1,23 @@
from fastapi import Depends, FastAPI
from .dependencies import get_query_token, get_token_header
from .internal import admin
from .routers import items, users
app = FastAPI(dependencies=[Depends(get_query_token)])
app.include_router(users.router)
app.include_router(items.router)
app.include_router(
admin.router,
prefix="/admin",
tags=["admin"],
dependencies=[Depends(get_token_header)],
responses={418: {"description": "I'm a teapot"}},
)
@app.get("/")
async def root():
return {"message": "Hello Bigger Applications!"}

View File

View File

@@ -0,0 +1,38 @@
from fastapi import APIRouter, Depends, HTTPException
from ..dependencies import get_token_header
router = APIRouter(
prefix="/items",
tags=["items"],
dependencies=[Depends(get_token_header)],
responses={404: {"description": "Not found"}},
)
fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}}
@router.get("/")
async def read_items():
return fake_items_db
@router.get("/{item_id}")
async def read_item(item_id: str):
if item_id not in fake_items_db:
raise HTTPException(status_code=404, detail="Item not found")
return {"name": fake_items_db[item_id]["name"], "item_id": item_id}
@router.put(
"/{item_id}",
tags=["custom"],
responses={403: {"description": "Operation forbidden"}},
)
async def update_item(item_id: str):
if item_id != "plumbus":
raise HTTPException(
status_code=403, detail="You can only update the item: plumbus"
)
return {"item_id": item_id, "name": "The great Plumbus"}

View File

@@ -0,0 +1,18 @@
from fastapi import APIRouter
router = APIRouter()
@router.get("/users/", tags=["users"])
async def read_users():
return [{"username": "Rick"}, {"username": "Morty"}]
@router.get("/users/me", tags=["users"])
async def read_user_me():
return {"username": "fakecurrentuser"}
@router.get("/users/{username}", tags=["users"])
async def read_user(username: str):
return {"username": username}

View File

@@ -0,0 +1,22 @@
from typing import Union
from fastapi import Body, FastAPI
from pydantic import BaseModel, Field
from typing_extensions import Annotated
app = FastAPI()
class Item(BaseModel):
name: str
description: Union[str, None] = Field(
default=None, title="The description of the item", max_length=300
)
price: float = Field(gt=0, description="The price must be greater than zero")
tax: Union[float, None] = None
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
results = {"item_id": item_id, "item": item}
return results

View File

@@ -0,0 +1,21 @@
from typing import Annotated
from fastapi import Body, FastAPI
from pydantic import BaseModel, Field
app = FastAPI()
class Item(BaseModel):
name: str
description: str | None = Field(
default=None, title="The description of the item", max_length=300
)
price: float = Field(gt=0, description="The price must be greater than zero")
tax: float | None = None
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
results = {"item_id": item_id, "item": item}
return results

View File

@@ -0,0 +1,21 @@
from typing import Annotated, Union
from fastapi import Body, FastAPI
from pydantic import BaseModel, Field
app = FastAPI()
class Item(BaseModel):
name: str
description: Union[str, None] = Field(
default=None, title="The description of the item", max_length=300
)
price: float = Field(gt=0, description="The price must be greater than zero")
tax: Union[float, None] = None
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
results = {"item_id": item_id, "item": item}
return results

View File

@@ -0,0 +1,28 @@
from typing import Union
from fastapi import FastAPI, Path
from pydantic import BaseModel
from typing_extensions import Annotated
app = FastAPI()
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
@app.put("/items/{item_id}")
async def update_item(
item_id: Annotated[int, Path(title="The ID of the item to get", ge=0, le=1000)],
q: Union[str, None] = None,
item: Union[Item, None] = None,
):
results = {"item_id": item_id}
if q:
results.update({"q": q})
if item:
results.update({"item": item})
return results

View File

@@ -0,0 +1,27 @@
from typing import Annotated
from fastapi import FastAPI, Path
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: str | None = None
price: float
tax: float | None = None
@app.put("/items/{item_id}")
async def update_item(
item_id: Annotated[int, Path(title="The ID of the item to get", ge=0, le=1000)],
q: str | None = None,
item: Item | None = None,
):
results = {"item_id": item_id}
if q:
results.update({"q": q})
if item:
results.update({"item": item})
return results

View File

@@ -0,0 +1,27 @@
from typing import Annotated, Union
from fastapi import FastAPI, Path
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
@app.put("/items/{item_id}")
async def update_item(
item_id: Annotated[int, Path(title="The ID of the item to get", ge=0, le=1000)],
q: Union[str, None] = None,
item: Union[Item, None] = None,
):
results = {"item_id": item_id}
if q:
results.update({"q": q})
if item:
results.update({"item": item})
return results

View File

@@ -0,0 +1,27 @@
from typing import Union
from fastapi import Body, FastAPI
from pydantic import BaseModel
from typing_extensions import Annotated
app = FastAPI()
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
class User(BaseModel):
username: str
full_name: Union[str, None] = None
@app.put("/items/{item_id}")
async def update_item(
item_id: int, item: Item, user: User, importance: Annotated[int, Body()]
):
results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
return results

View File

@@ -0,0 +1,26 @@
from typing import Annotated
from fastapi import Body, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: str | None = None
price: float
tax: float | None = None
class User(BaseModel):
username: str
full_name: str | None = None
@app.put("/items/{item_id}")
async def update_item(
item_id: int, item: Item, user: User, importance: Annotated[int, Body()]
):
results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
return results

View File

@@ -0,0 +1,26 @@
from typing import Annotated, Union
from fastapi import Body, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
class User(BaseModel):
username: str
full_name: Union[str, None] = None
@app.put("/items/{item_id}")
async def update_item(
item_id: int, item: Item, user: User, importance: Annotated[int, Body()]
):
results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
return results

View File

@@ -0,0 +1,34 @@
from typing import Union
from fastapi import Body, FastAPI
from pydantic import BaseModel
from typing_extensions import Annotated
app = FastAPI()
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
class User(BaseModel):
username: str
full_name: Union[str, None] = None
@app.put("/items/{item_id}")
async def update_item(
*,
item_id: int,
item: Item,
user: User,
importance: Annotated[int, Body(gt=0)],
q: Union[str, None] = None,
):
results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
if q:
results.update({"q": q})
return results

View File

@@ -0,0 +1,33 @@
from typing import Annotated
from fastapi import Body, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: str | None = None
price: float
tax: float | None = None
class User(BaseModel):
username: str
full_name: str | None = None
@app.put("/items/{item_id}")
async def update_item(
*,
item_id: int,
item: Item,
user: User,
importance: Annotated[int, Body(gt=0)],
q: str | None = None,
):
results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
if q:
results.update({"q": q})
return results

View File

@@ -0,0 +1,33 @@
from typing import Annotated, Union
from fastapi import Body, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
class User(BaseModel):
username: str
full_name: Union[str, None] = None
@app.put("/items/{item_id}")
async def update_item(
*,
item_id: int,
item: Item,
user: User,
importance: Annotated[int, Body(gt=0)],
q: Union[str, None] = None,
):
results = {"item_id": item_id, "item": item, "user": user, "importance": importance}
if q:
results.update({"q": q})
return results

View File

@@ -0,0 +1,20 @@
from typing import Union
from fastapi import Body, FastAPI
from pydantic import BaseModel
from typing_extensions import Annotated
app = FastAPI()
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
results = {"item_id": item_id, "item": item}
return results

View File

@@ -0,0 +1,19 @@
from typing import Annotated
from fastapi import Body, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: str | None = None
price: float
tax: float | None = None
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
results = {"item_id": item_id, "item": item}
return results

View File

@@ -0,0 +1,19 @@
from typing import Annotated, Union
from fastapi import Body, FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
description: Union[str, None] = None
price: float
tax: Union[float, None] = None
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Annotated[Item, Body(embed=True)]):
results = {"item_id": item_id, "item": item}
return results

View File

@@ -0,0 +1,11 @@
from typing import Union
from fastapi import Cookie, FastAPI
from typing_extensions import Annotated
app = FastAPI()
@app.get("/items/")
async def read_items(ads_id: Annotated[Union[str, None], Cookie()] = None):
return {"ads_id": ads_id}

View File

@@ -0,0 +1,10 @@
from typing import Annotated
from fastapi import Cookie, FastAPI
app = FastAPI()
@app.get("/items/")
async def read_items(ads_id: Annotated[str | None, Cookie()] = None):
return {"ads_id": ads_id}

View File

@@ -0,0 +1,10 @@
from typing import Annotated, Union
from fastapi import Cookie, FastAPI
app = FastAPI()
@app.get("/items/")
async def read_items(ads_id: Annotated[Union[str, None], Cookie()] = None):
return {"ads_id": ads_id}

View File

@@ -0,0 +1,25 @@
from typing import Union
from fastapi import Depends, FastAPI
from typing_extensions import Annotated
app = FastAPI()
async def common_parameters(
q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
return {"q": q, "skip": skip, "limit": limit}
CommonsDep = Annotated[dict, Depends(common_parameters)]
@app.get("/items/")
async def read_items(commons: CommonsDep):
return commons
@app.get("/users/")
async def read_users(commons: CommonsDep):
return commons

View File

@@ -0,0 +1,22 @@
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
CommonsDep = Annotated[dict, Depends(common_parameters)]
@app.get("/items/")
async def read_items(commons: CommonsDep):
return commons
@app.get("/users/")
async def read_users(commons: CommonsDep):
return commons

View File

@@ -0,0 +1,24 @@
from typing import Annotated, Union
from fastapi import Depends, FastAPI
app = FastAPI()
async def common_parameters(
q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
return {"q": q, "skip": skip, "limit": limit}
CommonsDep = Annotated[dict, Depends(common_parameters)]
@app.get("/items/")
async def read_items(commons: CommonsDep):
return commons
@app.get("/users/")
async def read_users(commons: CommonsDep):
return commons

View File

@@ -0,0 +1,22 @@
from typing import Union
from fastapi import Depends, FastAPI
from typing_extensions import Annotated
app = FastAPI()
async def common_parameters(
q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
return commons
@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
return commons

View File

@@ -0,0 +1,19 @@
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
return commons
@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
return commons

View File

@@ -0,0 +1,21 @@
from typing import Annotated, Union
from fastapi import Depends, FastAPI
app = FastAPI()
async def common_parameters(
q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
return commons
@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
return commons

View File

@@ -0,0 +1,26 @@
from typing import Union
from fastapi import Depends, FastAPI
from typing_extensions import Annotated
app = FastAPI()
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
class CommonQueryParams:
def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]):
response = {}
if commons.q:
response.update({"q": commons.q})
items = fake_items_db[commons.skip : commons.skip + commons.limit]
response.update({"items": items})
return response

View File

@@ -0,0 +1,25 @@
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
class CommonQueryParams:
def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]):
response = {}
if commons.q:
response.update({"q": commons.q})
items = fake_items_db[commons.skip : commons.skip + commons.limit]
response.update({"items": items})
return response

View File

@@ -0,0 +1,25 @@
from typing import Annotated, Union
from fastapi import Depends, FastAPI
app = FastAPI()
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
class CommonQueryParams:
def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends(CommonQueryParams)]):
response = {}
if commons.q:
response.update({"q": commons.q})
items = fake_items_db[commons.skip : commons.skip + commons.limit]
response.update({"items": items})
return response

View File

@@ -0,0 +1,26 @@
from typing import Any, Union
from fastapi import Depends, FastAPI
from typing_extensions import Annotated
app = FastAPI()
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
class CommonQueryParams:
def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(commons: Annotated[Any, Depends(CommonQueryParams)]):
response = {}
if commons.q:
response.update({"q": commons.q})
items = fake_items_db[commons.skip : commons.skip + commons.limit]
response.update({"items": items})
return response

View File

@@ -0,0 +1,25 @@
from typing import Annotated, Any
from fastapi import Depends, FastAPI
app = FastAPI()
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
class CommonQueryParams:
def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(commons: Annotated[Any, Depends(CommonQueryParams)]):
response = {}
if commons.q:
response.update({"q": commons.q})
items = fake_items_db[commons.skip : commons.skip + commons.limit]
response.update({"items": items})
return response

View File

@@ -0,0 +1,25 @@
from typing import Annotated, Any, Union
from fastapi import Depends, FastAPI
app = FastAPI()
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
class CommonQueryParams:
def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(commons: Annotated[Any, Depends(CommonQueryParams)]):
response = {}
if commons.q:
response.update({"q": commons.q})
items = fake_items_db[commons.skip : commons.skip + commons.limit]
response.update({"items": items})
return response

View File

@@ -0,0 +1,26 @@
from typing import Union
from fastapi import Depends, FastAPI
from typing_extensions import Annotated
app = FastAPI()
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
class CommonQueryParams:
def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends()]):
response = {}
if commons.q:
response.update({"q": commons.q})
items = fake_items_db[commons.skip : commons.skip + commons.limit]
response.update({"items": items})
return response

View File

@@ -0,0 +1,25 @@
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
class CommonQueryParams:
def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends()]):
response = {}
if commons.q:
response.update({"q": commons.q})
items = fake_items_db[commons.skip : commons.skip + commons.limit]
response.update({"items": items})
return response

View File

@@ -0,0 +1,25 @@
from typing import Annotated, Union
from fastapi import Depends, FastAPI
app = FastAPI()
fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
class CommonQueryParams:
def __init__(self, q: Union[str, None] = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(commons: Annotated[CommonQueryParams, Depends()]):
response = {}
if commons.q:
response.update({"q": commons.q})
items = fake_items_db[commons.skip : commons.skip + commons.limit]
response.update({"items": items})
return response

View File

@@ -0,0 +1,26 @@
from typing import Union
from fastapi import Cookie, Depends, FastAPI
from typing_extensions import Annotated
app = FastAPI()
def query_extractor(q: Union[str, None] = None):
return q
def query_or_cookie_extractor(
q: Annotated[str, Depends(query_extractor)],
last_query: Annotated[Union[str, None], Cookie()] = None,
):
if not q:
return last_query
return q
@app.get("/items/")
async def read_query(
query_or_default: Annotated[str, Depends(query_or_cookie_extractor)]
):
return {"q_or_cookie": query_or_default}

View File

@@ -0,0 +1,25 @@
from typing import Annotated
from fastapi import Cookie, Depends, FastAPI
app = FastAPI()
def query_extractor(q: str | None = None):
return q
def query_or_cookie_extractor(
q: Annotated[str, Depends(query_extractor)],
last_query: Annotated[str | None, Cookie()] = None,
):
if not q:
return last_query
return q
@app.get("/items/")
async def read_query(
query_or_default: Annotated[str, Depends(query_or_cookie_extractor)]
):
return {"q_or_cookie": query_or_default}

View File

@@ -0,0 +1,25 @@
from typing import Annotated, Union
from fastapi import Cookie, Depends, FastAPI
app = FastAPI()
def query_extractor(q: Union[str, None] = None):
return q
def query_or_cookie_extractor(
q: Annotated[str, Depends(query_extractor)],
last_query: Annotated[Union[str, None], Cookie()] = None,
):
if not q:
return last_query
return q
@app.get("/items/")
async def read_query(
query_or_default: Annotated[str, Depends(query_or_cookie_extractor)]
):
return {"q_or_cookie": query_or_default}

View File

@@ -0,0 +1,20 @@
from fastapi import Depends, FastAPI, Header, HTTPException
from typing_extensions import Annotated
app = FastAPI()
async def verify_token(x_token: Annotated[str, Header()]):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="X-Token header invalid")
async def verify_key(x_key: Annotated[str, Header()]):
if x_key != "fake-super-secret-key":
raise HTTPException(status_code=400, detail="X-Key header invalid")
return x_key
@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
return [{"item": "Foo"}, {"item": "Bar"}]

View File

@@ -0,0 +1,21 @@
from typing import Annotated
from fastapi import Depends, FastAPI, Header, HTTPException
app = FastAPI()
async def verify_token(x_token: Annotated[str, Header()]):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="X-Token header invalid")
async def verify_key(x_key: Annotated[str, Header()]):
if x_key != "fake-super-secret-key":
raise HTTPException(status_code=400, detail="X-Key header invalid")
return x_key
@app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_items():
return [{"item": "Foo"}, {"item": "Bar"}]

View File

@@ -0,0 +1,26 @@
from fastapi import Depends
from typing_extensions import Annotated
async def dependency_a():
dep_a = generate_dep_a()
try:
yield dep_a
finally:
dep_a.close()
async def dependency_b(dep_a: Annotated[DepA, Depends(dependency_a)]):
dep_b = generate_dep_b()
try:
yield dep_b
finally:
dep_b.close(dep_a)
async def dependency_c(dep_b: Annotated[DepB, Depends(dependency_b)]):
dep_c = generate_dep_c()
try:
yield dep_c
finally:
dep_c.close(dep_b)

View File

@@ -0,0 +1,27 @@
from typing import Annotated
from fastapi import Depends
async def dependency_a():
dep_a = generate_dep_a()
try:
yield dep_a
finally:
dep_a.close()
async def dependency_b(dep_a: Annotated[DepA, Depends(dependency_a)]):
dep_b = generate_dep_b()
try:
yield dep_b
finally:
dep_b.close(dep_a)
async def dependency_c(dep_b: Annotated[DepB, Depends(dependency_b)]):
dep_c = generate_dep_c()
try:
yield dep_c
finally:
dep_c.close(dep_b)

View File

@@ -0,0 +1,22 @@
from fastapi import Depends, FastAPI
from typing_extensions import Annotated
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
return {"fixed_content_in_query": fixed_content_included}

View File

@@ -0,0 +1,23 @@
from typing import Annotated
from fastapi import Depends, FastAPI
app = FastAPI()
class FixedContentQueryChecker:
def __init__(self, fixed_content: str):
self.fixed_content = fixed_content
def __call__(self, q: str = ""):
if q:
return self.fixed_content in q
return False
checker = FixedContentQueryChecker("bar")
@app.get("/query-checker/")
async def read_query_check(fixed_content_included: Annotated[bool, Depends(checker)]):
return {"fixed_content_in_query": fixed_content_included}

View File

@@ -0,0 +1,26 @@
from fastapi import Depends, FastAPI, Header, HTTPException
from typing_extensions import Annotated
async def verify_token(x_token: Annotated[str, Header()]):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="X-Token header invalid")
async def verify_key(x_key: Annotated[str, Header()]):
if x_key != "fake-super-secret-key":
raise HTTPException(status_code=400, detail="X-Key header invalid")
return x_key
app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])
@app.get("/items/")
async def read_items():
return [{"item": "Portal Gun"}, {"item": "Plumbus"}]
@app.get("/users/")
async def read_users():
return [{"username": "Rick"}, {"username": "Morty"}]

View File

@@ -0,0 +1,26 @@
from fastapi import Depends, FastAPI, Header, HTTPException
from typing_extensions import Annotated
async def verify_token(x_token: Annotated[str, Header()]):
if x_token != "fake-super-secret-token":
raise HTTPException(status_code=400, detail="X-Token header invalid")
async def verify_key(x_key: Annotated[str, Header()]):
if x_key != "fake-super-secret-key":
raise HTTPException(status_code=400, detail="X-Key header invalid")
return x_key
app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])
@app.get("/items/")
async def read_items():
return [{"item": "Portal Gun"}, {"item": "Plumbus"}]
@app.get("/users/")
async def read_users():
return [{"username": "Rick"}, {"username": "Morty"}]

View File

@@ -0,0 +1,60 @@
from typing import Union
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
from typing_extensions import Annotated
app = FastAPI()
async def common_parameters(
q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
return {"message": "Hello Items!", "params": commons}
@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
return {"message": "Hello Users!", "params": commons}
client = TestClient(app)
async def override_dependency(q: Union[str, None] = None):
return {"q": q, "skip": 5, "limit": 10}
app.dependency_overrides[common_parameters] = override_dependency
def test_override_in_items():
response = client.get("/items/")
assert response.status_code == 200
assert response.json() == {
"message": "Hello Items!",
"params": {"q": None, "skip": 5, "limit": 10},
}
def test_override_in_items_with_q():
response = client.get("/items/?q=foo")
assert response.status_code == 200
assert response.json() == {
"message": "Hello Items!",
"params": {"q": "foo", "skip": 5, "limit": 10},
}
def test_override_in_items_with_params():
response = client.get("/items/?q=foo&skip=100&limit=200")
assert response.status_code == 200
assert response.json() == {
"message": "Hello Items!",
"params": {"q": "foo", "skip": 5, "limit": 10},
}

View File

@@ -0,0 +1,57 @@
from typing import Annotated
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
return {"message": "Hello Items!", "params": commons}
@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
return {"message": "Hello Users!", "params": commons}
client = TestClient(app)
async def override_dependency(q: str | None = None):
return {"q": q, "skip": 5, "limit": 10}
app.dependency_overrides[common_parameters] = override_dependency
def test_override_in_items():
response = client.get("/items/")
assert response.status_code == 200
assert response.json() == {
"message": "Hello Items!",
"params": {"q": None, "skip": 5, "limit": 10},
}
def test_override_in_items_with_q():
response = client.get("/items/?q=foo")
assert response.status_code == 200
assert response.json() == {
"message": "Hello Items!",
"params": {"q": "foo", "skip": 5, "limit": 10},
}
def test_override_in_items_with_params():
response = client.get("/items/?q=foo&skip=100&limit=200")
assert response.status_code == 200
assert response.json() == {
"message": "Hello Items!",
"params": {"q": "foo", "skip": 5, "limit": 10},
}

View File

@@ -0,0 +1,59 @@
from typing import Annotated, Union
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
async def common_parameters(
q: Union[str, None] = None, skip: int = 0, limit: int = 100
):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
return {"message": "Hello Items!", "params": commons}
@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
return {"message": "Hello Users!", "params": commons}
client = TestClient(app)
async def override_dependency(q: Union[str, None] = None):
return {"q": q, "skip": 5, "limit": 10}
app.dependency_overrides[common_parameters] = override_dependency
def test_override_in_items():
response = client.get("/items/")
assert response.status_code == 200
assert response.json() == {
"message": "Hello Items!",
"params": {"q": None, "skip": 5, "limit": 10},
}
def test_override_in_items_with_q():
response = client.get("/items/?q=foo")
assert response.status_code == 200
assert response.json() == {
"message": "Hello Items!",
"params": {"q": "foo", "skip": 5, "limit": 10},
}
def test_override_in_items_with_params():
response = client.get("/items/?q=foo&skip=100&limit=200")
assert response.status_code == 200
assert response.json() == {
"message": "Hello Items!",
"params": {"q": "foo", "skip": 5, "limit": 10},
}

View File

@@ -0,0 +1,55 @@
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
app = FastAPI()
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
return {"message": "Hello Items!", "params": commons}
@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
return {"message": "Hello Users!", "params": commons}
client = TestClient(app)
async def override_dependency(q: str | None = None):
return {"q": q, "skip": 5, "limit": 10}
app.dependency_overrides[common_parameters] = override_dependency
def test_override_in_items():
response = client.get("/items/")
assert response.status_code == 200
assert response.json() == {
"message": "Hello Items!",
"params": {"q": None, "skip": 5, "limit": 10},
}
def test_override_in_items_with_q():
response = client.get("/items/?q=foo")
assert response.status_code == 200
assert response.json() == {
"message": "Hello Items!",
"params": {"q": "foo", "skip": 5, "limit": 10},
}
def test_override_in_items_with_params():
response = client.get("/items/?q=foo&skip=100&limit=200")
assert response.status_code == 200
assert response.json() == {
"message": "Hello Items!",
"params": {"q": "foo", "skip": 5, "limit": 10},
}

View File

@@ -0,0 +1,29 @@
from datetime import datetime, time, timedelta
from typing import Union
from uuid import UUID
from fastapi import Body, FastAPI
from typing_extensions import Annotated
app = FastAPI()
@app.put("/items/{item_id}")
async def read_items(
item_id: UUID,
start_datetime: Annotated[Union[datetime, None], Body()] = None,
end_datetime: Annotated[Union[datetime, None], Body()] = None,
repeat_at: Annotated[Union[time, None], Body()] = None,
process_after: Annotated[Union[timedelta, None], Body()] = None,
):
start_process = start_datetime + process_after
duration = end_datetime - start_process
return {
"item_id": item_id,
"start_datetime": start_datetime,
"end_datetime": end_datetime,
"repeat_at": repeat_at,
"process_after": process_after,
"start_process": start_process,
"duration": duration,
}

View File

@@ -0,0 +1,28 @@
from datetime import datetime, time, timedelta
from typing import Annotated
from uuid import UUID
from fastapi import Body, FastAPI
app = FastAPI()
@app.put("/items/{item_id}")
async def read_items(
item_id: UUID,
start_datetime: Annotated[datetime | None, Body()] = None,
end_datetime: Annotated[datetime | None, Body()] = None,
repeat_at: Annotated[time | None, Body()] = None,
process_after: Annotated[timedelta | None, Body()] = None,
):
start_process = start_datetime + process_after
duration = end_datetime - start_process
return {
"item_id": item_id,
"start_datetime": start_datetime,
"end_datetime": end_datetime,
"repeat_at": repeat_at,
"process_after": process_after,
"start_process": start_process,
"duration": duration,
}

View File

@@ -0,0 +1,28 @@
from datetime import datetime, time, timedelta
from typing import Annotated, Union
from uuid import UUID
from fastapi import Body, FastAPI
app = FastAPI()
@app.put("/items/{item_id}")
async def read_items(
item_id: UUID,
start_datetime: Annotated[Union[datetime, None], Body()] = None,
end_datetime: Annotated[Union[datetime, None], Body()] = None,
repeat_at: Annotated[Union[time, None], Body()] = None,
process_after: Annotated[Union[timedelta, None], Body()] = None,
):
start_process = start_datetime + process_after
duration = end_datetime - start_process
return {
"item_id": item_id,
"start_datetime": start_datetime,
"end_datetime": end_datetime,
"repeat_at": repeat_at,
"process_after": process_after,
"start_process": start_process,
"duration": duration,
}

View File

@@ -0,0 +1,11 @@
from typing import Union
from fastapi import FastAPI, Header
from typing_extensions import Annotated
app = FastAPI()
@app.get("/items/")
async def read_items(user_agent: Annotated[Union[str, None], Header()] = None):
return {"User-Agent": user_agent}

View File

@@ -0,0 +1,10 @@
from typing import Annotated
from fastapi import FastAPI, Header
app = FastAPI()
@app.get("/items/")
async def read_items(user_agent: Annotated[str | None, Header()] = None):
return {"User-Agent": user_agent}

View File

@@ -0,0 +1,10 @@
from typing import Annotated, Union
from fastapi import FastAPI, Header
app = FastAPI()
@app.get("/items/")
async def read_items(user_agent: Annotated[Union[str, None], Header()] = None):
return {"User-Agent": user_agent}

View File

@@ -0,0 +1,15 @@
from typing import Union
from fastapi import FastAPI, Header
from typing_extensions import Annotated
app = FastAPI()
@app.get("/items/")
async def read_items(
strange_header: Annotated[
Union[str, None], Header(convert_underscores=False)
] = None
):
return {"strange_header": strange_header}

View File

@@ -0,0 +1,12 @@
from typing import Annotated
from fastapi import FastAPI, Header
app = FastAPI()
@app.get("/items/")
async def read_items(
strange_header: Annotated[str | None, Header(convert_underscores=False)] = None
):
return {"strange_header": strange_header}

View File

@@ -0,0 +1,14 @@
from typing import Annotated, Union
from fastapi import FastAPI, Header
app = FastAPI()
@app.get("/items/")
async def read_items(
strange_header: Annotated[
Union[str, None], Header(convert_underscores=False)
] = None
):
return {"strange_header": strange_header}

View File

@@ -0,0 +1,11 @@
from typing import List, Union
from fastapi import FastAPI, Header
from typing_extensions import Annotated
app = FastAPI()
@app.get("/items/")
async def read_items(x_token: Annotated[Union[List[str], None], Header()] = None):
return {"X-Token values": x_token}

View File

@@ -0,0 +1,10 @@
from typing import Annotated
from fastapi import FastAPI, Header
app = FastAPI()
@app.get("/items/")
async def read_items(x_token: Annotated[list[str] | None, Header()] = None):
return {"X-Token values": x_token}

View File

@@ -0,0 +1,10 @@
from typing import Annotated, List, Union
from fastapi import FastAPI, Header
app = FastAPI()
@app.get("/items/")
async def read_items(x_token: Annotated[Union[List[str], None], Header()] = None):
return {"X-Token values": x_token}

View File

@@ -0,0 +1,17 @@
from typing import Union
from fastapi import FastAPI, Path, Query
from typing_extensions import Annotated
app = FastAPI()
@app.get("/items/{item_id}")
async def read_items(
item_id: Annotated[int, Path(title="The ID of the item to get")],
q: Annotated[Union[str, None], Query(alias="item-query")] = None,
):
results = {"item_id": item_id}
if q:
results.update({"q": q})
return results

View File

@@ -0,0 +1,16 @@
from typing import Annotated
from fastapi import FastAPI, Path, Query
app = FastAPI()
@app.get("/items/{item_id}")
async def read_items(
item_id: Annotated[int, Path(title="The ID of the item to get")],
q: Annotated[str | None, Query(alias="item-query")] = None,
):
results = {"item_id": item_id}
if q:
results.update({"q": q})
return results

Some files were not shown because too many files have changed in this diff Show More