Sebastián Ramírez
4 months ago
committed by
GitHub
GitHub
655 changed files with 3701 additions and 5658 deletions
@ -0,0 +1,22 @@ |
|||||
|
from fastapi import FastAPI |
||||
|
from fastapi.responses import JSONResponse |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
|
||||
|
class Item(BaseModel): |
||||
|
id: str |
||||
|
value: str |
||||
|
|
||||
|
|
||||
|
class Message(BaseModel): |
||||
|
message: str |
||||
|
|
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.get("/items/{item_id}", response_model=Item, responses={404: {"model": Message}}) |
||||
|
async def read_item(item_id: str): |
||||
|
if item_id == "foo": |
||||
|
return {"id": "foo", "value": "there goes my hero"} |
||||
|
return JSONResponse(status_code=404, content={"message": "Item not found"}) |
||||
@ -1,30 +0,0 @@ |
|||||
from typing import Union |
|
||||
|
|
||||
from fastapi import FastAPI |
|
||||
from fastapi.responses import FileResponse |
|
||||
from pydantic import BaseModel |
|
||||
|
|
||||
|
|
||||
class Item(BaseModel): |
|
||||
id: str |
|
||||
value: str |
|
||||
|
|
||||
|
|
||||
app = FastAPI() |
|
||||
|
|
||||
|
|
||||
@app.get( |
|
||||
"/items/{item_id}", |
|
||||
response_model=Item, |
|
||||
responses={ |
|
||||
200: { |
|
||||
"content": {"image/png": {}}, |
|
||||
"description": "Return the JSON item or an image.", |
|
||||
} |
|
||||
}, |
|
||||
) |
|
||||
async def read_item(item_id: str, img: Union[bool, None] = None): |
|
||||
if img: |
|
||||
return FileResponse("image.png", media_type="image/png") |
|
||||
else: |
|
||||
return {"id": "foo", "value": "there goes my hero"} |
|
||||
@ -0,0 +1,37 @@ |
|||||
|
from fastapi import FastAPI |
||||
|
from fastapi.responses import JSONResponse |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
|
||||
|
class Item(BaseModel): |
||||
|
id: str |
||||
|
value: str |
||||
|
|
||||
|
|
||||
|
class Message(BaseModel): |
||||
|
message: str |
||||
|
|
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.get( |
||||
|
"/items/{item_id}", |
||||
|
response_model=Item, |
||||
|
responses={ |
||||
|
404: {"model": Message, "description": "The item was not found"}, |
||||
|
200: { |
||||
|
"description": "Item requested by ID", |
||||
|
"content": { |
||||
|
"application/json": { |
||||
|
"example": {"id": "bar", "value": "The bar tenders"} |
||||
|
} |
||||
|
}, |
||||
|
}, |
||||
|
}, |
||||
|
) |
||||
|
async def read_item(item_id: str): |
||||
|
if item_id == "foo": |
||||
|
return {"id": "foo", "value": "there goes my hero"} |
||||
|
else: |
||||
|
return JSONResponse(status_code=404, content={"message": "Item not found"}) |
||||
@ -1,32 +0,0 @@ |
|||||
from typing import Union |
|
||||
|
|
||||
from fastapi import FastAPI |
|
||||
from fastapi.responses import FileResponse |
|
||||
from pydantic import BaseModel |
|
||||
|
|
||||
|
|
||||
class Item(BaseModel): |
|
||||
id: str |
|
||||
value: str |
|
||||
|
|
||||
|
|
||||
responses = { |
|
||||
404: {"description": "Item not found"}, |
|
||||
302: {"description": "The item was moved"}, |
|
||||
403: {"description": "Not enough privileges"}, |
|
||||
} |
|
||||
|
|
||||
|
|
||||
app = FastAPI() |
|
||||
|
|
||||
|
|
||||
@app.get( |
|
||||
"/items/{item_id}", |
|
||||
response_model=Item, |
|
||||
responses={**responses, 200: {"content": {"image/png": {}}}}, |
|
||||
) |
|
||||
async def read_item(item_id: str, img: Union[bool, None] = None): |
|
||||
if img: |
|
||||
return FileResponse("image.png", media_type="image/png") |
|
||||
else: |
|
||||
return {"id": "foo", "value": "there goes my hero"} |
|
||||
@ -1,25 +0,0 @@ |
|||||
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) |
|
||||
@ -1,25 +0,0 @@ |
|||||
from typing import 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: Union[str, None] = Body(default=None), |
|
||||
size: Union[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) |
|
||||
@ -0,0 +1,11 @@ |
|||||
|
from fastapi import FastAPI |
||||
|
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
app.add_middleware(HTTPSRedirectMiddleware) |
||||
|
|
||||
|
|
||||
|
@app.get("/") |
||||
|
async def main(): |
||||
|
return {"message": "Hello World"} |
||||
@ -0,0 +1,13 @@ |
|||||
|
from fastapi import FastAPI |
||||
|
from fastapi.middleware.trustedhost import TrustedHostMiddleware |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
app.add_middleware( |
||||
|
TrustedHostMiddleware, allowed_hosts=["example.com", "*.example.com"] |
||||
|
) |
||||
|
|
||||
|
|
||||
|
@app.get("/") |
||||
|
async def main(): |
||||
|
return {"message": "Hello World"} |
||||
@ -0,0 +1,11 @@ |
|||||
|
from fastapi import FastAPI |
||||
|
from fastapi.middleware.gzip import GZipMiddleware |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
app.add_middleware(GZipMiddleware, minimum_size=1000, compresslevel=5) |
||||
|
|
||||
|
|
||||
|
@app.get("/") |
||||
|
async def main(): |
||||
|
return "somebigcontent" |
||||
@ -0,0 +1,8 @@ |
|||||
|
from fastapi import FastAPI |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.get("/") |
||||
|
async def read_main(): |
||||
|
return {"msg": "Hello World"} |
||||
@ -0,0 +1,11 @@ |
|||||
|
from fastapi.testclient import TestClient |
||||
|
|
||||
|
from .main import app |
||||
|
|
||||
|
client = TestClient(app) |
||||
|
|
||||
|
|
||||
|
def test_read_main(): |
||||
|
response = client.get("/") |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"msg": "Hello World"} |
||||
@ -1,38 +0,0 @@ |
|||||
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/") |
|
||||
async def create_item(item: Item, x_token: Annotated[str, Header()]) -> Item: |
|
||||
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=409, detail="Item already exists") |
|
||||
fake_db[item.id] = item.model_dump() |
|
||||
return item |
|
||||
@ -1,65 +0,0 @@ |
|||||
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_nonexistent_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 == 409 |
|
||||
assert response.json() == {"detail": "Item already exists"} |
|
||||
@ -1,38 +0,0 @@ |
|||||
from typing import 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: 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/") |
|
||||
async def create_item(item: Item, x_token: str = Header()) -> Item: |
|
||||
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=409, detail="Item already exists") |
|
||||
fake_db[item.id] = item.model_dump() |
|
||||
return item |
|
||||
@ -1,65 +0,0 @@ |
|||||
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_nonexistent_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 == 409 |
|
||||
assert response.json() == {"detail": "Item already exists"} |
|
||||
@ -0,0 +1,18 @@ |
|||||
|
from fastapi import FastAPI |
||||
|
from fastapi.testclient import TestClient |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.get("/") |
||||
|
async def read_main(): |
||||
|
return {"msg": "Hello World"} |
||||
|
|
||||
|
|
||||
|
client = TestClient(app) |
||||
|
|
||||
|
|
||||
|
def test_read_main(): |
||||
|
response = client.get("/") |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"msg": "Hello World"} |
||||
@ -0,0 +1,31 @@ |
|||||
|
from fastapi import FastAPI |
||||
|
from fastapi.testclient import TestClient |
||||
|
from fastapi.websockets import WebSocket |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.get("/") |
||||
|
async def read_main(): |
||||
|
return {"msg": "Hello World"} |
||||
|
|
||||
|
|
||||
|
@app.websocket("/ws") |
||||
|
async def websocket(websocket: WebSocket): |
||||
|
await websocket.accept() |
||||
|
await websocket.send_json({"msg": "Hello WebSocket"}) |
||||
|
await websocket.close() |
||||
|
|
||||
|
|
||||
|
def test_read_main(): |
||||
|
client = TestClient(app) |
||||
|
response = client.get("/") |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"msg": "Hello World"} |
||||
|
|
||||
|
|
||||
|
def test_websocket(): |
||||
|
client = TestClient(app) |
||||
|
with client.websocket_connect("/ws") as websocket: |
||||
|
data = websocket.receive_json() |
||||
|
assert data == {"msg": "Hello WebSocket"} |
||||
@ -0,0 +1,24 @@ |
|||||
|
from fastapi import FastAPI |
||||
|
from fastapi.testclient import TestClient |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
items = {} |
||||
|
|
||||
|
|
||||
|
@app.on_event("startup") |
||||
|
async def startup_event(): |
||||
|
items["foo"] = {"name": "Fighters"} |
||||
|
items["bar"] = {"name": "Tenders"} |
||||
|
|
||||
|
|
||||
|
@app.get("/items/{item_id}") |
||||
|
async def read_items(item_id: str): |
||||
|
return items[item_id] |
||||
|
|
||||
|
|
||||
|
def test_read_items(): |
||||
|
with TestClient(app) as client: |
||||
|
response = client.get("/items/foo") |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"name": "Fighters"} |
||||
@ -0,0 +1,43 @@ |
|||||
|
from contextlib import asynccontextmanager |
||||
|
|
||||
|
from fastapi import FastAPI |
||||
|
from fastapi.testclient import TestClient |
||||
|
|
||||
|
items = {} |
||||
|
|
||||
|
|
||||
|
@asynccontextmanager |
||||
|
async def lifespan(app: FastAPI): |
||||
|
items["foo"] = {"name": "Fighters"} |
||||
|
items["bar"] = {"name": "Tenders"} |
||||
|
yield |
||||
|
# clean up items |
||||
|
items.clear() |
||||
|
|
||||
|
|
||||
|
app = FastAPI(lifespan=lifespan) |
||||
|
|
||||
|
|
||||
|
@app.get("/items/{item_id}") |
||||
|
async def read_items(item_id: str): |
||||
|
return items[item_id] |
||||
|
|
||||
|
|
||||
|
def test_read_items(): |
||||
|
# Before the lifespan starts, "items" is still empty |
||||
|
assert items == {} |
||||
|
|
||||
|
with TestClient(app) as client: |
||||
|
# Inside the "with TestClient" block, the lifespan starts and items added |
||||
|
assert items == {"foo": {"name": "Fighters"}, "bar": {"name": "Tenders"}} |
||||
|
|
||||
|
response = client.get("/items/foo") |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"name": "Fighters"} |
||||
|
|
||||
|
# After the requests is done, the items are still there |
||||
|
assert items == {"foo": {"name": "Fighters"}, "bar": {"name": "Tenders"}} |
||||
|
|
||||
|
# The end of the "with TestClient" block simulates terminating the app, so |
||||
|
# the lifespan ends and items are cleaned up |
||||
|
assert items == {} |
||||
@ -0,0 +1,8 @@ |
|||||
|
from fastapi import FastAPI |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.get("/") |
||||
|
async def root(): |
||||
|
return {"message": "Tomato"} |
||||
@ -0,0 +1,14 @@ |
|||||
|
import pytest |
||||
|
from httpx import ASGITransport, AsyncClient |
||||
|
|
||||
|
from .main import app |
||||
|
|
||||
|
|
||||
|
@pytest.mark.anyio |
||||
|
async def test_root(): |
||||
|
async with AsyncClient( |
||||
|
transport=ASGITransport(app=app), base_url="http://test" |
||||
|
) as ac: |
||||
|
response = await ac.get("/") |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"message": "Tomato"} |
||||
@ -0,0 +1,21 @@ |
|||||
|
from typing import Annotated |
||||
|
|
||||
|
from fastapi import Depends, FastAPI, HTTPException, status |
||||
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
class HTTPBearer403(HTTPBearer): |
||||
|
def make_not_authenticated_error(self) -> HTTPException: |
||||
|
return HTTPException( |
||||
|
status_code=status.HTTP_403_FORBIDDEN, detail="Not authenticated" |
||||
|
) |
||||
|
|
||||
|
|
||||
|
CredentialsDep = Annotated[HTTPAuthorizationCredentials, Depends(HTTPBearer403())] |
||||
|
|
||||
|
|
||||
|
@app.get("/me") |
||||
|
def read_me(credentials: CredentialsDep): |
||||
|
return {"message": "You are authenticated", "token": credentials.credentials} |
||||
@ -0,0 +1,15 @@ |
|||||
|
from fastapi import BackgroundTasks, FastAPI |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
def write_notification(email: str, message=""): |
||||
|
with open("log.txt", mode="w") as email_file: |
||||
|
content = f"notification for {email}: {message}" |
||||
|
email_file.write(content) |
||||
|
|
||||
|
|
||||
|
@app.post("/send-notification/{email}") |
||||
|
async def send_notification(email: str, background_tasks: BackgroundTasks): |
||||
|
background_tasks.add_task(write_notification, email, message="some notification") |
||||
|
return {"message": "Notification sent in the background"} |
||||
@ -1,26 +0,0 @@ |
|||||
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"} |
|
||||
@ -1,26 +0,0 @@ |
|||||
from typing import 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: str = Depends(get_query) |
|
||||
): |
|
||||
message = f"message to {email}\n" |
|
||||
background_tasks.add_task(write_log, message) |
|
||||
return {"message": "Message sent"} |
|
||||
@ -0,0 +1,8 @@ |
|||||
|
from fastapi import FastAPI |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.get("/items/") |
||||
|
def read_items(): |
||||
|
return ["plumbus", "portal gun"] |
||||
@ -0,0 +1,8 @@ |
|||||
|
from fastapi import FastAPI, Request |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.get("/app") |
||||
|
def read_main(request: Request): |
||||
|
return {"message": "Hello World", "root_path": request.scope.get("root_path")} |
||||
@ -0,0 +1,8 @@ |
|||||
|
from fastapi import FastAPI, Request |
||||
|
|
||||
|
app = FastAPI(root_path="/api/v1") |
||||
|
|
||||
|
|
||||
|
@app.get("/app") |
||||
|
def read_main(request: Request): |
||||
|
return {"message": "Hello World", "root_path": request.scope.get("root_path")} |
||||
@ -0,0 +1,14 @@ |
|||||
|
from fastapi import FastAPI, Request |
||||
|
|
||||
|
app = FastAPI( |
||||
|
servers=[ |
||||
|
{"url": "https://stag.example.com", "description": "Staging environment"}, |
||||
|
{"url": "https://prod.example.com", "description": "Production environment"}, |
||||
|
], |
||||
|
root_path="/api/v1", |
||||
|
) |
||||
|
|
||||
|
|
||||
|
@app.get("/app") |
||||
|
def read_main(request: Request): |
||||
|
return {"message": "Hello World", "root_path": request.scope.get("root_path")} |
||||
@ -0,0 +1,15 @@ |
|||||
|
from fastapi import FastAPI, Request |
||||
|
|
||||
|
app = FastAPI( |
||||
|
servers=[ |
||||
|
{"url": "https://stag.example.com", "description": "Staging environment"}, |
||||
|
{"url": "https://prod.example.com", "description": "Production environment"}, |
||||
|
], |
||||
|
root_path="/api/v1", |
||||
|
root_path_in_servers=False, |
||||
|
) |
||||
|
|
||||
|
|
||||
|
@app.get("/app") |
||||
|
def read_main(request: Request): |
||||
|
return {"message": "Hello World", "root_path": request.scope.get("root_path")} |
||||
@ -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") |
||||
@ -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!"} |
||||
@ -1,19 +0,0 @@ |
|||||
from typing import Union |
|
||||
|
|
||||
from fastapi import FastAPI |
|
||||
from pydantic import BaseModel |
|
||||
|
|
||||
|
|
||||
class Item(BaseModel): |
|
||||
name: str |
|
||||
description: Union[str, None] = None |
|
||||
price: float |
|
||||
tax: Union[float, None] = None |
|
||||
|
|
||||
|
|
||||
app = FastAPI() |
|
||||
|
|
||||
|
|
||||
@app.post("/items/") |
|
||||
async def create_item(item: Item): |
|
||||
return item |
|
||||
@ -1,23 +0,0 @@ |
|||||
from typing import Union |
|
||||
|
|
||||
from fastapi import FastAPI |
|
||||
from pydantic import BaseModel |
|
||||
|
|
||||
|
|
||||
class Item(BaseModel): |
|
||||
name: str |
|
||||
description: Union[str, None] = None |
|
||||
price: float |
|
||||
tax: Union[float, None] = None |
|
||||
|
|
||||
|
|
||||
app = FastAPI() |
|
||||
|
|
||||
|
|
||||
@app.post("/items/") |
|
||||
async def create_item(item: Item): |
|
||||
item_dict = item.model_dump() |
|
||||
if item.tax is not None: |
|
||||
price_with_tax = item.price + item.tax |
|
||||
item_dict.update({"price_with_tax": price_with_tax}) |
|
||||
return item_dict |
|
||||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue