--- /dev/null
+from fastapi import (
+ Body,
+ Cookie,
+ Depends,
+ FastAPI,
+ File,
+ Form,
+ Header,
+ Path,
+ Query,
+ Security,
+)
+from fastapi.security import (
+ HTTPBasic,
+ OAuth2,
+ OAuth2PasswordBearer,
+ OAuth2PasswordRequestForm,
+)
+from pydantic import BaseModel
+from starlette.responses import HTMLResponse, JSONResponse, PlainTextResponse
+from starlette.status import HTTP_202_ACCEPTED
+
+app = FastAPI()
+
+
+@app.get("/security")
+def get_security(sec=Security(HTTPBasic())):
+ return sec
+
+
+reusable_oauth2 = OAuth2(
+ flows={
+ "password": {
+ "tokenUrl": "/token",
+ "scopes": {"read:user": "Read a User", "write:user": "Create a user"},
+ }
+ }
+)
+
+
+@app.get("/security/oauth2")
+def get_security_oauth2(sec=Security(reusable_oauth2, scopes=["read:user"])):
+ return sec
+
+
+@app.post("/token")
+def post_token(request_data: OAuth2PasswordRequestForm = Form(...)):
+ data = request_data.parse()
+ access_token = data.username + ":" + data.password
+ return {"access_token": access_token}
+
+
+class Item(BaseModel):
+ name: str
+ price: float
+ is_offer: bool
+
+
+class FakeDB:
+ def __init__(self):
+ self.data = {
+ "johndoe": {
+ "username": "johndoe",
+ "password": "shouldbehashed",
+ "fist_name": "John",
+ "last_name": "Doe",
+ }
+ }
+
+
+class DBConnectionManager:
+ def __init__(self):
+ self.db = FakeDB()
+
+ def __call__(self):
+ return self.db
+
+
+connection_manager = DBConnectionManager()
+
+
+class TokenUserData(BaseModel):
+ username: str
+ password: str
+
+
+class UserInDB(BaseModel):
+ username: str
+ password: str
+ fist_name: str
+ last_name: str
+
+
+def require_token(
+ token: str = Security(reusable_oauth2, scopes=["read:user", "write:user"])
+):
+ raw_token = token.replace("Bearer ", "")
+ # Never do this plaintext password usage in production
+ username, password = raw_token.split(":")
+ return TokenUserData(username=username, password=password)
+
+
+def require_user(
+ db: FakeDB = Depends(connection_manager),
+ user_data: TokenUserData = Depends(require_token),
+):
+ return db.data[user_data.username]
+
+
+class UserOut(BaseModel):
+ username: str
+ fist_name: str
+ last_name: str
+
+
+@app.get("/dependency", response_model=UserOut)
+def get_dependency(user: UserInDB = Depends(require_user)):
+ return user
+++ /dev/null
-from fastapi import (
- Body,
- Cookie,
- Depends,
- FastAPI,
- File,
- Form,
- Header,
- Path,
- Query,
- Security,
-)
-from fastapi.security import (
- HTTPBasic,
- OAuth2,
- OAuth2PasswordBearer,
- OAuth2PasswordRequestForm,
-)
-from pydantic import BaseModel
-from starlette.responses import HTMLResponse, JSONResponse, PlainTextResponse
-from starlette.status import HTTP_202_ACCEPTED
-
-from .endpoints.a import router as router_a
-from .endpoints.b import router as router_b
-
-app = FastAPI()
-
-
-app.include_router(router_a)
-app.include_router(router_b, prefix="/b")
-
-
-@app.get("/cookie")
-def get_cookie(coo=Cookie(None)):
- return coo
-
-
-@app.get("/header")
-def get_header(head_name=Header(None)):
- return head_name
-
-
-@app.get("/header_under")
-def get_header(head_name=Header(None, convert_underscores=False)):
- return head_name
-
-
-@app.get("/security")
-def get_security(sec=Security(HTTPBasic())):
- return sec
-
-
-reusable_oauth2 = OAuth2(
- flows={
- "password": {
- "tokenUrl": "/token",
- "scopes": {"read:user": "Read a User", "write:user": "Create a user"},
- }
- }
-)
-
-
-@app.get("/security/oauth2")
-def get_security_oauth2(sec=Security(reusable_oauth2, scopes=["read:user"])):
- return sec
-
-
-@app.post("/token")
-def post_token(request_data: OAuth2PasswordRequestForm = Form(...)):
- data = request_data.parse()
- access_token = data.username + ":" + data.password
- return {"access_token": access_token}
-
-
-class Item(BaseModel):
- name: str
- price: float
- is_offer: bool
-
-
-@app.put("/items/{item_id}")
-def put_item(item_id: str, item: Item):
- return item
-
-
-@app.post("/items/")
-def post_item(item: Item):
- return item
-
-
-@app.post("/items-all-params/{item_id}")
-def post_items_all_params(
- item_id: str = Path(...),
- body: Item = Body(...),
- query_a: int = Query(None),
- query_b=Query(None),
- coo: str = Cookie(None),
- x_head: int = Header(None),
- x_under: str = Header(None, convert_underscores=False),
-):
- return {
- "item_id": item_id,
- "body": body,
- "query_a": query_a,
- "query_b": query_b,
- "coo": coo,
- "x_head": x_head,
- "x_under": x_under,
- }
-
-
-@app.post("/items-all-params-defaults/{item_id}")
-def post_items_all_params_default(
- item_id: str,
- body_item_a: Item,
- body_item_b: Item,
- query_a: int,
- query_b: int,
- coo: str = Cookie(None),
- x_head: int = Header(None),
- x_under: str = Header(None, convert_underscores=False),
-):
- return {
- "item_id": item_id,
- "body_item_a": body_item_a,
- "body_item_b": body_item_b,
- "query_a": query_a,
- "query_b": query_b,
- "coo": coo,
- "x_head": x_head,
- "x_under": x_under,
- }
-
-
-@app.delete("/items/{item_id}")
-def delete_item(item_id: str):
- return item_id
-
-
-@app.options("/options/")
-def options():
- return JSONResponse(headers={"x-fastapi": "fast"})
-
-
-@app.head("/head/")
-def head():
- return {"not sent": "nope"}
-
-
-@app.patch("/patch/{user_id}")
-def patch(user_id: str, increment: float):
- return {"user_id": user_id, "total": 5 + increment}
-
-
-@app.trace("/trace/")
-def trace():
- return PlainTextResponse(media_type="message/http")
-
-
-@app.get("/model", response_model=Item, status_code=HTTP_202_ACCEPTED)
-def model():
- return {"name": "Foo", "price": "5.0", "password": "not sent"}
-
-
-@app.get(
- "/metadata",
- tags=["tag1", "tag2"],
- summary="The summary",
- description="The description",
- response_description="Response description",
- deprecated=True,
- operation_id="a_very_long_and_strange_operation_id",
-)
-def get_meta():
- return "Foo"
-
-
-@app.get("/html", content_type=HTMLResponse)
-def get_html():
- return """
- <html>
- <body>
- <h1>
- Some text inside
- </h1>
- </body>
- </html>
- """
-
-
-class FakeDB:
- def __init__(self):
- self.data = {
- "johndoe": {
- "username": "johndoe",
- "password": "shouldbehashed",
- "fist_name": "John",
- "last_name": "Doe",
- }
- }
-
-
-class DBConnectionManager:
- def __init__(self):
- self.db = FakeDB()
-
- def __call__(self):
- return self.db
-
-
-connection_manager = DBConnectionManager()
-
-
-class TokenUserData(BaseModel):
- username: str
- password: str
-
-
-class UserInDB(BaseModel):
- username: str
- password: str
- fist_name: str
- last_name: str
-
-
-def require_token(
- token: str = Security(reusable_oauth2, scopes=["read:user", "write:user"])
-):
- raw_token = token.replace("Bearer ", "")
- # Never do this plaintext password usage in production
- username, password = raw_token.split(":")
- return TokenUserData(username=username, password=password)
-
-
-def require_user(
- db: FakeDB = Depends(connection_manager),
- user_data: TokenUserData = Depends(require_token),
-):
- return db.data[user_data.username]
-
-
-class UserOut(BaseModel):
- username: str
- fist_name: str
- last_name: str
-
-
-@app.get("/dependency", response_model=UserOut)
-def get_dependency(user: UserInDB = Depends(require_user)):
- return user
+++ /dev/null
-from fastapi import FastAPI
-from pydantic import BaseModel
-from starlette.responses import JSONResponse
-from starlette.testclient import TestClient
-
-app = FastAPI()
-
-
-class Item(BaseModel):
- name: str
- price: float = None
-
-
-@app.delete("/items/{item_id}")
-def delete_item(item_id: str, item: Item):
- return {"item_id": item_id, "item": item}
-
-
-@app.head("/items/{item_id}")
-def head_item(item_id: str):
- return JSONResponse(headers={"x-fastapi-item-id": item_id})
-
-
-@app.options("/items/{item_id}")
-def options_item(item_id: str):
- return JSONResponse(headers={"x-fastapi-item-id": item_id})
-
-
-@app.patch("/items/{item_id}")
-def patch_item(item_id: str, item: Item):
- return {"item_id": item_id, "item": item}
-
-
-@app.trace("/items/{item_id}")
-def trace_item(item_id: str):
- return JSONResponse(media_type="message/http")
-
-
-client = TestClient(app)
-
-openapi_schema = {
- "openapi": "3.0.2",
- "info": {"title": "Fast API", "version": "0.1.0"},
- "paths": {
- "/items/{item_id}": {
- "delete": {
- "responses": {
- "200": {
- "description": "Successful Response",
- "content": {"application/json": {"schema": {}}},
- },
- "422": {
- "description": "Validation Error",
- "content": {
- "application/json": {
- "schema": {
- "$ref": "#/components/schemas/HTTPValidationError"
- }
- }
- },
- },
- },
- "summary": "Delete Item Delete",
- "operationId": "delete_item_items__item_id__delete",
- "parameters": [
- {
- "required": True,
- "schema": {"title": "Item_Id", "type": "string"},
- "name": "item_id",
- "in": "path",
- }
- ],
- "requestBody": {
- "content": {
- "application/json": {
- "schema": {"$ref": "#/components/schemas/Item"}
- }
- },
- "required": True,
- },
- },
- "options": {
- "responses": {
- "200": {
- "description": "Successful Response",
- "content": {"application/json": {"schema": {}}},
- },
- "422": {
- "description": "Validation Error",
- "content": {
- "application/json": {
- "schema": {
- "$ref": "#/components/schemas/HTTPValidationError"
- }
- }
- },
- },
- },
- "summary": "Options Item Options",
- "operationId": "options_item_items__item_id__options",
- "parameters": [
- {
- "required": True,
- "schema": {"title": "Item_Id", "type": "string"},
- "name": "item_id",
- "in": "path",
- }
- ],
- },
- "head": {
- "responses": {
- "200": {
- "description": "Successful Response",
- "content": {"application/json": {"schema": {}}},
- },
- "422": {
- "description": "Validation Error",
- "content": {
- "application/json": {
- "schema": {
- "$ref": "#/components/schemas/HTTPValidationError"
- }
- }
- },
- },
- },
- "summary": "Head Item Head",
- "operationId": "head_item_items__item_id__head",
- "parameters": [
- {
- "required": True,
- "schema": {"title": "Item_Id", "type": "string"},
- "name": "item_id",
- "in": "path",
- }
- ],
- },
- "patch": {
- "responses": {
- "200": {
- "description": "Successful Response",
- "content": {"application/json": {"schema": {}}},
- },
- "422": {
- "description": "Validation Error",
- "content": {
- "application/json": {
- "schema": {
- "$ref": "#/components/schemas/HTTPValidationError"
- }
- }
- },
- },
- },
- "summary": "Patch Item Patch",
- "operationId": "patch_item_items__item_id__patch",
- "parameters": [
- {
- "required": True,
- "schema": {"title": "Item_Id", "type": "string"},
- "name": "item_id",
- "in": "path",
- }
- ],
- "requestBody": {
- "content": {
- "application/json": {
- "schema": {"$ref": "#/components/schemas/Item"}
- }
- },
- "required": True,
- },
- },
- "trace": {
- "responses": {
- "200": {
- "description": "Successful Response",
- "content": {"application/json": {"schema": {}}},
- },
- "422": {
- "description": "Validation Error",
- "content": {
- "application/json": {
- "schema": {
- "$ref": "#/components/schemas/HTTPValidationError"
- }
- }
- },
- },
- },
- "summary": "Trace Item Trace",
- "operationId": "trace_item_items__item_id__trace",
- "parameters": [
- {
- "required": True,
- "schema": {"title": "Item_Id", "type": "string"},
- "name": "item_id",
- "in": "path",
- }
- ],
- },
- }
- },
- "components": {
- "schemas": {
- "Item": {
- "title": "Item",
- "required": ["name"],
- "type": "object",
- "properties": {
- "name": {"title": "Name", "type": "string"},
- "price": {"title": "Price", "type": "number"},
- },
- },
- "ValidationError": {
- "title": "ValidationError",
- "required": ["loc", "msg", "type"],
- "type": "object",
- "properties": {
- "loc": {
- "title": "Location",
- "type": "array",
- "items": {"type": "string"},
- },
- "msg": {"title": "Message", "type": "string"},
- "type": {"title": "Error Type", "type": "string"},
- },
- },
- "HTTPValidationError": {
- "title": "HTTPValidationError",
- "type": "object",
- "properties": {
- "detail": {
- "title": "Detail",
- "type": "array",
- "items": {"$ref": "#/components/schemas/ValidationError"},
- }
- },
- },
- }
- },
-}
-
-
-def test_openapi_schema():
- response = client.get("/openapi.json")
- assert response.status_code == 200
- assert response.json() == openapi_schema
-
-
-def test_delete():
- response = client.delete("/items/foo", json={"name": "Foo"})
- assert response.status_code == 200
- assert response.json() == {"item_id": "foo", "item": {"name": "Foo", "price": None}}
-
-
-def test_head():
- response = client.head("/items/foo")
- assert response.status_code == 200
- assert response.headers["x-fastapi-item-id"] == "foo"
-
-
-def test_options():
- response = client.head("/items/foo")
- assert response.status_code == 200
- assert response.headers["x-fastapi-item-id"] == "foo"
-
-
-def test_patch():
- response = client.patch("/items/foo", json={"name": "Foo"})
- assert response.status_code == 200
- assert response.json() == {"item_id": "foo", "item": {"name": "Foo", "price": None}}
-
-
-def test_trace():
- response = client.request("trace", "/items/foo")
- assert response.status_code == 200
- assert response.headers["content-type"] == "message/http"
projects / thirdparty / fastapi / fastapi.git / commitdiff
