Sebastián Ramírez
5 months ago
111 changed files with 4026 additions and 1022 deletions
@ -0,0 +1,63 @@ |
|||||
|
# JSON with Bytes as Base64 { #json-with-bytes-as-base64 } |
||||
|
|
||||
|
If your app needs to receive and send JSON data, but you need to include binary data in it, you can encode it as base64. |
||||
|
|
||||
|
## Base64 vs Files { #base64-vs-files } |
||||
|
|
||||
|
Consider first if you can use [Request Files](../tutorial/request-files.md){.internal-link target=_blank} for uploading binary data and [Custom Response - FileResponse](./custom-response.md#fileresponse--fileresponse-){.internal-link target=_blank} for sending binary data, instead of encoding it in JSON. |
||||
|
|
||||
|
JSON can only contain UTF-8 encoded strings, so it can't contain raw bytes. |
||||
|
|
||||
|
Base64 can encode binary data in strings, but to do it, it needs to use more characters than the original binary data, so it would normally be less efficient than regular files. |
||||
|
|
||||
|
Use base64 only if you definitely need to include binary data in JSON, and you can't use files for that. |
||||
|
|
||||
|
## Pydantic `bytes` { #pydantic-bytes } |
||||
|
|
||||
|
You can declare a Pydantic model with `bytes` fields, and then use `val_json_bytes` in the model config to tell it to use base64 to *validate* input JSON data, as part of that validation it will decode the base64 string into bytes. |
||||
|
|
||||
|
{* ../../docs_src/json_base64_bytes/tutorial001_py310.py ln[1:9,29:35] hl[9] *} |
||||
|
|
||||
|
If you check the `/docs`, they will show that the field `data` expects base64 encoded bytes: |
||||
|
|
||||
|
<div class="screenshot"> |
||||
|
<img src="/img/tutorial/json-base64-bytes/image01.png"> |
||||
|
</div> |
||||
|
|
||||
|
You could send a request like: |
||||
|
|
||||
|
```json |
||||
|
{ |
||||
|
"description": "Some data", |
||||
|
"data": "aGVsbG8=" |
||||
|
} |
||||
|
``` |
||||
|
|
||||
|
/// tip |
||||
|
|
||||
|
`aGVsbG8=` is the base64 encoding of `hello`. |
||||
|
|
||||
|
/// |
||||
|
|
||||
|
And then Pydantic will decode the base64 string and give you the original bytes in the `data` field of the model. |
||||
|
|
||||
|
You will receive a response like: |
||||
|
|
||||
|
```json |
||||
|
{ |
||||
|
"description": "Some data", |
||||
|
"content": "hello" |
||||
|
} |
||||
|
``` |
||||
|
|
||||
|
## Pydantic `bytes` for Output Data { #pydantic-bytes-for-output-data } |
||||
|
|
||||
|
You can also use `bytes` fields with `ser_json_bytes` in the model config for output data, and Pydantic will *serialize* the bytes as base64 when generating the JSON response. |
||||
|
|
||||
|
{* ../../docs_src/json_base64_bytes/tutorial001_py310.py ln[1:2,12:16,29,38:41] hl[16] *} |
||||
|
|
||||
|
## Pydantic `bytes` for Input and Output Data { #pydantic-bytes-for-input-and-output-data } |
||||
|
|
||||
|
And of course, you can use the same model configured to use base64 to handle both input (*validate*) with `val_json_bytes` and output (*serialize*) with `ser_json_bytes` when receiving and sending JSON data. |
||||
|
|
||||
|
{* ../../docs_src/json_base64_bytes/tutorial001_py310.py ln[1:2,19:26,29,44:46] hl[23:26] *} |
||||
@ -0,0 +1,99 @@ |
|||||
|
# Stream Data { #stream-data } |
||||
|
|
||||
|
If you want to stream data that can be structured as JSON, you should [Stream JSON Lines](../tutorial/stream-json-lines.md){.internal-link target=_blank}. |
||||
|
|
||||
|
But if you want to **stream pure binary data** or strings, here's how you can do it. |
||||
|
|
||||
|
## Use Cases { #use-cases } |
||||
|
|
||||
|
You could use this if you want to stream pure strings, for example directly from the output of an **AI LLM** service. |
||||
|
|
||||
|
You could also use it to stream **large binary files**, where you stream each chunk of data as you read it, without having to read it all in memory at once. |
||||
|
|
||||
|
You could also stream **video** or **audio** this way, it could even be generated as you process and send it. |
||||
|
|
||||
|
## A `StreamingResponse` with `yield` { #a-streamingresponse-with-yield } |
||||
|
|
||||
|
If you declare a `response_class=StreamingResponse` in your *path operation function*, you can use `yield` to send each chunk of data in turn. |
||||
|
|
||||
|
{* ../../docs_src/stream_data/tutorial001_py310.py ln[1:23] hl[20,23] *} |
||||
|
|
||||
|
FastAPI will give each chunk of data to the `StreamingResponse` as is, it won't try to convert it to JSON or anything similar. |
||||
|
|
||||
|
### Non-async *path operation functions* { #non-async-path-operation-functions } |
||||
|
|
||||
|
You can also use regular `def` functions (without `async`), and use `yield` the same way. |
||||
|
|
||||
|
{* ../../docs_src/stream_data/tutorial001_py310.py ln[26:29] hl[27] *} |
||||
|
|
||||
|
### No Annotation { #no-annotation } |
||||
|
|
||||
|
You don't really need to declare the return type annotation for streaming binary data. |
||||
|
|
||||
|
As FastAPI will not try to convert the data to JSON with Pydantic or serialize it in any way, in this case, the type annotation is only for your editor and tools to use, it won't be used by FastAPI. |
||||
|
|
||||
|
{* ../../docs_src/stream_data/tutorial001_py310.py ln[32:35] hl[33] *} |
||||
|
|
||||
|
This also means that with `StreamingResponse` you have the **freedom** and **responsibility** to produce and encode the data bytes exactly as you need them to be sent, independent of the type annotations. 🤓 |
||||
|
|
||||
|
### Stream Bytes { #stream-bytes } |
||||
|
|
||||
|
One of the main use cases would be to stream `bytes` instead of strings, you can of course do it. |
||||
|
|
||||
|
{* ../../docs_src/stream_data/tutorial001_py310.py ln[44:47] hl[47] *} |
||||
|
|
||||
|
## A Custom `PNGStreamingResponse` { #a-custom-pngstreamingresponse } |
||||
|
|
||||
|
In the examples above, the data bytes were streamed, but the response didn't have a `Content-Type` header, so the client didn't know what type of data it was receiving. |
||||
|
|
||||
|
You can create a custom sub-class of `StreamingResponse` that sets the `Content-Type` header to the type of data you're streaming. |
||||
|
|
||||
|
For example, you can create a `PNGStreamingResponse` that sets the `Content-Type` header to `image/png` using the `media_type` attribute: |
||||
|
|
||||
|
{* ../../docs_src/stream_data/tutorial002_py310.py ln[6,19:20] hl[20] *} |
||||
|
|
||||
|
Then you can use this new class in `response_class=PNGStreamingResponse` in your *path operation function*: |
||||
|
|
||||
|
{* ../../docs_src/stream_data/tutorial002_py310.py ln[23:26] hl[23] *} |
||||
|
|
||||
|
### Simulate a File { #simulate-a-file } |
||||
|
|
||||
|
In this example, we are simulating a file with `io.BytesIO`, which is a file-like object that lives only in memory, but lets us use the same interface. |
||||
|
|
||||
|
For example, we can iterate over it to consume its contents, as we could with a file. |
||||
|
|
||||
|
{* ../../docs_src/stream_data/tutorial002_py310.py ln[1:26] hl[3,12:13,25] *} |
||||
|
|
||||
|
/// note | Technical Details |
||||
|
|
||||
|
The other two variables, `image_base64` and `binary_image`, are an image encoded in Base64, and then converted to bytes, to then pass it to `io.BytesIO`. |
||||
|
|
||||
|
Only so that it can live in the same file for this example and you can copy it and run it as is. 🥚 |
||||
|
|
||||
|
/// |
||||
|
|
||||
|
### Files and Async { #files-and-async } |
||||
|
|
||||
|
In most cases, file-like objects are not compatible with async and await by default. |
||||
|
|
||||
|
For example, they don't have an `await file.read()`, or `async for chunk in file`. |
||||
|
|
||||
|
And in many cases, reading them would be a blocking operation (that could block the event loop), because they are read from disk or from the network. |
||||
|
|
||||
|
/// info |
||||
|
|
||||
|
The example above is actually an exception, because the `io.BytesIO` object is already in memory, so reading it won't block anything. |
||||
|
|
||||
|
But in many cases reading a file or a file-like object would block. |
||||
|
|
||||
|
/// |
||||
|
|
||||
|
To avoid blocking the event loop, you can simply declare the *path operation function* with regular `def` instead of `async def`, that way FastAPI will run it on a threadpool worker, to avoid blocking the main loop. |
||||
|
|
||||
|
{* ../../docs_src/stream_data/tutorial002_py310.py ln[29:32] hl[30] *} |
||||
|
|
||||
|
/// tip |
||||
|
|
||||
|
If you need to call blocking code from inside of an async function, or an async function from inside of a blocking function, you could use <a href="https://asyncer.tiangolo.com" class="external-link" target="_blank">Asyncer</a>, a sibling library to FastAPI. |
||||
|
|
||||
|
/// |
||||
@ -0,0 +1,88 @@ |
|||||
|
# Strict Content-Type Checking { #strict-content-type-checking } |
||||
|
|
||||
|
By default, **FastAPI** uses strict `Content-Type` header checking for JSON request bodies, this means that JSON requests **must** include a valid `Content-Type` header (e.g. `application/json`) in order for the body to be parsed as JSON. |
||||
|
|
||||
|
## CSRF Risk { #csrf-risk } |
||||
|
|
||||
|
This default behavior provides protection against a class of **Cross-Site Request Forgery (CSRF)** attacks in a very specific scenario. |
||||
|
|
||||
|
These attacks exploit the fact that browsers allow scripts to send requests without doing any CORS preflight check when they: |
||||
|
|
||||
|
* don't have a `Content-Type` header (e.g. using `fetch()` with a `Blob` body) |
||||
|
* and don't send any authentication credentials. |
||||
|
|
||||
|
This type of attack is mainly relevant when: |
||||
|
|
||||
|
* the application is running locally (e.g. on `localhost`) or in an internal network |
||||
|
* and the application doesn't have any authentication, it expects that any request from the same network can be trusted. |
||||
|
|
||||
|
## Example Attack { #example-attack } |
||||
|
|
||||
|
Imagine you build a way to run a local AI agent. |
||||
|
|
||||
|
It provides an API at |
||||
|
|
||||
|
``` |
||||
|
http://localhost:8000/v1/agents/multivac |
||||
|
``` |
||||
|
|
||||
|
There's also a frontend at |
||||
|
|
||||
|
``` |
||||
|
http://localhost:8000 |
||||
|
``` |
||||
|
|
||||
|
/// tip |
||||
|
|
||||
|
Note that both have the same host. |
||||
|
|
||||
|
/// |
||||
|
|
||||
|
Then using the frontend you can make the AI agent do things on your behalf. |
||||
|
|
||||
|
As it's running **locally**, and not in the open internet, you decide to **not have any authentication** set up, just trusting the access to the local network. |
||||
|
|
||||
|
Then one of your users could install it and run it locally. |
||||
|
|
||||
|
Then they could open a malicious website, e.g. something like |
||||
|
|
||||
|
``` |
||||
|
https://evilhackers.example.com |
||||
|
``` |
||||
|
|
||||
|
And that malicious website sends requests using `fetch()` with a `Blob` body to the local API at |
||||
|
|
||||
|
``` |
||||
|
http://localhost:8000/v1/agents/multivac |
||||
|
``` |
||||
|
|
||||
|
Even though the host of the malicious website and the local app is different, the browser won't trigger a CORS preflight request because: |
||||
|
|
||||
|
* It's running without any authentication, it doesn't have to send any credentials. |
||||
|
* The browser thinks it's not sending JSON (because of the missing `Content-Type` header). |
||||
|
|
||||
|
Then the malicious website could make the local AI agent send angry messages to the user's ex-boss... or worse. 😅 |
||||
|
|
||||
|
## Open Internet { #open-internet } |
||||
|
|
||||
|
If your app is in the open internet, you wouldn't "trust the network" and let anyone send privileged requests without authentication. |
||||
|
|
||||
|
Attackers could simply run a script to send requests to your API, no need for browser interaction, so you are probably already securing any privileged endpoints. |
||||
|
|
||||
|
In that case **this attack / risk doesn't apply to you**. |
||||
|
|
||||
|
This risk and attack is mainly relevant when the app runs on the **local network** and that is the **only assumed protection**. |
||||
|
|
||||
|
## Allowing Requests Without Content-Type { #allowing-requests-without-content-type } |
||||
|
|
||||
|
If you need to support clients that don't send a `Content-Type` header, you can disable strict checking by setting `strict_content_type=False`: |
||||
|
|
||||
|
{* ../../docs_src/strict_content_type/tutorial001_py310.py hl[4] *} |
||||
|
|
||||
|
With this setting, requests without a `Content-Type` header will have their body parsed as JSON, which is the same behavior as older versions of FastAPI. |
||||
|
|
||||
|
/// info |
||||
|
|
||||
|
This behavior and configuration was added in FastAPI 0.132.0. |
||||
|
|
||||
|
/// |
||||
|
After Width: | Height: | Size: 71 KiB |
@ -0,0 +1,105 @@ |
|||||
|
# Stream JSON Lines { #stream-json-lines } |
||||
|
|
||||
|
You could have a sequence of data that you would like to send in a "**stream**", you could do it with **JSON Lines**. |
||||
|
|
||||
|
## What is a Stream? { #what-is-a-stream } |
||||
|
|
||||
|
"**Streaming**" data means that your app will start sending data items to the client without waiting for the entire sequence of items to be ready. |
||||
|
|
||||
|
So, it will send the first item, the client will receive and start processing it, and you might still be producing the next item. |
||||
|
|
||||
|
```mermaid |
||||
|
sequenceDiagram |
||||
|
participant App |
||||
|
participant Client |
||||
|
|
||||
|
App->>App: Produce Item 1 |
||||
|
App->>Client: Send Item 1 |
||||
|
App->>App: Produce Item 2 |
||||
|
Client->>Client: Process Item 1 |
||||
|
App->>Client: Send Item 2 |
||||
|
App->>App: Produce Item 3 |
||||
|
Client->>Client: Process Item 2 |
||||
|
App->>Client: Send Item 3 |
||||
|
Client->>Client: Process Item 3 |
||||
|
Note over App: Keeps producing... |
||||
|
Note over Client: Keeps consuming... |
||||
|
``` |
||||
|
|
||||
|
It could even be an infinite stream, where you keep sending data. |
||||
|
|
||||
|
## JSON Lines { #json-lines } |
||||
|
|
||||
|
In these cases, it's common to send "**JSON Lines**", which is a format where you send one JSON object per line. |
||||
|
|
||||
|
A response would have a content type of `application/jsonl` (instead of `application/json`) and the body would be something like: |
||||
|
|
||||
|
```json |
||||
|
{"name": "Plumbus", "description": "A multi-purpose household device."} |
||||
|
{"name": "Portal Gun", "description": "A portal opening device."} |
||||
|
{"name": "Meeseeks Box", "description": "A box that summons a Meeseeks."} |
||||
|
``` |
||||
|
|
||||
|
It's very similar to a JSON array (equivalent of a Python list), but instead of being wrapped in `[]` and having `,` between the items, it has **one JSON object per line**, they are separated by a new line character. |
||||
|
|
||||
|
/// info |
||||
|
|
||||
|
The important point is that your app will be able to produce each line in turn, while the client consumes the previous lines. |
||||
|
|
||||
|
/// |
||||
|
|
||||
|
/// note | Technical Details |
||||
|
|
||||
|
Because each JSON object will be separated by a new line, they can't contain literal new line characters in their content, but they can contain escaped new lines (`\n`), which is part of the JSON standard. |
||||
|
|
||||
|
But normally you won't have to worry about it, it's done automatically, continue reading. 🤓 |
||||
|
|
||||
|
/// |
||||
|
|
||||
|
## Use Cases { #use-cases } |
||||
|
|
||||
|
You could use this to stream data from an **AI LLM** service, from **logs** or **telemetry**, or from other types of data that can be structured in **JSON** items. |
||||
|
|
||||
|
/// tip |
||||
|
|
||||
|
If you want to stream binary data, for example video or audio, check the advanced guide: [Stream Data](../advanced/stream-data.md). |
||||
|
|
||||
|
/// |
||||
|
|
||||
|
## Stream JSON Lines with FastAPI { #stream-json-lines-with-fastapi } |
||||
|
|
||||
|
To stream JSON Lines with FastAPI you can, instead of using `return` in your *path operation function*, use `yield` to produce each item in turn. |
||||
|
|
||||
|
{* ../../docs_src/stream_json_lines/tutorial001_py310.py ln[1:24] hl[24] *} |
||||
|
|
||||
|
If each JSON item you want to send back is of type `Item` (a Pydantic model) and it's an async function, you can declare the return type as `AsyncIterable[Item]`: |
||||
|
|
||||
|
{* ../../docs_src/stream_json_lines/tutorial001_py310.py ln[1:24] hl[9:11,22] *} |
||||
|
|
||||
|
If you declare the return type, FastAPI will use it to **validate** the data, **document** it in OpenAPI, **filter** it, and **serialize** it using Pydantic. |
||||
|
|
||||
|
/// tip |
||||
|
|
||||
|
As Pydantic will serialize it in the **Rust** side, you will get much higher **performance** than if you don't declare a return type. |
||||
|
|
||||
|
/// |
||||
|
|
||||
|
### Non-async *path operation functions* { #non-async-path-operation-functions } |
||||
|
|
||||
|
You can also use regular `def` functions (without `async`), and use `yield` the same way. |
||||
|
|
||||
|
FastAPI will make sure it's run correctly so that it doesn't block the event loop. |
||||
|
|
||||
|
As in this case the function is not async, the right return type would be `Iterable[Item]`: |
||||
|
|
||||
|
{* ../../docs_src/stream_json_lines/tutorial001_py310.py ln[27:30] hl[28] *} |
||||
|
|
||||
|
### No Return Type { #no-return-type } |
||||
|
|
||||
|
You can also omit the return type. FastAPI will then use the [`jsonable_encoder`](./encoder.md){.internal-link target=_blank} to convert the data to something that can be serialized to JSON and then send it as JSON Lines. |
||||
|
|
||||
|
{* ../../docs_src/stream_json_lines/tutorial001_py310.py ln[33:36] hl[34] *} |
||||
|
|
||||
|
## Server Sent Events (SSE) { #server-sent-events-sse } |
||||
|
|
||||
|
A future version of FastAPI will also have first-class support for Server Sent Events (SSE), which are quite similar, but with a couple of extra details. 🤓 |
||||
@ -1,9 +1,9 @@ |
|||||
from fastapi import FastAPI |
from fastapi import FastAPI |
||||
from fastapi.responses import ORJSONResponse |
from fastapi.responses import HTMLResponse |
||||
|
|
||||
app = FastAPI(default_response_class=ORJSONResponse) |
app = FastAPI(default_response_class=HTMLResponse) |
||||
|
|
||||
|
|
||||
@app.get("/items/") |
@app.get("/items/") |
||||
async def read_items(): |
async def read_items(): |
||||
return [{"item_id": "Foo"}] |
return "<h1>Items</h1><p>This is a list of items.</p>" |
||||
|
|||||
@ -0,0 +1,46 @@ |
|||||
|
from fastapi import FastAPI |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
|
||||
|
class DataInput(BaseModel): |
||||
|
description: str |
||||
|
data: bytes |
||||
|
|
||||
|
model_config = {"val_json_bytes": "base64"} |
||||
|
|
||||
|
|
||||
|
class DataOutput(BaseModel): |
||||
|
description: str |
||||
|
data: bytes |
||||
|
|
||||
|
model_config = {"ser_json_bytes": "base64"} |
||||
|
|
||||
|
|
||||
|
class DataInputOutput(BaseModel): |
||||
|
description: str |
||||
|
data: bytes |
||||
|
|
||||
|
model_config = { |
||||
|
"val_json_bytes": "base64", |
||||
|
"ser_json_bytes": "base64", |
||||
|
} |
||||
|
|
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.post("/data") |
||||
|
def post_data(body: DataInput): |
||||
|
content = body.data.decode("utf-8") |
||||
|
return {"description": body.description, "content": content} |
||||
|
|
||||
|
|
||||
|
@app.get("/data") |
||||
|
def get_data() -> DataOutput: |
||||
|
data = "hello".encode("utf-8") |
||||
|
return DataOutput(description="A plumbus", data=data) |
||||
|
|
||||
|
|
||||
|
@app.post("/data-in-out") |
||||
|
def post_data_in_out(body: DataInputOutput) -> DataInputOutput: |
||||
|
return body |
||||
@ -1,2 +0,0 @@ |
|||||
def get_items(item_a: str, item_b: int, item_c: float, item_d: bool, item_e: bytes): |
|
||||
return item_a, item_b, item_c, item_d, item_e |
|
||||
@ -0,0 +1,65 @@ |
|||||
|
from collections.abc import AsyncIterable, Iterable |
||||
|
|
||||
|
from fastapi import FastAPI |
||||
|
from fastapi.responses import StreamingResponse |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
message = """ |
||||
|
Rick: (stumbles in drunkenly, and turns on the lights) Morty! You gotta come on. You got--... you gotta come with me. |
||||
|
Morty: (rubs his eyes) What, Rick? What's going on? |
||||
|
Rick: I got a surprise for you, Morty. |
||||
|
Morty: It's the middle of the night. What are you talking about? |
||||
|
Rick: (spills alcohol on Morty's bed) Come on, I got a surprise for you. (drags Morty by the ankle) Come on, hurry up. (pulls Morty out of his bed and into the hall) |
||||
|
Morty: Ow! Ow! You're tugging me too hard! |
||||
|
Rick: We gotta go, gotta get outta here, come on. Got a surprise for you Morty. |
||||
|
""" |
||||
|
|
||||
|
|
||||
|
@app.get("/story/stream", response_class=StreamingResponse) |
||||
|
async def stream_story() -> AsyncIterable[str]: |
||||
|
for line in message.splitlines(): |
||||
|
yield line |
||||
|
|
||||
|
|
||||
|
@app.get("/story/stream-no-async", response_class=StreamingResponse) |
||||
|
def stream_story_no_async() -> Iterable[str]: |
||||
|
for line in message.splitlines(): |
||||
|
yield line |
||||
|
|
||||
|
|
||||
|
@app.get("/story/stream-no-annotation", response_class=StreamingResponse) |
||||
|
async def stream_story_no_annotation(): |
||||
|
for line in message.splitlines(): |
||||
|
yield line |
||||
|
|
||||
|
|
||||
|
@app.get("/story/stream-no-async-no-annotation", response_class=StreamingResponse) |
||||
|
def stream_story_no_async_no_annotation(): |
||||
|
for line in message.splitlines(): |
||||
|
yield line |
||||
|
|
||||
|
|
||||
|
@app.get("/story/stream-bytes", response_class=StreamingResponse) |
||||
|
async def stream_story_bytes() -> AsyncIterable[bytes]: |
||||
|
for line in message.splitlines(): |
||||
|
yield line.encode("utf-8") |
||||
|
|
||||
|
|
||||
|
@app.get("/story/stream-no-async-bytes", response_class=StreamingResponse) |
||||
|
def stream_story_no_async_bytes() -> Iterable[bytes]: |
||||
|
for line in message.splitlines(): |
||||
|
yield line.encode("utf-8") |
||||
|
|
||||
|
|
||||
|
@app.get("/story/stream-no-annotation-bytes", response_class=StreamingResponse) |
||||
|
async def stream_story_no_annotation_bytes(): |
||||
|
for line in message.splitlines(): |
||||
|
yield line.encode("utf-8") |
||||
|
|
||||
|
|
||||
|
@app.get("/story/stream-no-async-no-annotation-bytes", response_class=StreamingResponse) |
||||
|
def stream_story_no_async_no_annotation_bytes(): |
||||
|
for line in message.splitlines(): |
||||
|
yield line.encode("utf-8") |
||||
@ -0,0 +1,44 @@ |
|||||
|
import base64 |
||||
|
from collections.abc import AsyncIterable, Iterable |
||||
|
from io import BytesIO |
||||
|
|
||||
|
from fastapi import FastAPI |
||||
|
from fastapi.responses import StreamingResponse |
||||
|
|
||||
|
image_base64 = "iVBORw0KGgoAAAANSUhEUgAAAB0AAAAdCAYAAABWk2cPAAAAbnpUWHRSYXcgcHJvZmlsZSB0eXBlIGV4aWYAAHjadYzRDYAwCET/mcIRDoq0jGOiJm7g+NJK0vjhS4DjIEfHfZ20DKqSrrWZmyFQV5ctRMOLACxglNCcXk7zVqFzJzF8kV6R5vOJ97yVH78HjfYAtg0ged033ZgAAAoCaVRYdFhNTDpjb20uYWRvYmUueG1wAAAAAAA8P3hwYWNrZXQgYmVnaW49Iu+7vyIgaWQ9Ilc1TTBNcENlaGlIenJlU3pOVGN6a2M5ZCI/Pgo8eDp4bXBtZXRhIHhtbG5zOng9ImFkb2JlOm5zOm1ldGEvIiB4OnhtcHRrPSJYTVAgQ29yZSA0LjQuMC1FeGl2MiI+CiA8cmRmOlJERiB4bWxuczpyZGY9Imh0dHA6Ly93d3cudzMub3JnLzE5OTkvMDIvMjItcmRmLXN5bnRheC1ucyMiPgogIDxyZGY6RGVzY3JpcHRpb24gcmRmOmFib3V0PSIiCiAgICB4bWxuczpleGlmPSJodHRwOi8vbnMuYWRvYmUuY29tL2V4aWYvMS4wLyIKICAgIHhtbG5zOnRpZmY9Imh0dHA6Ly9ucy5hZG9iZS5jb20vdGlmZi8xLjAvIgogICBleGlmOlBpeGVsWERpbWVuc2lvbj0iMjkiCiAgIGV4aWY6UGl4ZWxZRGltZW5zaW9uPSIyOSIKICAgdGlmZjpJbWFnZVdpZHRoPSIyOSIKICAgdGlmZjpJbWFnZUxlbmd0aD0iMjkiCiAgIHRpZmY6T3JpZW50YXRpb249IjEiLz4KIDwvcmRmOlJERj4KPC94OnhtcG1ldGE+CiAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAKICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgIAogICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgCiAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAKICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgIAogICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgCiAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAKICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgIAogICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgCiAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAKICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgIAogICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgCiAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAKICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgIAogICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgCiAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAKICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgIAogICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgCiAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAKICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgIAogICAgICAgICAgICAgICAgICAgICAgICAgICAKPD94cGFja2V0IGVuZD0idyI/PnQkBZAAAAAEc0JJVAgICAh8CGSIAAABoklEQVRIx8VXwY7FIAjE5iXWU+P/f6RHPNW9LIaOoHYP+0yMShVkwNGG1lqjfy4HfaF0oyEEt+oSQqBaa//m9Wd6PlqhhbRMDiEQM3e59FNKw5qZHpnQfuPaW6lazsztvu/eElFj5j63lNLlMz2ttbZtVMu1MTGo5Sujn93gMzOllKiUQjHGB9QxxneZhJ5iwZ1rL2fwenoGeL0q3wVGhBPHMz0PeFccIfASEeWcO8xEROd50q6eAV6s1s5XXoncas1EKqVQznnwUBdJJmm1l3hmmdlOMrGO8Vl5gZ56Y0y8IZF0BuqkQWM4B6HXrRCKa1SEqyzEo7KK59RT/VHDjX3ZvSefeW3CO6O6vsiA1NrwVkxxAcYTCcHyTjZmJd00pugBQoTnzjvn+kzLBh9GtRDjhleZFwbx3kugP3GvFzdkqRlbDYw0u/HxKjuOw2QxZCGL5V5f4l7cd6qsffUa1DcLM9N1XcTMvep5ul1e4jNPtZfWGIkE6dI8MquXg/dS2CGVJQ2ushd5GmlxFdOw+1tRa32MY4zDQ9yaZ60J3/iX+QG4U3qGrFHmswAAAABJRU5ErkJggg==" |
||||
|
binary_image = base64.b64decode(image_base64) |
||||
|
|
||||
|
|
||||
|
def read_image() -> BytesIO: |
||||
|
return BytesIO(binary_image) |
||||
|
|
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
class PNGStreamingResponse(StreamingResponse): |
||||
|
media_type = "image/png" |
||||
|
|
||||
|
|
||||
|
@app.get("/image/stream", response_class=PNGStreamingResponse) |
||||
|
async def stream_image() -> AsyncIterable[bytes]: |
||||
|
for chunk in read_image(): |
||||
|
yield chunk |
||||
|
|
||||
|
|
||||
|
@app.get("/image/stream-no-async", response_class=PNGStreamingResponse) |
||||
|
def stream_image_no_async() -> Iterable[bytes]: |
||||
|
for chunk in read_image(): |
||||
|
yield chunk |
||||
|
|
||||
|
|
||||
|
@app.get("/image/stream-no-annotation", response_class=PNGStreamingResponse) |
||||
|
async def stream_image_no_annotation(): |
||||
|
for chunk in read_image(): |
||||
|
yield chunk |
||||
|
|
||||
|
|
||||
|
@app.get("/image/stream-no-async-no-annotation", response_class=PNGStreamingResponse) |
||||
|
def stream_image_no_async_no_annotation(): |
||||
|
for chunk in read_image(): |
||||
|
yield chunk |
||||
@ -0,0 +1,42 @@ |
|||||
|
from collections.abc import AsyncIterable, Iterable |
||||
|
|
||||
|
from fastapi import FastAPI |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
class Item(BaseModel): |
||||
|
name: str |
||||
|
description: str | None |
||||
|
|
||||
|
|
||||
|
items = [ |
||||
|
Item(name="Plumbus", description="A multi-purpose household device."), |
||||
|
Item(name="Portal Gun", description="A portal opening device."), |
||||
|
Item(name="Meeseeks Box", description="A box that summons a Meeseeks."), |
||||
|
] |
||||
|
|
||||
|
|
||||
|
@app.get("/items/stream") |
||||
|
async def stream_items() -> AsyncIterable[Item]: |
||||
|
for item in items: |
||||
|
yield item |
||||
|
|
||||
|
|
||||
|
@app.get("/items/stream-no-async") |
||||
|
def stream_items_no_async() -> Iterable[Item]: |
||||
|
for item in items: |
||||
|
yield item |
||||
|
|
||||
|
|
||||
|
@app.get("/items/stream-no-annotation") |
||||
|
async def stream_items_no_annotation(): |
||||
|
for item in items: |
||||
|
yield item |
||||
|
|
||||
|
|
||||
|
@app.get("/items/stream-no-async-no-annotation") |
||||
|
def stream_items_no_async_no_annotation(): |
||||
|
for item in items: |
||||
|
yield item |
||||
@ -0,0 +1,14 @@ |
|||||
|
from fastapi import FastAPI |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
app = FastAPI(strict_content_type=False) |
||||
|
|
||||
|
|
||||
|
class Item(BaseModel): |
||||
|
name: str |
||||
|
price: float |
||||
|
|
||||
|
|
||||
|
@app.post("/items/") |
||||
|
async def create_item(item: Item): |
||||
|
return item |
||||
@ -0,0 +1,614 @@ |
|||||
|
--- |
||||
|
name: fastapi |
||||
|
description: FastAPI best practices and conventions. Use when working with FastAPI APIs and Pydantic models for them. Keeps FastAPI code clean and up to date with the latest features and patterns, updated with new versions. Write new code or refactor and update old code. |
||||
|
--- |
||||
|
|
||||
|
# FastAPI |
||||
|
|
||||
|
Official FastAPI skill to write code with best practices, keeping up to date with new versions and features. |
||||
|
|
||||
|
## Use the `fastapi` CLI |
||||
|
|
||||
|
Run the development server on localhost with reload: |
||||
|
|
||||
|
```bash |
||||
|
fastapi dev |
||||
|
``` |
||||
|
|
||||
|
|
||||
|
Run the production server: |
||||
|
|
||||
|
```bash |
||||
|
fastapi run |
||||
|
``` |
||||
|
|
||||
|
### Add an entrypoint in `pyproject.toml` |
||||
|
|
||||
|
FastAPI CLI will read the entrypoint in `pyproject.toml` to know where the FastAPI app is declared. |
||||
|
|
||||
|
```toml |
||||
|
[tool.fastapi] |
||||
|
entrypoint = "my_app.main:app" |
||||
|
``` |
||||
|
|
||||
|
### Use `fastapi` with a path |
||||
|
|
||||
|
When adding the entrypoint to `pyproject.toml` is not possible, or the user explicitly asks not to, or it's running an independent small app, you can pass the app file path to the `fastapi` command: |
||||
|
|
||||
|
```bash |
||||
|
fastapi dev my_app/main.py |
||||
|
``` |
||||
|
|
||||
|
Prefer to set the entrypoint in `pyproject.toml` when possible. |
||||
|
|
||||
|
## Use `Annotated` |
||||
|
|
||||
|
Always prefer the `Annotated` style for parameter and dependency declarations. |
||||
|
|
||||
|
It keeps the function signatures working in other contexts, respects the types, allows reusability. |
||||
|
|
||||
|
### In Parameter Declarations |
||||
|
|
||||
|
Use `Annotated` for parameter declarations, including `Path`, `Query`, `Header`, etc.: |
||||
|
|
||||
|
```python |
||||
|
from typing import Annotated |
||||
|
|
||||
|
from fastapi import FastAPI, Path, Query |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.get("/items/{item_id}") |
||||
|
async def read_item( |
||||
|
item_id: Annotated[int, Path(ge=1, description="The item ID")], |
||||
|
q: Annotated[str | None, Query(max_length=50)] = None, |
||||
|
): |
||||
|
return {"message": "Hello World"} |
||||
|
``` |
||||
|
|
||||
|
instead of: |
||||
|
|
||||
|
```python |
||||
|
# DO NOT DO THIS |
||||
|
@app.get("/items/{item_id}") |
||||
|
async def read_item( |
||||
|
item_id: int = Path(ge=1, description="The item ID"), |
||||
|
q: str | None = Query(default=None, max_length=50), |
||||
|
): |
||||
|
return {"message": "Hello World"} |
||||
|
``` |
||||
|
|
||||
|
### For Dependencies |
||||
|
|
||||
|
Use `Annotated` for dependencies with `Depends()`. |
||||
|
|
||||
|
Unless asked not to, create a new type alias for the dependency to allow re-using it. |
||||
|
|
||||
|
```python |
||||
|
from typing import Annotated |
||||
|
|
||||
|
from fastapi import Depends, FastAPI |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
def get_current_user(): |
||||
|
return {"username": "johndoe"} |
||||
|
|
||||
|
|
||||
|
CurrentUserDep = Annotated[dict, Depends(get_current_user)] |
||||
|
|
||||
|
|
||||
|
@app.get("/items/") |
||||
|
async def read_item(current_user: CurrentUserDep): |
||||
|
return {"message": "Hello World"} |
||||
|
``` |
||||
|
|
||||
|
instead of: |
||||
|
|
||||
|
```python |
||||
|
# DO NOT DO THIS |
||||
|
@app.get("/items/") |
||||
|
async def read_item(current_user: dict = Depends(get_current_user)): |
||||
|
return {"message": "Hello World"} |
||||
|
``` |
||||
|
|
||||
|
## Do not use Ellipsis for *path operations* or Pydantic models |
||||
|
|
||||
|
Do not use `...` as a default value for required parameters, it's not needed and not recommended. |
||||
|
|
||||
|
Do this, without Ellipsis (`...`): |
||||
|
|
||||
|
```python |
||||
|
from typing import Annotated |
||||
|
|
||||
|
from fastapi import FastAPI, Query |
||||
|
from pydantic import BaseModel, Field |
||||
|
|
||||
|
|
||||
|
class Item(BaseModel): |
||||
|
name: str |
||||
|
description: str | None = None |
||||
|
price: float = Field(gt=0) |
||||
|
|
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.post("/items/") |
||||
|
async def create_item(item: Item, project_id: Annotated[int, Query()]): ... |
||||
|
``` |
||||
|
|
||||
|
instead of this: |
||||
|
|
||||
|
```python |
||||
|
# DO NOT DO THIS |
||||
|
class Item(BaseModel): |
||||
|
name: str = ... |
||||
|
description: str | None = None |
||||
|
price: float = Field(..., gt=0) |
||||
|
|
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.post("/items/") |
||||
|
async def create_item(item: Item, project_id: Annotated[int, Query(...)]): ... |
||||
|
``` |
||||
|
|
||||
|
## Return Type or Response Model |
||||
|
|
||||
|
When possible, include a return type. It will be used to validate, filter, document, and serialize the response. |
||||
|
|
||||
|
```python |
||||
|
from fastapi import FastAPI |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
class Item(BaseModel): |
||||
|
name: str |
||||
|
description: str | None = None |
||||
|
|
||||
|
|
||||
|
@app.get("/items/me") |
||||
|
async def get_item() -> Item: |
||||
|
return Item(name="Plumbus", description="All-purpose home device") |
||||
|
``` |
||||
|
|
||||
|
**Important**: Return types or response models are what filter data ensuring no sensitive information is exposed. And they are used to serialize data with Pydantic (in Rust), this is the main idea that can increase response performance. |
||||
|
|
||||
|
The return type doesn't have to be a Pydantic model, it could be a different type, like a list of integers, or a dict, etc. |
||||
|
|
||||
|
### When to use `response_model` instead |
||||
|
|
||||
|
If the return type is not the same as the type that you want to use to validate, filter, or serialize, use the `response_model` parameter on the decorator instead. |
||||
|
|
||||
|
```python |
||||
|
from typing import Any |
||||
|
|
||||
|
from fastapi import FastAPI |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
class Item(BaseModel): |
||||
|
name: str |
||||
|
description: str | None = None |
||||
|
|
||||
|
|
||||
|
@app.get("/items/me", response_model=Item) |
||||
|
async def get_item() -> Any: |
||||
|
return {"name": "Foo", "description": "A very nice Item"} |
||||
|
``` |
||||
|
|
||||
|
This can be particularly useful when filtering data to expose only the public fields and avoid exposing sensitive information. |
||||
|
|
||||
|
```python |
||||
|
from typing import Any |
||||
|
|
||||
|
from fastapi import FastAPI |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
class InternalItem(BaseModel): |
||||
|
name: str |
||||
|
description: str | None = None |
||||
|
secret_key: str |
||||
|
|
||||
|
|
||||
|
class Item(BaseModel): |
||||
|
name: str |
||||
|
description: str | None = None |
||||
|
|
||||
|
|
||||
|
@app.get("/items/me", response_model=Item) |
||||
|
async def get_item() -> Any: |
||||
|
item = InternalItem( |
||||
|
name="Foo", description="A very nice Item", secret_key="supersecret" |
||||
|
) |
||||
|
return item |
||||
|
``` |
||||
|
|
||||
|
## Performance |
||||
|
|
||||
|
Do not use `ORJSONResponse` or `UJSONResponse`, they are deprecated. |
||||
|
|
||||
|
Instead, declare a return type or response model. Pydantic will handle the data serialization on the Rust side. |
||||
|
|
||||
|
## Including Routers |
||||
|
|
||||
|
When declaring routers, prefer to add router level parameters like prefix, tags, etc. to the router itself, instead of in `include_router()`. |
||||
|
|
||||
|
Do this: |
||||
|
|
||||
|
```python |
||||
|
from fastapi import APIRouter, FastAPI |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
router = APIRouter(prefix="/items", tags=["items"]) |
||||
|
|
||||
|
|
||||
|
@router.get("/") |
||||
|
async def list_items(): |
||||
|
return [] |
||||
|
|
||||
|
|
||||
|
# In main.py |
||||
|
app.include_router(router) |
||||
|
``` |
||||
|
|
||||
|
instead of this: |
||||
|
|
||||
|
```python |
||||
|
# DO NOT DO THIS |
||||
|
from fastapi import APIRouter, FastAPI |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
router = APIRouter() |
||||
|
|
||||
|
|
||||
|
@router.get("/") |
||||
|
async def list_items(): |
||||
|
return [] |
||||
|
|
||||
|
|
||||
|
# In main.py |
||||
|
app.include_router(router, prefix="/items", tags=["items"]) |
||||
|
``` |
||||
|
|
||||
|
There could be exceptions, but try to follow this convention. |
||||
|
|
||||
|
Apply shared dependencies at the router level via `dependencies=[Depends(...)]`. |
||||
|
|
||||
|
## Dependency Injection |
||||
|
|
||||
|
Use dependencies when: |
||||
|
|
||||
|
* They can't be declared in Pydantic validation and require additional logic |
||||
|
* The logic depends on external resources or could block in any other way |
||||
|
* Other dependencies need their results (it's a sub-dependency) |
||||
|
* The logic can be shared by multiple endpoints to do things like error early, authentication, etc. |
||||
|
* They need to handle cleanup (e.g., DB sessions, file handles), using dependencies with `yield` |
||||
|
* Their logic needs input data from the request, like headers, query parameters, etc. |
||||
|
|
||||
|
### Dependencies with `yield` and `scope` |
||||
|
|
||||
|
When using dependencies with `yield`, they can have a `scope` that defines when the exit code is run. |
||||
|
|
||||
|
Use the default scope `"request"` to run the exit code after the response is sent back. |
||||
|
|
||||
|
```python |
||||
|
from typing import Annotated |
||||
|
|
||||
|
from fastapi import Depends, FastAPI |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
def get_db(): |
||||
|
db = DBSession() |
||||
|
try: |
||||
|
yield db |
||||
|
finally: |
||||
|
db.close() |
||||
|
|
||||
|
|
||||
|
DBDep = Annotated[DBSession, Depends(get_db)] |
||||
|
|
||||
|
|
||||
|
@app.get("/items/") |
||||
|
async def read_items(db: DBDep): |
||||
|
return db.query(Item).all() |
||||
|
``` |
||||
|
|
||||
|
Use the scope `"function"` when they should run the exit code after the response data is generated but before the response is sent back to the client. |
||||
|
|
||||
|
```python |
||||
|
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") |
||||
|
|
||||
|
UserNameDep = Annotated[str, Depends(get_username, scope="function")] |
||||
|
|
||||
|
@app.get("/users/me") |
||||
|
def get_user_me(username: UserNameDep): |
||||
|
return username |
||||
|
``` |
||||
|
|
||||
|
### Class Dependencies |
||||
|
|
||||
|
Avoid creating class dependencies when possible. |
||||
|
|
||||
|
If a class is needed, instead create a regular function dependency that returns a class instance. |
||||
|
|
||||
|
Do this: |
||||
|
|
||||
|
```python |
||||
|
from dataclasses import dataclass |
||||
|
from typing import Annotated |
||||
|
|
||||
|
from fastapi import Depends, FastAPI |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@dataclass |
||||
|
class DatabasePaginator: |
||||
|
offset: int = 0 |
||||
|
limit: int = 100 |
||||
|
q: str | None = None |
||||
|
|
||||
|
def get_page(self) -> dict: |
||||
|
# Simulate a page of data |
||||
|
return { |
||||
|
"offset": self.offset, |
||||
|
"limit": self.limit, |
||||
|
"q": self.q, |
||||
|
"items": [], |
||||
|
} |
||||
|
|
||||
|
|
||||
|
def get_db_paginator( |
||||
|
offset: int = 0, limit: int = 100, q: str | None = None |
||||
|
) -> DatabasePaginator: |
||||
|
return DatabasePaginator(offset=offset, limit=limit, q=q) |
||||
|
|
||||
|
|
||||
|
PaginatorDep = Annotated[DatabasePaginator, Depends(get_db_paginator)] |
||||
|
|
||||
|
|
||||
|
@app.get("/items/") |
||||
|
async def read_items(paginator: PaginatorDep): |
||||
|
return paginator.get_page() |
||||
|
``` |
||||
|
|
||||
|
instead of this: |
||||
|
|
||||
|
```python |
||||
|
# DO NOT DO THIS |
||||
|
from typing import Annotated |
||||
|
|
||||
|
from fastapi import Depends, FastAPI |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
class DatabasePaginator: |
||||
|
def __init__(self, offset: int = 0, limit: int = 100, q: str | None = None): |
||||
|
self.offset = offset |
||||
|
self.limit = limit |
||||
|
self.q = q |
||||
|
|
||||
|
def get_page(self) -> dict: |
||||
|
# Simulate a page of data |
||||
|
return { |
||||
|
"offset": self.offset, |
||||
|
"limit": self.limit, |
||||
|
"q": self.q, |
||||
|
"items": [], |
||||
|
} |
||||
|
|
||||
|
|
||||
|
@app.get("/items/") |
||||
|
async def read_items(paginator: Annotated[DatabasePaginator, Depends()]): |
||||
|
return paginator.get_page() |
||||
|
``` |
||||
|
|
||||
|
## Async vs Sync *path operations* |
||||
|
|
||||
|
Use `async` *path operations* only when fully certain that the logic called inside is compatible with async and await (it's called with `await`) or that doesn't block. |
||||
|
|
||||
|
```python |
||||
|
from fastapi import FastAPI |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
# Use async def when calling async code |
||||
|
@app.get("/async-items/") |
||||
|
async def read_async_items(): |
||||
|
data = await some_async_library.fetch_items() |
||||
|
return data |
||||
|
|
||||
|
|
||||
|
# Use plain def when calling blocking/sync code or when in doubt |
||||
|
@app.get("/items/") |
||||
|
def read_items(): |
||||
|
data = some_blocking_library.fetch_items() |
||||
|
return data |
||||
|
``` |
||||
|
|
||||
|
In case of doubt, or by default, use regular `def` functions, those will be run in a threadpool so they don't block the event loop. |
||||
|
|
||||
|
The same rules apply to dependencies. |
||||
|
|
||||
|
Make sure blocking code is not run inside of `async` functions. The logic will work, but will damage the performance heavily. |
||||
|
|
||||
|
### Asyncer |
||||
|
|
||||
|
When needing to run blocking code inside of async functions, or async code inside of blocking functions, suggest using Asyncer. |
||||
|
|
||||
|
Install: |
||||
|
|
||||
|
```bash |
||||
|
uv add asyncer |
||||
|
``` |
||||
|
|
||||
|
Run blocking sync code inside of async with `asyncify()`: |
||||
|
|
||||
|
```python |
||||
|
from asyncer import asyncify |
||||
|
from fastapi import FastAPI |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
def do_blocking_work(name: str) -> str: |
||||
|
# Some blocking I/O operation |
||||
|
return f"Hello {name}" |
||||
|
|
||||
|
|
||||
|
@app.get("/items/") |
||||
|
async def read_items(): |
||||
|
result = await asyncify(do_blocking_work)(name="World") |
||||
|
return {"message": result} |
||||
|
``` |
||||
|
|
||||
|
And run async code inside of blocking sync code with `syncify()`: |
||||
|
|
||||
|
```python |
||||
|
from asyncer import syncify |
||||
|
from fastapi import FastAPI |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
async def do_async_work(name: str) -> str: |
||||
|
return f"Hello {name}" |
||||
|
|
||||
|
|
||||
|
@app.get("/items/") |
||||
|
def read_items(): |
||||
|
result = syncify(do_async_work)(name="World") |
||||
|
return {"message": result} |
||||
|
``` |
||||
|
|
||||
|
## Use uv, ruff, ty |
||||
|
|
||||
|
If uv is available, use it to manage dependencies. |
||||
|
|
||||
|
If Ruff is available, use it to lint and format the code. Consider enabling the FastAPI rules. |
||||
|
|
||||
|
If ty is available, use it to check types. |
||||
|
|
||||
|
## SQLModel for SQL databases |
||||
|
|
||||
|
When working with SQL databases, prefer using SQLModel as it is integrated with Pydantic and will allow declaring data validation with the same models. |
||||
|
|
||||
|
## Do not use Pydantic RootModels |
||||
|
|
||||
|
Do not use Pydantic `RootModel`, instead use regular type annotations with `Annotated` and Pydantic validation utilities. |
||||
|
|
||||
|
For example, for a list with validations you could do: |
||||
|
|
||||
|
```python |
||||
|
from typing import Annotated |
||||
|
|
||||
|
from fastapi import Body, FastAPI |
||||
|
from pydantic import Field |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.post("/items/") |
||||
|
async def create_items(items: Annotated[list[int], Field(min_length=1), Body()]): |
||||
|
return items |
||||
|
``` |
||||
|
|
||||
|
instead of: |
||||
|
|
||||
|
```python |
||||
|
# DO NOT DO THIS |
||||
|
from typing import Annotated |
||||
|
|
||||
|
from fastapi import FastAPI |
||||
|
from pydantic import Field, RootModel |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
class ItemList(RootModel[Annotated[list[int], Field(min_length=1)]]): |
||||
|
pass |
||||
|
|
||||
|
|
||||
|
@app.post("/items/") |
||||
|
async def create_items(items: ItemList): |
||||
|
return items |
||||
|
|
||||
|
``` |
||||
|
|
||||
|
FastAPI supports these type annotations and will create a Pydantic `TypeAdapter` for them, so that types can work as normally and there's no need for the custom logic and types in RootModels. |
||||
|
|
||||
|
## Use one HTTP operation per function |
||||
|
|
||||
|
Don't mix HTTP operations in a single function, having one function per HTTP operation helps separate concerns and organize the code. |
||||
|
|
||||
|
Do this: |
||||
|
|
||||
|
```python |
||||
|
from fastapi import FastAPI |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
class Item(BaseModel): |
||||
|
name: str |
||||
|
|
||||
|
|
||||
|
@app.get("/items/") |
||||
|
async def list_items(): |
||||
|
return [] |
||||
|
|
||||
|
|
||||
|
@app.post("/items/") |
||||
|
async def create_item(item: Item): |
||||
|
return item |
||||
|
``` |
||||
|
|
||||
|
instead of this: |
||||
|
|
||||
|
```python |
||||
|
# DO NOT DO THIS |
||||
|
from fastapi import FastAPI, Request |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
class Item(BaseModel): |
||||
|
name: str |
||||
|
|
||||
|
|
||||
|
@app.api_route("/items/", methods=["GET", "POST"]) |
||||
|
async def handle_items(request: Request): |
||||
|
if request.method == "GET": |
||||
|
return [] |
||||
|
``` |
||||
@ -1,40 +0,0 @@ |
|||||
import os |
|
||||
from typing import Any |
|
||||
|
|
||||
from pdm.backend.hooks import Context |
|
||||
|
|
||||
TIANGOLO_BUILD_PACKAGE = os.getenv("TIANGOLO_BUILD_PACKAGE") |
|
||||
|
|
||||
|
|
||||
def pdm_build_initialize(context: Context) -> None: |
|
||||
metadata = context.config.metadata |
|
||||
# Get main version |
|
||||
version = metadata["version"] |
|
||||
# Get custom config for the current package, from the env var |
|
||||
all_configs_config: dict[str, Any] = context.config.data["tool"]["tiangolo"][ |
|
||||
"_internal-slim-build" |
|
||||
]["packages"] |
|
||||
|
|
||||
if TIANGOLO_BUILD_PACKAGE not in all_configs_config: |
|
||||
return |
|
||||
|
|
||||
config = all_configs_config[TIANGOLO_BUILD_PACKAGE] |
|
||||
project_config: dict[str, Any] = config["project"] |
|
||||
# Override main [project] configs with custom configs for this package |
|
||||
for key, value in project_config.items(): |
|
||||
metadata[key] = value |
|
||||
# Get custom build config for the current package |
|
||||
build_config: dict[str, Any] = ( |
|
||||
config.get("tool", {}).get("pdm", {}).get("build", {}) |
|
||||
) |
|
||||
# Override PDM build config with custom build config for this package |
|
||||
for key, value in build_config.items(): |
|
||||
context.config.build_config[key] = value |
|
||||
# Get main dependencies |
|
||||
dependencies: list[str] = metadata.get("dependencies", []) |
|
||||
# Sync versions in dependencies |
|
||||
new_dependencies = [] |
|
||||
for dep in dependencies: |
|
||||
new_dep = f"{dep}>={version}" |
|
||||
new_dependencies.append(new_dep) |
|
||||
metadata["dependencies"] = new_dependencies |
|
||||
@ -0,0 +1,37 @@ |
|||||
|
import subprocess |
||||
|
import time |
||||
|
|
||||
|
import httpx |
||||
|
from playwright.sync_api import Playwright, sync_playwright |
||||
|
|
||||
|
|
||||
|
# Run playwright codegen to generate the code below, copy paste the sections in run() |
||||
|
def run(playwright: Playwright) -> None: |
||||
|
browser = playwright.chromium.launch(headless=False) |
||||
|
# Update the viewport manually |
||||
|
context = browser.new_context(viewport={"width": 960, "height": 1080}) |
||||
|
page = context.new_page() |
||||
|
page.goto("http://localhost:8000/docs") |
||||
|
page.get_by_role("button", name="POST /data Post Data").click() |
||||
|
# Manually add the screenshot |
||||
|
page.screenshot(path="docs/en/docs/img/tutorial/json-base64-bytes/image01.png") |
||||
|
|
||||
|
# --------------------- |
||||
|
context.close() |
||||
|
browser.close() |
||||
|
|
||||
|
|
||||
|
process = subprocess.Popen( |
||||
|
["fastapi", "run", "docs_src/json_base64_bytes/tutorial001_py310.py"] |
||||
|
) |
||||
|
try: |
||||
|
for _ in range(3): |
||||
|
try: |
||||
|
response = httpx.get("http://localhost:8000/docs") |
||||
|
except httpx.ConnectError: |
||||
|
time.sleep(1) |
||||
|
break |
||||
|
with sync_playwright() as playwright: |
||||
|
run(playwright) |
||||
|
finally: |
||||
|
process.terminate() |
||||
@ -0,0 +1,6 @@ |
|||||
|
#!/usr/bin/env bash |
||||
|
|
||||
|
set -e |
||||
|
set -x |
||||
|
|
||||
|
bash scripts/test.sh --cov --cov-context=test ${@} |
||||
@ -0,0 +1,73 @@ |
|||||
|
import warnings |
||||
|
|
||||
|
import pytest |
||||
|
from fastapi import FastAPI |
||||
|
from fastapi.exceptions import FastAPIDeprecationWarning |
||||
|
from fastapi.responses import ORJSONResponse, UJSONResponse |
||||
|
from fastapi.testclient import TestClient |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
|
||||
|
class Item(BaseModel): |
||||
|
name: str |
||||
|
price: float |
||||
|
|
||||
|
|
||||
|
# ORJSON |
||||
|
|
||||
|
|
||||
|
def _make_orjson_app() -> FastAPI: |
||||
|
with warnings.catch_warnings(): |
||||
|
warnings.simplefilter("ignore", FastAPIDeprecationWarning) |
||||
|
app = FastAPI(default_response_class=ORJSONResponse) |
||||
|
|
||||
|
@app.get("/items") |
||||
|
def get_items() -> Item: |
||||
|
return Item(name="widget", price=9.99) |
||||
|
|
||||
|
return app |
||||
|
|
||||
|
|
||||
|
def test_orjson_response_returns_correct_data(): |
||||
|
app = _make_orjson_app() |
||||
|
client = TestClient(app) |
||||
|
with warnings.catch_warnings(): |
||||
|
warnings.simplefilter("ignore", FastAPIDeprecationWarning) |
||||
|
response = client.get("/items") |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"name": "widget", "price": 9.99} |
||||
|
|
||||
|
|
||||
|
def test_orjson_response_emits_deprecation_warning(): |
||||
|
with pytest.warns(FastAPIDeprecationWarning, match="ORJSONResponse is deprecated"): |
||||
|
ORJSONResponse(content={"hello": "world"}) |
||||
|
|
||||
|
|
||||
|
# UJSON |
||||
|
|
||||
|
|
||||
|
def _make_ujson_app() -> FastAPI: |
||||
|
with warnings.catch_warnings(): |
||||
|
warnings.simplefilter("ignore", FastAPIDeprecationWarning) |
||||
|
app = FastAPI(default_response_class=UJSONResponse) |
||||
|
|
||||
|
@app.get("/items") |
||||
|
def get_items() -> Item: |
||||
|
return Item(name="widget", price=9.99) |
||||
|
|
||||
|
return app |
||||
|
|
||||
|
|
||||
|
def test_ujson_response_returns_correct_data(): |
||||
|
app = _make_ujson_app() |
||||
|
client = TestClient(app) |
||||
|
with warnings.catch_warnings(): |
||||
|
warnings.simplefilter("ignore", FastAPIDeprecationWarning) |
||||
|
response = client.get("/items") |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"name": "widget", "price": 9.99} |
||||
|
|
||||
|
|
||||
|
def test_ujson_response_emits_deprecation_warning(): |
||||
|
with pytest.warns(FastAPIDeprecationWarning, match="UJSONResponse is deprecated"): |
||||
|
UJSONResponse(content={"hello": "world"}) |
||||
@ -0,0 +1,51 @@ |
|||||
|
from unittest.mock import patch |
||||
|
|
||||
|
from fastapi import FastAPI |
||||
|
from fastapi.responses import JSONResponse |
||||
|
from fastapi.testclient import TestClient |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
|
||||
|
class Item(BaseModel): |
||||
|
name: str |
||||
|
price: float |
||||
|
|
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.get("/default") |
||||
|
def get_default() -> Item: |
||||
|
return Item(name="widget", price=9.99) |
||||
|
|
||||
|
|
||||
|
@app.get("/explicit", response_class=JSONResponse) |
||||
|
def get_explicit() -> Item: |
||||
|
return Item(name="widget", price=9.99) |
||||
|
|
||||
|
|
||||
|
client = TestClient(app) |
||||
|
|
||||
|
|
||||
|
def test_default_response_class_skips_json_dumps(): |
||||
|
"""When no response_class is set, the fast path serializes directly to |
||||
|
JSON bytes via Pydantic's dump_json and never calls json.dumps.""" |
||||
|
with patch( |
||||
|
"starlette.responses.json.dumps", wraps=__import__("json").dumps |
||||
|
) as mock_dumps: |
||||
|
response = client.get("/default") |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"name": "widget", "price": 9.99} |
||||
|
mock_dumps.assert_not_called() |
||||
|
|
||||
|
|
||||
|
def test_explicit_response_class_uses_json_dumps(): |
||||
|
"""When response_class is explicitly set to JSONResponse, the normal path |
||||
|
is used and json.dumps is called via JSONResponse.render().""" |
||||
|
with patch( |
||||
|
"starlette.responses.json.dumps", wraps=__import__("json").dumps |
||||
|
) as mock_dumps: |
||||
|
response = client.get("/explicit") |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"name": "widget", "price": 9.99} |
||||
|
mock_dumps.assert_called_once() |
||||
@ -0,0 +1,75 @@ |
|||||
|
from fastapi import FastAPI |
||||
|
from fastapi.testclient import TestClient |
||||
|
|
||||
|
|
||||
|
def test_root_path_does_not_persist_across_requests(): |
||||
|
app = FastAPI() |
||||
|
|
||||
|
@app.get("/") |
||||
|
def read_root(): # pragma: no cover |
||||
|
return {"ok": True} |
||||
|
|
||||
|
# Attacker request with a spoofed root_path |
||||
|
attacker_client = TestClient(app, root_path="/evil-api") |
||||
|
response1 = attacker_client.get("/openapi.json") |
||||
|
data1 = response1.json() |
||||
|
assert any(s.get("url") == "/evil-api" for s in data1.get("servers", [])) |
||||
|
|
||||
|
# Subsequent legitimate request with no root_path |
||||
|
clean_client = TestClient(app) |
||||
|
response2 = clean_client.get("/openapi.json") |
||||
|
data2 = response2.json() |
||||
|
servers = [s.get("url") for s in data2.get("servers", [])] |
||||
|
assert "/evil-api" not in servers |
||||
|
|
||||
|
|
||||
|
def test_multiple_different_root_paths_do_not_accumulate(): |
||||
|
app = FastAPI() |
||||
|
|
||||
|
@app.get("/") |
||||
|
def read_root(): # pragma: no cover |
||||
|
return {"ok": True} |
||||
|
|
||||
|
for prefix in ["/path-a", "/path-b", "/path-c"]: |
||||
|
c = TestClient(app, root_path=prefix) |
||||
|
c.get("/openapi.json") |
||||
|
|
||||
|
# A clean request should not have any of them |
||||
|
clean_client = TestClient(app) |
||||
|
response = clean_client.get("/openapi.json") |
||||
|
data = response.json() |
||||
|
servers = [s.get("url") for s in data.get("servers", [])] |
||||
|
for prefix in ["/path-a", "/path-b", "/path-c"]: |
||||
|
assert prefix not in servers, ( |
||||
|
f"root_path '{prefix}' leaked into clean request: {servers}" |
||||
|
) |
||||
|
|
||||
|
|
||||
|
def test_legitimate_root_path_still_appears(): |
||||
|
app = FastAPI() |
||||
|
|
||||
|
@app.get("/") |
||||
|
def read_root(): # pragma: no cover |
||||
|
return {"ok": True} |
||||
|
|
||||
|
client = TestClient(app, root_path="/api/v1") |
||||
|
response = client.get("/openapi.json") |
||||
|
data = response.json() |
||||
|
servers = [s.get("url") for s in data.get("servers", [])] |
||||
|
assert "/api/v1" in servers |
||||
|
|
||||
|
|
||||
|
def test_configured_servers_not_mutated(): |
||||
|
configured_servers = [{"url": "https://prod.example.com"}] |
||||
|
app = FastAPI(servers=configured_servers) |
||||
|
|
||||
|
@app.get("/") |
||||
|
def read_root(): # pragma: no cover |
||||
|
return {"ok": True} |
||||
|
|
||||
|
# Request with a rogue root_path |
||||
|
attacker_client = TestClient(app, root_path="/evil") |
||||
|
attacker_client.get("/openapi.json") |
||||
|
|
||||
|
# The original servers list must be untouched |
||||
|
assert configured_servers == [{"url": "https://prod.example.com"}] |
||||
@ -0,0 +1,42 @@ |
|||||
|
import json |
||||
|
from typing import AsyncIterable, Iterable # noqa: UP035 to test coverage |
||||
|
|
||||
|
from fastapi import FastAPI |
||||
|
from fastapi.testclient import TestClient |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
|
||||
|
class Item(BaseModel): |
||||
|
name: str |
||||
|
|
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.get("/items/stream-bare-async") |
||||
|
async def stream_bare_async() -> AsyncIterable: |
||||
|
yield {"name": "foo"} |
||||
|
|
||||
|
|
||||
|
@app.get("/items/stream-bare-sync") |
||||
|
def stream_bare_sync() -> Iterable: |
||||
|
yield {"name": "bar"} |
||||
|
|
||||
|
|
||||
|
client = TestClient(app) |
||||
|
|
||||
|
|
||||
|
def test_stream_bare_async_iterable(): |
||||
|
response = client.get("/items/stream-bare-async") |
||||
|
assert response.status_code == 200 |
||||
|
assert response.headers["content-type"] == "application/jsonl" |
||||
|
lines = [json.loads(line) for line in response.text.strip().splitlines()] |
||||
|
assert lines == [{"name": "foo"}] |
||||
|
|
||||
|
|
||||
|
def test_stream_bare_sync_iterable(): |
||||
|
response = client.get("/items/stream-bare-sync") |
||||
|
assert response.status_code == 200 |
||||
|
assert response.headers["content-type"] == "application/jsonl" |
||||
|
lines = [json.loads(line) for line in response.text.strip().splitlines()] |
||||
|
assert lines == [{"name": "bar"}] |
||||
@ -0,0 +1,88 @@ |
|||||
|
""" |
||||
|
Test that async streaming endpoints can be cancelled without hanging. |
||||
|
|
||||
|
Ref: https://github.com/fastapi/fastapi/issues/14680 |
||||
|
""" |
||||
|
|
||||
|
from collections.abc import AsyncIterable |
||||
|
|
||||
|
import anyio |
||||
|
import pytest |
||||
|
from fastapi import FastAPI |
||||
|
from fastapi.responses import StreamingResponse |
||||
|
|
||||
|
pytestmark = [ |
||||
|
pytest.mark.anyio, |
||||
|
pytest.mark.filterwarnings("ignore::pytest.PytestUnraisableExceptionWarning"), |
||||
|
] |
||||
|
|
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.get("/stream-raw", response_class=StreamingResponse) |
||||
|
async def stream_raw() -> AsyncIterable[str]: |
||||
|
"""Async generator with no internal await - would hang without checkpoint.""" |
||||
|
i = 0 |
||||
|
while True: |
||||
|
yield f"item {i}\n" |
||||
|
i += 1 |
||||
|
|
||||
|
|
||||
|
@app.get("/stream-jsonl") |
||||
|
async def stream_jsonl() -> AsyncIterable[int]: |
||||
|
"""JSONL async generator with no internal await.""" |
||||
|
i = 0 |
||||
|
while True: |
||||
|
yield i |
||||
|
i += 1 |
||||
|
|
||||
|
|
||||
|
async def _run_asgi_and_cancel(app: FastAPI, path: str, timeout: float) -> bool: |
||||
|
"""Call the ASGI app for *path* and cancel after *timeout* seconds. |
||||
|
|
||||
|
Returns `True` if the cancellation was delivered (i.e. it did not hang). |
||||
|
""" |
||||
|
chunks: list[bytes] = [] |
||||
|
|
||||
|
async def receive(): # type: ignore[no-untyped-def] |
||||
|
# Simulate a client that never disconnects, rely on cancellation |
||||
|
await anyio.sleep(float("inf")) |
||||
|
return {"type": "http.disconnect"} # pragma: no cover |
||||
|
|
||||
|
async def send(message: dict) -> None: # type: ignore[type-arg] |
||||
|
if message["type"] == "http.response.body": |
||||
|
chunks.append(message.get("body", b"")) |
||||
|
|
||||
|
scope = { |
||||
|
"type": "http", |
||||
|
"asgi": {"version": "3.0", "spec_version": "2.0"}, |
||||
|
"http_version": "1.1", |
||||
|
"method": "GET", |
||||
|
"path": path, |
||||
|
"query_string": b"", |
||||
|
"root_path": "", |
||||
|
"headers": [], |
||||
|
"server": ("test", 80), |
||||
|
} |
||||
|
|
||||
|
with anyio.move_on_after(timeout) as cancel_scope: |
||||
|
await app(scope, receive, send) # type: ignore[arg-type] |
||||
|
|
||||
|
# If we got here within the timeout the generator was cancellable. |
||||
|
# cancel_scope.cancelled_caught is True when move_on_after fired. |
||||
|
return cancel_scope.cancelled_caught or len(chunks) > 0 |
||||
|
|
||||
|
|
||||
|
async def test_raw_stream_cancellation() -> None: |
||||
|
"""Raw streaming endpoint should be cancellable within a reasonable time.""" |
||||
|
cancelled = await _run_asgi_and_cancel(app, "/stream-raw", timeout=3.0) |
||||
|
# The key assertion: we reached this line at all (didn't hang). |
||||
|
# cancelled will be True because the infinite generator was interrupted. |
||||
|
assert cancelled |
||||
|
|
||||
|
|
||||
|
async def test_jsonl_stream_cancellation() -> None: |
||||
|
"""JSONL streaming endpoint should be cancellable within a reasonable time.""" |
||||
|
cancelled = await _run_asgi_and_cancel(app, "/stream-jsonl", timeout=3.0) |
||||
|
assert cancelled |
||||
@ -0,0 +1,40 @@ |
|||||
|
from collections.abc import AsyncIterable, Iterable |
||||
|
|
||||
|
import pytest |
||||
|
from fastapi import FastAPI |
||||
|
from fastapi.exceptions import ResponseValidationError |
||||
|
from fastapi.testclient import TestClient |
||||
|
from pydantic import BaseModel |
||||
|
|
||||
|
|
||||
|
class Item(BaseModel): |
||||
|
name: str |
||||
|
price: float |
||||
|
|
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app.get("/items/stream-invalid") |
||||
|
async def stream_items_invalid() -> AsyncIterable[Item]: |
||||
|
yield {"name": "valid", "price": 1.0} |
||||
|
yield {"name": "invalid", "price": "not-a-number"} |
||||
|
|
||||
|
|
||||
|
@app.get("/items/stream-invalid-sync") |
||||
|
def stream_items_invalid_sync() -> Iterable[Item]: |
||||
|
yield {"name": "valid", "price": 1.0} |
||||
|
yield {"name": "invalid", "price": "not-a-number"} |
||||
|
|
||||
|
|
||||
|
client = TestClient(app) |
||||
|
|
||||
|
|
||||
|
def test_stream_json_validation_error_async(): |
||||
|
with pytest.raises(ResponseValidationError): |
||||
|
client.get("/items/stream-invalid") |
||||
|
|
||||
|
|
||||
|
def test_stream_json_validation_error_sync(): |
||||
|
with pytest.raises(ResponseValidationError): |
||||
|
client.get("/items/stream-invalid-sync") |
||||
@ -0,0 +1,44 @@ |
|||||
|
from fastapi import FastAPI |
||||
|
from fastapi.testclient import TestClient |
||||
|
|
||||
|
app_default = FastAPI() |
||||
|
|
||||
|
|
||||
|
@app_default.post("/items/") |
||||
|
async def app_default_post(data: dict): |
||||
|
return data |
||||
|
|
||||
|
|
||||
|
app_lax = FastAPI(strict_content_type=False) |
||||
|
|
||||
|
|
||||
|
@app_lax.post("/items/") |
||||
|
async def app_lax_post(data: dict): |
||||
|
return data |
||||
|
|
||||
|
|
||||
|
client_default = TestClient(app_default) |
||||
|
client_lax = TestClient(app_lax) |
||||
|
|
||||
|
|
||||
|
def test_default_strict_rejects_no_content_type(): |
||||
|
response = client_default.post("/items/", content='{"key": "value"}') |
||||
|
assert response.status_code == 422 |
||||
|
|
||||
|
|
||||
|
def test_default_strict_accepts_json_content_type(): |
||||
|
response = client_default.post("/items/", json={"key": "value"}) |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"key": "value"} |
||||
|
|
||||
|
|
||||
|
def test_lax_accepts_no_content_type(): |
||||
|
response = client_lax.post("/items/", content='{"key": "value"}') |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"key": "value"} |
||||
|
|
||||
|
|
||||
|
def test_lax_accepts_json_content_type(): |
||||
|
response = client_lax.post("/items/", json={"key": "value"}) |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"key": "value"} |
||||
@ -0,0 +1,91 @@ |
|||||
|
from fastapi import APIRouter, FastAPI |
||||
|
from fastapi.testclient import TestClient |
||||
|
|
||||
|
# Lax app with nested routers, inner overrides to strict |
||||
|
|
||||
|
app_nested = FastAPI(strict_content_type=False) # lax app |
||||
|
outer_router = APIRouter(prefix="/outer") # inherits lax from app |
||||
|
inner_strict = APIRouter(prefix="/strict", strict_content_type=True) |
||||
|
inner_default = APIRouter(prefix="/default") |
||||
|
|
||||
|
|
||||
|
@inner_strict.post("/items/") |
||||
|
async def inner_strict_post(data: dict): |
||||
|
return data |
||||
|
|
||||
|
|
||||
|
@inner_default.post("/items/") |
||||
|
async def inner_default_post(data: dict): |
||||
|
return data |
||||
|
|
||||
|
|
||||
|
outer_router.include_router(inner_strict) |
||||
|
outer_router.include_router(inner_default) |
||||
|
app_nested.include_router(outer_router) |
||||
|
|
||||
|
client_nested = TestClient(app_nested) |
||||
|
|
||||
|
|
||||
|
def test_strict_inner_on_lax_app_rejects_no_content_type(): |
||||
|
response = client_nested.post("/outer/strict/items/", content='{"key": "value"}') |
||||
|
assert response.status_code == 422 |
||||
|
|
||||
|
|
||||
|
def test_default_inner_inherits_lax_from_app(): |
||||
|
response = client_nested.post("/outer/default/items/", content='{"key": "value"}') |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"key": "value"} |
||||
|
|
||||
|
|
||||
|
def test_strict_inner_accepts_json_content_type(): |
||||
|
response = client_nested.post("/outer/strict/items/", json={"key": "value"}) |
||||
|
assert response.status_code == 200 |
||||
|
|
||||
|
|
||||
|
def test_default_inner_accepts_json_content_type(): |
||||
|
response = client_nested.post("/outer/default/items/", json={"key": "value"}) |
||||
|
assert response.status_code == 200 |
||||
|
|
||||
|
|
||||
|
# Strict app -> lax outer router -> strict inner router |
||||
|
|
||||
|
app_mixed = FastAPI(strict_content_type=True) |
||||
|
mixed_outer = APIRouter(prefix="/outer", strict_content_type=False) |
||||
|
mixed_inner = APIRouter(prefix="/inner", strict_content_type=True) |
||||
|
|
||||
|
|
||||
|
@mixed_outer.post("/items/") |
||||
|
async def mixed_outer_post(data: dict): |
||||
|
return data |
||||
|
|
||||
|
|
||||
|
@mixed_inner.post("/items/") |
||||
|
async def mixed_inner_post(data: dict): |
||||
|
return data |
||||
|
|
||||
|
|
||||
|
mixed_outer.include_router(mixed_inner) |
||||
|
app_mixed.include_router(mixed_outer) |
||||
|
|
||||
|
client_mixed = TestClient(app_mixed) |
||||
|
|
||||
|
|
||||
|
def test_lax_outer_on_strict_app_accepts_no_content_type(): |
||||
|
response = client_mixed.post("/outer/items/", content='{"key": "value"}') |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"key": "value"} |
||||
|
|
||||
|
|
||||
|
def test_strict_inner_on_lax_outer_rejects_no_content_type(): |
||||
|
response = client_mixed.post("/outer/inner/items/", content='{"key": "value"}') |
||||
|
assert response.status_code == 422 |
||||
|
|
||||
|
|
||||
|
def test_lax_outer_accepts_json_content_type(): |
||||
|
response = client_mixed.post("/outer/items/", json={"key": "value"}) |
||||
|
assert response.status_code == 200 |
||||
|
|
||||
|
|
||||
|
def test_strict_inner_on_lax_outer_accepts_json_content_type(): |
||||
|
response = client_mixed.post("/outer/inner/items/", json={"key": "value"}) |
||||
|
assert response.status_code == 200 |
||||
@ -0,0 +1,61 @@ |
|||||
|
from fastapi import APIRouter, FastAPI |
||||
|
from fastapi.testclient import TestClient |
||||
|
|
||||
|
app = FastAPI() |
||||
|
|
||||
|
router_lax = APIRouter(prefix="/lax", strict_content_type=False) |
||||
|
router_strict = APIRouter(prefix="/strict", strict_content_type=True) |
||||
|
router_default = APIRouter(prefix="/default") |
||||
|
|
||||
|
|
||||
|
@router_lax.post("/items/") |
||||
|
async def router_lax_post(data: dict): |
||||
|
return data |
||||
|
|
||||
|
|
||||
|
@router_strict.post("/items/") |
||||
|
async def router_strict_post(data: dict): |
||||
|
return data |
||||
|
|
||||
|
|
||||
|
@router_default.post("/items/") |
||||
|
async def router_default_post(data: dict): |
||||
|
return data |
||||
|
|
||||
|
|
||||
|
app.include_router(router_lax) |
||||
|
app.include_router(router_strict) |
||||
|
app.include_router(router_default) |
||||
|
|
||||
|
client = TestClient(app) |
||||
|
|
||||
|
|
||||
|
def test_lax_router_on_strict_app_accepts_no_content_type(): |
||||
|
response = client.post("/lax/items/", content='{"key": "value"}') |
||||
|
assert response.status_code == 200 |
||||
|
assert response.json() == {"key": "value"} |
||||
|
|
||||
|
|
||||
|
def test_strict_router_on_strict_app_rejects_no_content_type(): |
||||
|
response = client.post("/strict/items/", content='{"key": "value"}') |
||||
|
assert response.status_code == 422 |
||||
|
|
||||
|
|
||||
|
def test_default_router_inherits_strict_from_app(): |
||||
|
response = client.post("/default/items/", content='{"key": "value"}') |
||||
|
assert response.status_code == 422 |
||||
|
|
||||
|
|
||||
|
def test_lax_router_accepts_json_content_type(): |
||||
|
response = client.post("/lax/items/", json={"key": "value"}) |
||||
|
assert response.status_code == 200 |
||||
|
|
||||
|
|
||||
|
def test_strict_router_accepts_json_content_type(): |
||||
|
response = client.post("/strict/items/", json={"key": "value"}) |
||||
|
assert response.status_code == 200 |
||||
|
|
||||
|
|
||||
|
def test_default_router_accepts_json_content_type(): |
||||
|
response = client.post("/default/items/", json={"key": "value"}) |
||||
|
assert response.status_code == 200 |
||||
@ -0,0 +1,37 @@ |
|||||
|
from fastapi.openapi.docs import get_swagger_ui_html |
||||
|
|
||||
|
|
||||
|
def test_init_oauth_html_chars_are_escaped(): |
||||
|
xss_payload = "Evil</script><script>alert(1)</script>" |
||||
|
html = get_swagger_ui_html( |
||||
|
openapi_url="/openapi.json", |
||||
|
title="Test", |
||||
|
init_oauth={"appName": xss_payload}, |
||||
|
) |
||||
|
body = html.body.decode() |
||||
|
|
||||
|
assert "</script><script>" not in body |
||||
|
assert "\\u003c/script\\u003e\\u003cscript\\u003e" in body |
||||
|
|
||||
|
|
||||
|
def test_swagger_ui_parameters_html_chars_are_escaped(): |
||||
|
html = get_swagger_ui_html( |
||||
|
openapi_url="/openapi.json", |
||||
|
title="Test", |
||||
|
swagger_ui_parameters={"customKey": "<img src=x onerror=alert(1)>"}, |
||||
|
) |
||||
|
body = html.body.decode() |
||||
|
assert "<img src=x onerror=alert(1)>" not in body |
||||
|
assert "\\u003cimg" in body |
||||
|
|
||||
|
|
||||
|
def test_normal_init_oauth_still_works(): |
||||
|
html = get_swagger_ui_html( |
||||
|
openapi_url="/openapi.json", |
||||
|
title="Test", |
||||
|
init_oauth={"clientId": "my-client", "appName": "My App"}, |
||||
|
) |
||||
|
body = html.body.decode() |
||||
|
assert '"clientId": "my-client"' in body |
||||
|
assert '"appName": "My App"' in body |
||||
|
assert "ui.initOAuth" in body |
||||
@ -0,0 +1,50 @@ |
|||||
|
import importlib |
||||
|
|
||||
|
import pytest |
||||
|
from fastapi.testclient import TestClient |
||||
|
from inline_snapshot import snapshot |
||||
|
|
||||
|
|
||||
|
@pytest.fixture( |
||||
|
name="client", |
||||
|
params=[ |
||||
|
pytest.param("tutorial010_py310"), |
||||
|
], |
||||
|
) |
||||
|
def get_client(request: pytest.FixtureRequest): |
||||
|
mod = importlib.import_module(f"docs_src.custom_response.{request.param}") |
||||
|
client = TestClient(mod.app) |
||||
|
return client |
||||
|
|
||||
|
|
||||
|
def test_get_custom_response(client: TestClient): |
||||
|
response = client.get("/items/") |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.text == snapshot("<h1>Items</h1><p>This is a list of items.</p>") |
||||
|
|
||||
|
|
||||
|
def test_openapi_schema(client: TestClient): |
||||
|
response = client.get("/openapi.json") |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == snapshot( |
||||
|
{ |
||||
|
"openapi": "3.1.0", |
||||
|
"info": {"title": "FastAPI", "version": "0.1.0"}, |
||||
|
"paths": { |
||||
|
"/items/": { |
||||
|
"get": { |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
"content": { |
||||
|
"text/html": {"schema": {"type": "string"}} |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
"summary": "Read Items", |
||||
|
"operationId": "read_items_items__get", |
||||
|
} |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
) |
||||
@ -0,0 +1,225 @@ |
|||||
|
import importlib |
||||
|
|
||||
|
import pytest |
||||
|
from fastapi.testclient import TestClient |
||||
|
from inline_snapshot import snapshot |
||||
|
|
||||
|
from tests.utils import needs_py310 |
||||
|
|
||||
|
|
||||
|
@pytest.fixture( |
||||
|
name="client", |
||||
|
params=[pytest.param("tutorial001_py310", marks=needs_py310)], |
||||
|
) |
||||
|
def get_client(request: pytest.FixtureRequest): |
||||
|
mod = importlib.import_module(f"docs_src.json_base64_bytes.{request.param}") |
||||
|
|
||||
|
client = TestClient(mod.app) |
||||
|
return client |
||||
|
|
||||
|
|
||||
|
def test_post_data(client: TestClient): |
||||
|
response = client.post( |
||||
|
"/data", |
||||
|
json={ |
||||
|
"description": "A file", |
||||
|
"data": "SGVsbG8sIFdvcmxkIQ==", |
||||
|
}, |
||||
|
) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == {"description": "A file", "content": "Hello, World!"} |
||||
|
|
||||
|
|
||||
|
def test_get_data(client: TestClient): |
||||
|
response = client.get("/data") |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == {"description": "A plumbus", "data": "aGVsbG8="} |
||||
|
|
||||
|
|
||||
|
def test_post_data_in_out(client: TestClient): |
||||
|
response = client.post( |
||||
|
"/data-in-out", |
||||
|
json={ |
||||
|
"description": "A plumbus", |
||||
|
"data": "SGVsbG8sIFdvcmxkIQ==", |
||||
|
}, |
||||
|
) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == { |
||||
|
"description": "A plumbus", |
||||
|
"data": "SGVsbG8sIFdvcmxkIQ==", |
||||
|
} |
||||
|
|
||||
|
|
||||
|
def test_openapi_schema(client: TestClient): |
||||
|
response = client.get("/openapi.json") |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == snapshot( |
||||
|
{ |
||||
|
"openapi": "3.1.0", |
||||
|
"info": {"title": "FastAPI", "version": "0.1.0"}, |
||||
|
"paths": { |
||||
|
"/data": { |
||||
|
"get": { |
||||
|
"summary": "Get Data", |
||||
|
"operationId": "get_data_data_get", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
"content": { |
||||
|
"application/json": { |
||||
|
"schema": { |
||||
|
"$ref": "#/components/schemas/DataOutput" |
||||
|
} |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
}, |
||||
|
"post": { |
||||
|
"summary": "Post Data", |
||||
|
"operationId": "post_data_data_post", |
||||
|
"requestBody": { |
||||
|
"content": { |
||||
|
"application/json": { |
||||
|
"schema": {"$ref": "#/components/schemas/DataInput"} |
||||
|
} |
||||
|
}, |
||||
|
"required": True, |
||||
|
}, |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
"content": {"application/json": {"schema": {}}}, |
||||
|
}, |
||||
|
"422": { |
||||
|
"description": "Validation Error", |
||||
|
"content": { |
||||
|
"application/json": { |
||||
|
"schema": { |
||||
|
"$ref": "#/components/schemas/HTTPValidationError" |
||||
|
} |
||||
|
} |
||||
|
}, |
||||
|
}, |
||||
|
}, |
||||
|
}, |
||||
|
}, |
||||
|
"/data-in-out": { |
||||
|
"post": { |
||||
|
"summary": "Post Data In Out", |
||||
|
"operationId": "post_data_in_out_data_in_out_post", |
||||
|
"requestBody": { |
||||
|
"content": { |
||||
|
"application/json": { |
||||
|
"schema": { |
||||
|
"$ref": "#/components/schemas/DataInputOutput" |
||||
|
} |
||||
|
} |
||||
|
}, |
||||
|
"required": True, |
||||
|
}, |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
"content": { |
||||
|
"application/json": { |
||||
|
"schema": { |
||||
|
"$ref": "#/components/schemas/DataInputOutput" |
||||
|
} |
||||
|
} |
||||
|
}, |
||||
|
}, |
||||
|
"422": { |
||||
|
"description": "Validation Error", |
||||
|
"content": { |
||||
|
"application/json": { |
||||
|
"schema": { |
||||
|
"$ref": "#/components/schemas/HTTPValidationError" |
||||
|
} |
||||
|
} |
||||
|
}, |
||||
|
}, |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
}, |
||||
|
"components": { |
||||
|
"schemas": { |
||||
|
"DataInput": { |
||||
|
"properties": { |
||||
|
"description": {"type": "string", "title": "Description"}, |
||||
|
"data": { |
||||
|
"type": "string", |
||||
|
"contentEncoding": "base64", |
||||
|
"contentMediaType": "application/octet-stream", |
||||
|
"title": "Data", |
||||
|
}, |
||||
|
}, |
||||
|
"type": "object", |
||||
|
"required": ["description", "data"], |
||||
|
"title": "DataInput", |
||||
|
}, |
||||
|
"DataInputOutput": { |
||||
|
"properties": { |
||||
|
"description": {"type": "string", "title": "Description"}, |
||||
|
"data": { |
||||
|
"type": "string", |
||||
|
"contentEncoding": "base64", |
||||
|
"contentMediaType": "application/octet-stream", |
||||
|
"title": "Data", |
||||
|
}, |
||||
|
}, |
||||
|
"type": "object", |
||||
|
"required": ["description", "data"], |
||||
|
"title": "DataInputOutput", |
||||
|
}, |
||||
|
"DataOutput": { |
||||
|
"properties": { |
||||
|
"description": {"type": "string", "title": "Description"}, |
||||
|
"data": { |
||||
|
"type": "string", |
||||
|
"contentEncoding": "base64", |
||||
|
"contentMediaType": "application/octet-stream", |
||||
|
"title": "Data", |
||||
|
}, |
||||
|
}, |
||||
|
"type": "object", |
||||
|
"required": ["description", "data"], |
||||
|
"title": "DataOutput", |
||||
|
}, |
||||
|
"HTTPValidationError": { |
||||
|
"properties": { |
||||
|
"detail": { |
||||
|
"items": { |
||||
|
"$ref": "#/components/schemas/ValidationError" |
||||
|
}, |
||||
|
"type": "array", |
||||
|
"title": "Detail", |
||||
|
} |
||||
|
}, |
||||
|
"type": "object", |
||||
|
"title": "HTTPValidationError", |
||||
|
}, |
||||
|
"ValidationError": { |
||||
|
"properties": { |
||||
|
"ctx": {"title": "Context", "type": "object"}, |
||||
|
"input": {"title": "Input"}, |
||||
|
"loc": { |
||||
|
"items": { |
||||
|
"anyOf": [{"type": "string"}, {"type": "integer"}] |
||||
|
}, |
||||
|
"type": "array", |
||||
|
"title": "Location", |
||||
|
}, |
||||
|
"msg": {"type": "string", "title": "Message"}, |
||||
|
"type": {"type": "string", "title": "Error Type"}, |
||||
|
}, |
||||
|
"type": "object", |
||||
|
"required": ["loc", "msg", "type"], |
||||
|
"title": "ValidationError", |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
) |
||||
@ -22,7 +22,7 @@ def get_mod_name(request: pytest.FixtureRequest): |
|||||
@pytest.fixture(name="client") |
@pytest.fixture(name="client") |
||||
def get_test_client(mod_name: str, monkeypatch: MonkeyPatch) -> TestClient: |
def get_test_client(mod_name: str, monkeypatch: MonkeyPatch) -> TestClient: |
||||
if mod_name in sys.modules: |
if mod_name in sys.modules: |
||||
del sys.modules[mod_name] |
del sys.modules[mod_name] # pragma: no cover |
||||
monkeypatch.setenv("ADMIN_EMAIL", "[email protected]") |
monkeypatch.setenv("ADMIN_EMAIL", "[email protected]") |
||||
main_mod = importlib.import_module(mod_name) |
main_mod = importlib.import_module(mod_name) |
||||
return TestClient(main_mod.app) |
return TestClient(main_mod.app) |
||||
|
|||||
@ -0,0 +1,154 @@ |
|||||
|
import importlib |
||||
|
|
||||
|
import pytest |
||||
|
from fastapi.testclient import TestClient |
||||
|
from inline_snapshot import snapshot |
||||
|
|
||||
|
|
||||
|
@pytest.fixture( |
||||
|
name="client", |
||||
|
params=[ |
||||
|
pytest.param("tutorial001_py310"), |
||||
|
], |
||||
|
) |
||||
|
def get_client(request: pytest.FixtureRequest): |
||||
|
mod = importlib.import_module(f"docs_src.stream_data.{request.param}") |
||||
|
|
||||
|
client = TestClient(mod.app) |
||||
|
return client |
||||
|
|
||||
|
|
||||
|
expected_text = ( |
||||
|
"" |
||||
|
"Rick: (stumbles in drunkenly, and turns on the lights)" |
||||
|
" Morty! You gotta come on. You got--... you gotta come with me." |
||||
|
"Morty: (rubs his eyes) What, Rick? What's going on?" |
||||
|
"Rick: I got a surprise for you, Morty." |
||||
|
"Morty: It's the middle of the night. What are you talking about?" |
||||
|
"Rick: (spills alcohol on Morty's bed) Come on, I got a surprise for you." |
||||
|
" (drags Morty by the ankle) Come on, hurry up." |
||||
|
" (pulls Morty out of his bed and into the hall)" |
||||
|
"Morty: Ow! Ow! You're tugging me too hard!" |
||||
|
"Rick: We gotta go, gotta get outta here, come on." |
||||
|
" Got a surprise for you Morty." |
||||
|
) |
||||
|
|
||||
|
|
||||
|
@pytest.mark.parametrize( |
||||
|
"path", |
||||
|
[ |
||||
|
"/story/stream", |
||||
|
"/story/stream-no-async", |
||||
|
"/story/stream-no-annotation", |
||||
|
"/story/stream-no-async-no-annotation", |
||||
|
"/story/stream-bytes", |
||||
|
"/story/stream-no-async-bytes", |
||||
|
"/story/stream-no-annotation-bytes", |
||||
|
"/story/stream-no-async-no-annotation-bytes", |
||||
|
], |
||||
|
) |
||||
|
def test_stream_story(client: TestClient, path: str): |
||||
|
response = client.get(path) |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.text == expected_text |
||||
|
|
||||
|
|
||||
|
def test_openapi_schema(client: TestClient): |
||||
|
response = client.get("/openapi.json") |
||||
|
assert response.status_code == 200, response.text |
||||
|
assert response.json() == snapshot( |
||||
|
{ |
||||
|
"openapi": "3.1.0", |
||||
|
"info": {"title": "FastAPI", "version": "0.1.0"}, |
||||
|
"paths": { |
||||
|
"/story/stream": { |
||||
|
"get": { |
||||
|
"summary": "Stream Story", |
||||
|
"operationId": "stream_story_story_stream_get", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
"/story/stream-no-async": { |
||||
|
"get": { |
||||
|
"summary": "Stream Story No Async", |
||||
|
"operationId": "stream_story_no_async_story_stream_no_async_get", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
"/story/stream-no-annotation": { |
||||
|
"get": { |
||||
|
"summary": "Stream Story No Annotation", |
||||
|
"operationId": "stream_story_no_annotation_story_stream_no_annotation_get", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
"/story/stream-no-async-no-annotation": { |
||||
|
"get": { |
||||
|
"summary": "Stream Story No Async No Annotation", |
||||
|
"operationId": "stream_story_no_async_no_annotation_story_stream_no_async_no_annotation_get", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
"/story/stream-bytes": { |
||||
|
"get": { |
||||
|
"summary": "Stream Story Bytes", |
||||
|
"operationId": "stream_story_bytes_story_stream_bytes_get", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
"/story/stream-no-async-bytes": { |
||||
|
"get": { |
||||
|
"summary": "Stream Story No Async Bytes", |
||||
|
"operationId": "stream_story_no_async_bytes_story_stream_no_async_bytes_get", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
"/story/stream-no-annotation-bytes": { |
||||
|
"get": { |
||||
|
"summary": "Stream Story No Annotation Bytes", |
||||
|
"operationId": "stream_story_no_annotation_bytes_story_stream_no_annotation_bytes_get", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
"/story/stream-no-async-no-annotation-bytes": { |
||||
|
"get": { |
||||
|
"summary": "Stream Story No Async No Annotation Bytes", |
||||
|
"operationId": "stream_story_no_async_no_annotation_bytes_story_stream_no_async_no_annotation_bytes_get", |
||||
|
"responses": { |
||||
|
"200": { |
||||
|
"description": "Successful Response", |
||||
|
} |
||||
|
}, |
||||
|
} |
||||
|
}, |
||||
|
}, |
||||
|
} |
||||
|
) |
||||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue