+++ /dev/null
-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"})
+++ /dev/null
-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"})
+++ /dev/null
-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"}
+++ /dev/null
-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"}
+++ /dev/null
-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"
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.get("/")
-async def read_main():
- return {"msg": "Hello World"}
+++ /dev/null
-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"}
+++ /dev/null
-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"}
+++ /dev/null
-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"}
+++ /dev/null
-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"}
+++ /dev/null
-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 == {}
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.get("/")
-async def root():
- return {"message": "Tomato"}
+++ /dev/null
-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"}
+++ /dev/null
-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}
+++ /dev/null
-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"}
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.get("/items/")
-def read_items():
- return ["plumbus", "portal gun"]
+++ /dev/null
-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")}
+++ /dev/null
-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")}
+++ /dev/null
-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")}
+++ /dev/null
-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")}
+++ /dev/null
-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")
+++ /dev/null
-from fastapi import APIRouter
-
-router = APIRouter()
-
-
-@router.post("/")
-async def update_admin():
- return {"message": "Admin getting schwifty"}
+++ /dev/null
-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!"}
+++ /dev/null
-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"}
+++ /dev/null
-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}
+++ /dev/null
-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")
+++ /dev/null
-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!"}
+++ /dev/null
-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
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.post("/index-weights/")
-async def create_index_weights(weights: dict[int, float]):
- return weights
+++ /dev/null
-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"}
+++ /dev/null
-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}"}
+++ /dev/null
-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}"}
+++ /dev/null
-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}"}
+++ /dev/null
-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"}
+++ /dev/null
-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}"}
+++ /dev/null
-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}"}
+++ /dev/null
-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"}]
+++ /dev/null
-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"}])
+++ /dev/null
-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>
- """
+++ /dev/null
-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)
+++ /dev/null
-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()
+++ /dev/null
-from fastapi import FastAPI
-from fastapi.responses import PlainTextResponse
-
-app = FastAPI()
-
-
-@app.get("/", response_class=PlainTextResponse)
-async def main():
- return "Hello World"
+++ /dev/null
-from fastapi import FastAPI
-from fastapi.responses import RedirectResponse
-
-app = FastAPI()
-
-
-@app.get("/typer")
-async def redirect_typer():
- return RedirectResponse("https://typer.tiangolo.com")
+++ /dev/null
-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"
+++ /dev/null
-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/"
+++ /dev/null
-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())
+++ /dev/null
-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")
+++ /dev/null
-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)
+++ /dev/null
-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
+++ /dev/null
-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"}
+++ /dev/null
-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"}]
+++ /dev/null
-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)
+++ /dev/null
-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"}]
+++ /dev/null
-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"}]
+++ /dev/null
-async def get_db():
- db = DBSession()
- try:
- yield db
- finally:
- db.close()
+++ /dev/null
-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)
+++ /dev/null
-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)
+++ /dev/null
-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
+++ /dev/null
-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
+++ /dev/null
-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
+++ /dev/null
-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
+++ /dev/null
-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
+++ /dev/null
-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
+++ /dev/null
-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
+++ /dev/null
-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
+++ /dev/null
-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
+++ /dev/null
-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}
+++ /dev/null
-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}
+++ /dev/null
-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"}]
+++ /dev/null
-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"}]
+++ /dev/null
-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]
+++ /dev/null
-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"}]
+++ /dev/null
-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}
+++ /dev/null
-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
+++ /dev/null
-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
+++ /dev/null
-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}
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.get("/")
-async def root():
- return {"message": "Hello World"}
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.get("/")
-def root():
- return {"message": "Hello World"}
+++ /dev/null
-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},
- ]
+++ /dev/null
-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"}
+++ /dev/null
-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"}
+++ /dev/null
-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))
+++ /dev/null
-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")
+++ /dev/null
-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]}
+++ /dev/null
-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]}
+++ /dev/null
-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}
+++ /dev/null
-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}
+++ /dev/null
-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
+++ /dev/null
-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}
+++ /dev/null
-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": "dp@x-force.example.com",
- },
- license_info={
- "name": "Apache 2.0",
- "identifier": "Apache-2.0",
- },
-)
-
-
-@app.get("/items/")
-async def read_items():
- return [{"name": "Katana"}]
+++ /dev/null
-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": "dp@x-force.example.com",
- },
- 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"}]
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI(openapi_url="/api/v1/openapi.json")
-
-
-@app.get("/items/")
-async def read_items():
- return [{"name": "Foo"}]
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI(docs_url="/documentation", redoc_url=None)
-
-
-@app.get("/items/")
-async def read_items():
- return [{"name": "Foo"}]
+++ /dev/null
-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"}]
+++ /dev/null
-import time
-
-from fastapi import FastAPI, Request
-
-app = FastAPI()
-
-
-@app.middleware("http")
-async def add_process_time_header(request: Request, call_next):
- start_time = time.perf_counter()
- response = await call_next(request)
- process_time = time.perf_counter() - start_time
- response.headers["X-Process-Time"] = str(process_time)
- return response
+++ /dev/null
-from datetime import datetime
-
-from fastapi import FastAPI
-from pydantic import BaseModel
-
-app = FastAPI()
-
-
-class Subscription(BaseModel):
- username: str
- monthly_fee: float
- start_date: datetime
-
-
-@app.webhooks.post("new-subscription")
-def new_subscription(body: Subscription):
- """
- When a new user subscribes to your service we'll send you a POST request with this
- data to the URL that you register for the event `new-subscription` in the dashboard.
- """
-
-
-@app.get("/users/")
-def read_users():
- return ["Rick", "Morty"]
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.get("/items/", operation_id="some_specific_id_you_define")
-async def read_items():
- return [{"item_id": "Foo"}]
+++ /dev/null
-from fastapi import FastAPI
-from fastapi.routing import APIRoute
-
-app = FastAPI()
-
-
-@app.get("/items/")
-async def read_items():
- return [{"item_id": "Foo"}]
-
-
-def use_route_names_as_operation_ids(app: FastAPI) -> None:
- """
- Simplify operation IDs so that generated API clients have simpler function
- names.
-
- Should be called only after all routes have been added.
- """
- for route in app.routes:
- if isinstance(route, APIRoute):
- route.operation_id = route.name # in this case, 'read_items'
-
-
-use_route_names_as_operation_ids(app)
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.get("/items/", include_in_schema=False)
-async def read_items():
- return [{"item_id": "Foo"}]
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.get("/items/", openapi_extra={"x-aperture-labs-portal": "blue"})
-async def read_items():
- return [{"item_id": "portal-gun"}]
+++ /dev/null
-from fastapi import FastAPI, Request
-
-app = FastAPI()
-
-
-def magic_data_reader(raw_body: bytes):
- return {
- "size": len(raw_body),
- "content": {
- "name": "Maaaagic",
- "price": 42,
- "description": "Just kiddin', no magic here. β¨",
- },
- }
-
-
-@app.post(
- "/items/",
- openapi_extra={
- "requestBody": {
- "content": {
- "application/json": {
- "schema": {
- "required": ["name", "price"],
- "type": "object",
- "properties": {
- "name": {"type": "string"},
- "price": {"type": "number"},
- "description": {"type": "string"},
- },
- }
- }
- },
- "required": True,
- },
- },
-)
-async def create_item(request: Request):
- raw_body = await request.body()
- data = magic_data_reader(raw_body)
- return data
+++ /dev/null
-import yaml
-from fastapi import FastAPI, HTTPException, Request
-from pydantic import BaseModel, ValidationError
-
-app = FastAPI()
-
-
-class Item(BaseModel):
- name: str
- tags: list[str]
-
-
-@app.post(
- "/items/",
- openapi_extra={
- "requestBody": {
- "content": {"application/x-yaml": {"schema": Item.model_json_schema()}},
- "required": True,
- },
- },
-)
-async def create_item(request: Request):
- raw_body = await request.body()
- try:
- data = yaml.safe_load(raw_body)
- except yaml.YAMLError:
- raise HTTPException(status_code=422, detail="Invalid YAML")
- try:
- item = Item.model_validate(data)
- except ValidationError as e:
- raise HTTPException(status_code=422, detail=e.errors(include_url=False))
- return item
+++ /dev/null
-from enum import Enum
-
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-class Tags(Enum):
- items = "items"
- users = "users"
-
-
-@app.get("/items/", tags=[Tags.items])
-async def get_items():
- return ["Portal gun", "Plumbus"]
-
-
-@app.get("/users/", tags=[Tags.users])
-async def read_users():
- return ["Rick", "Morty"]
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.get("/items/", tags=["items"])
-async def read_items():
- return [{"name": "Foo", "price": 42}]
-
-
-@app.get("/users/", tags=["users"])
-async def read_users():
- return [{"username": "johndoe"}]
-
-
-@app.get("/elements/", tags=["items"], deprecated=True)
-async def read_elements():
- return [{"item_id": "Foo"}]
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.get("/items/{item_id}")
-async def read_item(item_id):
- return {"item_id": item_id}
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.get("/items/{item_id}")
-async def read_item(item_id: int):
- return {"item_id": item_id}
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.get("/users/me")
-async def read_user_me():
- return {"user_id": "the current user"}
-
-
-@app.get("/users/{user_id}")
-async def read_user(user_id: str):
- return {"user_id": user_id}
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.get("/users")
-async def read_users():
- return ["Rick", "Morty"]
-
-
-@app.get("/users")
-async def read_users2():
- return ["Bean", "Elfo"]
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.get("/files/{file_path:path}")
-async def read_file(file_path: str):
- return {"file_path": file_path}
+++ /dev/null
-from enum import Enum
-
-from fastapi import FastAPI
-
-
-class ModelName(str, Enum):
- alexnet = "alexnet"
- resnet = "resnet"
- lenet = "lenet"
-
-
-app = FastAPI()
-
-
-@app.get("/models/{model_name}")
-async def get_model(model_name: ModelName):
- if model_name is ModelName.alexnet:
- return {"model_name": model_name, "message": "Deep Learning FTW!"}
-
- if model_name.value == "lenet":
- return {"model_name": model_name, "message": "LeCNN all the images"}
-
- return {"model_name": model_name, "message": "Have some residuals"}
+++ /dev/null
-from typing import Annotated
-
-from fastapi import FastAPI, Path
-
-app = FastAPI()
-
-
-@app.get("/items/{item_id}")
-async def read_items(
- q: str, item_id: Annotated[int, Path(title="The ID of the item to get")]
-):
- results = {"item_id": item_id}
- if q:
- results.update({"q": q})
- return results
+++ /dev/null
-from fastapi import FastAPI, Path
-
-app = FastAPI()
-
-
-@app.get("/items/{item_id}")
-async def read_items(q: str, item_id: int = Path(title="The ID of the item to get")):
- results = {"item_id": item_id}
- if q:
- results.update({"q": q})
- return results
+++ /dev/null
-from typing import Annotated
-
-from fastapi import FastAPI, Path
-
-app = FastAPI()
-
-
-@app.get("/items/{item_id}")
-async def read_items(
- item_id: Annotated[int, Path(title="The ID of the item to get")], q: str
-):
- results = {"item_id": item_id}
- if q:
- results.update({"q": q})
- return results
+++ /dev/null
-from fastapi import FastAPI, Path
-
-app = FastAPI()
-
-
-@app.get("/items/{item_id}")
-async def read_items(*, item_id: int = Path(title="The ID of the item to get"), q: str):
- results = {"item_id": item_id}
- if q:
- results.update({"q": q})
- return results
+++ /dev/null
-from typing import Annotated
-
-from fastapi import FastAPI, Path
-
-app = FastAPI()
-
-
-@app.get("/items/{item_id}")
-async def read_items(
- item_id: Annotated[int, Path(title="The ID of the item to get", ge=1)], q: str
-):
- results = {"item_id": item_id}
- if q:
- results.update({"q": q})
- return results
+++ /dev/null
-from fastapi import FastAPI, Path
-
-app = FastAPI()
-
-
-@app.get("/items/{item_id}")
-async def read_items(
- *, item_id: int = Path(title="The ID of the item to get", ge=1), q: str
-):
- results = {"item_id": item_id}
- if q:
- results.update({"q": q})
- return results
+++ /dev/null
-from typing import Annotated
-
-from fastapi import FastAPI, Path
-
-app = FastAPI()
-
-
-@app.get("/items/{item_id}")
-async def read_items(
- item_id: Annotated[int, Path(title="The ID of the item to get", gt=0, le=1000)],
- q: str,
-):
- results = {"item_id": item_id}
- if q:
- results.update({"q": q})
- return results
+++ /dev/null
-from fastapi import FastAPI, Path
-
-app = FastAPI()
-
-
-@app.get("/items/{item_id}")
-async def read_items(
- *,
- item_id: int = Path(title="The ID of the item to get", gt=0, le=1000),
- q: str,
-):
- results = {"item_id": item_id}
- if q:
- results.update({"q": q})
- return results
+++ /dev/null
-from typing import Annotated
-
-from fastapi import FastAPI, Path, Query
-
-app = FastAPI()
-
-
-@app.get("/items/{item_id}")
-async def read_items(
- *,
- item_id: Annotated[int, Path(title="The ID of the item to get", ge=0, le=1000)],
- q: str,
- size: Annotated[float, Query(gt=0, lt=10.5)],
-):
- results = {"item_id": item_id}
- if q:
- results.update({"q": q})
- if size:
- results.update({"size": size})
- return results
+++ /dev/null
-from fastapi import FastAPI, Path, Query
-
-app = FastAPI()
-
-
-@app.get("/items/{item_id}")
-async def read_items(
- *,
- item_id: int = Path(title="The ID of the item to get", ge=0, le=1000),
- q: str,
- size: float = Query(gt=0, lt=10.5),
-):
- results = {"item_id": item_id}
- if q:
- results.update({"q": q})
- if size:
- results.update({"size": size})
- return results
+++ /dev/null
-def get_full_name(first_name, last_name):
- full_name = first_name.title() + " " + last_name.title()
- return full_name
-
-
-print(get_full_name("john", "doe"))
+++ /dev/null
-def get_full_name(first_name: str, last_name: str):
- full_name = first_name.title() + " " + last_name.title()
- return full_name
-
-
-print(get_full_name("john", "doe"))
+++ /dev/null
-def get_name_with_age(name: str, age: int):
- name_with_age = name + " is this old: " + age
- return name_with_age
+++ /dev/null
-def get_name_with_age(name: str, age: int):
- name_with_age = name + " is this old: " + str(age)
- return name_with_age
+++ /dev/null
-def process_items(items: list[str]):
- for item in items:
- print(item)
+++ /dev/null
-def process_items(items_t: tuple[int, int, str], items_s: set[bytes]):
- return items_t, items_s
+++ /dev/null
-def process_items(prices: dict[str, float]):
- for item_name, item_price in prices.items():
- print(item_name)
- print(item_price)
+++ /dev/null
-from typing import Union
-
-
-def process_item(item: Union[int, str]):
- print(item)
+++ /dev/null
-from typing import Optional
-
-
-def say_hi(name: Optional[str] = None):
- if name is not None:
- print(f"Hey {name}!")
- else:
- print("Hello World")
+++ /dev/null
-from typing import Union
-
-
-def say_hi(name: Union[str, None] = None):
- if name is not None:
- print(f"Hey {name}!")
- else:
- print("Hello World")
+++ /dev/null
-def say_hi(name: str | None):
- print(f"Hey {name}!")
+++ /dev/null
-from typing import Optional
-
-
-def say_hi(name: Optional[str]):
- print(f"Hey {name}!")
+++ /dev/null
-class Person:
- def __init__(self, name: str):
- self.name = name
-
-
-def get_person_name(one_person: Person):
- return one_person.name
+++ /dev/null
-from typing import Annotated
-
-
-def say_hello(name: Annotated[str, "this is just metadata"]) -> str:
- return f"Hello {name}"
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
-
-
-@app.get("/items/")
-async def read_item(skip: int = 0, limit: int = 10):
- return fake_items_db[skip : skip + limit]
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.get("/items/{item_id}")
-async def read_user_item(item_id: str, needy: str):
- item = {"item_id": item_id, "needy": needy}
- return item
+++ /dev/null
-from typing import Annotated
-
-from fastapi import FastAPI, Query
-
-app = FastAPI()
-
-
-@app.get("/items/")
-async def read_items(q: Annotated[str, Query(min_length=3)] = "fixedquery"):
- results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
- if q:
- results.update({"q": q})
- return results
+++ /dev/null
-from fastapi import FastAPI, Query
-
-app = FastAPI()
-
-
-@app.get("/items/")
-async def read_items(q: str = Query(default="fixedquery", min_length=3)):
- results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
- if q:
- results.update({"q": q})
- return results
+++ /dev/null
-from typing import Annotated
-
-from fastapi import FastAPI, Query
-
-app = FastAPI()
-
-
-@app.get("/items/")
-async def read_items(q: Annotated[str, Query(min_length=3)]):
- results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
- if q:
- results.update({"q": q})
- return results
+++ /dev/null
-from fastapi import FastAPI, Query
-
-app = FastAPI()
-
-
-@app.get("/items/")
-async def read_items(q: str = Query(min_length=3)):
- results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
- if q:
- results.update({"q": q})
- return results
+++ /dev/null
-from typing import Annotated
-
-from fastapi import FastAPI, Query
-
-app = FastAPI()
-
-
-@app.get("/items/")
-async def read_items(q: Annotated[list[str], Query()] = ["foo", "bar"]):
- query_items = {"q": q}
- return query_items
+++ /dev/null
-from fastapi import FastAPI, Query
-
-app = FastAPI()
-
-
-@app.get("/items/")
-async def read_items(q: list[str] = Query(default=["foo", "bar"])):
- query_items = {"q": q}
- return query_items
+++ /dev/null
-from typing import Annotated
-
-from fastapi import FastAPI, Query
-
-app = FastAPI()
-
-
-@app.get("/items/")
-async def read_items(q: Annotated[list, Query()] = []):
- query_items = {"q": q}
- return query_items
+++ /dev/null
-from fastapi import FastAPI, Query
-
-app = FastAPI()
-
-
-@app.get("/items/")
-async def read_items(q: list = Query(default=[])):
- query_items = {"q": q}
- return query_items
+++ /dev/null
-from typing import Annotated
-
-from fastapi import FastAPI, File, UploadFile
-
-app = FastAPI()
-
-
-@app.post("/files/")
-async def create_file(file: Annotated[bytes, File(description="A file read as bytes")]):
- return {"file_size": len(file)}
-
-
-@app.post("/uploadfile/")
-async def create_upload_file(
- file: Annotated[UploadFile, File(description="A file read as UploadFile")],
-):
- return {"filename": file.filename}
+++ /dev/null
-from fastapi import FastAPI, File, UploadFile
-
-app = FastAPI()
-
-
-@app.post("/files/")
-async def create_file(file: bytes = File(description="A file read as bytes")):
- return {"file_size": len(file)}
-
-
-@app.post("/uploadfile/")
-async def create_upload_file(
- file: UploadFile = File(description="A file read as UploadFile"),
-):
- return {"filename": file.filename}
+++ /dev/null
-from typing import Annotated
-
-from fastapi import FastAPI, File, UploadFile
-
-app = FastAPI()
-
-
-@app.post("/files/")
-async def create_file(file: Annotated[bytes, File()]):
- return {"file_size": len(file)}
-
-
-@app.post("/uploadfile/")
-async def create_upload_file(file: UploadFile):
- return {"filename": file.filename}
+++ /dev/null
-from fastapi import FastAPI, File, UploadFile
-
-app = FastAPI()
-
-
-@app.post("/files/")
-async def create_file(file: bytes = File()):
- return {"file_size": len(file)}
-
-
-@app.post("/uploadfile/")
-async def create_upload_file(file: UploadFile):
- return {"filename": file.filename}
+++ /dev/null
-from typing import Annotated
-
-from fastapi import FastAPI, File, UploadFile
-from fastapi.responses import HTMLResponse
-
-app = FastAPI()
-
-
-@app.post("/files/")
-async def create_files(files: Annotated[list[bytes], File()]):
- return {"file_sizes": [len(file) for file in files]}
-
-
-@app.post("/uploadfiles/")
-async def create_upload_files(files: list[UploadFile]):
- return {"filenames": [file.filename for file in files]}
-
-
-@app.get("/")
-async def main():
- content = """
-<body>
-<form action="/files/" enctype="multipart/form-data" method="post">
-<input name="files" type="file" multiple>
-<input type="submit">
-</form>
-<form action="/uploadfiles/" enctype="multipart/form-data" method="post">
-<input name="files" type="file" multiple>
-<input type="submit">
-</form>
-</body>
- """
- return HTMLResponse(content=content)
+++ /dev/null
-from fastapi import FastAPI, File, UploadFile
-from fastapi.responses import HTMLResponse
-
-app = FastAPI()
-
-
-@app.post("/files/")
-async def create_files(files: list[bytes] = File()):
- return {"file_sizes": [len(file) for file in files]}
-
-
-@app.post("/uploadfiles/")
-async def create_upload_files(files: list[UploadFile]):
- return {"filenames": [file.filename for file in files]}
-
-
-@app.get("/")
-async def main():
- content = """
-<body>
-<form action="/files/" enctype="multipart/form-data" method="post">
-<input name="files" type="file" multiple>
-<input type="submit">
-</form>
-<form action="/uploadfiles/" enctype="multipart/form-data" method="post">
-<input name="files" type="file" multiple>
-<input type="submit">
-</form>
-</body>
- """
- return HTMLResponse(content=content)
+++ /dev/null
-from typing import Annotated
-
-from fastapi import FastAPI, File, UploadFile
-from fastapi.responses import HTMLResponse
-
-app = FastAPI()
-
-
-@app.post("/files/")
-async def create_files(
- files: Annotated[list[bytes], File(description="Multiple files as bytes")],
-):
- return {"file_sizes": [len(file) for file in files]}
-
-
-@app.post("/uploadfiles/")
-async def create_upload_files(
- files: Annotated[
- list[UploadFile], File(description="Multiple files as UploadFile")
- ],
-):
- return {"filenames": [file.filename for file in files]}
-
-
-@app.get("/")
-async def main():
- content = """
-<body>
-<form action="/files/" enctype="multipart/form-data" method="post">
-<input name="files" type="file" multiple>
-<input type="submit">
-</form>
-<form action="/uploadfiles/" enctype="multipart/form-data" method="post">
-<input name="files" type="file" multiple>
-<input type="submit">
-</form>
-</body>
- """
- return HTMLResponse(content=content)
+++ /dev/null
-from fastapi import FastAPI, File, UploadFile
-from fastapi.responses import HTMLResponse
-
-app = FastAPI()
-
-
-@app.post("/files/")
-async def create_files(
- files: list[bytes] = File(description="Multiple files as bytes"),
-):
- return {"file_sizes": [len(file) for file in files]}
-
-
-@app.post("/uploadfiles/")
-async def create_upload_files(
- files: list[UploadFile] = File(description="Multiple files as UploadFile"),
-):
- return {"filenames": [file.filename for file in files]}
-
-
-@app.get("/")
-async def main():
- content = """
-<body>
-<form action="/files/" enctype="multipart/form-data" method="post">
-<input name="files" type="file" multiple>
-<input type="submit">
-</form>
-<form action="/uploadfiles/" enctype="multipart/form-data" method="post">
-<input name="files" type="file" multiple>
-<input type="submit">
-</form>
-</body>
- """
- return HTMLResponse(content=content)
+++ /dev/null
-from typing import Annotated
-
-from fastapi import FastAPI, Form
-from pydantic import BaseModel
-
-app = FastAPI()
-
-
-class FormData(BaseModel):
- username: str
- password: str
-
-
-@app.post("/login/")
-async def login(data: Annotated[FormData, Form()]):
- return data
+++ /dev/null
-from fastapi import FastAPI, Form
-from pydantic import BaseModel
-
-app = FastAPI()
-
-
-class FormData(BaseModel):
- username: str
- password: str
-
-
-@app.post("/login/")
-async def login(data: FormData = Form()):
- return data
+++ /dev/null
-from typing import Annotated
-
-from fastapi import FastAPI, Form
-from pydantic import BaseModel
-
-app = FastAPI()
-
-
-class FormData(BaseModel):
- username: str
- password: str
- model_config = {"extra": "forbid"}
-
-
-@app.post("/login/")
-async def login(data: Annotated[FormData, Form()]):
- return data
+++ /dev/null
-from fastapi import FastAPI, Form
-from pydantic import BaseModel
-
-app = FastAPI()
-
-
-class FormData(BaseModel):
- username: str
- password: str
- model_config = {"extra": "forbid"}
-
-
-@app.post("/login/")
-async def login(data: FormData = Form()):
- return data
+++ /dev/null
-from typing import Annotated
-
-from fastapi import FastAPI, Form
-
-app = FastAPI()
-
-
-@app.post("/login/")
-async def login(username: Annotated[str, Form()], password: Annotated[str, Form()]):
- return {"username": username}
+++ /dev/null
-from fastapi import FastAPI, Form
-
-app = FastAPI()
-
-
-@app.post("/login/")
-async def login(username: str = Form(), password: str = Form()):
- return {"username": username}
+++ /dev/null
-from typing import Annotated
-
-from fastapi import FastAPI, File, Form, UploadFile
-
-app = FastAPI()
-
-
-@app.post("/files/")
-async def create_file(
- file: Annotated[bytes, File()],
- fileb: Annotated[UploadFile, File()],
- token: Annotated[str, Form()],
-):
- return {
- "file_size": len(file),
- "token": token,
- "fileb_content_type": fileb.content_type,
- }
+++ /dev/null
-from fastapi import FastAPI, File, Form, UploadFile
-
-app = FastAPI()
-
-
-@app.post("/files/")
-async def create_file(
- file: bytes = File(), fileb: UploadFile = File(), token: str = Form()
-):
- return {
- "file_size": len(file),
- "token": token,
- "fileb_content_type": fileb.content_type,
- }
+++ /dev/null
-from fastapi import FastAPI, Response, status
-
-app = FastAPI()
-
-tasks = {"foo": "Listen to the Bar Fighters"}
-
-
-@app.put("/get-or-create-task/{task_id}", status_code=200)
-def get_or_create_task(task_id: str, response: Response):
- if task_id not in tasks:
- tasks[task_id] = "This didn't exist before"
- response.status_code = status.HTTP_201_CREATED
- return tasks[task_id]
+++ /dev/null
-from fastapi import FastAPI
-from fastapi.responses import JSONResponse
-
-app = FastAPI()
-
-
-@app.post("/cookie/")
-def create_cookie():
- content = {"message": "Come to the dark side, we have cookies"}
- response = JSONResponse(content=content)
- response.set_cookie(key="fakesession", value="fake-cookie-session-value")
- return response
+++ /dev/null
-from fastapi import FastAPI, Response
-
-app = FastAPI()
-
-
-@app.post("/cookie-and-object/")
-def create_cookie(response: Response):
- response.set_cookie(key="fakesession", value="fake-cookie-session-value")
- return {"message": "Come to the dark side, we have cookies"}
+++ /dev/null
-from fastapi import FastAPI, Response
-
-app = FastAPI()
-
-
-@app.get("/legacy/")
-def get_legacy_data():
- data = """<?xml version="1.0"?>
- <shampoo>
- <Header>
- Apply shampoo here.
- </Header>
- <Body>
- You'll have to use soap here.
- </Body>
- </shampoo>
- """
- return Response(content=data, media_type="application/xml")
+++ /dev/null
-from fastapi import FastAPI
-from fastapi.responses import JSONResponse
-
-app = FastAPI()
-
-
-@app.get("/headers/")
-def get_headers():
- content = {"message": "Hello World"}
- headers = {"X-Cat-Dog": "alone in the world", "Content-Language": "en-US"}
- return JSONResponse(content=content, headers=headers)
+++ /dev/null
-from fastapi import FastAPI, Response
-
-app = FastAPI()
-
-
-@app.get("/headers-and-object/")
-def get_headers(response: Response):
- response.headers["X-Cat-Dog"] = "alone in the world"
- return {"message": "Hello World"}
+++ /dev/null
-from fastapi import FastAPI, Response
-from fastapi.responses import JSONResponse, RedirectResponse
-
-app = FastAPI()
-
-
-@app.get("/portal")
-async def get_portal(teleport: bool = False) -> Response:
- if teleport:
- return RedirectResponse(url="https://www.youtube.com/watch?v=dQw4w9WgXcQ")
- return JSONResponse(content={"message": "Here's your interdimensional portal."})
+++ /dev/null
-from fastapi import FastAPI
-from fastapi.responses import RedirectResponse
-
-app = FastAPI()
-
-
-@app.get("/teleport")
-async def get_teleport() -> RedirectResponse:
- return RedirectResponse(url="https://www.youtube.com/watch?v=dQw4w9WgXcQ")
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.post("/items/", status_code=201)
-async def create_item(name: str):
- return {"name": name}
+++ /dev/null
-from fastapi import FastAPI, status
-
-app = FastAPI()
-
-
-@app.post("/items/", status_code=status.HTTP_201_CREATED)
-async def create_item(name: str):
- return {"name": name}
+++ /dev/null
-from typing import Annotated
-
-from fastapi import Depends, FastAPI
-from fastapi.security import OAuth2PasswordBearer
-
-app = FastAPI()
-
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
-
-
-@app.get("/items/")
-async def read_items(token: Annotated[str, Depends(oauth2_scheme)]):
- return {"token": token}
+++ /dev/null
-from fastapi import Depends, FastAPI
-from fastapi.security import OAuth2PasswordBearer
-
-app = FastAPI()
-
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
-
-
-@app.get("/items/")
-async def read_items(token: str = Depends(oauth2_scheme)):
- return {"token": token}
+++ /dev/null
-from typing import Annotated
-
-from fastapi import Depends, FastAPI
-from fastapi.security import HTTPBasic, HTTPBasicCredentials
-
-app = FastAPI()
-
-security = HTTPBasic()
-
-
-@app.get("/users/me")
-def read_current_user(credentials: Annotated[HTTPBasicCredentials, Depends(security)]):
- return {"username": credentials.username, "password": credentials.password}
+++ /dev/null
-from fastapi import Depends, FastAPI
-from fastapi.security import HTTPBasic, HTTPBasicCredentials
-
-app = FastAPI()
-
-security = HTTPBasic()
-
-
-@app.get("/users/me")
-def read_current_user(credentials: HTTPBasicCredentials = Depends(security)):
- return {"username": credentials.username, "password": credentials.password}
+++ /dev/null
-import secrets
-from typing import Annotated
-
-from fastapi import Depends, FastAPI, HTTPException, status
-from fastapi.security import HTTPBasic, HTTPBasicCredentials
-
-app = FastAPI()
-
-security = HTTPBasic()
-
-
-def get_current_username(
- credentials: Annotated[HTTPBasicCredentials, Depends(security)],
-):
- current_username_bytes = credentials.username.encode("utf8")
- correct_username_bytes = b"stanleyjobson"
- is_correct_username = secrets.compare_digest(
- current_username_bytes, correct_username_bytes
- )
- current_password_bytes = credentials.password.encode("utf8")
- correct_password_bytes = b"swordfish"
- is_correct_password = secrets.compare_digest(
- current_password_bytes, correct_password_bytes
- )
- if not (is_correct_username and is_correct_password):
- raise HTTPException(
- status_code=status.HTTP_401_UNAUTHORIZED,
- detail="Incorrect username or password",
- headers={"WWW-Authenticate": "Basic"},
- )
- return credentials.username
-
-
-@app.get("/users/me")
-def read_current_user(username: Annotated[str, Depends(get_current_username)]):
- return {"username": username}
+++ /dev/null
-import secrets
-
-from fastapi import Depends, FastAPI, HTTPException, status
-from fastapi.security import HTTPBasic, HTTPBasicCredentials
-
-app = FastAPI()
-
-security = HTTPBasic()
-
-
-def get_current_username(credentials: HTTPBasicCredentials = Depends(security)):
- current_username_bytes = credentials.username.encode("utf8")
- correct_username_bytes = b"stanleyjobson"
- is_correct_username = secrets.compare_digest(
- current_username_bytes, correct_username_bytes
- )
- current_password_bytes = credentials.password.encode("utf8")
- correct_password_bytes = b"swordfish"
- is_correct_password = secrets.compare_digest(
- current_password_bytes, correct_password_bytes
- )
- if not (is_correct_username and is_correct_password):
- raise HTTPException(
- status_code=status.HTTP_401_UNAUTHORIZED,
- detail="Incorrect username or password",
- headers={"WWW-Authenticate": "Basic"},
- )
- return credentials.username
-
-
-@app.get("/users/me")
-def read_current_user(username: str = Depends(get_current_username)):
- return {"username": username}
+++ /dev/null
-from pydantic_settings import BaseSettings
-
-
-class Settings(BaseSettings):
- app_name: str = "Awesome API"
- admin_email: str
- items_per_user: int = 50
-
-
-settings = Settings()
+++ /dev/null
-from fastapi import FastAPI
-
-from .config import settings
-
-app = FastAPI()
-
-
-@app.get("/info")
-async def info():
- return {
- "app_name": settings.app_name,
- "admin_email": settings.admin_email,
- "items_per_user": settings.items_per_user,
- }
+++ /dev/null
-from pydantic_settings import BaseSettings
-
-
-class Settings(BaseSettings):
- app_name: str = "Awesome API"
- admin_email: str
- items_per_user: int = 50
+++ /dev/null
-from functools import lru_cache
-from typing import Annotated
-
-from fastapi import Depends, FastAPI
-
-from .config import Settings
-
-app = FastAPI()
-
-
-@lru_cache
-def get_settings():
- return Settings()
-
-
-@app.get("/info")
-async def info(settings: Annotated[Settings, Depends(get_settings)]):
- return {
- "app_name": settings.app_name,
- "admin_email": settings.admin_email,
- "items_per_user": settings.items_per_user,
- }
+++ /dev/null
-from fastapi.testclient import TestClient
-
-from .config import Settings
-from .main import app, get_settings
-
-client = TestClient(app)
-
-
-def get_settings_override():
- return Settings(admin_email="testing_admin@example.com")
-
-
-app.dependency_overrides[get_settings] = get_settings_override
-
-
-def test_app():
- response = client.get("/info")
- data = response.json()
- assert data == {
- "app_name": "Awesome API",
- "admin_email": "testing_admin@example.com",
- "items_per_user": 50,
- }
+++ /dev/null
-from pydantic_settings import BaseSettings
-
-
-class Settings(BaseSettings):
- app_name: str = "Awesome API"
- admin_email: str
- items_per_user: int = 50
+++ /dev/null
-from functools import lru_cache
-
-from fastapi import Depends, FastAPI
-
-from .config import Settings
-
-app = FastAPI()
-
-
-@lru_cache
-def get_settings():
- return Settings()
-
-
-@app.get("/info")
-async def info(settings: Settings = Depends(get_settings)):
- return {
- "app_name": settings.app_name,
- "admin_email": settings.admin_email,
- "items_per_user": settings.items_per_user,
- }
+++ /dev/null
-from fastapi.testclient import TestClient
-
-from .config import Settings
-from .main import app, get_settings
-
-client = TestClient(app)
-
-
-def get_settings_override():
- return Settings(admin_email="testing_admin@example.com")
-
-
-app.dependency_overrides[get_settings] = get_settings_override
-
-
-def test_app():
- response = client.get("/info")
- data = response.json()
- assert data == {
- "app_name": "Awesome API",
- "admin_email": "testing_admin@example.com",
- "items_per_user": 50,
- }
+++ /dev/null
-from pydantic_settings import BaseSettings, SettingsConfigDict
-
-
-class Settings(BaseSettings):
- app_name: str = "Awesome API"
- admin_email: str
- items_per_user: int = 50
-
- model_config = SettingsConfigDict(env_file=".env")
+++ /dev/null
-from functools import lru_cache
-from typing import Annotated
-
-from fastapi import Depends, FastAPI
-
-from . import config
-
-app = FastAPI()
-
-
-@lru_cache
-def get_settings():
- return config.Settings()
-
-
-@app.get("/info")
-async def info(settings: Annotated[config.Settings, Depends(get_settings)]):
- return {
- "app_name": settings.app_name,
- "admin_email": settings.admin_email,
- "items_per_user": settings.items_per_user,
- }
+++ /dev/null
-from pydantic_settings import BaseSettings, SettingsConfigDict
-
-
-class Settings(BaseSettings):
- app_name: str = "Awesome API"
- admin_email: str
- items_per_user: int = 50
-
- model_config = SettingsConfigDict(env_file=".env")
+++ /dev/null
-from functools import lru_cache
-
-from fastapi import Depends, FastAPI
-
-from . import config
-
-app = FastAPI()
-
-
-@lru_cache
-def get_settings():
- return config.Settings()
-
-
-@app.get("/info")
-async def info(settings: config.Settings = Depends(get_settings)):
- return {
- "app_name": settings.app_name,
- "admin_email": settings.admin_email,
- "items_per_user": settings.items_per_user,
- }
+++ /dev/null
-from fastapi import FastAPI
-from pydantic_settings import BaseSettings
-
-
-class Settings(BaseSettings):
- app_name: str = "Awesome API"
- admin_email: str
- items_per_user: int = 50
-
-
-settings = Settings()
-app = FastAPI()
-
-
-@app.get("/info")
-async def info():
- return {
- "app_name": settings.app_name,
- "admin_email": settings.admin_email,
- "items_per_user": settings.items_per_user,
- }
+++ /dev/null
-from fastapi import FastAPI
-from fastapi.staticfiles import StaticFiles
-
-app = FastAPI()
-
-app.mount("/static", StaticFiles(directory="static"), name="static")
+++ /dev/null
-from fastapi import FastAPI
-
-app = FastAPI()
-
-
-@app.get("/app")
-def read_main():
- return {"message": "Hello World from main app"}
-
-
-subapi = FastAPI()
-
-
-@subapi.get("/sub")
-def read_sub():
- return {"message": "Hello World from sub API"}
-
-
-app.mount("/subapi", subapi)
+++ /dev/null
-from fastapi import FastAPI, Request
-from fastapi.responses import HTMLResponse
-from fastapi.staticfiles import StaticFiles
-from fastapi.templating import Jinja2Templates
-
-app = FastAPI()
-
-app.mount("/static", StaticFiles(directory="static"), name="static")
-
-
-templates = Jinja2Templates(directory="templates")
-
-
-@app.get("/items/{id}", response_class=HTMLResponse)
-async def read_item(request: Request, id: str):
- return templates.TemplateResponse(
- request=request, name="item.html", context={"id": id}
- )
+++ /dev/null
-from fastapi import FastAPI, Request
-
-app = FastAPI()
-
-
-@app.get("/items/{item_id}")
-def read_root(item_id: str, request: Request):
- client_host = request.client.host
- return {"client_host": client_host, "item_id": item_id}
+++ /dev/null
-from fastapi import FastAPI, WebSocket
-from fastapi.responses import HTMLResponse
-
-app = FastAPI()
-
-html = """
-<!DOCTYPE html>
-<html>
- <head>
- <title>Chat</title>
- </head>
- <body>
- <h1>WebSocket Chat</h1>
- <form action="" onsubmit="sendMessage(event)">
- <input type="text" id="messageText" autocomplete="off"/>
- <button>Send</button>
- </form>
- <ul id='messages'>
- </ul>
- <script>
- var ws = new WebSocket("ws://localhost:8000/ws");
- ws.onmessage = function(event) {
- var messages = document.getElementById('messages')
- var message = document.createElement('li')
- var content = document.createTextNode(event.data)
- message.appendChild(content)
- messages.appendChild(message)
- };
- function sendMessage(event) {
- var input = document.getElementById("messageText")
- ws.send(input.value)
- input.value = ''
- event.preventDefault()
- }
- </script>
- </body>
-</html>
-"""
-
-
-@app.get("/")
-async def get():
- return HTMLResponse(html)
-
-
-@app.websocket("/ws")
-async def websocket_endpoint(websocket: WebSocket):
- await websocket.accept()
- while True:
- data = await websocket.receive_text()
- await websocket.send_text(f"Message text was: {data}")
+++ /dev/null
-from fastapi import FastAPI, WebSocket, WebSocketDisconnect
-from fastapi.responses import HTMLResponse
-
-app = FastAPI()
-
-html = """
-<!DOCTYPE html>
-<html>
- <head>
- <title>Chat</title>
- </head>
- <body>
- <h1>WebSocket Chat</h1>
- <h2>Your ID: <span id="ws-id"></span></h2>
- <form action="" onsubmit="sendMessage(event)">
- <input type="text" id="messageText" autocomplete="off"/>
- <button>Send</button>
- </form>
- <ul id='messages'>
- </ul>
- <script>
- var client_id = Date.now()
- document.querySelector("#ws-id").textContent = client_id;
- var ws = new WebSocket(`ws://localhost:8000/ws/${client_id}`);
- ws.onmessage = function(event) {
- var messages = document.getElementById('messages')
- var message = document.createElement('li')
- var content = document.createTextNode(event.data)
- message.appendChild(content)
- messages.appendChild(message)
- };
- function sendMessage(event) {
- var input = document.getElementById("messageText")
- ws.send(input.value)
- input.value = ''
- event.preventDefault()
- }
- </script>
- </body>
-</html>
-"""
-
-
-class ConnectionManager:
- def __init__(self):
- self.active_connections: list[WebSocket] = []
-
- async def connect(self, websocket: WebSocket):
- await websocket.accept()
- self.active_connections.append(websocket)
-
- def disconnect(self, websocket: WebSocket):
- self.active_connections.remove(websocket)
-
- async def send_personal_message(self, message: str, websocket: WebSocket):
- await websocket.send_text(message)
-
- async def broadcast(self, message: str):
- for connection in self.active_connections:
- await connection.send_text(message)
-
-
-manager = ConnectionManager()
-
-
-@app.get("/")
-async def get():
- return HTMLResponse(html)
-
-
-@app.websocket("/ws/{client_id}")
-async def websocket_endpoint(websocket: WebSocket, client_id: int):
- await manager.connect(websocket)
- try:
- while True:
- data = await websocket.receive_text()
- await manager.send_personal_message(f"You wrote: {data}", websocket)
- await manager.broadcast(f"Client #{client_id} says: {data}")
- except WebSocketDisconnect:
- manager.disconnect(websocket)
- await manager.broadcast(f"Client #{client_id} left the chat")
+++ /dev/null
-from a2wsgi import WSGIMiddleware
-from fastapi import FastAPI
-from flask import Flask, request
-from markupsafe import escape
-
-flask_app = Flask(__name__)
-
-
-@flask_app.route("/")
-def flask_main():
- name = request.args.get("name", "World")
- return f"Hello, {escape(name)} from Flask!"
-
-
-app = FastAPI()
-
-
-@app.get("/v2")
-def read_main():
- return {"message": "Hello World"}
-
-
-app.mount("/v1", WSGIMiddleware(flask_app))
Some text
-{* ../../docs_src/bigger_applications/app_an_py39/internal/admin.py hl[3] title["app/internal/admin.py"] *}
+{* ../../docs_src/bigger_applications/app_an_py310/internal/admin.py hl[3] title["app/internal/admin.py"] *}
Some more text
Some text
-{* ../../docs_src/bigger_applications/app_an_py39/internal/admin.py hl[3] title["app/internal/admin.py"] *}
+{* ../../docs_src/bigger_applications/app_an_py310/internal/admin.py hl[3] title["app/internal/admin.py"] *}
Some more text
And even more text
-{* ../../docs_src/python_types/tutorial001_py39.py *}
+{* ../../docs_src/python_types/tutorial001_py310.py *}
Some text
-{* ../../docs_src/bigger_applications/app_an_py39/internal/admin.py hl[3] title["app/internal/admin.py"] *}
+{* ../../docs_src/bigger_applications/app_an_py310/internal/admin.py hl[3] title["app/internal/admin.py"] *}
Some more text
## Simple code includes { #simple-code-includes }
-{* ../../docs_src/python_types/tutorial001_py39.py *}
+{* ../../docs_src/python_types/tutorial001_py310.py *}
-{* ../../docs_src/python_types/tutorial002_py39.py *}
+{* ../../docs_src/python_types/tutorial002_py310.py *}
## Code includes with highlighting { #code-includes-with-highlighting }
-{* ../../docs_src/python_types/tutorial002_py39.py hl[1] *}
+{* ../../docs_src/python_types/tutorial002_py310.py hl[1] *}
-{* ../../docs_src/python_types/tutorial006_py39.py hl[10] *}
+{* ../../docs_src/python_types/tutorial006_py310.py hl[10] *}
## Code includes with line ranges { #code-includes-with-line-ranges }
## Code includes qith title { #code-includes-with-title }
-{* ../../docs_src/bigger_applications/app_an_py39/routers/users.py hl[1,3] title["app/routers/users.py"] *}
+{* ../../docs_src/bigger_applications/app_an_py310/routers/users.py hl[1,3] title["app/routers/users.py"] *}
-{* ../../docs_src/bigger_applications/app_an_py39/internal/admin.py hl[3] title["app/internal/admin.py"] *}
+{* ../../docs_src/bigger_applications/app_an_py310/internal/admin.py hl[3] title["app/internal/admin.py"] *}
## Code includes with unknown attributes { #code-includes-with-unknown-attributes }
-{* ../../docs_src/python_types/tutorial001_py39.py unknown[123] *}
+{* ../../docs_src/python_types/tutorial001_py310.py unknown[123] *}
## Some more code includes to test fixing { #some-more-code-includes-to-test-fixing }
{* ../../docs_src/dependencies/tutorial013_an_py310.py ln[19:21] *}
-{* ../../docs_src/bigger_applications/app_an_py39/internal/admin.py hl[3] title["app/internal/admin.py"] *}
+{* ../../docs_src/bigger_applications/app_an_py310/internal/admin.py hl[3] title["app/internal/admin.py"] *}
{* ../../docs_src/dependencies/tutorial013_an_py310.py ln[30:38] hl[31:33] *}
## ΠΡΠΎΡΡΡΠ΅ Π²ΠΊΠ»ΡΡΠ΅Π½ΠΈΡ ΠΊΠΎΠ΄Π° { #simple-code-includes }
-{* ../../docs_src/python_types/tutorial001_py39.py *}
+{* ../../docs_src/python_types/tutorial001_py310.py *}
-{* ../../docs_src/python_types/tutorial002_py39.py *}
+{* ../../docs_src/python_types/tutorial002_py310.py *}
## ΠΠΊΠ»ΡΡΠ΅Π½ΠΈΡ ΠΊΠΎΠ΄Π° Ρ ΠΏΠΎΠ΄ΡΠ²Π΅ΡΠΊΠΎΠΉ { #code-includes-with-highlighting }
-{* ../../docs_src/python_types/tutorial002_py39.py hl[1] *}
+{* ../../docs_src/python_types/tutorial002_py310.py hl[1] *}
-{* ../../docs_src/python_types/tutorial006_py39.py hl[10] *}
+{* ../../docs_src/python_types/tutorial006_py310.py hl[10] *}
## ΠΠΊΠ»ΡΡΠ΅Π½ΠΈΡ ΠΊΠΎΠ΄Π° Ρ Π΄ΠΈΠ°ΠΏΠ°Π·ΠΎΠ½Π°ΠΌΠΈ ΡΡΡΠΎΠΊ { #code-includes-with-line-ranges }
## ΠΠΊΠ»ΡΡΠ΅Π½ΠΈΡ ΠΊΠΎΠ΄Π° Ρ Π·Π°Π³ΠΎΠ»ΠΎΠ²ΠΊΠΎΠΌ { #code-includes-with-title }
-{* ../../docs_src/bigger_applications/app_an_py39/routers/users.py hl[1,3] title["app/routers/users.py"] *}
+{* ../../docs_src/bigger_applications/app_an_py310/routers/users.py hl[1,3] title["app/routers/users.py"] *}
-{* ../../docs_src/bigger_applications/app_an_py39/internal/admin.py hl[3] title["app/internal/admin.py"] *}
+{* ../../docs_src/bigger_applications/app_an_py310/internal/admin.py hl[3] title["app/internal/admin.py"] *}
## ΠΠΊΠ»ΡΡΠ΅Π½ΠΈΡ ΠΊΠΎΠ΄Π° Ρ Π½Π΅ΠΈΠ·Π²Π΅ΡΡΠ½ΡΠΌΠΈ Π°ΡΡΠΈΠ±ΡΡΠ°ΠΌΠΈ { #code-includes-with-unknown-attributes }
-{* ../../docs_src/python_types/tutorial001_py39.py unknown[123] *}
+{* ../../docs_src/python_types/tutorial001_py310.py unknown[123] *}
## ΠΡΡ Π²ΠΊΠ»ΡΡΠ΅Π½ΠΈΡ ΠΊΠΎΠ΄Π° Π΄Π»Ρ ΡΠ΅ΡΡΠΈΡΠΎΠ²Π°Π½ΠΈΡ ΠΈΡΠΏΡΠ°Π²Π»Π΅Π½ΠΈΡ { #some-more-code-includes-to-test-fixing }
{* ../../docs_src/dependencies/tutorial013_an_py310.py ln[19 : 21] *}
-{* ../../docs_src/bigger_applications/app_an_py39/wrong.py hl[3] title["app/internal/admin.py"] *}
+{* ../../docs_src/bigger_applications/app_an_py310/wrong.py hl[3] title["app/internal/admin.py"] *}
{* ../../docs_src/dependencies/tutorial013_an_py310.py ln[1:30] hl[1:10] *}
## ΠΡΠΎΡΡΡΠ΅ Π²ΠΊΠ»ΡΡΠ΅Π½ΠΈΡ ΠΊΠΎΠ΄Π° { #simple-code-includes }
-{* ../../docs_src/python_types/tutorial001_py39.py *}
+{* ../../docs_src/python_types/tutorial001_py310.py *}
-{* ../../docs_src/python_types/tutorial002_py39.py *}
+{* ../../docs_src/python_types/tutorial002_py310.py *}
## ΠΠΊΠ»ΡΡΠ΅Π½ΠΈΡ ΠΊΠΎΠ΄Π° Ρ ΠΏΠΎΠ΄ΡΠ²Π΅ΡΠΊΠΎΠΉ { #code-includes-with-highlighting }
-{* ../../docs_src/python_types/tutorial002_py39.py hl[1] *}
+{* ../../docs_src/python_types/tutorial002_py310.py hl[1] *}
-{* ../../docs_src/python_types/tutorial006_py39.py hl[10] *}
+{* ../../docs_src/python_types/tutorial006_py310.py hl[10] *}
## ΠΠΊΠ»ΡΡΠ΅Π½ΠΈΡ ΠΊΠΎΠ΄Π° Ρ Π΄ΠΈΠ°ΠΏΠ°Π·ΠΎΠ½Π°ΠΌΠΈ ΡΡΡΠΎΠΊ { #code-includes-with-line-ranges }
## ΠΠΊΠ»ΡΡΠ΅Π½ΠΈΡ ΠΊΠΎΠ΄Π° Ρ Π·Π°Π³ΠΎΠ»ΠΎΠ²ΠΊΠΎΠΌ { #code-includes-with-title }
-{* ../../docs_src/bigger_applications/app_an_py39/routers/users.py hl[1,3] title["app/routers/users.py"] *}
+{* ../../docs_src/bigger_applications/app_an_py310/routers/users.py hl[1,3] title["app/routers/users.py"] *}
-{* ../../docs_src/bigger_applications/app_an_py39/internal/admin.py hl[3] title["app/internal/admin.py"] *}
+{* ../../docs_src/bigger_applications/app_an_py310/internal/admin.py hl[3] title["app/internal/admin.py"] *}
## ΠΠΊΠ»ΡΡΠ΅Π½ΠΈΡ ΠΊΠΎΠ΄Π° Ρ Π½Π΅ΠΈΠ·Π²Π΅ΡΡΠ½ΡΠΌΠΈ Π°ΡΡΠΈΠ±ΡΡΠ°ΠΌΠΈ { #code-includes-with-unknown-attributes }
-{* ../../docs_src/python_types/tutorial001_py39.py unknown[123] *}
+{* ../../docs_src/python_types/tutorial001_py310.py unknown[123] *}
## ΠΡΡ Π²ΠΊΠ»ΡΡΠ΅Π½ΠΈΡ ΠΊΠΎΠ΄Π° Π΄Π»Ρ ΡΠ΅ΡΡΠΈΡΠΎΠ²Π°Π½ΠΈΡ ΠΈΡΠΏΡΠ°Π²Π»Π΅Π½ΠΈΡ { #some-more-code-includes-to-test-fixing }
{* ../../docs_src/dependencies/tutorial013_an_py310.py ln[19:21] *}
-{* ../../docs_src/bigger_applications/app_an_py39/internal/admin.py hl[3] title["app/internal/admin.py"] *}
+{* ../../docs_src/bigger_applications/app_an_py310/internal/admin.py hl[3] title["app/internal/admin.py"] *}
{* ../../docs_src/dependencies/tutorial013_an_py310.py ln[30:38] hl[31:33] *}
+++ /dev/null
-import importlib
-import re
-from types import ModuleType
-from unittest.mock import patch
-
-import pytest
-
-from ...utils import needs_py310
-
-
-@pytest.fixture(
- name="module",
- params=[
- pytest.param("tutorial009c_py310"),
- pytest.param("tutorial009c_py310", marks=needs_py310),
- ],
-)
-def get_module(request: pytest.FixtureRequest):
- mod = importlib.import_module(f"docs_src.python_types.{request.param}")
- return mod
-
-
-def test_say_hi(module: ModuleType):
- with patch("builtins.print") as mock_print:
- module.say_hi("FastAPI")
-
- mock_print.assert_called_once_with("Hey FastAPI!")
-
- with pytest.raises(
- TypeError,
- match=re.escape("say_hi() missing 1 required positional argument: 'name'"),
- ):
- module.say_hi()
projects / thirdparty / fastapi / fastapi.git / commitdiff
