committed by
GitHub
214 changed files with 28 additions and 3393 deletions
@ -1,22 +0,0 @@ |
|||
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,37 +0,0 @@ |
|||
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,11 +0,0 @@ |
|||
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"} |
|||
@ -1,13 +0,0 @@ |
|||
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"} |
|||
@ -1,11 +0,0 @@ |
|||
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" |
|||
@ -1,8 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/") |
|||
async def read_main(): |
|||
return {"msg": "Hello World"} |
|||
@ -1,11 +0,0 @@ |
|||
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,18 +0,0 @@ |
|||
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"} |
|||
@ -1,31 +0,0 @@ |
|||
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"} |
|||
@ -1,24 +0,0 @@ |
|||
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"} |
|||
@ -1,43 +0,0 @@ |
|||
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 == {} |
|||
@ -1,8 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/") |
|||
async def root(): |
|||
return {"message": "Tomato"} |
|||
@ -1,14 +0,0 @@ |
|||
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"} |
|||
@ -1,21 +0,0 @@ |
|||
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} |
|||
@ -1,15 +0,0 @@ |
|||
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,8 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/items/") |
|||
def read_items(): |
|||
return ["plumbus", "portal gun"] |
|||
@ -1,8 +0,0 @@ |
|||
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")} |
|||
@ -1,8 +0,0 @@ |
|||
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")} |
|||
@ -1,14 +0,0 @@ |
|||
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")} |
|||
@ -1,15 +0,0 @@ |
|||
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")} |
|||
@ -1,13 +0,0 @@ |
|||
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") |
|||
@ -1,8 +0,0 @@ |
|||
from fastapi import APIRouter |
|||
|
|||
router = APIRouter() |
|||
|
|||
|
|||
@router.post("/") |
|||
async def update_admin(): |
|||
return {"message": "Admin getting schwifty"} |
|||
@ -1,23 +0,0 @@ |
|||
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,38 +0,0 @@ |
|||
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"} |
|||
@ -1,18 +0,0 @@ |
|||
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} |
|||
@ -1,11 +0,0 @@ |
|||
from fastapi import Header, HTTPException |
|||
|
|||
|
|||
async def get_token_header(x_token: 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") |
|||
@ -1,23 +0,0 @@ |
|||
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,14 +0,0 @@ |
|||
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 |
|||
@ -1,8 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.post("/index-weights/") |
|||
async def create_index_weights(weights: dict[int, float]): |
|||
return weights |
|||
@ -1,16 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic_settings import BaseSettings |
|||
|
|||
|
|||
class Settings(BaseSettings): |
|||
openapi_url: str = "/openapi.json" |
|||
|
|||
|
|||
settings = Settings() |
|||
|
|||
app = FastAPI(openapi_url=settings.openapi_url) |
|||
|
|||
|
|||
@app.get("/") |
|||
def root(): |
|||
return {"message": "Hello World"} |
|||
@ -1,8 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI(swagger_ui_parameters={"syntaxHighlight": False}) |
|||
|
|||
|
|||
@app.get("/users/{username}") |
|||
async def read_user(username: str): |
|||
return {"message": f"Hello {username}"} |
|||
@ -1,8 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI(swagger_ui_parameters={"syntaxHighlight": {"theme": "obsidian"}}) |
|||
|
|||
|
|||
@app.get("/users/{username}") |
|||
async def read_user(username: str): |
|||
return {"message": f"Hello {username}"} |
|||
@ -1,8 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI(swagger_ui_parameters={"deepLinking": False}) |
|||
|
|||
|
|||
@app.get("/users/{username}") |
|||
async def read_user(username: str): |
|||
return {"message": f"Hello {username}"} |
|||
@ -1,24 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.middleware.cors import CORSMiddleware |
|||
|
|||
app = FastAPI() |
|||
|
|||
origins = [ |
|||
"http://localhost.tiangolo.com", |
|||
"https://localhost.tiangolo.com", |
|||
"http://localhost", |
|||
"http://localhost:8080", |
|||
] |
|||
|
|||
app.add_middleware( |
|||
CORSMiddleware, |
|||
allow_origins=origins, |
|||
allow_credentials=True, |
|||
allow_methods=["*"], |
|||
allow_headers=["*"], |
|||
) |
|||
|
|||
|
|||
@app.get("/") |
|||
async def main(): |
|||
return {"message": "Hello World"} |
|||
@ -1,38 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.openapi.docs import ( |
|||
get_redoc_html, |
|||
get_swagger_ui_html, |
|||
get_swagger_ui_oauth2_redirect_html, |
|||
) |
|||
|
|||
app = FastAPI(docs_url=None, redoc_url=None) |
|||
|
|||
|
|||
@app.get("/docs", include_in_schema=False) |
|||
async def custom_swagger_ui_html(): |
|||
return get_swagger_ui_html( |
|||
openapi_url=app.openapi_url, |
|||
title=app.title + " - Swagger UI", |
|||
oauth2_redirect_url=app.swagger_ui_oauth2_redirect_url, |
|||
swagger_js_url="https://unpkg.com/swagger-ui-dist@5/swagger-ui-bundle.js", |
|||
swagger_css_url="https://unpkg.com/swagger-ui-dist@5/swagger-ui.css", |
|||
) |
|||
|
|||
|
|||
@app.get(app.swagger_ui_oauth2_redirect_url, include_in_schema=False) |
|||
async def swagger_ui_redirect(): |
|||
return get_swagger_ui_oauth2_redirect_html() |
|||
|
|||
|
|||
@app.get("/redoc", include_in_schema=False) |
|||
async def redoc_html(): |
|||
return get_redoc_html( |
|||
openapi_url=app.openapi_url, |
|||
title=app.title + " - ReDoc", |
|||
redoc_js_url="https://unpkg.com/redoc@2/bundles/redoc.standalone.js", |
|||
) |
|||
|
|||
|
|||
@app.get("/users/{username}") |
|||
async def read_user(username: str): |
|||
return {"message": f"Hello {username}"} |
|||
@ -1,41 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.openapi.docs import ( |
|||
get_redoc_html, |
|||
get_swagger_ui_html, |
|||
get_swagger_ui_oauth2_redirect_html, |
|||
) |
|||
from fastapi.staticfiles import StaticFiles |
|||
|
|||
app = FastAPI(docs_url=None, redoc_url=None) |
|||
|
|||
app.mount("/static", StaticFiles(directory="static"), name="static") |
|||
|
|||
|
|||
@app.get("/docs", include_in_schema=False) |
|||
async def custom_swagger_ui_html(): |
|||
return get_swagger_ui_html( |
|||
openapi_url=app.openapi_url, |
|||
title=app.title + " - Swagger UI", |
|||
oauth2_redirect_url=app.swagger_ui_oauth2_redirect_url, |
|||
swagger_js_url="/static/swagger-ui-bundle.js", |
|||
swagger_css_url="/static/swagger-ui.css", |
|||
) |
|||
|
|||
|
|||
@app.get(app.swagger_ui_oauth2_redirect_url, include_in_schema=False) |
|||
async def swagger_ui_redirect(): |
|||
return get_swagger_ui_oauth2_redirect_html() |
|||
|
|||
|
|||
@app.get("/redoc", include_in_schema=False) |
|||
async def redoc_html(): |
|||
return get_redoc_html( |
|||
openapi_url=app.openapi_url, |
|||
title=app.title + " - ReDoc", |
|||
redoc_js_url="/static/redoc.standalone.js", |
|||
) |
|||
|
|||
|
|||
@app.get("/users/{username}") |
|||
async def read_user(username: str): |
|||
return {"message": f"Hello {username}"} |
|||
@ -1,9 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import UJSONResponse |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/items/", response_class=UJSONResponse) |
|||
async def read_items(): |
|||
return [{"item_id": "Foo"}] |
|||
@ -1,9 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import ORJSONResponse |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/items/", response_class=ORJSONResponse) |
|||
async def read_items(): |
|||
return ORJSONResponse([{"item_id": "Foo"}]) |
|||
@ -1,18 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import HTMLResponse |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/items/", response_class=HTMLResponse) |
|||
async def read_items(): |
|||
return """ |
|||
<html> |
|||
<head> |
|||
<title>Some HTML in here</title> |
|||
</head> |
|||
<body> |
|||
<h1>Look ma! HTML!</h1> |
|||
</body> |
|||
</html> |
|||
""" |
|||
@ -1,19 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import HTMLResponse |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(): |
|||
html_content = """ |
|||
<html> |
|||
<head> |
|||
<title>Some HTML in here</title> |
|||
</head> |
|||
<body> |
|||
<h1>Look ma! HTML!</h1> |
|||
</body> |
|||
</html> |
|||
""" |
|||
return HTMLResponse(content=html_content, status_code=200) |
|||
@ -1,23 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import HTMLResponse |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
def generate_html_response(): |
|||
html_content = """ |
|||
<html> |
|||
<head> |
|||
<title>Some HTML in here</title> |
|||
</head> |
|||
<body> |
|||
<h1>Look ma! HTML!</h1> |
|||
</body> |
|||
</html> |
|||
""" |
|||
return HTMLResponse(content=html_content, status_code=200) |
|||
|
|||
|
|||
@app.get("/items/", response_class=HTMLResponse) |
|||
async def read_items(): |
|||
return generate_html_response() |
|||
@ -1,9 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import PlainTextResponse |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/", response_class=PlainTextResponse) |
|||
async def main(): |
|||
return "Hello World" |
|||
@ -1,9 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import RedirectResponse |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/typer") |
|||
async def redirect_typer(): |
|||
return RedirectResponse("https://typer.tiangolo.com") |
|||
@ -1,9 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import RedirectResponse |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/fastapi", response_class=RedirectResponse) |
|||
async def redirect_fastapi(): |
|||
return "https://fastapi.tiangolo.com" |
|||
@ -1,9 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import RedirectResponse |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/pydantic", response_class=RedirectResponse, status_code=302) |
|||
async def redirect_pydantic(): |
|||
return "https://docs.pydantic.dev/" |
|||
@ -1,14 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import StreamingResponse |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
async def fake_video_streamer(): |
|||
for i in range(10): |
|||
yield b"some fake video bytes" |
|||
|
|||
|
|||
@app.get("/") |
|||
async def main(): |
|||
return StreamingResponse(fake_video_streamer()) |
|||
@ -1,14 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import StreamingResponse |
|||
|
|||
some_file_path = "large-video-file.mp4" |
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/") |
|||
def main(): |
|||
def iterfile(): # (1) |
|||
with open(some_file_path, mode="rb") as file_like: # (2) |
|||
yield from file_like # (3) |
|||
|
|||
return StreamingResponse(iterfile(), media_type="video/mp4") |
|||
@ -1,10 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import FileResponse |
|||
|
|||
some_file_path = "large-video-file.mp4" |
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/") |
|||
async def main(): |
|||
return FileResponse(some_file_path) |
|||
@ -1,10 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import FileResponse |
|||
|
|||
some_file_path = "large-video-file.mp4" |
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/", response_class=FileResponse) |
|||
async def main(): |
|||
return some_file_path |
|||
@ -1,19 +0,0 @@ |
|||
from typing import Any |
|||
|
|||
import orjson |
|||
from fastapi import FastAPI, Response |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class CustomORJSONResponse(Response): |
|||
media_type = "application/json" |
|||
|
|||
def render(self, content: Any) -> bytes: |
|||
assert orjson is not None, "orjson must be installed" |
|||
return orjson.dumps(content, option=orjson.OPT_INDENT_2) |
|||
|
|||
|
|||
@app.get("/", response_class=CustomORJSONResponse) |
|||
async def main(): |
|||
return {"message": "Hello World"} |
|||
@ -1,9 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import ORJSONResponse |
|||
|
|||
app = FastAPI(default_response_class=ORJSONResponse) |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(): |
|||
return [{"item_id": "Foo"}] |
|||
@ -1,15 +0,0 @@ |
|||
import uvicorn |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/") |
|||
def root(): |
|||
a = "a" |
|||
b = "b" + a |
|||
return {"hello world": b} |
|||
|
|||
|
|||
if __name__ == "__main__": |
|||
uvicorn.run(app, host="0.0.0.0", port=8000) |
|||
@ -1,21 +0,0 @@ |
|||
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"}] |
|||
@ -1,19 +0,0 @@ |
|||
from fastapi import Depends, FastAPI, Header, HTTPException |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
async def verify_token(x_token: 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: 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"}] |
|||
@ -1,6 +0,0 @@ |
|||
async def get_db(): |
|||
db = DBSession() |
|||
try: |
|||
yield db |
|||
finally: |
|||
db.close() |
|||
@ -1,27 +0,0 @@ |
|||
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) |
|||
@ -1,25 +0,0 @@ |
|||
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=Depends(dependency_a)): |
|||
dep_b = generate_dep_b() |
|||
try: |
|||
yield dep_b |
|||
finally: |
|||
dep_b.close(dep_a) |
|||
|
|||
|
|||
async def dependency_c(dep_b=Depends(dependency_b)): |
|||
dep_c = generate_dep_c() |
|||
try: |
|||
yield dep_c |
|||
finally: |
|||
dep_c.close(dep_b) |
|||
@ -1,32 +0,0 @@ |
|||
from typing import Annotated |
|||
|
|||
from fastapi import Depends, FastAPI, HTTPException |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
data = { |
|||
"plumbus": {"description": "Freshly pickled plumbus", "owner": "Morty"}, |
|||
"portal-gun": {"description": "Gun to create portals", "owner": "Rick"}, |
|||
} |
|||
|
|||
|
|||
class OwnerError(Exception): |
|||
pass |
|||
|
|||
|
|||
def get_username(): |
|||
try: |
|||
yield "Rick" |
|||
except OwnerError as e: |
|||
raise HTTPException(status_code=400, detail=f"Owner error: {e}") |
|||
|
|||
|
|||
@app.get("/items/{item_id}") |
|||
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]): |
|||
if item_id not in data: |
|||
raise HTTPException(status_code=404, detail="Item not found") |
|||
item = data[item_id] |
|||
if item["owner"] != username: |
|||
raise OwnerError(username) |
|||
return item |
|||
@ -1,30 +0,0 @@ |
|||
from fastapi import Depends, FastAPI, HTTPException |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
data = { |
|||
"plumbus": {"description": "Freshly pickled plumbus", "owner": "Morty"}, |
|||
"portal-gun": {"description": "Gun to create portals", "owner": "Rick"}, |
|||
} |
|||
|
|||
|
|||
class OwnerError(Exception): |
|||
pass |
|||
|
|||
|
|||
def get_username(): |
|||
try: |
|||
yield "Rick" |
|||
except OwnerError as e: |
|||
raise HTTPException(status_code=400, detail=f"Owner error: {e}") |
|||
|
|||
|
|||
@app.get("/items/{item_id}") |
|||
def get_item(item_id: str, username: str = Depends(get_username)): |
|||
if item_id not in data: |
|||
raise HTTPException(status_code=404, detail="Item not found") |
|||
item = data[item_id] |
|||
if item["owner"] != username: |
|||
raise OwnerError(username) |
|||
return item |
|||
@ -1,29 +0,0 @@ |
|||
from typing import Annotated |
|||
|
|||
from fastapi import Depends, FastAPI, HTTPException |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class InternalError(Exception): |
|||
pass |
|||
|
|||
|
|||
def get_username(): |
|||
try: |
|||
yield "Rick" |
|||
except InternalError: |
|||
print("Oops, we didn't raise again, Britney 😱") |
|||
|
|||
|
|||
@app.get("/items/{item_id}") |
|||
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]): |
|||
if item_id == "portal-gun": |
|||
raise InternalError( |
|||
f"The portal gun is too dangerous to be owned by {username}" |
|||
) |
|||
if item_id != "plumbus": |
|||
raise HTTPException( |
|||
status_code=404, detail="Item not found, there's only a plumbus here" |
|||
) |
|||
return item_id |
|||
@ -1,27 +0,0 @@ |
|||
from fastapi import Depends, FastAPI, HTTPException |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class InternalError(Exception): |
|||
pass |
|||
|
|||
|
|||
def get_username(): |
|||
try: |
|||
yield "Rick" |
|||
except InternalError: |
|||
print("Oops, we didn't raise again, Britney 😱") |
|||
|
|||
|
|||
@app.get("/items/{item_id}") |
|||
def get_item(item_id: str, username: str = Depends(get_username)): |
|||
if item_id == "portal-gun": |
|||
raise InternalError( |
|||
f"The portal gun is too dangerous to be owned by {username}" |
|||
) |
|||
if item_id != "plumbus": |
|||
raise HTTPException( |
|||
status_code=404, detail="Item not found, there's only a plumbus here" |
|||
) |
|||
return item_id |
|||
@ -1,30 +0,0 @@ |
|||
from typing import Annotated |
|||
|
|||
from fastapi import Depends, FastAPI, HTTPException |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class InternalError(Exception): |
|||
pass |
|||
|
|||
|
|||
def get_username(): |
|||
try: |
|||
yield "Rick" |
|||
except InternalError: |
|||
print("We don't swallow the internal error here, we raise again 😎") |
|||
raise |
|||
|
|||
|
|||
@app.get("/items/{item_id}") |
|||
def get_item(item_id: str, username: Annotated[str, Depends(get_username)]): |
|||
if item_id == "portal-gun": |
|||
raise InternalError( |
|||
f"The portal gun is too dangerous to be owned by {username}" |
|||
) |
|||
if item_id != "plumbus": |
|||
raise HTTPException( |
|||
status_code=404, detail="Item not found, there's only a plumbus here" |
|||
) |
|||
return item_id |
|||
@ -1,28 +0,0 @@ |
|||
from fastapi import Depends, FastAPI, HTTPException |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class InternalError(Exception): |
|||
pass |
|||
|
|||
|
|||
def get_username(): |
|||
try: |
|||
yield "Rick" |
|||
except InternalError: |
|||
print("We don't swallow the internal error here, we raise again 😎") |
|||
raise |
|||
|
|||
|
|||
@app.get("/items/{item_id}") |
|||
def get_item(item_id: str, username: str = Depends(get_username)): |
|||
if item_id == "portal-gun": |
|||
raise InternalError( |
|||
f"The portal gun is too dangerous to be owned by {username}" |
|||
) |
|||
if item_id != "plumbus": |
|||
raise HTTPException( |
|||
status_code=404, detail="Item not found, there's only a plumbus here" |
|||
) |
|||
return item_id |
|||
@ -1,17 +0,0 @@ |
|||
from typing import Annotated |
|||
|
|||
from fastapi import Depends, FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
def get_username(): |
|||
try: |
|||
yield "Rick" |
|||
finally: |
|||
print("Cleanup up before response is sent") |
|||
|
|||
|
|||
@app.get("/users/me") |
|||
def get_user_me(username: Annotated[str, Depends(get_username, scope="function")]): |
|||
return username |
|||
@ -1,15 +0,0 @@ |
|||
from fastapi import Depends, FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
def get_username(): |
|||
try: |
|||
yield "Rick" |
|||
finally: |
|||
print("Cleanup up before response is sent") |
|||
|
|||
|
|||
@app.get("/users/me") |
|||
def get_user_me(username: str = Depends(get_username, scope="function")): |
|||
return username |
|||
@ -1,14 +0,0 @@ |
|||
class MySuperContextManager: |
|||
def __init__(self): |
|||
self.db = DBSession() |
|||
|
|||
def __enter__(self): |
|||
return self.db |
|||
|
|||
def __exit__(self, exc_type, exc_value, traceback): |
|||
self.db.close() |
|||
|
|||
|
|||
async def get_db(): |
|||
with MySuperContextManager() as db: |
|||
yield db |
|||
@ -1,23 +0,0 @@ |
|||
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} |
|||
@ -1,21 +0,0 @@ |
|||
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: bool = Depends(checker)): |
|||
return {"fixed_content_in_query": fixed_content_included} |
|||
@ -1,27 +0,0 @@ |
|||
from typing import Annotated |
|||
|
|||
from fastapi import Depends, FastAPI, Header, HTTPException |
|||
|
|||
|
|||
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"}] |
|||
@ -1,25 +0,0 @@ |
|||
from fastapi import Depends, FastAPI, Header, HTTPException |
|||
|
|||
|
|||
async def verify_token(x_token: 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: 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"}] |
|||
@ -1,16 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
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] |
|||
@ -1,14 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.on_event("shutdown") |
|||
def shutdown_event(): |
|||
with open("log.txt", mode="a") as log: |
|||
log.write("Application shutdown") |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(): |
|||
return [{"name": "Foo"}] |
|||
@ -1,28 +0,0 @@ |
|||
from contextlib import asynccontextmanager |
|||
|
|||
from fastapi import FastAPI |
|||
|
|||
|
|||
def fake_answer_to_everything_ml_model(x: float): |
|||
return x * 42 |
|||
|
|||
|
|||
ml_models = {} |
|||
|
|||
|
|||
@asynccontextmanager |
|||
async def lifespan(app: FastAPI): |
|||
# Load the ML model |
|||
ml_models["answer_to_everything"] = fake_answer_to_everything_ml_model |
|||
yield |
|||
# Clean up the ML models and release the resources |
|||
ml_models.clear() |
|||
|
|||
|
|||
app = FastAPI(lifespan=lifespan) |
|||
|
|||
|
|||
@app.get("/predict") |
|||
async def predict(x: float): |
|||
result = ml_models["answer_to_everything"](x) |
|||
return {"result": result} |
|||
@ -1,29 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.openapi.utils import get_openapi |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(): |
|||
return [{"name": "Foo"}] |
|||
|
|||
|
|||
def custom_openapi(): |
|||
if app.openapi_schema: |
|||
return app.openapi_schema |
|||
openapi_schema = get_openapi( |
|||
title="Custom title", |
|||
version="2.5.0", |
|||
summary="This is a very custom OpenAPI schema", |
|||
description="Here's a longer description of the custom **OpenAPI** schema", |
|||
routes=app.routes, |
|||
) |
|||
openapi_schema["info"]["x-logo"] = { |
|||
"url": "https://fastapi.tiangolo.com/img/logo-margin/logo-teal.png" |
|||
} |
|||
app.openapi_schema = openapi_schema |
|||
return app.openapi_schema |
|||
|
|||
|
|||
app.openapi = custom_openapi |
|||
@ -1,20 +0,0 @@ |
|||
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 |
|||
@ -1,8 +0,0 @@ |
|||
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} |
|||
@ -1,8 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/") |
|||
async def root(): |
|||
return {"message": "Hello World"} |
|||
@ -1,8 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/") |
|||
def root(): |
|||
return {"message": "Hello World"} |
|||
@ -1,26 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
price: float |
|||
|
|||
|
|||
class ResponseMessage(BaseModel): |
|||
message: str |
|||
|
|||
|
|||
@app.post("/items/", response_model=ResponseMessage) |
|||
async def create_item(item: Item): |
|||
return {"message": "item received"} |
|||
|
|||
|
|||
@app.get("/items/", response_model=list[Item]) |
|||
async def get_items(): |
|||
return [ |
|||
{"name": "Plumbus", "price": 3}, |
|||
{"name": "Portal Gun", "price": 9001}, |
|||
] |
|||
@ -1,36 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
price: float |
|||
|
|||
|
|||
class ResponseMessage(BaseModel): |
|||
message: str |
|||
|
|||
|
|||
class User(BaseModel): |
|||
username: str |
|||
email: str |
|||
|
|||
|
|||
@app.post("/items/", response_model=ResponseMessage, tags=["items"]) |
|||
async def create_item(item: Item): |
|||
return {"message": "Item received"} |
|||
|
|||
|
|||
@app.get("/items/", response_model=list[Item], tags=["items"]) |
|||
async def get_items(): |
|||
return [ |
|||
{"name": "Plumbus", "price": 3}, |
|||
{"name": "Portal Gun", "price": 9001}, |
|||
] |
|||
|
|||
|
|||
@app.post("/users/", response_model=ResponseMessage, tags=["users"]) |
|||
async def create_user(user: User): |
|||
return {"message": "User received"} |
|||
@ -1,42 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.routing import APIRoute |
|||
from pydantic import BaseModel |
|||
|
|||
|
|||
def custom_generate_unique_id(route: APIRoute): |
|||
return f"{route.tags[0]}-{route.name}" |
|||
|
|||
|
|||
app = FastAPI(generate_unique_id_function=custom_generate_unique_id) |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
price: float |
|||
|
|||
|
|||
class ResponseMessage(BaseModel): |
|||
message: str |
|||
|
|||
|
|||
class User(BaseModel): |
|||
username: str |
|||
email: str |
|||
|
|||
|
|||
@app.post("/items/", response_model=ResponseMessage, tags=["items"]) |
|||
async def create_item(item: Item): |
|||
return {"message": "Item received"} |
|||
|
|||
|
|||
@app.get("/items/", response_model=list[Item], tags=["items"]) |
|||
async def get_items(): |
|||
return [ |
|||
{"name": "Plumbus", "price": 3}, |
|||
{"name": "Portal Gun", "price": 9001}, |
|||
] |
|||
|
|||
|
|||
@app.post("/users/", response_model=ResponseMessage, tags=["users"]) |
|||
async def create_user(user: User): |
|||
return {"message": "User received"} |
|||
@ -1,15 +0,0 @@ |
|||
import json |
|||
from pathlib import Path |
|||
|
|||
file_path = Path("./openapi.json") |
|||
openapi_content = json.loads(file_path.read_text()) |
|||
|
|||
for path_data in openapi_content["paths"].values(): |
|||
for operation in path_data.values(): |
|||
tag = operation["tags"][0] |
|||
operation_id = operation["operationId"] |
|||
to_remove = f"{tag}-" |
|||
new_operation_id = operation_id[len(to_remove) :] |
|||
operation["operationId"] = new_operation_id |
|||
|
|||
file_path.write_text(json.dumps(openapi_content)) |
|||
@ -1,25 +0,0 @@ |
|||
import strawberry |
|||
from fastapi import FastAPI |
|||
from strawberry.fastapi import GraphQLRouter |
|||
|
|||
|
|||
@strawberry.type |
|||
class User: |
|||
name: str |
|||
age: int |
|||
|
|||
|
|||
@strawberry.type |
|||
class Query: |
|||
@strawberry.field |
|||
def user(self) -> User: |
|||
return User(name="Patrick", age=100) |
|||
|
|||
|
|||
schema = strawberry.Schema(query=Query) |
|||
|
|||
|
|||
graphql_app = GraphQLRouter(schema) |
|||
|
|||
app = FastAPI() |
|||
app.include_router(graphql_app, prefix="/graphql") |
|||
@ -1,12 +0,0 @@ |
|||
from fastapi import FastAPI, HTTPException |
|||
|
|||
app = FastAPI() |
|||
|
|||
items = {"foo": "The Foo Wrestlers"} |
|||
|
|||
|
|||
@app.get("/items/{item_id}") |
|||
async def read_item(item_id: str): |
|||
if item_id not in items: |
|||
raise HTTPException(status_code=404, detail="Item not found") |
|||
return {"item": items[item_id]} |
|||
@ -1,16 +0,0 @@ |
|||
from fastapi import FastAPI, HTTPException |
|||
|
|||
app = FastAPI() |
|||
|
|||
items = {"foo": "The Foo Wrestlers"} |
|||
|
|||
|
|||
@app.get("/items-header/{item_id}") |
|||
async def read_item_header(item_id: str): |
|||
if item_id not in items: |
|||
raise HTTPException( |
|||
status_code=404, |
|||
detail="Item not found", |
|||
headers={"X-Error": "There goes my error"}, |
|||
) |
|||
return {"item": items[item_id]} |
|||
@ -1,25 +0,0 @@ |
|||
from fastapi import FastAPI, Request |
|||
from fastapi.responses import JSONResponse |
|||
|
|||
|
|||
class UnicornException(Exception): |
|||
def __init__(self, name: str): |
|||
self.name = name |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.exception_handler(UnicornException) |
|||
async def unicorn_exception_handler(request: Request, exc: UnicornException): |
|||
return JSONResponse( |
|||
status_code=418, |
|||
content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."}, |
|||
) |
|||
|
|||
|
|||
@app.get("/unicorns/{name}") |
|||
async def read_unicorn(name: str): |
|||
if name == "yolo": |
|||
raise UnicornException(name=name) |
|||
return {"unicorn_name": name} |
|||
@ -1,26 +0,0 @@ |
|||
from fastapi import FastAPI, HTTPException |
|||
from fastapi.exceptions import RequestValidationError |
|||
from fastapi.responses import PlainTextResponse |
|||
from starlette.exceptions import HTTPException as StarletteHTTPException |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.exception_handler(StarletteHTTPException) |
|||
async def http_exception_handler(request, exc): |
|||
return PlainTextResponse(str(exc.detail), status_code=exc.status_code) |
|||
|
|||
|
|||
@app.exception_handler(RequestValidationError) |
|||
async def validation_exception_handler(request, exc: RequestValidationError): |
|||
message = "Validation errors:" |
|||
for error in exc.errors(): |
|||
message += f"\nField: {error['loc']}, Error: {error['msg']}" |
|||
return PlainTextResponse(message, status_code=400) |
|||
|
|||
|
|||
@app.get("/items/{item_id}") |
|||
async def read_item(item_id: int): |
|||
if item_id == 3: |
|||
raise HTTPException(status_code=418, detail="Nope! I don't like 3.") |
|||
return {"item_id": item_id} |
|||
@ -1,25 +0,0 @@ |
|||
from fastapi import FastAPI, Request |
|||
from fastapi.encoders import jsonable_encoder |
|||
from fastapi.exceptions import RequestValidationError |
|||
from fastapi.responses import JSONResponse |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.exception_handler(RequestValidationError) |
|||
async def validation_exception_handler(request: Request, exc: RequestValidationError): |
|||
return JSONResponse( |
|||
status_code=422, |
|||
content=jsonable_encoder({"detail": exc.errors(), "body": exc.body}), |
|||
) |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
title: str |
|||
size: int |
|||
|
|||
|
|||
@app.post("/items/") |
|||
async def create_item(item: Item): |
|||
return item |
|||
@ -1,28 +0,0 @@ |
|||
from fastapi import FastAPI, HTTPException |
|||
from fastapi.exception_handlers import ( |
|||
http_exception_handler, |
|||
request_validation_exception_handler, |
|||
) |
|||
from fastapi.exceptions import RequestValidationError |
|||
from starlette.exceptions import HTTPException as StarletteHTTPException |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.exception_handler(StarletteHTTPException) |
|||
async def custom_http_exception_handler(request, exc): |
|||
print(f"OMG! An HTTP error!: {repr(exc)}") |
|||
return await http_exception_handler(request, exc) |
|||
|
|||
|
|||
@app.exception_handler(RequestValidationError) |
|||
async def validation_exception_handler(request, exc): |
|||
print(f"OMG! The client sent invalid data!: {exc}") |
|||
return await request_validation_exception_handler(request, exc) |
|||
|
|||
|
|||
@app.get("/items/{item_id}") |
|||
async def read_item(item_id: int): |
|||
if item_id == 3: |
|||
raise HTTPException(status_code=418, detail="Nope! I don't like 3.") |
|||
return {"item_id": item_id} |
|||
@ -1,38 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
description = """ |
|||
ChimichangApp API helps you do awesome stuff. 🚀 |
|||
|
|||
## Items |
|||
|
|||
You can **read items**. |
|||
|
|||
## Users |
|||
|
|||
You will be able to: |
|||
|
|||
* **Create users** (_not implemented_). |
|||
* **Read users** (_not implemented_). |
|||
""" |
|||
|
|||
app = FastAPI( |
|||
title="ChimichangApp", |
|||
description=description, |
|||
summary="Deadpool's favorite app. Nuff said.", |
|||
version="0.0.1", |
|||
terms_of_service="http://example.com/terms/", |
|||
contact={ |
|||
"name": "Deadpoolio the Amazing", |
|||
"url": "http://x-force.example.com/contact/", |
|||
"email": "[email protected]", |
|||
}, |
|||
license_info={ |
|||
"name": "Apache 2.0", |
|||
"identifier": "Apache-2.0", |
|||
}, |
|||
) |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(): |
|||
return [{"name": "Katana"}] |
|||
@ -1,38 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
description = """ |
|||
ChimichangApp API helps you do awesome stuff. 🚀 |
|||
|
|||
## Items |
|||
|
|||
You can **read items**. |
|||
|
|||
## Users |
|||
|
|||
You will be able to: |
|||
|
|||
* **Create users** (_not implemented_). |
|||
* **Read users** (_not implemented_). |
|||
""" |
|||
|
|||
app = FastAPI( |
|||
title="ChimichangApp", |
|||
description=description, |
|||
summary="Deadpool's favorite app. Nuff said.", |
|||
version="0.0.1", |
|||
terms_of_service="http://example.com/terms/", |
|||
contact={ |
|||
"name": "Deadpoolio the Amazing", |
|||
"url": "http://x-force.example.com/contact/", |
|||
"email": "[email protected]", |
|||
}, |
|||
license_info={ |
|||
"name": "Apache 2.0", |
|||
"url": "https://www.apache.org/licenses/LICENSE-2.0.html", |
|||
}, |
|||
) |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(): |
|||
return [{"name": "Katana"}] |
|||
@ -1,8 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI(openapi_url="/api/v1/openapi.json") |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(): |
|||
return [{"name": "Foo"}] |
|||
@ -1,8 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
app = FastAPI(docs_url="/documentation", redoc_url=None) |
|||
|
|||
|
|||
@app.get("/items/") |
|||
async def read_items(): |
|||
return [{"name": "Foo"}] |
|||
@ -1,28 +0,0 @@ |
|||
from fastapi import FastAPI |
|||
|
|||
tags_metadata = [ |
|||
{ |
|||
"name": "users", |
|||
"description": "Operations with users. The **login** logic is also here.", |
|||
}, |
|||
{ |
|||
"name": "items", |
|||
"description": "Manage items. So _fancy_ they have their own docs.", |
|||
"externalDocs": { |
|||
"description": "Items external docs", |
|||
"url": "https://fastapi.tiangolo.com/", |
|||
}, |
|||
}, |
|||
] |
|||
|
|||
app = FastAPI(openapi_tags=tags_metadata) |
|||
|
|||
|
|||
@app.get("/users/", tags=["users"]) |
|||
async def get_users(): |
|||
return [{"name": "Harry"}, {"name": "Ron"}] |
|||
|
|||
|
|||
@app.get("/items/", tags=["items"]) |
|||
async def get_items(): |
|||
return [{"name": "wand"}, {"name": "flying broom"}] |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue