Motov Yurii
4 months ago
committed by
GitHub
GitHub
655 changed files with 3919 additions and 5636 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 |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue