committed by
GitHub
219 changed files with 11562 additions and 452 deletions
@ -1,6 +1,6 @@ |
|||
from fastapi.testclient import TestClient |
|||
|
|||
from .main_b import app |
|||
from .main import app |
|||
|
|||
client = TestClient(app) |
|||
|
@ -0,0 +1,36 @@ |
|||
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: 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: 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 |
@ -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"} |
@ -0,0 +1,24 @@ |
|||
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: str = Depends(get_query) |
|||
): |
|||
message = f"message to {email}\n" |
|||
background_tasks.add_task(write_log, message) |
|||
return {"message": "Message sent"} |
@ -0,0 +1,17 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: str | None = None |
|||
price: float |
|||
tax: float | None = None |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.post("/items/") |
|||
async def create_item(item: Item): |
|||
return item |
@ -0,0 +1,21 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: str | None = None |
|||
price: float |
|||
tax: float | None = None |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.post("/items/") |
|||
async def create_item(item: Item): |
|||
item_dict = item.dict() |
|||
if item.tax: |
|||
price_with_tax = item.price + item.tax |
|||
item_dict.update({"price_with_tax": price_with_tax}) |
|||
return item_dict |
@ -0,0 +1,17 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: str | None = None |
|||
price: float |
|||
tax: float | None = None |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.put("/items/{item_id}") |
|||
async def create_item(item_id: int, item: Item): |
|||
return {"item_id": item_id, **item.dict()} |
@ -0,0 +1,20 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: str | None = None |
|||
price: float |
|||
tax: float | None = None |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.put("/items/{item_id}") |
|||
async def create_item(item_id: int, item: Item, q: str | None = None): |
|||
result = {"item_id": item_id, **item.dict()} |
|||
if q: |
|||
result.update({"q": q}) |
|||
return result |
@ -0,0 +1,19 @@ |
|||
from fastapi import Body, FastAPI |
|||
from pydantic import BaseModel, Field |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: str | None = Field( |
|||
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: Item = Body(..., embed=True)): |
|||
results = {"item_id": item_id, "item": item} |
|||
return results |
@ -0,0 +1,26 @@ |
|||
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: 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 |
@ -0,0 +1,22 @@ |
|||
from fastapi import 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): |
|||
results = {"item_id": item_id, "item": item, "user": user} |
|||
return results |
@ -0,0 +1,24 @@ |
|||
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: int = Body(...) |
|||
): |
|||
results = {"item_id": item_id, "item": item, "user": user, "importance": importance} |
|||
return results |
@ -0,0 +1,31 @@ |
|||
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: 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 |
@ -0,0 +1,17 @@ |
|||
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: Item = Body(..., embed=True)): |
|||
results = {"item_id": item_id, "item": item} |
|||
return results |
@ -0,0 +1,18 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: str | None = None |
|||
price: float |
|||
tax: float | None = None |
|||
tags: list = [] |
|||
|
|||
|
|||
@app.put("/items/{item_id}") |
|||
async def update_item(item_id: int, item: Item): |
|||
results = {"item_id": item_id, "item": item} |
|||
return results |
@ -0,0 +1,18 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: str | None = None |
|||
price: float |
|||
tax: float | None = None |
|||
tags: list[str] = [] |
|||
|
|||
|
|||
@app.put("/items/{item_id}") |
|||
async def update_item(item_id: int, item: Item): |
|||
results = {"item_id": item_id, "item": item} |
|||
return results |
@ -0,0 +1,20 @@ |
|||
from typing import Optional |
|||
|
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: Optional[str] = None |
|||
price: float |
|||
tax: Optional[float] = None |
|||
tags: list[str] = [] |
|||
|
|||
|
|||
@app.put("/items/{item_id}") |
|||
async def update_item(item_id: int, item: Item): |
|||
results = {"item_id": item_id, "item": item} |
|||
return results |
@ -0,0 +1,18 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: str | None = None |
|||
price: float |
|||
tax: float | None = None |
|||
tags: set[str] = set() |
|||
|
|||
|
|||
@app.put("/items/{item_id}") |
|||
async def update_item(item_id: int, item: Item): |
|||
results = {"item_id": item_id, "item": item} |
|||
return results |
@ -0,0 +1,20 @@ |
|||
from typing import Optional |
|||
|
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: Optional[str] = None |
|||
price: float |
|||
tax: Optional[float] = None |
|||
tags: set[str] = set() |
|||
|
|||
|
|||
@app.put("/items/{item_id}") |
|||
async def update_item(item_id: int, item: Item): |
|||
results = {"item_id": item_id, "item": item} |
|||
return results |
@ -0,0 +1,24 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Image(BaseModel): |
|||
url: str |
|||
name: str |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: str | None = None |
|||
price: float |
|||
tax: float | None = None |
|||
tags: set[str] = [] |
|||
image: Image | None = None |
|||
|
|||
|
|||
@app.put("/items/{item_id}") |
|||
async def update_item(item_id: int, item: Item): |
|||
results = {"item_id": item_id, "item": item} |
|||
return results |
@ -0,0 +1,26 @@ |
|||
from typing import Optional |
|||
|
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Image(BaseModel): |
|||
url: str |
|||
name: str |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: Optional[str] = None |
|||
price: float |
|||
tax: Optional[float] = None |
|||
tags: set[str] = [] |
|||
image: Optional[Image] = None |
|||
|
|||
|
|||
@app.put("/items/{item_id}") |
|||
async def update_item(item_id: int, item: Item): |
|||
results = {"item_id": item_id, "item": item} |
|||
return results |
@ -0,0 +1,24 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel, HttpUrl |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Image(BaseModel): |
|||
url: HttpUrl |
|||
name: str |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: str | None = None |
|||
price: float |
|||
tax: float | None = None |
|||
tags: set[str] = set() |
|||
image: Image | None = None |
|||
|
|||
|
|||
@app.put("/items/{item_id}") |
|||
async def update_item(item_id: int, item: Item): |
|||
results = {"item_id": item_id, "item": item} |
|||
return results |
@ -0,0 +1,26 @@ |
|||
from typing import Optional |
|||
|
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel, HttpUrl |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Image(BaseModel): |
|||
url: HttpUrl |
|||
name: str |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: Optional[str] = None |
|||
price: float |
|||
tax: Optional[float] = None |
|||
tags: set[str] = set() |
|||
image: Optional[Image] = None |
|||
|
|||
|
|||
@app.put("/items/{item_id}") |
|||
async def update_item(item_id: int, item: Item): |
|||
results = {"item_id": item_id, "item": item} |
|||
return results |
@ -0,0 +1,24 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel, HttpUrl |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Image(BaseModel): |
|||
url: HttpUrl |
|||
name: str |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: str | None = None |
|||
price: float |
|||
tax: float | None = None |
|||
tags: set[str] = set() |
|||
images: list[Image] | None = None |
|||
|
|||
|
|||
@app.put("/items/{item_id}") |
|||
async def update_item(item_id: int, item: Item): |
|||
results = {"item_id": item_id, "item": item} |
|||
return results |
@ -0,0 +1,26 @@ |
|||
from typing import Optional |
|||
|
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel, HttpUrl |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Image(BaseModel): |
|||
url: HttpUrl |
|||
name: str |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: Optional[str] = None |
|||
price: float |
|||
tax: Optional[float] = None |
|||
tags: set[str] = set() |
|||
images: Optional[list[Image]] = None |
|||
|
|||
|
|||
@app.put("/items/{item_id}") |
|||
async def update_item(item_id: int, item: Item): |
|||
results = {"item_id": item_id, "item": item} |
|||
return results |
@ -0,0 +1,30 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel, HttpUrl |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Image(BaseModel): |
|||
url: HttpUrl |
|||
name: str |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: str | None = None |
|||
price: float |
|||
tax: float | None = None |
|||
tags: set[str] = set() |
|||
images: list[Image] | None = None |
|||
|
|||
|
|||
class Offer(BaseModel): |
|||
name: str |
|||
description: str | None = None |
|||
price: float |
|||
items: list[Item] |
|||
|
|||
|
|||
@app.post("/offers/") |
|||
async def create_offer(offer: Offer): |
|||
return offer |
@ -0,0 +1,32 @@ |
|||
from typing import Optional |
|||
|
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel, HttpUrl |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Image(BaseModel): |
|||
url: HttpUrl |
|||
name: str |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: Optional[str] = None |
|||
price: float |
|||
tax: Optional[float] = None |
|||
tags: set[str] = set() |
|||
images: Optional[list[Image]] = None |
|||
|
|||
|
|||
class Offer(BaseModel): |
|||
name: str |
|||
description: Optional[str] = None |
|||
price: float |
|||
items: list[Item] |
|||
|
|||
|
|||
@app.post("/offers/") |
|||
async def create_offer(offer: Offer): |
|||
return offer |
@ -0,0 +1,14 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel, HttpUrl |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Image(BaseModel): |
|||
url: HttpUrl |
|||
name: str |
|||
|
|||
|
|||
@app.post("/images/multiple/") |
|||
async def create_multiple_images(images: list[Image]): |
|||
return images |
@ -0,0 +1,8 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.post("/index-weights/") |
|||
async def create_index_weights(weights: dict[int, float]): |
|||
return weights |
@ -0,0 +1,32 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.encoders import jsonable_encoder |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str | None = None |
|||
description: str | None = None |
|||
price: float | None = None |
|||
tax: float = 10.5 |
|||
tags: list[str] = [] |
|||
|
|||
|
|||
items = { |
|||
"foo": {"name": "Foo", "price": 50.2}, |
|||
"bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2}, |
|||
"baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []}, |
|||
} |
|||
|
|||
|
|||
@app.get("/items/{item_id}", response_model=Item) |
|||
async def read_item(item_id: str): |
|||
return items[item_id] |
|||
|
|||
|
|||
@app.put("/items/{item_id}", response_model=Item) |
|||
async def update_item(item_id: str, item: Item): |
|||
update_item_encoded = jsonable_encoder(item) |
|||
items[item_id] = update_item_encoded |
|||
return update_item_encoded |
@ -0,0 +1,34 @@ |
|||
from typing import Optional |
|||
|
|||
from fastapi import FastAPI |
|||
from fastapi.encoders import jsonable_encoder |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: Optional[str] = None |
|||
description: Optional[str] = None |
|||
price: Optional[float] = None |
|||
tax: float = 10.5 |
|||
tags: list[str] = [] |
|||
|
|||
|
|||
items = { |
|||
"foo": {"name": "Foo", "price": 50.2}, |
|||
"bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2}, |
|||
"baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []}, |
|||
} |
|||
|
|||
|
|||
@app.get("/items/{item_id}", response_model=Item) |
|||
async def read_item(item_id: str): |
|||
return items[item_id] |
|||
|
|||
|
|||
@app.put("/items/{item_id}", response_model=Item) |
|||
async def update_item(item_id: str, item: Item): |
|||
update_item_encoded = jsonable_encoder(item) |
|||
items[item_id] = update_item_encoded |
|||
return update_item_encoded |
@ -0,0 +1,35 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.encoders import jsonable_encoder |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str | None = None |
|||
description: str | None = None |
|||
price: float | None = None |
|||
tax: float = 10.5 |
|||
tags: list[str] = [] |
|||
|
|||
|
|||
items = { |
|||
"foo": {"name": "Foo", "price": 50.2}, |
|||
"bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2}, |
|||
"baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []}, |
|||
} |
|||
|
|||
|
|||
@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): |
|||
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) |
|||
return updated_item |
@ -0,0 +1,37 @@ |
|||
from typing import Optional |
|||
|
|||
from fastapi import FastAPI |
|||
from fastapi.encoders import jsonable_encoder |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: Optional[str] = None |
|||
description: Optional[str] = None |
|||
price: Optional[float] = None |
|||
tax: float = 10.5 |
|||
tags: list[str] = [] |
|||
|
|||
|
|||
items = { |
|||
"foo": {"name": "Foo", "price": 50.2}, |
|||
"bar": {"name": "Bar", "description": "The bartenders", "price": 62, "tax": 20.2}, |
|||
"baz": {"name": "Baz", "description": None, "price": 50.2, "tax": 10.5, "tags": []}, |
|||
} |
|||
|
|||
|
|||
@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): |
|||
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) |
|||
return updated_item |
@ -0,0 +1,8 @@ |
|||
from fastapi import Cookie, FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(ads_id: str | None = Cookie(None)): |
|||
return {"ads_id": ads_id} |
@ -0,0 +1,17 @@ |
|||
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: dict = Depends(common_parameters)): |
|||
return commons |
|||
|
|||
|
|||
@app.get("/users/") |
|||
async def read_users(commons: dict = Depends(common_parameters)): |
|||
return commons |
@ -0,0 +1,23 @@ |
|||
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: 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 |
@ -0,0 +1,23 @@ |
|||
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=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 |
@ -0,0 +1,23 @@ |
|||
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: 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 |
@ -0,0 +1,20 @@ |
|||
from fastapi import Cookie, Depends, FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
def query_extractor(q: str | None = None): |
|||
return q |
|||
|
|||
|
|||
def query_or_cookie_extractor( |
|||
q: str = Depends(query_extractor), last_query: str | None = Cookie(None) |
|||
): |
|||
if not q: |
|||
return last_query |
|||
return q |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)): |
|||
return {"q_or_cookie": query_or_default} |
@ -0,0 +1,22 @@ |
|||
from datetime import datetime |
|||
|
|||
from fastapi import FastAPI |
|||
from fastapi.encoders import jsonable_encoder |
|||
from pydantic import BaseModel |
|||
|
|||
fake_db = {} |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
title: str |
|||
timestamp: datetime |
|||
description: str | None = None |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.put("/items/{id}") |
|||
def update_item(id: str, item: Item): |
|||
json_compatible_item_data = jsonable_encoder(item) |
|||
fake_db[id] = json_compatible_item_data |
@ -0,0 +1,27 @@ |
|||
from datetime import datetime, time, timedelta |
|||
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: datetime | None = Body(None), |
|||
end_datetime: datetime | None = Body(None), |
|||
repeat_at: time | None = Body(None), |
|||
process_after: 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, |
|||
} |
@ -0,0 +1,41 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel, EmailStr |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class UserIn(BaseModel): |
|||
username: str |
|||
password: str |
|||
email: EmailStr |
|||
full_name: str | None = None |
|||
|
|||
|
|||
class UserOut(BaseModel): |
|||
username: str |
|||
email: EmailStr |
|||
full_name: str | None = None |
|||
|
|||
|
|||
class UserInDB(BaseModel): |
|||
username: str |
|||
hashed_password: str |
|||
email: EmailStr |
|||
full_name: str | None = None |
|||
|
|||
|
|||
def fake_password_hasher(raw_password: str): |
|||
return "supersecret" + raw_password |
|||
|
|||
|
|||
def fake_save_user(user_in: UserIn): |
|||
hashed_password = fake_password_hasher(user_in.password) |
|||
user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password) |
|||
print("User saved! ..not really") |
|||
return user_in_db |
|||
|
|||
|
|||
@app.post("/user/", response_model=UserOut) |
|||
async def create_user(user_in: UserIn): |
|||
user_saved = fake_save_user(user_in) |
|||
return user_saved |
@ -0,0 +1,39 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel, EmailStr |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class UserBase(BaseModel): |
|||
username: str |
|||
email: EmailStr |
|||
full_name: str | None = None |
|||
|
|||
|
|||
class UserIn(UserBase): |
|||
password: str |
|||
|
|||
|
|||
class UserOut(UserBase): |
|||
pass |
|||
|
|||
|
|||
class UserInDB(UserBase): |
|||
hashed_password: str |
|||
|
|||
|
|||
def fake_password_hasher(raw_password: str): |
|||
return "supersecret" + raw_password |
|||
|
|||
|
|||
def fake_save_user(user_in: UserIn): |
|||
hashed_password = fake_password_hasher(user_in.password) |
|||
user_in_db = UserInDB(**user_in.dict(), hashed_password=hashed_password) |
|||
print("User saved! ..not really") |
|||
return user_in_db |
|||
|
|||
|
|||
@app.post("/user/", response_model=UserOut) |
|||
async def create_user(user_in: UserIn): |
|||
user_saved = fake_save_user(user_in) |
|||
return user_saved |
@ -0,0 +1,35 @@ |
|||
from typing import Union |
|||
|
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class BaseItem(BaseModel): |
|||
description: str |
|||
type: str |
|||
|
|||
|
|||
class CarItem(BaseItem): |
|||
type = "car" |
|||
|
|||
|
|||
class PlaneItem(BaseItem): |
|||
type = "plane" |
|||
size: int |
|||
|
|||
|
|||
items = { |
|||
"item1": {"description": "All my friends drive a low rider", "type": "car"}, |
|||
"item2": { |
|||
"description": "Music is my aeroplane, it's my aeroplane", |
|||
"type": "plane", |
|||
"size": 5, |
|||
}, |
|||
} |
|||
|
|||
|
|||
@app.get("/items/{item_id}", response_model=Union[PlaneItem, CarItem]) |
|||
async def read_item(item_id: str): |
|||
return items[item_id] |
@ -0,0 +1,20 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: str |
|||
|
|||
|
|||
items = [ |
|||
{"name": "Foo", "description": "There comes my hero"}, |
|||
{"name": "Red", "description": "It's my aeroplane"}, |
|||
] |
|||
|
|||
|
|||
@app.get("/items/", response_model=list[Item]) |
|||
async def read_items(): |
|||
return items |
@ -0,0 +1,8 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/keyword-weights/", response_model=dict[str, float]) |
|||
async def read_keyword_weights(): |
|||
return {"foo": 2.3, "bar": 3.4} |
@ -0,0 +1,8 @@ |
|||
from fastapi import FastAPI, Header |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(user_agent: str | None = Header(None)): |
|||
return {"User-Agent": user_agent} |
@ -0,0 +1,10 @@ |
|||
from fastapi import FastAPI, Header |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items( |
|||
strange_header: str | None = Header(None, convert_underscores=False) |
|||
): |
|||
return {"strange_header": strange_header} |
@ -0,0 +1,8 @@ |
|||
from fastapi import FastAPI, Header |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(x_token: list[str] | None = Header(None)): |
|||
return {"X-Token values": x_token} |
@ -0,0 +1,10 @@ |
|||
from typing import Optional |
|||
|
|||
from fastapi import FastAPI, Header |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(x_token: Optional[list[str]] = Header(None)): |
|||
return {"X-Token values": x_token} |
@ -0,0 +1,17 @@ |
|||
from fastapi import FastAPI, status |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: str | None = None |
|||
price: float |
|||
tax: float | None = None |
|||
tags: set[str] = set() |
|||
|
|||
|
|||
@app.post("/items/", response_model=Item, status_code=status.HTTP_201_CREATED) |
|||
async def create_item(item: Item): |
|||
return item |
@ -0,0 +1,19 @@ |
|||
from typing import Optional |
|||
|
|||
from fastapi import FastAPI, status |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: Optional[str] = None |
|||
price: float |
|||
tax: Optional[float] = None |
|||
tags: set[str] = set() |
|||
|
|||
|
|||
@app.post("/items/", response_model=Item, status_code=status.HTTP_201_CREATED) |
|||
async def create_item(item: Item): |
|||
return item |
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue