Motov Yurii
6 months ago
committed by
GitHub
GitHub
62 changed files with 1643 additions and 278 deletions
@ -0,0 +1,28 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import FileResponse |
|||
from pydantic import BaseModel |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
id: str |
|||
value: str |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get( |
|||
"/items/{item_id}", |
|||
response_model=Item, |
|||
responses={ |
|||
200: { |
|||
"content": {"image/png": {}}, |
|||
"description": "Return the JSON item or an image.", |
|||
} |
|||
}, |
|||
) |
|||
async def read_item(item_id: str, img: bool | None = None): |
|||
if img: |
|||
return FileResponse("image.png", media_type="image/png") |
|||
else: |
|||
return {"id": "foo", "value": "there goes my hero"} |
|||
@ -0,0 +1,30 @@ |
|||
from fastapi import FastAPI |
|||
from fastapi.responses import FileResponse |
|||
from pydantic import BaseModel |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
id: str |
|||
value: str |
|||
|
|||
|
|||
responses = { |
|||
404: {"description": "Item not found"}, |
|||
302: {"description": "The item was moved"}, |
|||
403: {"description": "Not enough privileges"}, |
|||
} |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get( |
|||
"/items/{item_id}", |
|||
response_model=Item, |
|||
responses={**responses, 200: {"content": {"image/png": {}}}}, |
|||
) |
|||
async def read_item(item_id: str, img: bool | None = None): |
|||
if img: |
|||
return FileResponse("image.png", media_type="image/png") |
|||
else: |
|||
return {"id": "foo", "value": "there goes my hero"} |
|||
@ -0,0 +1,36 @@ |
|||
import gzip |
|||
from typing import Callable, List |
|||
|
|||
from fastapi import Body, FastAPI, Request, Response |
|||
from fastapi.routing import APIRoute |
|||
from typing_extensions import Annotated |
|||
|
|||
|
|||
class GzipRequest(Request): |
|||
async def body(self) -> bytes: |
|||
if not hasattr(self, "_body"): |
|||
body = await super().body() |
|||
if "gzip" in self.headers.getlist("Content-Encoding"): |
|||
body = gzip.decompress(body) |
|||
self._body = body |
|||
return self._body |
|||
|
|||
|
|||
class GzipRoute(APIRoute): |
|||
def get_route_handler(self) -> Callable: |
|||
original_route_handler = super().get_route_handler() |
|||
|
|||
async def custom_route_handler(request: Request) -> Response: |
|||
request = GzipRequest(request.scope, request.receive) |
|||
return await original_route_handler(request) |
|||
|
|||
return custom_route_handler |
|||
|
|||
|
|||
app = FastAPI() |
|||
app.router.route_class = GzipRoute |
|||
|
|||
|
|||
@app.post("/sum") |
|||
async def sum_numbers(numbers: Annotated[List[int], Body()]): |
|||
return {"sum": sum(numbers)} |
|||
@ -0,0 +1,36 @@ |
|||
import gzip |
|||
from collections.abc import Callable |
|||
from typing import Annotated |
|||
|
|||
from fastapi import Body, FastAPI, Request, Response |
|||
from fastapi.routing import APIRoute |
|||
|
|||
|
|||
class GzipRequest(Request): |
|||
async def body(self) -> bytes: |
|||
if not hasattr(self, "_body"): |
|||
body = await super().body() |
|||
if "gzip" in self.headers.getlist("Content-Encoding"): |
|||
body = gzip.decompress(body) |
|||
self._body = body |
|||
return self._body |
|||
|
|||
|
|||
class GzipRoute(APIRoute): |
|||
def get_route_handler(self) -> Callable: |
|||
original_route_handler = super().get_route_handler() |
|||
|
|||
async def custom_route_handler(request: Request) -> Response: |
|||
request = GzipRequest(request.scope, request.receive) |
|||
return await original_route_handler(request) |
|||
|
|||
return custom_route_handler |
|||
|
|||
|
|||
app = FastAPI() |
|||
app.router.route_class = GzipRoute |
|||
|
|||
|
|||
@app.post("/sum") |
|||
async def sum_numbers(numbers: Annotated[list[int], Body()]): |
|||
return {"sum": sum(numbers)} |
|||
@ -0,0 +1,35 @@ |
|||
import gzip |
|||
from typing import Annotated, Callable |
|||
|
|||
from fastapi import Body, FastAPI, Request, Response |
|||
from fastapi.routing import APIRoute |
|||
|
|||
|
|||
class GzipRequest(Request): |
|||
async def body(self) -> bytes: |
|||
if not hasattr(self, "_body"): |
|||
body = await super().body() |
|||
if "gzip" in self.headers.getlist("Content-Encoding"): |
|||
body = gzip.decompress(body) |
|||
self._body = body |
|||
return self._body |
|||
|
|||
|
|||
class GzipRoute(APIRoute): |
|||
def get_route_handler(self) -> Callable: |
|||
original_route_handler = super().get_route_handler() |
|||
|
|||
async def custom_route_handler(request: Request) -> Response: |
|||
request = GzipRequest(request.scope, request.receive) |
|||
return await original_route_handler(request) |
|||
|
|||
return custom_route_handler |
|||
|
|||
|
|||
app = FastAPI() |
|||
app.router.route_class = GzipRoute |
|||
|
|||
|
|||
@app.post("/sum") |
|||
async def sum_numbers(numbers: Annotated[list[int], Body()]): |
|||
return {"sum": sum(numbers)} |
|||
@ -0,0 +1,35 @@ |
|||
import gzip |
|||
from collections.abc import Callable |
|||
|
|||
from fastapi import Body, FastAPI, Request, Response |
|||
from fastapi.routing import APIRoute |
|||
|
|||
|
|||
class GzipRequest(Request): |
|||
async def body(self) -> bytes: |
|||
if not hasattr(self, "_body"): |
|||
body = await super().body() |
|||
if "gzip" in self.headers.getlist("Content-Encoding"): |
|||
body = gzip.decompress(body) |
|||
self._body = body |
|||
return self._body |
|||
|
|||
|
|||
class GzipRoute(APIRoute): |
|||
def get_route_handler(self) -> Callable: |
|||
original_route_handler = super().get_route_handler() |
|||
|
|||
async def custom_route_handler(request: Request) -> Response: |
|||
request = GzipRequest(request.scope, request.receive) |
|||
return await original_route_handler(request) |
|||
|
|||
return custom_route_handler |
|||
|
|||
|
|||
app = FastAPI() |
|||
app.router.route_class = GzipRoute |
|||
|
|||
|
|||
@app.post("/sum") |
|||
async def sum_numbers(numbers: list[int] = Body()): |
|||
return {"sum": sum(numbers)} |
|||
@ -0,0 +1,35 @@ |
|||
import gzip |
|||
from typing import Callable |
|||
|
|||
from fastapi import Body, FastAPI, Request, Response |
|||
from fastapi.routing import APIRoute |
|||
|
|||
|
|||
class GzipRequest(Request): |
|||
async def body(self) -> bytes: |
|||
if not hasattr(self, "_body"): |
|||
body = await super().body() |
|||
if "gzip" in self.headers.getlist("Content-Encoding"): |
|||
body = gzip.decompress(body) |
|||
self._body = body |
|||
return self._body |
|||
|
|||
|
|||
class GzipRoute(APIRoute): |
|||
def get_route_handler(self) -> Callable: |
|||
original_route_handler = super().get_route_handler() |
|||
|
|||
async def custom_route_handler(request: Request) -> Response: |
|||
request = GzipRequest(request.scope, request.receive) |
|||
return await original_route_handler(request) |
|||
|
|||
return custom_route_handler |
|||
|
|||
|
|||
app = FastAPI() |
|||
app.router.route_class = GzipRoute |
|||
|
|||
|
|||
@app.post("/sum") |
|||
async def sum_numbers(numbers: list[int] = Body()): |
|||
return {"sum": sum(numbers)} |
|||
@ -0,0 +1,30 @@ |
|||
from typing import Callable, List |
|||
|
|||
from fastapi import Body, FastAPI, HTTPException, Request, Response |
|||
from fastapi.exceptions import RequestValidationError |
|||
from fastapi.routing import APIRoute |
|||
from typing_extensions import Annotated |
|||
|
|||
|
|||
class ValidationErrorLoggingRoute(APIRoute): |
|||
def get_route_handler(self) -> Callable: |
|||
original_route_handler = super().get_route_handler() |
|||
|
|||
async def custom_route_handler(request: Request) -> Response: |
|||
try: |
|||
return await original_route_handler(request) |
|||
except RequestValidationError as exc: |
|||
body = await request.body() |
|||
detail = {"errors": exc.errors(), "body": body.decode()} |
|||
raise HTTPException(status_code=422, detail=detail) |
|||
|
|||
return custom_route_handler |
|||
|
|||
|
|||
app = FastAPI() |
|||
app.router.route_class = ValidationErrorLoggingRoute |
|||
|
|||
|
|||
@app.post("/") |
|||
async def sum_numbers(numbers: Annotated[List[int], Body()]): |
|||
return sum(numbers) |
|||
@ -0,0 +1,30 @@ |
|||
from collections.abc import Callable |
|||
from typing import Annotated |
|||
|
|||
from fastapi import Body, FastAPI, HTTPException, Request, Response |
|||
from fastapi.exceptions import RequestValidationError |
|||
from fastapi.routing import APIRoute |
|||
|
|||
|
|||
class ValidationErrorLoggingRoute(APIRoute): |
|||
def get_route_handler(self) -> Callable: |
|||
original_route_handler = super().get_route_handler() |
|||
|
|||
async def custom_route_handler(request: Request) -> Response: |
|||
try: |
|||
return await original_route_handler(request) |
|||
except RequestValidationError as exc: |
|||
body = await request.body() |
|||
detail = {"errors": exc.errors(), "body": body.decode()} |
|||
raise HTTPException(status_code=422, detail=detail) |
|||
|
|||
return custom_route_handler |
|||
|
|||
|
|||
app = FastAPI() |
|||
app.router.route_class = ValidationErrorLoggingRoute |
|||
|
|||
|
|||
@app.post("/") |
|||
async def sum_numbers(numbers: Annotated[list[int], Body()]): |
|||
return sum(numbers) |
|||
@ -0,0 +1,29 @@ |
|||
from typing import Annotated, Callable |
|||
|
|||
from fastapi import Body, FastAPI, HTTPException, Request, Response |
|||
from fastapi.exceptions import RequestValidationError |
|||
from fastapi.routing import APIRoute |
|||
|
|||
|
|||
class ValidationErrorLoggingRoute(APIRoute): |
|||
def get_route_handler(self) -> Callable: |
|||
original_route_handler = super().get_route_handler() |
|||
|
|||
async def custom_route_handler(request: Request) -> Response: |
|||
try: |
|||
return await original_route_handler(request) |
|||
except RequestValidationError as exc: |
|||
body = await request.body() |
|||
detail = {"errors": exc.errors(), "body": body.decode()} |
|||
raise HTTPException(status_code=422, detail=detail) |
|||
|
|||
return custom_route_handler |
|||
|
|||
|
|||
app = FastAPI() |
|||
app.router.route_class = ValidationErrorLoggingRoute |
|||
|
|||
|
|||
@app.post("/") |
|||
async def sum_numbers(numbers: Annotated[list[int], Body()]): |
|||
return sum(numbers) |
|||
@ -0,0 +1,29 @@ |
|||
from collections.abc import Callable |
|||
|
|||
from fastapi import Body, FastAPI, HTTPException, Request, Response |
|||
from fastapi.exceptions import RequestValidationError |
|||
from fastapi.routing import APIRoute |
|||
|
|||
|
|||
class ValidationErrorLoggingRoute(APIRoute): |
|||
def get_route_handler(self) -> Callable: |
|||
original_route_handler = super().get_route_handler() |
|||
|
|||
async def custom_route_handler(request: Request) -> Response: |
|||
try: |
|||
return await original_route_handler(request) |
|||
except RequestValidationError as exc: |
|||
body = await request.body() |
|||
detail = {"errors": exc.errors(), "body": body.decode()} |
|||
raise HTTPException(status_code=422, detail=detail) |
|||
|
|||
return custom_route_handler |
|||
|
|||
|
|||
app = FastAPI() |
|||
app.router.route_class = ValidationErrorLoggingRoute |
|||
|
|||
|
|||
@app.post("/") |
|||
async def sum_numbers(numbers: list[int] = Body()): |
|||
return sum(numbers) |
|||
@ -0,0 +1,29 @@ |
|||
from typing import Callable |
|||
|
|||
from fastapi import Body, FastAPI, HTTPException, Request, Response |
|||
from fastapi.exceptions import RequestValidationError |
|||
from fastapi.routing import APIRoute |
|||
|
|||
|
|||
class ValidationErrorLoggingRoute(APIRoute): |
|||
def get_route_handler(self) -> Callable: |
|||
original_route_handler = super().get_route_handler() |
|||
|
|||
async def custom_route_handler(request: Request) -> Response: |
|||
try: |
|||
return await original_route_handler(request) |
|||
except RequestValidationError as exc: |
|||
body = await request.body() |
|||
detail = {"errors": exc.errors(), "body": body.decode()} |
|||
raise HTTPException(status_code=422, detail=detail) |
|||
|
|||
return custom_route_handler |
|||
|
|||
|
|||
app = FastAPI() |
|||
app.router.route_class = ValidationErrorLoggingRoute |
|||
|
|||
|
|||
@app.post("/") |
|||
async def sum_numbers(numbers: list[int] = Body()): |
|||
return sum(numbers) |
|||
@ -0,0 +1,39 @@ |
|||
import time |
|||
from collections.abc import Callable |
|||
|
|||
from fastapi import APIRouter, FastAPI, Request, Response |
|||
from fastapi.routing import APIRoute |
|||
|
|||
|
|||
class TimedRoute(APIRoute): |
|||
def get_route_handler(self) -> Callable: |
|||
original_route_handler = super().get_route_handler() |
|||
|
|||
async def custom_route_handler(request: Request) -> Response: |
|||
before = time.time() |
|||
response: Response = await original_route_handler(request) |
|||
duration = time.time() - before |
|||
response.headers["X-Response-Time"] = str(duration) |
|||
print(f"route duration: {duration}") |
|||
print(f"route response: {response}") |
|||
print(f"route response headers: {response.headers}") |
|||
return response |
|||
|
|||
return custom_route_handler |
|||
|
|||
|
|||
app = FastAPI() |
|||
router = APIRouter(route_class=TimedRoute) |
|||
|
|||
|
|||
@app.get("/") |
|||
async def not_timed(): |
|||
return {"message": "Not timed"} |
|||
|
|||
|
|||
@router.get("/timed") |
|||
async def timed(): |
|||
return {"message": "It's the time of my life"} |
|||
|
|||
|
|||
app.include_router(router) |
|||
@ -0,0 +1,19 @@ |
|||
from dataclasses import dataclass |
|||
|
|||
from fastapi import FastAPI |
|||
|
|||
|
|||
@dataclass |
|||
class Item: |
|||
name: str |
|||
price: float |
|||
description: str | None = None |
|||
tax: float | None = None |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.post("/items/") |
|||
async def create_item(item: Item): |
|||
return item |
|||
@ -0,0 +1,25 @@ |
|||
from dataclasses import dataclass, field |
|||
|
|||
from fastapi import FastAPI |
|||
|
|||
|
|||
@dataclass |
|||
class Item: |
|||
name: str |
|||
price: float |
|||
tags: list[str] = field(default_factory=list) |
|||
description: str | None = None |
|||
tax: float | None = None |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/items/next", response_model=Item) |
|||
async def read_next_item(): |
|||
return { |
|||
"name": "Island In The Moon", |
|||
"price": 12.99, |
|||
"description": "A place to be playin' and havin' fun", |
|||
"tags": ["breater"], |
|||
} |
|||
@ -0,0 +1,26 @@ |
|||
from dataclasses import dataclass, field |
|||
from typing import Union |
|||
|
|||
from fastapi import FastAPI |
|||
|
|||
|
|||
@dataclass |
|||
class Item: |
|||
name: str |
|||
price: float |
|||
tags: list[str] = field(default_factory=list) |
|||
description: Union[str, None] = None |
|||
tax: Union[float, None] = None |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.get("/items/next", response_model=Item) |
|||
async def read_next_item(): |
|||
return { |
|||
"name": "Island In The Moon", |
|||
"price": 12.99, |
|||
"description": "A place to be playin' and havin' fun", |
|||
"tags": ["breater"], |
|||
} |
|||
@ -0,0 +1,54 @@ |
|||
from dataclasses import field # (1) |
|||
|
|||
from fastapi import FastAPI |
|||
from pydantic.dataclasses import dataclass # (2) |
|||
|
|||
|
|||
@dataclass |
|||
class Item: |
|||
name: str |
|||
description: str | None = None |
|||
|
|||
|
|||
@dataclass |
|||
class Author: |
|||
name: str |
|||
items: list[Item] = field(default_factory=list) # (3) |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.post("/authors/{author_id}/items/", response_model=Author) # (4) |
|||
async def create_author_items(author_id: str, items: list[Item]): # (5) |
|||
return {"name": author_id, "items": items} # (6) |
|||
|
|||
|
|||
@app.get("/authors/", response_model=list[Author]) # (7) |
|||
def get_authors(): # (8) |
|||
return [ # (9) |
|||
{ |
|||
"name": "Breaters", |
|||
"items": [ |
|||
{ |
|||
"name": "Island In The Moon", |
|||
"description": "A place to be playin' and havin' fun", |
|||
}, |
|||
{"name": "Holy Buddies"}, |
|||
], |
|||
}, |
|||
{ |
|||
"name": "System of an Up", |
|||
"items": [ |
|||
{ |
|||
"name": "Salt", |
|||
"description": "The kombucha mushroom people's favorite", |
|||
}, |
|||
{"name": "Pad Thai"}, |
|||
{ |
|||
"name": "Lonely Night", |
|||
"description": "The mostests lonliest nightiest of allest", |
|||
}, |
|||
], |
|||
}, |
|||
] |
|||
@ -0,0 +1,55 @@ |
|||
from dataclasses import field # (1) |
|||
from typing import Union |
|||
|
|||
from fastapi import FastAPI |
|||
from pydantic.dataclasses import dataclass # (2) |
|||
|
|||
|
|||
@dataclass |
|||
class Item: |
|||
name: str |
|||
description: Union[str, None] = None |
|||
|
|||
|
|||
@dataclass |
|||
class Author: |
|||
name: str |
|||
items: list[Item] = field(default_factory=list) # (3) |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.post("/authors/{author_id}/items/", response_model=Author) # (4) |
|||
async def create_author_items(author_id: str, items: list[Item]): # (5) |
|||
return {"name": author_id, "items": items} # (6) |
|||
|
|||
|
|||
@app.get("/authors/", response_model=list[Author]) # (7) |
|||
def get_authors(): # (8) |
|||
return [ # (9) |
|||
{ |
|||
"name": "Breaters", |
|||
"items": [ |
|||
{ |
|||
"name": "Island In The Moon", |
|||
"description": "A place to be playin' and havin' fun", |
|||
}, |
|||
{"name": "Holy Buddies"}, |
|||
], |
|||
}, |
|||
{ |
|||
"name": "System of an Up", |
|||
"items": [ |
|||
{ |
|||
"name": "Salt", |
|||
"description": "The kombucha mushroom people's favorite", |
|||
}, |
|||
{"name": "Pad Thai"}, |
|||
{ |
|||
"name": "Lonely Night", |
|||
"description": "The mostests lonliest nightiest of allest", |
|||
}, |
|||
], |
|||
}, |
|||
] |
|||
@ -0,0 +1,51 @@ |
|||
from fastapi import APIRouter, FastAPI |
|||
from pydantic import BaseModel, HttpUrl |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Invoice(BaseModel): |
|||
id: str |
|||
title: str | None = None |
|||
customer: str |
|||
total: float |
|||
|
|||
|
|||
class InvoiceEvent(BaseModel): |
|||
description: str |
|||
paid: bool |
|||
|
|||
|
|||
class InvoiceEventReceived(BaseModel): |
|||
ok: bool |
|||
|
|||
|
|||
invoices_callback_router = APIRouter() |
|||
|
|||
|
|||
@invoices_callback_router.post( |
|||
"{$callback_url}/invoices/{$request.body.id}", response_model=InvoiceEventReceived |
|||
) |
|||
def invoice_notification(body: InvoiceEvent): |
|||
pass |
|||
|
|||
|
|||
@app.post("/invoices/", callbacks=invoices_callback_router.routes) |
|||
def create_invoice(invoice: Invoice, callback_url: HttpUrl | None = None): |
|||
""" |
|||
Create an invoice. |
|||
|
|||
This will (let's imagine) let the API user (some external developer) create an |
|||
invoice. |
|||
|
|||
And this path operation will: |
|||
|
|||
* Send the invoice to the client. |
|||
* Collect the money from the client. |
|||
* Send a notification back to the API user (the external developer), as a callback. |
|||
* At this point is that the API will somehow send a POST request to the |
|||
external API with the notification of the invoice event |
|||
(e.g. "payment successful"). |
|||
""" |
|||
# Send the invoice, collect the money, send the notification (the callback) |
|||
return {"msg": "Invoice received"} |
|||
@ -0,0 +1,28 @@ |
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: str | None = None |
|||
price: float |
|||
tax: float | None = None |
|||
tags: set[str] = set() |
|||
|
|||
|
|||
@app.post("/items/", response_model=Item, summary="Create an item") |
|||
async def create_item(item: Item): |
|||
""" |
|||
Create an item with all the information: |
|||
|
|||
- **name**: each item must have a name |
|||
- **description**: a long description |
|||
- **price**: required |
|||
- **tax**: if the item doesn't have tax, you can omit this |
|||
- **tags**: a set of unique tag strings for this item |
|||
\f |
|||
:param item: User input. |
|||
""" |
|||
return item |
|||
@ -0,0 +1,30 @@ |
|||
from typing import Union |
|||
|
|||
from fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
name: str |
|||
description: Union[str, None] = None |
|||
price: float |
|||
tax: Union[float, None] = None |
|||
tags: set[str] = set() |
|||
|
|||
|
|||
@app.post("/items/", response_model=Item, summary="Create an item") |
|||
async def create_item(item: Item): |
|||
""" |
|||
Create an item with all the information: |
|||
|
|||
- **name**: each item must have a name |
|||
- **description**: a long description |
|||
- **price**: required |
|||
- **tax**: if the item doesn't have tax, you can omit this |
|||
- **tags**: a set of unique tag strings for this item |
|||
\f |
|||
:param item: User input. |
|||
""" |
|||
return item |
|||
@ -0,0 +1,32 @@ |
|||
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.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.parse_obj(data) |
|||
except ValidationError as e: |
|||
raise HTTPException(status_code=422, detail=e.errors()) |
|||
return item |
|||
@ -0,0 +1,32 @@ |
|||
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 |
|||
@ -0,0 +1,21 @@ |
|||
from datetime import datetime |
|||
|
|||
from fastapi import FastAPI |
|||
from fastapi.encoders import jsonable_encoder |
|||
from fastapi.responses import JSONResponse |
|||
from pydantic import BaseModel |
|||
|
|||
|
|||
class Item(BaseModel): |
|||
title: str |
|||
timestamp: datetime |
|||
description: str | None = None |
|||
|
|||
|
|||
app = FastAPI() |
|||
|
|||
|
|||
@app.put("/items/{id}") |
|||
def update_item(id: str, item: Item): |
|||
json_compatible_item_data = jsonable_encoder(item) |
|||
return JSONResponse(content=json_compatible_item_data) |
|||
@ -0,0 +1,10 @@ |
|||
from pydantic import BaseSettings |
|||
|
|||
|
|||
class Settings(BaseSettings): |
|||
app_name: str = "Awesome API" |
|||
admin_email: str |
|||
items_per_user: int = 50 |
|||
|
|||
class Config: |
|||
env_file = ".env" |
|||
@ -0,0 +1,10 @@ |
|||
from pydantic import BaseSettings |
|||
|
|||
|
|||
class Settings(BaseSettings): |
|||
app_name: str = "Awesome API" |
|||
admin_email: str |
|||
items_per_user: int = 50 |
|||
|
|||
class Config: |
|||
env_file = ".env" |
|||
@ -0,0 +1,288 @@ |
|||
import importlib |
|||
|
|||
import pytest |
|||
from fastapi.testclient import TestClient |
|||
|
|||
from ...utils import needs_py310, needs_pydanticv1, needs_pydanticv2 |
|||
|
|||
|
|||
@pytest.fixture( |
|||
name="client", |
|||
params=[ |
|||
pytest.param("tutorial001"), |
|||
pytest.param("tutorial001_py310", marks=needs_py310), |
|||
], |
|||
) |
|||
def get_client(request: pytest.FixtureRequest): |
|||
mod = importlib.import_module(f"docs_src.response_directly.{request.param}") |
|||
|
|||
client = TestClient(mod.app) |
|||
return client |
|||
|
|||
|
|||
def test_path_operation(client: TestClient): |
|||
response = client.put( |
|||
"/items/1", |
|||
json={ |
|||
"title": "Foo", |
|||
"timestamp": "2023-01-01T12:00:00", |
|||
"description": "A test item", |
|||
}, |
|||
) |
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == { |
|||
"description": "A test item", |
|||
"timestamp": "2023-01-01T12:00:00", |
|||
"title": "Foo", |
|||
} |
|||
|
|||
|
|||
@needs_pydanticv2 |
|||
def test_openapi_schema_pv2(client: TestClient): |
|||
response = client.get("/openapi.json") |
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == { |
|||
"info": { |
|||
"title": "FastAPI", |
|||
"version": "0.1.0", |
|||
}, |
|||
"openapi": "3.1.0", |
|||
"paths": { |
|||
"/items/{id}": { |
|||
"put": { |
|||
"operationId": "update_item_items__id__put", |
|||
"parameters": [ |
|||
{ |
|||
"in": "path", |
|||
"name": "id", |
|||
"required": True, |
|||
"schema": {"title": "Id", "type": "string"}, |
|||
}, |
|||
], |
|||
"requestBody": { |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/Item", |
|||
}, |
|||
}, |
|||
}, |
|||
"required": True, |
|||
}, |
|||
"responses": { |
|||
"200": { |
|||
"content": { |
|||
"application/json": {"schema": {}}, |
|||
}, |
|||
"description": "Successful Response", |
|||
}, |
|||
"422": { |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/HTTPValidationError", |
|||
}, |
|||
}, |
|||
}, |
|||
"description": "Validation Error", |
|||
}, |
|||
}, |
|||
"summary": "Update Item", |
|||
}, |
|||
}, |
|||
}, |
|||
"components": { |
|||
"schemas": { |
|||
"HTTPValidationError": { |
|||
"properties": { |
|||
"detail": { |
|||
"items": { |
|||
"$ref": "#/components/schemas/ValidationError", |
|||
}, |
|||
"title": "Detail", |
|||
"type": "array", |
|||
}, |
|||
}, |
|||
"title": "HTTPValidationError", |
|||
"type": "object", |
|||
}, |
|||
"Item": { |
|||
"properties": { |
|||
"description": { |
|||
"anyOf": [ |
|||
{"type": "string"}, |
|||
{"type": "null"}, |
|||
], |
|||
"title": "Description", |
|||
}, |
|||
"timestamp": { |
|||
"format": "date-time", |
|||
"title": "Timestamp", |
|||
"type": "string", |
|||
}, |
|||
"title": {"title": "Title", "type": "string"}, |
|||
}, |
|||
"required": [ |
|||
"title", |
|||
"timestamp", |
|||
], |
|||
"title": "Item", |
|||
"type": "object", |
|||
}, |
|||
"ValidationError": { |
|||
"properties": { |
|||
"loc": { |
|||
"items": { |
|||
"anyOf": [ |
|||
{"type": "string"}, |
|||
{"type": "integer"}, |
|||
], |
|||
}, |
|||
"title": "Location", |
|||
"type": "array", |
|||
}, |
|||
"msg": {"title": "Message", "type": "string"}, |
|||
"type": {"title": "Error Type", "type": "string"}, |
|||
}, |
|||
"required": ["loc", "msg", "type"], |
|||
"title": "ValidationError", |
|||
"type": "object", |
|||
}, |
|||
}, |
|||
}, |
|||
} |
|||
|
|||
|
|||
@needs_pydanticv1 |
|||
def test_openapi_schema_pv1(client: TestClient): |
|||
response = client.get("/openapi.json") |
|||
assert response.status_code == 200, response.text |
|||
assert response.json() == { |
|||
"info": { |
|||
"title": "FastAPI", |
|||
"version": "0.1.0", |
|||
}, |
|||
"openapi": "3.1.0", |
|||
"paths": { |
|||
"/items/{id}": { |
|||
"put": { |
|||
"operationId": "update_item_items__id__put", |
|||
"parameters": [ |
|||
{ |
|||
"in": "path", |
|||
"name": "id", |
|||
"required": True, |
|||
"schema": { |
|||
"title": "Id", |
|||
"type": "string", |
|||
}, |
|||
}, |
|||
], |
|||
"requestBody": { |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/Item", |
|||
}, |
|||
}, |
|||
}, |
|||
"required": True, |
|||
}, |
|||
"responses": { |
|||
"200": { |
|||
"content": { |
|||
"application/json": { |
|||
"schema": {}, |
|||
}, |
|||
}, |
|||
"description": "Successful Response", |
|||
}, |
|||
"422": { |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/HTTPValidationError", |
|||
}, |
|||
}, |
|||
}, |
|||
"description": "Validation Error", |
|||
}, |
|||
}, |
|||
"summary": "Update Item", |
|||
}, |
|||
}, |
|||
}, |
|||
"components": { |
|||
"schemas": { |
|||
"HTTPValidationError": { |
|||
"properties": { |
|||
"detail": { |
|||
"items": { |
|||
"$ref": "#/components/schemas/ValidationError", |
|||
}, |
|||
"title": "Detail", |
|||
"type": "array", |
|||
}, |
|||
}, |
|||
"title": "HTTPValidationError", |
|||
"type": "object", |
|||
}, |
|||
"Item": { |
|||
"properties": { |
|||
"description": { |
|||
"title": "Description", |
|||
"type": "string", |
|||
}, |
|||
"timestamp": { |
|||
"format": "date-time", |
|||
"title": "Timestamp", |
|||
"type": "string", |
|||
}, |
|||
"title": { |
|||
"title": "Title", |
|||
"type": "string", |
|||
}, |
|||
}, |
|||
"required": [ |
|||
"title", |
|||
"timestamp", |
|||
], |
|||
"title": "Item", |
|||
"type": "object", |
|||
}, |
|||
"ValidationError": { |
|||
"properties": { |
|||
"loc": { |
|||
"items": { |
|||
"anyOf": [ |
|||
{ |
|||
"type": "string", |
|||
}, |
|||
{ |
|||
"type": "integer", |
|||
}, |
|||
], |
|||
}, |
|||
"title": "Location", |
|||
"type": "array", |
|||
}, |
|||
"msg": { |
|||
"title": "Message", |
|||
"type": "string", |
|||
}, |
|||
"type": { |
|||
"title": "Error Type", |
|||
"type": "string", |
|||
}, |
|||
}, |
|||
"required": [ |
|||
"loc", |
|||
"msg", |
|||
"type", |
|||
], |
|||
"title": "ValidationError", |
|||
"type": "object", |
|||
}, |
|||
}, |
|||
}, |
|||
} |
|||
@ -1,20 +1,45 @@ |
|||
import importlib |
|||
from types import ModuleType |
|||
|
|||
import pytest |
|||
from pytest import MonkeyPatch |
|||
|
|||
from ...utils import needs_pydanticv2 |
|||
from ...utils import needs_py39, needs_pydanticv2 |
|||
|
|||
|
|||
@needs_pydanticv2 |
|||
def test_settings(monkeypatch: MonkeyPatch): |
|||
from docs_src.settings.app02 import main |
|||
@pytest.fixture( |
|||
name="mod_path", |
|||
params=[ |
|||
pytest.param("app02"), |
|||
pytest.param("app02_an"), |
|||
pytest.param("app02_an_py39", marks=needs_py39), |
|||
], |
|||
) |
|||
def get_mod_path(request: pytest.FixtureRequest): |
|||
mod_path = f"docs_src.settings.{request.param}" |
|||
return mod_path |
|||
|
|||
|
|||
@pytest.fixture(name="main_mod") |
|||
def get_main_mod(mod_path: str) -> ModuleType: |
|||
main_mod = importlib.import_module(f"{mod_path}.main") |
|||
return main_mod |
|||
|
|||
|
|||
@pytest.fixture(name="test_main_mod") |
|||
def get_test_main_mod(mod_path: str) -> ModuleType: |
|||
test_main_mod = importlib.import_module(f"{mod_path}.test_main") |
|||
return test_main_mod |
|||
|
|||
|
|||
@needs_pydanticv2 |
|||
def test_settings(main_mod: ModuleType, monkeypatch: MonkeyPatch): |
|||
monkeypatch.setenv("ADMIN_EMAIL", "[email protected]") |
|||
settings = main.get_settings() |
|||
settings = main_mod.get_settings() |
|||
assert settings.app_name == "Awesome API" |
|||
assert settings.items_per_user == 50 |
|||
|
|||
|
|||
@needs_pydanticv2 |
|||
def test_override_settings(): |
|||
from docs_src.settings.app02 import test_main |
|||
|
|||
test_main.test_app() |
|||
def test_override_settings(test_main_mod: ModuleType): |
|||
test_main_mod.test_app() |
|||
|
|||
@ -0,0 +1,59 @@ |
|||
import importlib |
|||
from types import ModuleType |
|||
|
|||
import pytest |
|||
from fastapi.testclient import TestClient |
|||
from pytest import MonkeyPatch |
|||
|
|||
from ...utils import needs_py39, needs_pydanticv1, needs_pydanticv2 |
|||
|
|||
|
|||
@pytest.fixture( |
|||
name="mod_path", |
|||
params=[ |
|||
pytest.param("app03"), |
|||
pytest.param("app03_an"), |
|||
pytest.param("app03_an_py39", marks=needs_py39), |
|||
], |
|||
) |
|||
def get_mod_path(request: pytest.FixtureRequest): |
|||
mod_path = f"docs_src.settings.{request.param}" |
|||
return mod_path |
|||
|
|||
|
|||
@pytest.fixture(name="main_mod") |
|||
def get_main_mod(mod_path: str) -> ModuleType: |
|||
main_mod = importlib.import_module(f"{mod_path}.main") |
|||
return main_mod |
|||
|
|||
|
|||
@needs_pydanticv2 |
|||
def test_settings(main_mod: ModuleType, monkeypatch: MonkeyPatch): |
|||
monkeypatch.setenv("ADMIN_EMAIL", "[email protected]") |
|||
settings = main_mod.get_settings() |
|||
assert settings.app_name == "Awesome API" |
|||
assert settings.admin_email == "[email protected]" |
|||
assert settings.items_per_user == 50 |
|||
|
|||
|
|||
@needs_pydanticv1 |
|||
def test_settings_pv1(mod_path: str, monkeypatch: MonkeyPatch): |
|||
monkeypatch.setenv("ADMIN_EMAIL", "[email protected]") |
|||
config_mod = importlib.import_module(f"{mod_path}.config_pv1") |
|||
settings = config_mod.Settings() |
|||
assert settings.app_name == "Awesome API" |
|||
assert settings.admin_email == "[email protected]" |
|||
assert settings.items_per_user == 50 |
|||
|
|||
|
|||
@needs_pydanticv2 |
|||
def test_endpoint(main_mod: ModuleType, monkeypatch: MonkeyPatch): |
|||
monkeypatch.setenv("ADMIN_EMAIL", "[email protected]") |
|||
client = TestClient(main_mod.app) |
|||
response = client.get("/info") |
|||
assert response.status_code == 200 |
|||
assert response.json() == { |
|||
"app_name": "Awesome API", |
|||
"admin_email": "[email protected]", |
|||
"items_per_user": 50, |
|||
} |
|||
@ -0,0 +1,112 @@ |
|||
from fastapi.testclient import TestClient |
|||
|
|||
from docs_src.using_request_directly.tutorial001 import app |
|||
|
|||
client = TestClient(app) |
|||
|
|||
|
|||
def test_path_operation(): |
|||
response = client.get("/items/foo") |
|||
assert response.status_code == 200 |
|||
assert response.json() == {"client_host": "testclient", "item_id": "foo"} |
|||
|
|||
|
|||
def test_openapi(): |
|||
response = client.get("/openapi.json") |
|||
assert response.status_code == 200 |
|||
assert response.json() == { |
|||
"info": { |
|||
"title": "FastAPI", |
|||
"version": "0.1.0", |
|||
}, |
|||
"openapi": "3.1.0", |
|||
"paths": { |
|||
"/items/{item_id}": { |
|||
"get": { |
|||
"operationId": "read_root_items__item_id__get", |
|||
"parameters": [ |
|||
{ |
|||
"in": "path", |
|||
"name": "item_id", |
|||
"required": True, |
|||
"schema": { |
|||
"title": "Item Id", |
|||
"type": "string", |
|||
}, |
|||
}, |
|||
], |
|||
"responses": { |
|||
"200": { |
|||
"content": { |
|||
"application/json": { |
|||
"schema": {}, |
|||
}, |
|||
}, |
|||
"description": "Successful Response", |
|||
}, |
|||
"422": { |
|||
"content": { |
|||
"application/json": { |
|||
"schema": { |
|||
"$ref": "#/components/schemas/HTTPValidationError", |
|||
}, |
|||
}, |
|||
}, |
|||
"description": "Validation Error", |
|||
}, |
|||
}, |
|||
"summary": "Read Root", |
|||
}, |
|||
}, |
|||
}, |
|||
"components": { |
|||
"schemas": { |
|||
"HTTPValidationError": { |
|||
"properties": { |
|||
"detail": { |
|||
"items": { |
|||
"$ref": "#/components/schemas/ValidationError", |
|||
}, |
|||
"title": "Detail", |
|||
"type": "array", |
|||
}, |
|||
}, |
|||
"title": "HTTPValidationError", |
|||
"type": "object", |
|||
}, |
|||
"ValidationError": { |
|||
"properties": { |
|||
"loc": { |
|||
"items": { |
|||
"anyOf": [ |
|||
{ |
|||
"type": "string", |
|||
}, |
|||
{ |
|||
"type": "integer", |
|||
}, |
|||
], |
|||
}, |
|||
"title": "Location", |
|||
"type": "array", |
|||
}, |
|||
"msg": { |
|||
"title": "Message", |
|||
"type": "string", |
|||
}, |
|||
"type": { |
|||
"title": "Error Type", |
|||
"type": "string", |
|||
}, |
|||
}, |
|||
"required": [ |
|||
"loc", |
|||
"msg", |
|||
"type", |
|||
], |
|||
"title": "ValidationError", |
|||
"type": "object", |
|||
}, |
|||
}, |
|||
}, |
|||
} |
|||
@ -1,50 +0,0 @@ |
|||
import pytest |
|||
from fastapi import FastAPI |
|||
from fastapi.testclient import TestClient |
|||
|
|||
from ...utils import needs_py39 |
|||
|
|||
|
|||
@pytest.fixture(name="app") |
|||
def get_app(): |
|||
from docs_src.websockets.tutorial003_py39 import app |
|||
|
|||
return app |
|||
|
|||
|
|||
@pytest.fixture(name="html") |
|||
def get_html(): |
|||
from docs_src.websockets.tutorial003_py39 import html |
|||
|
|||
return html |
|||
|
|||
|
|||
@pytest.fixture(name="client") |
|||
def get_client(app: FastAPI): |
|||
client = TestClient(app) |
|||
|
|||
return client |
|||
|
|||
|
|||
@needs_py39 |
|||
def test_get(client: TestClient, html: str): |
|||
response = client.get("/") |
|||
assert response.text == html |
|||
|
|||
|
|||
@needs_py39 |
|||
def test_websocket_handle_disconnection(client: TestClient): |
|||
with client.websocket_connect("/ws/1234") as connection, client.websocket_connect( |
|||
"/ws/5678" |
|||
) as connection_two: |
|||
connection.send_text("Hello from 1234") |
|||
data1 = connection.receive_text() |
|||
assert data1 == "You wrote: Hello from 1234" |
|||
data2 = connection_two.receive_text() |
|||
client1_says = "Client #1234 says: Hello from 1234" |
|||
assert data2 == client1_says |
|||
data1 = connection.receive_text() |
|||
assert data1 == client1_says |
|||
connection_two.close() |
|||
data1 = connection.receive_text() |
|||
assert data1 == "Client #5678 left the chat" |
|||
Loading…
Reference in new issue