Compare commits

...

35 Commits

Author SHA1 Message Date
Sebastián Ramírez
e68a68c97c 🔖 Release 0.12.0, add additional responses 2019-04-05 14:35:01 +04:00
Sebastián Ramírez
907e613ff2 🔧 Update test-conv-html.sh to allow extra params 2019-04-05 14:29:36 +04:00
Sebastián Ramírez
f0fc2fad2c 📝 Update release notes 2019-04-05 14:28:30 +04:00
Sebastián Ramírez
ad471307e2 Additional Responses (#97)
Add additional responses to OpenAPI, including Pydantic models or schemas directly, custom status codes, media types, extending `response_model`, etc.
2019-04-05 14:18:28 +04:00
Sebastián Ramírez
2bd775988f Add/refactor addditional responses, tests, docs 2019-04-05 13:54:00 +04:00
Sebastián Ramírez
c59ddc8a24 🔖 Release 0.11.0 2019-04-03 15:51:44 +04:00
Sebastián Ramírez
378b39bbbc 📝 Update release notes 2019-04-03 15:49:58 +04:00
Sebastián Ramírez
37e0306517 📝 Update default error response in SQL tutorial 2019-04-03 15:49:40 +04:00
Sebastián Ramírez
fad3a9e1dc Add auto_error to security utils (#134)
to allow them to be optional, also allowing the declaration of multiple security schemes
2019-04-03 15:44:52 +04:00
Sebastián Ramírez
b35b0a9a90 📝 Update release notes 2019-03-31 22:05:03 +04:00
Alex Iribarren
1426b6200a 🗃️ Close the DB even if exceptions are raised (#89)
* Close the DB even if exceptions are raised

* 📝 Add note about closing DB in finally
2019-03-31 22:01:32 +04:00
Sebastián Ramírez
40e5f3764e 📝 Update release notes 2019-03-31 20:56:52 +04:00
Alif Jahan
e5c75807ce 🔥 removed duplicate dependency in pyproject.toml (#128) 2019-03-31 20:55:12 +04:00
Sebastián Ramírez
deff2b6678 🔖 Release 0.10.3 2019-03-30 21:54:00 +04:00
Sebastián Ramírez
7c572fdb3a 📝 Update release notes 2019-03-30 21:32:24 +04:00
Daniel Hahler
ae970638cf 👷 Set Travis to use dist=xenial and Python 3.7 instead of 3.7-dev (#92) 2019-03-30 21:30:31 +04:00
Sebastián Ramírez
deae92bba1 📝 Update release notes 2019-03-30 21:09:47 +04:00
yihuang
f806ba642a 🔥 Remove repeated param declaration (#123) 2019-03-30 21:07:41 +04:00
Sebastián Ramírez
5a3cf863da 📝 Update release notes 2019-03-30 19:55:01 +04:00
Sebastián Ramírez
dd6ab23b62 Add docs/tests extending OpenAPI (#126) 2019-03-30 19:53:44 +04:00
Sebastián Ramírez
0449499188 📝 Update release notes 2019-03-30 18:32:23 +04:00
The Gitter Badger
4dc7b32861 📝 Add a Gitter chat badge and links (#117)
* Add Gitter badge

* 📝 Add links and badges to Gitter chat
2019-03-30 18:30:02 +04:00
Sebastián Ramírez
08d849d5c5 📝 Update release notes 2019-03-29 19:06:02 +04:00
James Saunders
714e68b5f0 📝 Add note in response model docs: why not return type annotations (#109)
* Update response model documentation to explain design choice

Closes #101

* 📝 Update note about return function type annotation
2019-03-29 19:02:53 +04:00
Sebastián Ramírez
3d4f59f35a 📝 Udpate release notes 2019-03-29 18:43:40 +04:00
Stratos Gerakakis
3ce2920fef 🐛 fix name of shutdown_event in docs (#105)
Fix name copy/paste name error in docs source for startup/shutdown events.
2019-03-29 18:39:57 +04:00
Mohammed
eda9b28338 files formatting 2019-03-23 13:10:45 +03:00
Mohammed
7514ac6fb0 100% test coverage 2019-03-23 13:01:53 +03:00
Mohammed
25fb4239cc increase test coverage 2019-03-23 01:13:09 +03:00
Mohammed
65568065e0 Remove extra code. 2019-03-23 00:47:32 +03:00
Mohammed
95679ca5e6 Fix: adding additional_responses on .include_router() 2019-03-23 00:37:10 +03:00
Mohammed
84a300ef84 Formatting according to guide 2019-03-22 22:54:48 +03:00
Mohammed
c6d28c8209 Accept Multiple Additional Responses 2019-03-22 22:50:47 +03:00
Mohammed
3984e9b8ac Additional Responses test 2019-03-22 22:40:46 +03:00
Mohammed
aa0bca7bb2 Additional Responses implementation 2019-03-22 22:40:07 +03:00
54 changed files with 2354 additions and 98 deletions

View File

@@ -1,10 +1,12 @@
dist: xenial
language: python
cache: pip
python:
- "3.6"
- "3.7-dev"
- "3.7"
install:
- pip install flit

View File

@@ -14,6 +14,9 @@
<a href="https://pypi.org/project/fastapi" target="_blank">
<img src="https://badge.fury.io/py/fastapi.svg" alt="Package version">
</a>
<a href="https://gitter.im/tiangolo/fastapi?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge" target="_blank">
<img src="https://badges.gitter.im/tiangolo/fastapi.svg" alt="Join the chat at https://gitter.im/tiangolo/fastapi">
</a>
</p>
---

View File

@@ -24,6 +24,15 @@ There you can select "Releases only".
Doing it, you will receive notifications (in your email) whenever there's a new release (a new version) of **FastAPI** with bug fixes and new features.
## Join the chat
<a href="https://gitter.im/tiangolo/fastapi?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge" target="_blank">
<img src="https://badges.gitter.im/tiangolo/fastapi.svg" alt="Join the chat at https://gitter.im/tiangolo/fastapi">
</a>
Join the chat on Gitter: <a href="https://gitter.im/tiangolo/fastapi" target="_blank">https://gitter.im/tiangolo/fastapi</a>.
There you can ask quick questions, help others, share ideas, etc.
## Connect with the author

View File

Binary file not shown.

After

Width:  |  Height:  |  Size: 80 KiB

View File

Binary file not shown.

Before

Width:  |  Height:  |  Size: 60 KiB

After

Width:  |  Height:  |  Size: 76 KiB

View File

Binary file not shown.

After

Width:  |  Height:  |  Size: 63 KiB

View File

@@ -14,6 +14,9 @@
<a href="https://pypi.org/project/fastapi" target="_blank">
<img src="https://badge.fury.io/py/fastapi.svg" alt="Package version">
</a>
<a href="https://gitter.im/tiangolo/fastapi?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge" target="_blank">
<img src="https://badges.gitter.im/tiangolo/fastapi.svg" alt="Join the chat at https://gitter.im/tiangolo/fastapi">
</a>
</p>
---

View File

@@ -1,5 +1,38 @@
## Next release
## 0.12.0
* Add additional `responses` parameter to *path operation decorators* to extend responses in OpenAPI (and API docs).
* It also allows extending existing responses generated from `response_model`, declare other media types (like images), etc.
* The new documentation is here: <a href="https://fastapi.tiangolo.com/tutorial/additional-responses/" target="_blank">Additional Responses</a>.
* `responses` can also be added to `.include_router()`, the updated docs are here: <a href="https://fastapi.tiangolo.com/tutorial/bigger-applications/#add-some-custom-tags-and-responses" target="_blank">Bigger Applications</a>.
* PR <a href="https://github.com/tiangolo/fastapi/pull/97" target="_blank">#97</a> originally initiated by <a href="https://github.com/barsi" target="_blank">@barsi</a>.
* Update `scripts/test-cov-html.sh` to allow passing extra parameters like `-vv`, for development.
## 0.11.0
* Add `auto_error` parameter to security utility functions. Allowing them to be optional. Also allowing to have multiple alternative security schemes that are then checked in a single dependency instead of each one verifying and returning the error to the client automatically when not satisfied. PR <a href="https://github.com/tiangolo/fastapi/pull/134" target="_blank">#134</a>.
* Update <a href="https://fastapi.tiangolo.com/tutorial/sql-databases/#create-a-middleware-to-handle-sessions" target="_blank">SQL Tutorial</a> to close database sessions even when there are exceptions. PR <a href="https://github.com/tiangolo/fastapi/pull/89" target="_blank">#89</a> by <a href="https://github.com/alexiri" target="_blank">@alexiri</a>.
* Fix duplicate dependency in `pyproject.toml`. PR <a href="https://github.com/tiangolo/fastapi/pull/128" target="_blank">#128</a> by <a href="https://github.com/zxalif" target="_blank">@zxalif</a>.
## 0.10.3
* Add Gitter chat, badge, links, etc. <a href="https://gitter.im/tiangolo/fastapi" target="_blank">https://gitter.im/tiangolo/fastapi
</a>. PR <a href="https://github.com/tiangolo/fastapi/pull/117" target="_blank">#117</a>.
* Add docs about <a href="https://fastapi.tiangolo.com/tutorial/extending-openapi/" target="_blank">Extending OpenAPI</a>. PR <a href="https://github.com/tiangolo/fastapi/pull/126" target="_blank">#126</a>.
* Make Travis run Ubuntu Xenial (newer version) and Python 3.7 instead of Python 3.7-dev. PR <a href="https://github.com/tiangolo/fastapi/pull/92" target="_blank">#92</a> by <a href="https://github.com/blueyed" target="_blank">@blueyed</a>.
* Fix duplicated param variable creation. PR <a href="https://github.com/tiangolo/fastapi/pull/123" target="_blank">#123</a> by <a href="https://github.com/yihuang" target="_blank">@yihuang</a>.
* Add note in <a href="https://fastapi.tiangolo.com/tutorial/response-model/" target="_blank">Response Model docs</a> about why using a function parameter instead of a function return type annotation. PR <a href="https://github.com/tiangolo/fastapi/pull/109" target="_blank">#109</a> by <a href="https://github.com/JHSaunders" target="_blank">@JHSaunders</a>.
* Fix event docs (startup/shutdown) function name. PR <a href="https://github.com/tiangolo/fastapi/pull/105" target="_blank">#105</a> by <a href="https://github.com/stratosgear" target="_blank">@stratosgear</a>.
## 0.10.2
* Fix OpenAPI (JSON Schema) for declarations of Python `Union` (JSON Schema `additionalProperties`). PR <a href="https://github.com/tiangolo/fastapi/pull/121" target="_blank">#121</a>.

View File

@@ -0,0 +1,23 @@
from fastapi import FastAPI
from pydantic import BaseModel
from starlette.responses import JSONResponse
class Item(BaseModel):
id: str
value: str
class Message(BaseModel):
message: str
app = FastAPI()
@app.get("/items/{item_id}", response_model=Item, responses={404: {"model": Message}})
async def read_item(item_id: str):
if item_id == "foo":
return {"id": "foo", "value": "there goes my hero"}
else:
return JSONResponse(status_code=404, content={"message": "Item not found"})

View File

@@ -0,0 +1,28 @@
from fastapi import FastAPI
from pydantic import BaseModel
from starlette.responses import FileResponse
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):
if img:
return FileResponse("image.png", media_type="image/png")
else:
return {"id": "foo", "value": "there goes my hero"}

View File

@@ -0,0 +1,37 @@
from fastapi import FastAPI
from pydantic import BaseModel
from starlette.responses import JSONResponse
class Item(BaseModel):
id: str
value: str
class Message(BaseModel):
message: str
app = FastAPI()
@app.get(
"/items/{item_id}",
response_model=Item,
responses={
404: {"model": Message, "description": "The item was not found"},
200: {
"description": "Item requested by ID",
"content": {
"application/json": {
"example": {"id": "bar", "value": "The bar tenders"}
}
},
},
},
)
async def read_item(item_id: str):
if item_id == "foo":
return {"id": "foo", "value": "there goes my hero"}
else:
return JSONResponse(status_code=404, content={"message": "Item not found"})

View File

@@ -0,0 +1,30 @@
from fastapi import FastAPI
from pydantic import BaseModel
from starlette.responses import FileResponse
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):
if img:
return FileResponse("image.png", media_type="image/png")
else:
return {"id": "foo", "value": "there goes my hero"}

View File

@@ -5,4 +5,9 @@ from .routers import items, users
app = FastAPI()
app.include_router(users.router)
app.include_router(items.router, prefix="/items", tags=["items"])
app.include_router(
items.router,
prefix="/items",
tags=["items"],
responses={404: {"description": "Not found"}},
)

View File

@@ -1,4 +1,4 @@
from fastapi import APIRouter
from fastapi import APIRouter, HTTPException
router = APIRouter()
@@ -11,3 +11,14 @@ async def read_items():
@router.get("/{item_id}")
async def read_item(item_id: str):
return {"name": "Fake Specific Item", "item_id": item_id}
@router.put(
"/{item_id}",
tags=["custom"],
responses={403: {"description": "Operation forbidden"}},
)
async def update_item(item_id: str):
if item_id != "foo":
raise HTTPException(status_code=403, detail="You can only update the item: foo")
return {"item_id": item_id, "name": "The Fighters"}

View File

@@ -4,7 +4,7 @@ app = FastAPI()
@app.on_event("shutdown")
def startup_event():
def shutdown_event():
with open("log.txt", mode="a") as log:
log.write("Application shutdown")

View File

@@ -0,0 +1,28 @@
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi
app = FastAPI()
@app.get("/items/")
async def read_items():
return [{"name": "Foo"}]
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title="Custom title",
version="2.5.0",
description="This is a very custom OpenAPI schema",
routes=app.routes,
)
openapi_schema["info"]["x-logo"] = {
"url": "https://fastapi.tiangolo.com/img/logo-margin/logo-teal.png"
}
app.openapi_schema = openapi_schema
return app.openapi_schema
app.openapi = custom_openapi

View File

@@ -3,6 +3,7 @@ from sqlalchemy import Boolean, Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.orm import Session, sessionmaker
from starlette.requests import Request
from starlette.responses import Response
# SQLAlchemy specific code, as with any other app
SQLALCHEMY_DATABASE_URI = "sqlite:///./test.db"
@@ -66,7 +67,10 @@ def read_user(user_id: int, db: Session = Depends(get_db)):
@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
request.state.db = SessionLocal()
response = await call_next(request)
request.state.db.close()
response = Response("Internal server error", status_code=500)
try:
request.state.db = SessionLocal()
response = await call_next(request)
finally:
request.state.db.close()
return response

View File

@@ -0,0 +1,235 @@
!!! warning
This is a rather advanced topic.
If you are starting with **FastAPI**, you might not need this.
You can declare additional responses, with additional status codes, media types, descriptions, etc.
Those additional responses will be included in the OpenAPI schema, so they will also appear in the API docs.
But for those additional responses you have to make sure you return a `Response` like `JSONResponse` directly, with your status code and content.
## Additional Response with `model`
You can pass to your *path operation decorators* a parameter `responses`.
It receives a `dict`, the keys are status codes for each response, like `200`, and the values are other `dict`s with the information for each of them.
Each of those response `dict`s can have a key `model`, containing a Pydantic model, just like `response_model`.
**FastAPI** will take that model, generate its JSON Schema and include it in the correct place in OpenAPI.
For example, to declare another response with a status code `404` and a Pydantic model `Message`, you can write:
```Python hl_lines="18 23"
{!./src/additional_responses/tutorial001.py!}
```
!!! note
Have in mind that you have to return the `JSONResponse` directly.
!!! info
The `model` key is not part of OpenAPI.
**FastAPI** will take the Pydantic model from there, generate the `JSON Schema`, and put it in the correct place.
The correct place is:
* In the key `content`, that has as value another JSON object (`dict`) that contains:
* A key with the media type, e.g. `application/json`, that contains as value another JSON object, that contains:
* A key `schema`, that has as the value the JSON Schema from the model, here's the correct place.
* **FastAPI** adds a reference here to the global JSON Schemas in another place in your OpenAPI instead of including it directly. This way, other applications and clients can use those JSON Schemas directly, provide better code generation tools, etc.
The generated responses in the OpenAPI for this *path operation* will be:
```JSON hl_lines="3 4 5 6 7 8 9 10 11 12"
{
"responses": {
"404": {
"description": "Additional Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Message"
}
}
}
},
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/Item"
}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
```
The schemas are referenced to another place inside the OpenAPI schema:
```JSON hl_lines="4 5 6 7 8 9 10 11 12 13 14 15 16"
{
"components": {
"schemas": {
"Message": {
"title": "Message",
"required": [
"message"
],
"type": "object",
"properties": {
"message": {
"title": "Message",
"type": "string"
}
}
},
"Item": {
"title": "Item",
"required": [
"id",
"value"
],
"type": "object",
"properties": {
"id": {
"title": "Id",
"type": "string"
},
"value": {
"title": "Value",
"type": "string"
}
}
},
"ValidationError": {
"title": "ValidationError",
"required": [
"loc",
"msg",
"type"
],
"type": "object",
"properties": {
"loc": {
"title": "Location",
"type": "array",
"items": {
"type": "string"
}
},
"msg": {
"title": "Message",
"type": "string"
},
"type": {
"title": "Error Type",
"type": "string"
}
}
},
"HTTPValidationError": {
"title": "HTTPValidationError",
"type": "object",
"properties": {
"detail": {
"title": "Detail",
"type": "array",
"items": {
"$ref": "#/components/schemas/ValidationError"
}
}
}
}
}
}
}
```
## Additional media types for the main response
You can use this same `responses` parameter to add different media types for the same main response.
For example, you can add an additional media type of `image/png`, declaring that your *path operation* can return a JSON object (with media type `application/json`) or a PNG image:
```Python hl_lines="17 18 19 20 21 22 23 24 28"
{!./src/additional_responses/tutorial002.py!}
```
!!! note
Notice that you have to return the image using a `FileResponse` directly.
## Combining information
You can also combine response information from multiple places, including the `response_model`, `status_code`, and `responses` parameters.
You can declare a `response_model`, using the default status code `200` (or a custom one if you need), and then declare additional information for that same response in `responses`, directly in the OpenAPI schema.
**FastAPI** will keep the additional information from `responses`, and combine it with the JSON Schema from your model.
For example, you can declare a response with a status code `404` that uses a Pydantic model and has a custom `description`.
And a response with a status code `200` that uses your `response_model`, but includes a custom `example`:
```Python hl_lines="20 21 22 23 24 25 26 27 28 29 30 31"
{!./src/additional_responses/tutorial003.py!}
```
It will all be combined and included in your OpenAPI, and shown in the API docs:
<img src="/img/tutorial/additional-responses/image01.png">
## Combine predefined responses and custom ones
You might want to have some predefined responses that apply to many *path operations*, but you want to combine them with custom responses needed by each *path operation*.
For those cases, you can use the Python technique of "unpacking" a `dict` with `**dict_to_unpack`:
```Python
old_dict = {
"old key": "old value",
"second old key": "second old value",
}
new_dict = {**old_dict, "new key": "new value"}
```
Here, `new_dict` will contain all the key-value pairs from `old_dict` plus the new key-value pair:
```Python
{
"old key": "old value",
"second old key": "second old value",
"new key": "new value",
}
```
You can use that technique to re-use some predefined responses in your *path operations* and combine them with additional custom ones.
For example:
```Python hl_lines="11 12 13 14 15 24"
{!./src/additional_responses/tutorial004.py!}
```
## More information about OpenAPI responses
To see what exactly you can include in the responses, you can check these sections in the OpenAPI specification:
* <a href="https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.2.md#responsesObject" target="_blank">OpenAPI Responses Object</a>, it includes the `Response Object`.
* <a href="https://github.com/OAI/OpenAPI-Specification/blob/master/versions/3.0.2.md#responseObject" target="_blank">OpenAPI Response Object</a>, you can include anything from this directly in each response inside your `responses` parameter. Including `description`, `headers`, `content` (inside of this is that you declare different media types and JSON Schemas), and `links`.

View File

@@ -103,7 +103,17 @@ But let's say that this time we are more lazy.
And we don't want to have to explicitly type `/items/` and `tags=["items"]` in every *path operation* (we will be able to do it later):
```Python hl_lines="6 11 16"
```Python hl_lines="6 11"
{!./src/bigger_applications/app/routers/items.py!}
```
### Add some custom `tags` and `responses`
We are not adding the prefix `/items/` nor the `tags=["items"]` to add them later.
But we can add custom `tags` and `responses` that will be applied to a specific *path operation*:
```Python hl_lines="18 19"
{!./src/bigger_applications/app/routers/items.py!}
```
@@ -192,7 +202,7 @@ So, to be able to use both of them in the same file, we import the submodules di
Now, let's include the `router` from the submodule `users`:
```Python hl_lines="8"
```Python hl_lines="7"
{!./src/bigger_applications/app/main.py!}
```
@@ -217,7 +227,7 @@ It will include all the routes from that router as part of it.
So it won't affect performance.
### Include an `APIRouter` with a prefix
### Include an `APIRouter` with a `prefix`, `tags`, and `responses`
Now, let's include the router form the `items` submodule.
@@ -237,9 +247,11 @@ async def read_item(item_id: str):
So, the prefix in this case would be `/items`.
And we can also add a list of `tags` that will be applied to all the *path operations* included in this router:
We can also add a list of `tags` that will be applied to all the *path operations* included in this router.
```Python hl_lines="9"
And we can add predefined `responses` that will be included in all the *path operations* too.
```Python hl_lines="8 9 10 11 12 13"
{!./src/bigger_applications/app/main.py!}
```
@@ -250,12 +262,18 @@ The end result is that the item paths are now:
...as we intended.
And they are marked with a list of tags that contain a single string `"items"`.
They will be marked with a list of tags that contain a single string `"items"`.
The *path operation* that declared a `"custom"` tag will have both tags, `items` and `custom`.
These "tags" are especially useful for the automatic interactive documentation systems (using OpenAPI).
And all of them will include the the predefined `responses`.
The *path operation* that declared a custom `403` response will have both the predefined responses (`404`) and the `403` declared in it directly.
!!! check
The `prefix` and `tags` parameters are (as in many other cases) just a feature from **FastAPI** to help you avoid code duplication.
The `prefix`, `tags`, and `responses` parameters are (as in many other cases) just a feature from **FastAPI** to help you avoid code duplication.
!!! tip

View File

@@ -0,0 +1,90 @@
!!! warning
This is a rather advanced feature. You probably can skip it.
If you are just following the tutorial - user guide, you can probably skip this section.
If you already know that you need to modify the generated OpenAPI schema, continue reading.
There are some cases where you might need to modify the generated OpenAPI schema.
In this section you will see how.
## The normal process
The normal (default) process, is as follows.
A `FastAPI` application (instance) has an `.openapi()` method that is expected to return the OpenAPI schema.
As part of the application object creation, a *path operation* for `/openapi.json` (or for whatever you set your `openapi_url`) is registered.
It just returns a JSON response with the result of the application's `.openapi()` method.
By default, what the method `.openapi()` does is check the property `.openapi_schema` to see if it has contents and return them.
If it doesn't, it generates them using the utility function at `fastapi.openapi.utils.get_openapi`.
And that function `get_openapi()` receives as parameters:
* `title`: The OpenAPI title, shown in the docs.
* `version`: The version of your API, e.g. `2.5.0`.
* `openapi_version`: The version of the OpenAPI specification used. By default, the latest: `3.0.2`.
* `description`: The description of your API.
* `routes`: A list of routes, these are each of the registered *path operations*. They are taken from `app.routes`.
* `openapi_prefix`: The URL prefix to be used in your OpenAPI.
## Overriding the defaults
Using the information above, you can use the same utility function to generate the OpenAPI schema and override each part that you need.
For example, let's add <a href="https://github.com/Rebilly/ReDoc/blob/master/docs/redoc-vendor-extensions.md#x-logo" target="_blank">ReDoc's OpenAPI extension to include a custom logo</a>.
### Normal **FastAPI**
First, write all your **FastAPI** application as normally:
```Python hl_lines="1 4 7 8 9"
{!./src/extending_openapi/tutorial001.py!}
```
### Generate the OpenAPI schema
Then, use the same utility function to generate the OpenAPI schema, inside a `custom_openapi()` function:
```Python hl_lines="2 15 16 17 18 19 20"
{!./src/extending_openapi/tutorial001.py!}
```
### Modify the OpenAPI schema
Now you can add the ReDoc extension, adding a custom `x-logo` to the `info` "object" in the OpenAPI schema:
```Python hl_lines="21 22 23"
{!./src/extending_openapi/tutorial001.py!}
```
### Cache the OpenAPI schema
You can use the property `.openapi_schema` as a "cache", to store your generated schema.
That way, your application won't have to generate the schema every time a user opens your API docs.
It will be generated only once, and then the same cached schema will be used for the next requests.
```Python hl_lines="13 14 24 25"
{!./src/extending_openapi/tutorial001.py!}
```
### Override the method
Now you can replace the `.openapi()` method with your new function.
```Python hl_lines="28"
{!./src/extending_openapi/tutorial001.py!}
```
### Check it
Once you go to <a href="http://127.0.0.1:8000/redoc" target="_blank">http://127.0.0.1:8000/redoc</a> you will see that you are using your custom logo (in this example, **FastAPI**'s logo):
<img src="/img/tutorial/extending-openapi/image01.png">

View File

@@ -24,6 +24,9 @@ But most importantly:
* Will limit the output data to that of the model. We'll see how that's important below.
!!! note "Technical Details"
The response model is declared in this parameter instead of as a function return type annotation, because the path function may not actually return that response model but rather return a `dict`, database object or some other model, and then use the `response_model` to perform the field limiting and serialization.
## Return the same input data
Here we are declaring a `UserIn` model, it will contain a plaintext password:

View File

@@ -37,7 +37,7 @@ For now, don't pay attention to the rest, only the imports:
Define the database that SQLAlchemy should "connect" to:
```Python hl_lines="8"
```Python hl_lines="9"
{!./src/sql_databases/tutorial001.py!}
```
@@ -59,7 +59,7 @@ SQLALCHEMY_DATABASE_URI = "postgresql://user:password@postgresserver/db"
## Create the SQLAlchemy `engine`
```Python hl_lines="11 12 13"
```Python hl_lines="12 13 14"
{!./src/sql_databases/tutorial001.py!}
```
@@ -90,7 +90,7 @@ We will use `Session` to declare types later and getter better editor support an
For now, create the `SessionLocal`:
```Python hl_lines="14"
```Python hl_lines="15"
{!./src/sql_databases/tutorial001.py!}
```
@@ -108,10 +108,17 @@ A "middleware" is a function that is always executed for each request, and have
This middleware (just a function) will create a new SQLAlchemy `SessionLocal` for each request, add it to the request and then close it once the request is finished.
```Python hl_lines="67 68 69 70 71 72"
```Python hl_lines="68 69 70 71 72 73 74 75 76"
{!./src/sql_databases/tutorial001.py!}
```
!!! info
We put the creation of the `SessionLocal()` and handling of the requests in a `try` block.
And then we close it in the `finally` block.
This way we make sure the database session is always closed after the request. Even if there was an exception in the middle.
### About `request.state`
<a href="https://www.starlette.io/requests/#other-state" target="_blank">`request.state` is a property of each Starlette `Request` object</a>, it is there to store arbitrary objects attached to the request itself, like the database session in this case.
@@ -126,7 +133,7 @@ And when using the dependency in a path operation function, we declare it with t
This will then give us better editor support inside the path operation function, because the editor will know that the `db` parameter is of type `Session`.
```Python hl_lines="53 54 68"
```Python hl_lines="54 55 69"
{!./src/sql_databases/tutorial001.py!}
```
@@ -145,13 +152,13 @@ That way you don't have to declare them explicitly in every model.
So, your models will behave very similarly to, for example, Flask-SQLAlchemy.
```Python hl_lines="17 18 19 20 21"
```Python hl_lines="18 19 20 21 22"
{!./src/sql_databases/tutorial001.py!}
```
## Create the SQLAlchemy `Base` model
```Python hl_lines="24"
```Python hl_lines="25"
{!./src/sql_databases/tutorial001.py!}
```
@@ -161,7 +168,7 @@ Now this is finally code specific to your app.
Here's a user model that will be a table in the database:
```Python hl_lines="27 28 29 30 31"
```Python hl_lines="28 29 30 31 32"
{!./src/sql_databases/tutorial001.py!}
```
@@ -169,7 +176,7 @@ Here's a user model that will be a table in the database:
In a very simplistic way, initialize your database (create the tables, etc) and make sure you have a first user:
```Python hl_lines="34 36 38 39 40 41 42 44"
```Python hl_lines="35 37 39 40 41 42 43 45"
{!./src/sql_databases/tutorial001.py!}
```
@@ -197,7 +204,7 @@ Also, as all the functionality is self-contained in the same code, you can copy
By creating a function that is only dedicated to getting your user from a `user_id` (or any other parameter) independent of your path operation function, you can more easily re-use it in multiple parts and also add <abbr title="Automated tests, written in code, that check if another piece of code is working correctly.">unit tests</abbr> for it:
```Python hl_lines="48 49"
```Python hl_lines="49 50"
{!./src/sql_databases/tutorial001.py!}
```
@@ -207,7 +214,7 @@ Now, finally, here's the standard **FastAPI** code.
Create your app and path operation function:
```Python hl_lines="58 61 62 63 64"
```Python hl_lines="59 62 63 64 65"
{!./src/sql_databases/tutorial001.py!}
```
@@ -243,7 +250,7 @@ user = get_user(db_session, user_id=user_id)
Then we should declare the path operation without `async def`, just with a normal `def`:
```Python hl_lines="62"
```Python hl_lines="63"
{!./src/sql_databases/tutorial001.py!}
```

View File

@@ -1,6 +1,6 @@
"""FastAPI framework, high performance, easy to learn, fast to code, ready for production"""
__version__ = "0.10.2"
__version__ = "0.12.0"
from starlette.background import BackgroundTasks

View File

@@ -1,4 +1,4 @@
from typing import Any, Callable, Dict, List, Optional, Type
from typing import Any, Callable, Dict, List, Optional, Type, Union
from fastapi import routing
from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_html
@@ -114,6 +114,7 @@ class FastAPI(Starlette):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
methods: List[str] = None,
operation_id: str = None,
@@ -130,6 +131,7 @@ class FastAPI(Starlette):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=methods,
operation_id=operation_id,
@@ -148,6 +150,7 @@ class FastAPI(Starlette):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
methods: List[str] = None,
operation_id: str = None,
@@ -165,6 +168,7 @@ class FastAPI(Starlette):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=methods,
operation_id=operation_id,
@@ -177,9 +181,16 @@ class FastAPI(Starlette):
return decorator
def include_router(
self, router: routing.APIRouter, *, prefix: str = "", tags: List[str] = None
self,
router: routing.APIRouter,
*,
prefix: str = "",
tags: List[str] = None,
responses: Dict[Union[int, str], Dict[str, Any]] = None,
) -> None:
self.router.include_router(router, prefix=prefix, tags=tags)
self.router.include_router(
router, prefix=prefix, tags=tags, responses=responses or {}
)
def get(
self,
@@ -191,6 +202,7 @@ class FastAPI(Starlette):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
include_in_schema: bool = True,
@@ -205,6 +217,7 @@ class FastAPI(Starlette):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
operation_id=operation_id,
include_in_schema=include_in_schema,
@@ -222,6 +235,7 @@ class FastAPI(Starlette):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
include_in_schema: bool = True,
@@ -236,6 +250,7 @@ class FastAPI(Starlette):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
operation_id=operation_id,
include_in_schema=include_in_schema,
@@ -253,6 +268,7 @@ class FastAPI(Starlette):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
include_in_schema: bool = True,
@@ -267,6 +283,7 @@ class FastAPI(Starlette):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
operation_id=operation_id,
include_in_schema=include_in_schema,
@@ -284,6 +301,7 @@ class FastAPI(Starlette):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
include_in_schema: bool = True,
@@ -298,6 +316,7 @@ class FastAPI(Starlette):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
operation_id=operation_id,
include_in_schema=include_in_schema,
@@ -315,6 +334,7 @@ class FastAPI(Starlette):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
include_in_schema: bool = True,
@@ -329,6 +349,7 @@ class FastAPI(Starlette):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
operation_id=operation_id,
include_in_schema=include_in_schema,
@@ -346,6 +367,7 @@ class FastAPI(Starlette):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
include_in_schema: bool = True,
@@ -360,6 +382,7 @@ class FastAPI(Starlette):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
operation_id=operation_id,
include_in_schema=include_in_schema,
@@ -377,6 +400,7 @@ class FastAPI(Starlette):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
include_in_schema: bool = True,
@@ -391,6 +415,7 @@ class FastAPI(Starlette):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
operation_id=operation_id,
include_in_schema=include_in_schema,
@@ -408,6 +433,7 @@ class FastAPI(Starlette):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
include_in_schema: bool = True,
@@ -422,6 +448,7 @@ class FastAPI(Starlette):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
operation_id=operation_id,
include_in_schema=include_in_schema,

View File

@@ -100,7 +100,6 @@ def get_dependant(*, path: str, call: Callable, name: str = None) -> Dependant:
lenient_issubclass(param.annotation, param_supported_types)
or param.annotation == param.empty
), f"Path params must be of one of the supported types"
param = signature_params[param_name]
add_param_to_fields(
param=param,
dependant=dependant,

View File

@@ -178,6 +178,23 @@ def get_openapi_path(
definitions[
"HTTPValidationError"
] = validation_error_response_definition
if route.responses:
for (additional_status_code, response) in route.responses.items():
assert isinstance(
response, dict
), "An additional response must be a dict"
field = route.response_fields.get(additional_status_code)
if field:
response_schema, _ = field_schema(
field, model_name_map=model_name_map, ref_prefix=REF_PREFIX
)
response.setdefault("content", {}).setdefault(
"application/json", {}
)["schema"] = response_schema
response.setdefault("description", "Additional Response")
operation.setdefault("responses", {})[
str(additional_status_code)
] = response
status_code = str(route.status_code)
response_schema = {"type": "string"}
if lenient_issubclass(route.content_type, JSONResponse):
@@ -189,13 +206,14 @@ def get_openapi_path(
)
else:
response_schema = {}
content = {route.content_type.media_type: {"schema": response_schema}}
operation["responses"] = {
status_code: {
"description": route.response_description,
"content": content,
}
}
operation.setdefault("responses", {}).setdefault(status_code, {})[
"description"
] = route.response_description
operation.setdefault("responses", {}).setdefault(
status_code, {}
).setdefault("content", {}).setdefault(route.content_type.media_type, {})[
"schema"
] = response_schema
if all_route_params or route.body_field:
operation["responses"][str(HTTP_422_UNPROCESSABLE_ENTITY)] = {
"description": "Validation Error",

View File

@@ -1,7 +1,7 @@
import asyncio
import inspect
import logging
from typing import Any, Callable, List, Optional, Type
from typing import Any, Callable, Dict, List, Optional, Type, Union
from fastapi import params
from fastapi.dependencies.models import Dependant
@@ -110,6 +110,7 @@ class APIRoute(routing.Route):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
name: str = None,
methods: List[str] = None,
@@ -143,6 +144,30 @@ class APIRoute(routing.Route):
self.summary = summary
self.description = description or self.endpoint.__doc__
self.response_description = response_description
self.responses = responses or {}
response_fields = {}
for additional_status_code, response in self.responses.items():
assert isinstance(response, dict), "An additional response must be a dict"
model = response.get("model")
if model:
assert lenient_issubclass(
model, BaseModel
), "A response model must be a Pydantic model"
response_name = f"Response_{additional_status_code}_{self.name}"
response_field = Field(
name=response_name,
type_=model,
class_validators=None,
default=None,
required=False,
model_config=UnconstrainedConfig,
schema=Schema(None),
)
response_fields[additional_status_code] = response_field
if response_fields:
self.response_fields: Dict[Union[int, str], Field] = response_fields
else:
self.response_fields = {}
self.deprecated = deprecated
if methods is None:
methods = ["GET"]
@@ -180,6 +205,7 @@ class APIRouter(routing.Router):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
methods: List[str] = None,
operation_id: str = None,
@@ -196,6 +222,7 @@ class APIRouter(routing.Router):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=methods,
operation_id=operation_id,
@@ -215,6 +242,7 @@ class APIRouter(routing.Router):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
methods: List[str] = None,
operation_id: str = None,
@@ -232,6 +260,7 @@ class APIRouter(routing.Router):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=methods,
operation_id=operation_id,
@@ -244,7 +273,12 @@ class APIRouter(routing.Router):
return decorator
def include_router(
self, router: "APIRouter", *, prefix: str = "", tags: List[str] = None
self,
router: "APIRouter",
*,
prefix: str = "",
tags: List[str] = None,
responses: Dict[Union[int, str], Dict[str, Any]] = None,
) -> None:
if prefix:
assert prefix.startswith("/"), "A path prefix must start with '/'"
@@ -253,6 +287,9 @@ class APIRouter(routing.Router):
), "A path prefix must not end with '/', as the routes will start with '/'"
for route in router.routes:
if isinstance(route, APIRoute):
if responses is None:
responses = {}
responses = {**responses, **route.responses}
self.add_api_route(
prefix + route.path,
route.endpoint,
@@ -262,6 +299,7 @@ class APIRouter(routing.Router):
summary=route.summary,
description=route.description,
response_description=route.response_description,
responses=responses,
deprecated=route.deprecated,
methods=route.methods,
operation_id=route.operation_id,
@@ -292,6 +330,7 @@ class APIRouter(routing.Router):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
include_in_schema: bool = True,
@@ -306,6 +345,7 @@ class APIRouter(routing.Router):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=["GET"],
operation_id=operation_id,
@@ -324,6 +364,7 @@ class APIRouter(routing.Router):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
include_in_schema: bool = True,
@@ -338,6 +379,7 @@ class APIRouter(routing.Router):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=["PUT"],
operation_id=operation_id,
@@ -356,6 +398,7 @@ class APIRouter(routing.Router):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
include_in_schema: bool = True,
@@ -370,6 +413,7 @@ class APIRouter(routing.Router):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=["POST"],
operation_id=operation_id,
@@ -388,6 +432,7 @@ class APIRouter(routing.Router):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
include_in_schema: bool = True,
@@ -402,6 +447,7 @@ class APIRouter(routing.Router):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=["DELETE"],
operation_id=operation_id,
@@ -420,6 +466,7 @@ class APIRouter(routing.Router):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
include_in_schema: bool = True,
@@ -434,6 +481,7 @@ class APIRouter(routing.Router):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=["OPTIONS"],
operation_id=operation_id,
@@ -452,6 +500,7 @@ class APIRouter(routing.Router):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
include_in_schema: bool = True,
@@ -466,6 +515,7 @@ class APIRouter(routing.Router):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=["HEAD"],
operation_id=operation_id,
@@ -484,6 +534,7 @@ class APIRouter(routing.Router):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
include_in_schema: bool = True,
@@ -498,6 +549,7 @@ class APIRouter(routing.Router):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=["PATCH"],
operation_id=operation_id,
@@ -516,6 +568,7 @@ class APIRouter(routing.Router):
summary: str = None,
description: str = None,
response_description: str = "Successful Response",
responses: Dict[Union[int, str], Dict[str, Any]] = None,
deprecated: bool = None,
operation_id: str = None,
include_in_schema: bool = True,
@@ -530,6 +583,7 @@ class APIRouter(routing.Router):
summary=summary,
description=description,
response_description=response_description,
responses=responses or {},
deprecated=deprecated,
methods=["TRACE"],
operation_id=operation_id,

View File

@@ -1,3 +1,5 @@
from typing import Optional
from fastapi.openapi.models import APIKey, APIKeyIn
from fastapi.security.base import SecurityBase
from starlette.exceptions import HTTPException
@@ -10,42 +12,54 @@ class APIKeyBase(SecurityBase):
class APIKeyQuery(APIKeyBase):
def __init__(self, *, name: str, scheme_name: str = None):
self.model = APIKey(**{"in": APIKeyIn.query}, name=name)
def __init__(self, *, name: str, scheme_name: str = None, auto_error: bool = True):
self.model: APIKey = APIKey(**{"in": APIKeyIn.query}, name=name)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
async def __call__(self, request: Request) -> str:
async def __call__(self, request: Request) -> Optional[str]:
api_key: str = request.query_params.get(self.model.name)
if not api_key:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
if self.auto_error:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
return None
return api_key
class APIKeyHeader(APIKeyBase):
def __init__(self, *, name: str, scheme_name: str = None):
self.model = APIKey(**{"in": APIKeyIn.header}, name=name)
def __init__(self, *, name: str, scheme_name: str = None, auto_error: bool = True):
self.model: APIKey = APIKey(**{"in": APIKeyIn.header}, name=name)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
async def __call__(self, request: Request) -> str:
async def __call__(self, request: Request) -> Optional[str]:
api_key: str = request.headers.get(self.model.name)
if not api_key:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
if self.auto_error:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
return None
return api_key
class APIKeyCookie(APIKeyBase):
def __init__(self, *, name: str, scheme_name: str = None):
self.model = APIKey(**{"in": APIKeyIn.cookie}, name=name)
def __init__(self, *, name: str, scheme_name: str = None, auto_error: bool = True):
self.model: APIKey = APIKey(**{"in": APIKeyIn.cookie}, name=name)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
async def __call__(self, request: Request) -> str:
async def __call__(self, request: Request) -> Optional[str]:
api_key: str = request.cookies.get(self.model.name)
if not api_key:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
if self.auto_error:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
return None
return api_key

View File

@@ -1,5 +1,6 @@
import binascii
from base64 import b64decode
from typing import Optional
from fastapi.openapi.models import (
HTTPBase as HTTPBaseModel,
@@ -24,27 +25,38 @@ class HTTPAuthorizationCredentials(BaseModel):
class HTTPBase(SecurityBase):
def __init__(self, *, scheme: str, scheme_name: str = None):
def __init__(
self, *, scheme: str, scheme_name: str = None, auto_error: bool = True
):
self.model = HTTPBaseModel(scheme=scheme)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
async def __call__(self, request: Request) -> str:
async def __call__(
self, request: Request
) -> Optional[HTTPAuthorizationCredentials]:
authorization: str = request.headers.get("Authorization")
scheme, credentials = get_authorization_scheme_param(authorization)
if not (authorization and scheme and credentials):
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
if self.auto_error:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
return None
return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)
class HTTPBasic(HTTPBase):
def __init__(self, *, scheme_name: str = None, realm: str = None):
def __init__(
self, *, scheme_name: str = None, realm: str = None, auto_error: bool = True
):
self.model = HTTPBaseModel(scheme="basic")
self.scheme_name = scheme_name or self.__class__.__name__
self.realm = realm
self.auto_error = auto_error
async def __call__(self, request: Request) -> str:
async def __call__(self, request: Request) -> Optional[HTTPBasicCredentials]:
authorization: str = request.headers.get("Authorization")
scheme, param = get_authorization_scheme_param(authorization)
# before implementing headers with 401 errors, wait for: https://github.com/encode/starlette/issues/295
@@ -53,9 +65,12 @@ class HTTPBasic(HTTPBase):
status_code=HTTP_403_FORBIDDEN, detail="Invalid authentication credentials"
)
if not authorization or scheme.lower() != "basic":
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
if self.auto_error:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
return None
try:
data = b64decode(param).decode("ascii")
except (ValueError, UnicodeDecodeError, binascii.Error):
@@ -67,17 +82,29 @@ class HTTPBasic(HTTPBase):
class HTTPBearer(HTTPBase):
def __init__(self, *, bearerFormat: str = None, scheme_name: str = None):
def __init__(
self,
*,
bearerFormat: str = None,
scheme_name: str = None,
auto_error: bool = True
):
self.model = HTTPBearerModel(bearerFormat=bearerFormat)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
async def __call__(self, request: Request) -> str:
async def __call__(
self, request: Request
) -> Optional[HTTPAuthorizationCredentials]:
authorization: str = request.headers.get("Authorization")
scheme, credentials = get_authorization_scheme_param(authorization)
if not (authorization and scheme and credentials):
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
if self.auto_error:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
return None
if scheme.lower() != "bearer":
raise HTTPException(
status_code=HTTP_403_FORBIDDEN,
@@ -87,17 +114,23 @@ class HTTPBearer(HTTPBase):
class HTTPDigest(HTTPBase):
def __init__(self, *, scheme_name: str = None):
def __init__(self, *, scheme_name: str = None, auto_error: bool = True):
self.model = HTTPBaseModel(scheme="digest")
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
async def __call__(self, request: Request) -> str:
async def __call__(
self, request: Request
) -> Optional[HTTPAuthorizationCredentials]:
authorization: str = request.headers.get("Authorization")
scheme, credentials = get_authorization_scheme_param(authorization)
if not (authorization and scheme and credentials):
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
if self.auto_error:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
return None
if scheme.lower() != "digest":
raise HTTPException(
status_code=HTTP_403_FORBIDDEN,

View File

@@ -113,32 +113,49 @@ class OAuth2PasswordRequestFormStrict(OAuth2PasswordRequestForm):
class OAuth2(SecurityBase):
def __init__(
self, *, flows: OAuthFlowsModel = OAuthFlowsModel(), scheme_name: str = None
self,
*,
flows: OAuthFlowsModel = OAuthFlowsModel(),
scheme_name: str = None,
auto_error: bool = True
):
self.model = OAuth2Model(flows=flows)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
async def __call__(self, request: Request) -> str:
async def __call__(self, request: Request) -> Optional[str]:
authorization: str = request.headers.get("Authorization")
if not authorization:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
if self.auto_error:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
return None
return authorization
class OAuth2PasswordBearer(OAuth2):
def __init__(self, tokenUrl: str, scheme_name: str = None, scopes: dict = None):
def __init__(
self,
tokenUrl: str,
scheme_name: str = None,
scopes: dict = None,
auto_error: bool = True,
):
if not scopes:
scopes = {}
flows = OAuthFlowsModel(password={"tokenUrl": tokenUrl, "scopes": scopes})
super().__init__(flows=flows, scheme_name=scheme_name)
super().__init__(flows=flows, scheme_name=scheme_name, auto_error=auto_error)
async def __call__(self, request: Request) -> str:
async def __call__(self, request: Request) -> Optional[str]:
authorization: str = request.headers.get("Authorization")
scheme, param = get_authorization_scheme_param(authorization)
if not authorization or scheme.lower() != "bearer":
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
if self.auto_error:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
return None
return param

View File

@@ -1,3 +1,5 @@
from typing import Optional
from fastapi.openapi.models import OpenIdConnect as OpenIdConnectModel
from fastapi.security.base import SecurityBase
from starlette.exceptions import HTTPException
@@ -6,14 +8,20 @@ from starlette.status import HTTP_403_FORBIDDEN
class OpenIdConnect(SecurityBase):
def __init__(self, *, openIdConnectUrl: str, scheme_name: str = None):
def __init__(
self, *, openIdConnectUrl: str, scheme_name: str = None, auto_error: bool = True
):
self.model = OpenIdConnectModel(openIdConnectUrl=openIdConnectUrl)
self.scheme_name = scheme_name or self.__class__.__name__
self.auto_error = auto_error
async def __call__(self, request: Request) -> str:
async def __call__(self, request: Request) -> Optional[str]:
authorization: str = request.headers.get("Authorization")
if not authorization:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
if self.auto_error:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
else:
return None
return authorization

View File

@@ -30,6 +30,8 @@ def get_flat_models_from_routes(
body_fields_from_routes.append(route.body_field)
if route.response_field:
responses_from_routes.append(route.response_field)
if route.response_fields:
responses_from_routes.extend(route.response_fields.values())
flat_models = get_flat_models_from_fields(
body_fields_from_routes + responses_from_routes
)

View File

@@ -44,6 +44,7 @@ nav:
- Path Operation Configuration: 'tutorial/path-operation-configuration.md'
- Path Operation Advanced Configuration: 'tutorial/path-operation-advanced-configuration.md'
- Custom Response: 'tutorial/custom-response.md'
- Additional Responses: 'tutorial/additional-responses.md'
- Dependencies:
- First Steps: 'tutorial/dependencies/first-steps.md'
- Classes as Dependencies: 'tutorial/dependencies/classes-as-dependencies.md'
@@ -67,6 +68,7 @@ nav:
- WebSockets: 'tutorial/websockets.md'
- 'Events: startup - shutdown': 'tutorial/events.md'
- Debugging: 'tutorial/debugging.md'
- Extending OpenAPI: 'tutorial/extending-openapi.md'
- Concurrency and async / await: 'async.md'
- Deployment: 'deployment.md'
- Project Generation - Template: 'project-generation.md'

View File

@@ -58,7 +58,6 @@ all = [
"pyyaml",
"graphene",
"ujson",
"ujson",
"email_validator",
"uvicorn",
]

View File

@@ -3,4 +3,4 @@
set -e
set -x
bash scripts/test.sh --cov-report=html
bash scripts/test.sh --cov-report=html ${@}

View File

@@ -0,0 +1,52 @@
from fastapi import APIRouter, FastAPI
from starlette.testclient import TestClient
router = APIRouter()
sub_router = APIRouter()
app = FastAPI()
@sub_router.get("/")
def read_item():
return {"id": "foo"}
router.include_router(sub_router, prefix="/items")
app.include_router(router)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {
"/items/": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"summary": "Read Item Get",
"operationId": "read_item_items__get",
}
}
},
}
client = TestClient(app)
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_path_operation():
response = client.get("/items/")
assert response.status_code == 200
assert response.json() == {"id": "foo"}

View File

@@ -0,0 +1,75 @@
from typing import Optional
from fastapi import Depends, FastAPI, Security
from fastapi.security import APIKeyCookie
from pydantic import BaseModel
from starlette.testclient import TestClient
app = FastAPI()
api_key = APIKeyCookie(name="key", auto_error=False)
class User(BaseModel):
username: str
def get_current_user(oauth_header: Optional[str] = Security(api_key)):
if oauth_header is None:
return None
user = User(username=oauth_header)
return user
@app.get("/users/me")
def read_current_user(current_user: User = Depends(get_current_user)):
if current_user is None:
return {"msg": "Create an account first"}
else:
return current_user
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {
"/users/me": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"summary": "Read Current User Get",
"operationId": "read_current_user_users_me_get",
"security": [{"APIKeyCookie": []}],
}
}
},
"components": {
"securitySchemes": {
"APIKeyCookie": {"type": "apiKey", "name": "key", "in": "cookie"}
}
},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_security_api_key():
response = client.get("/users/me", cookies={"key": "secret"})
assert response.status_code == 200
assert response.json() == {"username": "secret"}
def test_security_api_key_no_key():
response = client.get("/users/me")
assert response.status_code == 200
assert response.json() == {"msg": "Create an account first"}

View File

@@ -0,0 +1,74 @@
from typing import Optional
from fastapi import Depends, FastAPI, Security
from fastapi.security import APIKeyHeader
from pydantic import BaseModel
from starlette.testclient import TestClient
app = FastAPI()
api_key = APIKeyHeader(name="key", auto_error=False)
class User(BaseModel):
username: str
def get_current_user(oauth_header: Optional[str] = Security(api_key)):
if oauth_header is None:
return None
user = User(username=oauth_header)
return user
@app.get("/users/me")
def read_current_user(current_user: Optional[User] = Depends(get_current_user)):
if current_user is None:
return {"msg": "Create an account first"}
return current_user
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {
"/users/me": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"summary": "Read Current User Get",
"operationId": "read_current_user_users_me_get",
"security": [{"APIKeyHeader": []}],
}
}
},
"components": {
"securitySchemes": {
"APIKeyHeader": {"type": "apiKey", "name": "key", "in": "header"}
}
},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_security_api_key():
response = client.get("/users/me", headers={"key": "secret"})
assert response.status_code == 200
assert response.json() == {"username": "secret"}
def test_security_api_key_no_key():
response = client.get("/users/me")
assert response.status_code == 200
assert response.json() == {"msg": "Create an account first"}

View File

@@ -0,0 +1,74 @@
from typing import Optional
from fastapi import Depends, FastAPI, Security
from fastapi.security import APIKeyQuery
from pydantic import BaseModel
from starlette.testclient import TestClient
app = FastAPI()
api_key = APIKeyQuery(name="key", auto_error=False)
class User(BaseModel):
username: str
def get_current_user(oauth_header: Optional[str] = Security(api_key)):
if oauth_header is None:
return None
user = User(username=oauth_header)
return user
@app.get("/users/me")
def read_current_user(current_user: Optional[User] = Depends(get_current_user)):
if current_user is None:
return {"msg": "Create an account first"}
return current_user
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {
"/users/me": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"summary": "Read Current User Get",
"operationId": "read_current_user_users_me_get",
"security": [{"APIKeyQuery": []}],
}
}
},
"components": {
"securitySchemes": {
"APIKeyQuery": {"type": "apiKey", "name": "key", "in": "query"}
}
},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_security_api_key():
response = client.get("/users/me?key=secret")
assert response.status_code == 200
assert response.json() == {"username": "secret"}
def test_security_api_key_no_key():
response = client.get("/users/me")
assert response.status_code == 200
assert response.json() == {"msg": "Create an account first"}

View File

@@ -0,0 +1,62 @@
from typing import Optional
from fastapi import FastAPI, Security
from fastapi.security.http import HTTPAuthorizationCredentials, HTTPBase
from starlette.testclient import TestClient
app = FastAPI()
security = HTTPBase(scheme="Other", auto_error=False)
@app.get("/users/me")
def read_current_user(
credentials: Optional[HTTPAuthorizationCredentials] = Security(security)
):
if credentials is None:
return {"msg": "Create an account first"}
return {"scheme": credentials.scheme, "credentials": credentials.credentials}
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {
"/users/me": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"summary": "Read Current User Get",
"operationId": "read_current_user_users_me_get",
"security": [{"HTTPBase": []}],
}
}
},
"components": {
"securitySchemes": {"HTTPBase": {"type": "http", "scheme": "Other"}}
},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_security_http_base():
response = client.get("/users/me", headers={"Authorization": "Other foobar"})
assert response.status_code == 200
assert response.json() == {"scheme": "Other", "credentials": "foobar"}
def test_security_http_base_no_credentials():
response = client.get("/users/me")
assert response.status_code == 200
assert response.json() == {"msg": "Create an account first"}

View File

@@ -0,0 +1,79 @@
from base64 import b64encode
from typing import Optional
from fastapi import FastAPI, Security
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from requests.auth import HTTPBasicAuth
from starlette.testclient import TestClient
app = FastAPI()
security = HTTPBasic(auto_error=False)
@app.get("/users/me")
def read_current_user(credentials: Optional[HTTPBasicCredentials] = Security(security)):
if credentials is None:
return {"msg": "Create an account first"}
return {"username": credentials.username, "password": credentials.password}
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {
"/users/me": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"summary": "Read Current User Get",
"operationId": "read_current_user_users_me_get",
"security": [{"HTTPBasic": []}],
}
}
},
"components": {
"securitySchemes": {"HTTPBasic": {"type": "http", "scheme": "basic"}}
},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_security_http_basic():
auth = HTTPBasicAuth(username="john", password="secret")
response = client.get("/users/me", auth=auth)
assert response.status_code == 200
assert response.json() == {"username": "john", "password": "secret"}
def test_security_http_basic_no_credentials():
response = client.get("/users/me")
assert response.status_code == 200
assert response.json() == {"msg": "Create an account first"}
def test_security_http_basic_invalid_credentials():
response = client.get(
"/users/me", headers={"Authorization": "Basic notabase64token"}
)
assert response.status_code == 403
assert response.json() == {"detail": "Invalid authentication credentials"}
def test_security_http_basic_non_basic_credentials():
payload = b64encode(b"johnsecret").decode("ascii")
auth_header = f"Basic {payload}"
response = client.get("/users/me", headers={"Authorization": auth_header})
assert response.status_code == 403
assert response.json() == {"detail": "Invalid authentication credentials"}

View File

@@ -0,0 +1,68 @@
from typing import Optional
from fastapi import FastAPI, Security
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from starlette.testclient import TestClient
app = FastAPI()
security = HTTPBearer(auto_error=False)
@app.get("/users/me")
def read_current_user(
credentials: Optional[HTTPAuthorizationCredentials] = Security(security)
):
if credentials is None:
return {"msg": "Create an account first"}
return {"scheme": credentials.scheme, "credentials": credentials.credentials}
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {
"/users/me": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"summary": "Read Current User Get",
"operationId": "read_current_user_users_me_get",
"security": [{"HTTPBearer": []}],
}
}
},
"components": {
"securitySchemes": {"HTTPBearer": {"type": "http", "scheme": "bearer"}}
},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_security_http_bearer():
response = client.get("/users/me", headers={"Authorization": "Bearer foobar"})
assert response.status_code == 200
assert response.json() == {"scheme": "Bearer", "credentials": "foobar"}
def test_security_http_bearer_no_credentials():
response = client.get("/users/me")
assert response.status_code == 200
assert response.json() == {"msg": "Create an account first"}
def test_security_http_bearer_incorrect_scheme_credentials():
response = client.get("/users/me", headers={"Authorization": "Basic notreally"})
assert response.status_code == 403
assert response.json() == {"detail": "Invalid authentication credentials"}

View File

@@ -0,0 +1,70 @@
from typing import Optional
from fastapi import FastAPI, Security
from fastapi.security import HTTPAuthorizationCredentials, HTTPDigest
from starlette.testclient import TestClient
app = FastAPI()
security = HTTPDigest(auto_error=False)
@app.get("/users/me")
def read_current_user(
credentials: Optional[HTTPAuthorizationCredentials] = Security(security)
):
if credentials is None:
return {"msg": "Create an account first"}
return {"scheme": credentials.scheme, "credentials": credentials.credentials}
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {
"/users/me": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"summary": "Read Current User Get",
"operationId": "read_current_user_users_me_get",
"security": [{"HTTPDigest": []}],
}
}
},
"components": {
"securitySchemes": {"HTTPDigest": {"type": "http", "scheme": "digest"}}
},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_security_http_digest():
response = client.get("/users/me", headers={"Authorization": "Digest foobar"})
assert response.status_code == 200
assert response.json() == {"scheme": "Digest", "credentials": "foobar"}
def test_security_http_digest_no_credentials():
response = client.get("/users/me")
assert response.status_code == 200
assert response.json() == {"msg": "Create an account first"}
def test_security_http_digest_incorrect_scheme_credentials():
response = client.get(
"/users/me", headers={"Authorization": "Other invalidauthorization"}
)
assert response.status_code == 403
assert response.json() == {"detail": "Invalid authentication credentials"}

View File

@@ -0,0 +1,254 @@
from typing import Optional
import pytest
from fastapi import Depends, FastAPI, Security
from fastapi.security import OAuth2
from fastapi.security.oauth2 import OAuth2PasswordRequestFormStrict
from pydantic import BaseModel
from starlette.testclient import TestClient
app = FastAPI()
reusable_oauth2 = OAuth2(
flows={
"password": {
"tokenUrl": "/token",
"scopes": {"read:users": "Read the users", "write:users": "Create users"},
}
},
auto_error=False,
)
class User(BaseModel):
username: str
def get_current_user(oauth_header: Optional[str] = Security(reusable_oauth2)):
if oauth_header is None:
return None
user = User(username=oauth_header)
return user
@app.post("/login")
def read_current_user(form_data: OAuth2PasswordRequestFormStrict = Depends()):
return form_data
@app.get("/users/me")
def read_current_user(current_user: Optional[User] = Depends(get_current_user)):
if current_user is None:
return {"msg": "Create an account first"}
return current_user
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {
"/login": {
"post": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
"summary": "Read Current User Post",
"operationId": "read_current_user_login_post",
"requestBody": {
"content": {
"application/x-www-form-urlencoded": {
"schema": {
"$ref": "#/components/schemas/Body_read_current_user"
}
}
},
"required": True,
},
}
},
"/users/me": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"summary": "Read Current User Get",
"operationId": "read_current_user_users_me_get",
"security": [{"OAuth2": []}],
}
},
},
"components": {
"schemas": {
"Body_read_current_user": {
"title": "Body_read_current_user",
"required": ["grant_type", "username", "password"],
"type": "object",
"properties": {
"grant_type": {
"title": "Grant_Type",
"pattern": "password",
"type": "string",
},
"username": {"title": "Username", "type": "string"},
"password": {"title": "Password", "type": "string"},
"scope": {"title": "Scope", "type": "string", "default": ""},
"client_id": {"title": "Client_Id", "type": "string"},
"client_secret": {"title": "Client_Secret", "type": "string"},
},
},
"ValidationError": {
"title": "ValidationError",
"required": ["loc", "msg", "type"],
"type": "object",
"properties": {
"loc": {
"title": "Location",
"type": "array",
"items": {"type": "string"},
},
"msg": {"title": "Message", "type": "string"},
"type": {"title": "Error Type", "type": "string"},
},
},
"HTTPValidationError": {
"title": "HTTPValidationError",
"type": "object",
"properties": {
"detail": {
"title": "Detail",
"type": "array",
"items": {"$ref": "#/components/schemas/ValidationError"},
}
},
},
},
"securitySchemes": {
"OAuth2": {
"type": "oauth2",
"flows": {
"password": {
"scopes": {
"read:users": "Read the users",
"write:users": "Create users",
},
"tokenUrl": "/token",
}
},
}
},
},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_security_oauth2():
response = client.get("/users/me", headers={"Authorization": "Bearer footokenbar"})
assert response.status_code == 200
assert response.json() == {"username": "Bearer footokenbar"}
def test_security_oauth2_password_other_header():
response = client.get("/users/me", headers={"Authorization": "Other footokenbar"})
assert response.status_code == 200
assert response.json() == {"username": "Other footokenbar"}
def test_security_oauth2_password_bearer_no_header():
response = client.get("/users/me")
assert response.status_code == 200
assert response.json() == {"msg": "Create an account first"}
required_params = {
"detail": [
{
"loc": ["body", "grant_type"],
"msg": "field required",
"type": "value_error.missing",
},
{
"loc": ["body", "username"],
"msg": "field required",
"type": "value_error.missing",
},
{
"loc": ["body", "password"],
"msg": "field required",
"type": "value_error.missing",
},
]
}
grant_type_required = {
"detail": [
{
"loc": ["body", "grant_type"],
"msg": "field required",
"type": "value_error.missing",
}
]
}
grant_type_incorrect = {
"detail": [
{
"loc": ["body", "grant_type"],
"msg": 'string does not match regex "password"',
"type": "value_error.str.regex",
"ctx": {"pattern": "password"},
}
]
}
@pytest.mark.parametrize(
"data,expected_status,expected_response",
[
(None, 422, required_params),
({"username": "johndoe", "password": "secret"}, 422, grant_type_required),
(
{"username": "johndoe", "password": "secret", "grant_type": "incorrect"},
422,
grant_type_incorrect,
),
(
{"username": "johndoe", "password": "secret", "grant_type": "password"},
200,
{
"grant_type": "password",
"username": "johndoe",
"password": "secret",
"scopes": [],
"client_id": None,
"client_secret": None,
},
),
],
)
def test_strict_login(data, expected_status, expected_response):
response = client.post("/login", data=data)
assert response.status_code == expected_status
assert response.json() == expected_response

View File

@@ -0,0 +1,71 @@
from typing import Optional
from fastapi import FastAPI, Security
from fastapi.security import OAuth2PasswordBearer
from starlette.testclient import TestClient
app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token", auto_error=False)
@app.get("/items/")
async def read_items(token: Optional[str] = Security(oauth2_scheme)):
if token is None:
return {"msg": "Create an account first"}
return {"token": token}
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {
"/items/": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"summary": "Read Items Get",
"operationId": "read_items_items__get",
"security": [{"OAuth2PasswordBearer": []}],
}
}
},
"components": {
"securitySchemes": {
"OAuth2PasswordBearer": {
"type": "oauth2",
"flows": {"password": {"scopes": {}, "tokenUrl": "/token"}},
}
}
},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_no_token():
response = client.get("/items")
assert response.status_code == 200
assert response.json() == {"msg": "Create an account first"}
def test_token():
response = client.get("/items", headers={"Authorization": "Bearer testtoken"})
assert response.status_code == 200
assert response.json() == {"token": "testtoken"}
def test_incorrect_token():
response = client.get("/items", headers={"Authorization": "Notexistent testtoken"})
assert response.status_code == 200
assert response.json() == {"msg": "Create an account first"}

View File

@@ -0,0 +1,80 @@
from typing import Optional
from fastapi import Depends, FastAPI, Security
from fastapi.security.open_id_connect_url import OpenIdConnect
from pydantic import BaseModel
from starlette.testclient import TestClient
app = FastAPI()
oid = OpenIdConnect(openIdConnectUrl="/openid", auto_error=False)
class User(BaseModel):
username: str
def get_current_user(oauth_header: Optional[str] = Security(oid)):
if oauth_header is None:
return None
user = User(username=oauth_header)
return user
@app.get("/users/me")
def read_current_user(current_user: Optional[User] = Depends(get_current_user)):
if current_user is None:
return {"msg": "Create an account first"}
return current_user
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {
"/users/me": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"summary": "Read Current User Get",
"operationId": "read_current_user_users_me_get",
"security": [{"OpenIdConnect": []}],
}
}
},
"components": {
"securitySchemes": {
"OpenIdConnect": {"type": "openIdConnect", "openIdConnectUrl": "/openid"}
}
},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_security_oauth2():
response = client.get("/users/me", headers={"Authorization": "Bearer footokenbar"})
assert response.status_code == 200
assert response.json() == {"username": "Bearer footokenbar"}
def test_security_oauth2_password_other_header():
response = client.get("/users/me", headers={"Authorization": "Other footokenbar"})
assert response.status_code == 200
assert response.json() == {"username": "Other footokenbar"}
def test_security_oauth2_password_bearer_no_header():
response = client.get("/users/me")
assert response.status_code == 200
assert response.json() == {"msg": "Create an account first"}

View File

View File

@@ -0,0 +1,116 @@
from starlette.testclient import TestClient
from additional_responses.tutorial001 import app
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {
"/items/{item_id}": {
"get": {
"responses": {
"404": {
"description": "Additional Response",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/Message"}
}
},
},
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/Item"}
}
},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
"summary": "Read Item Get",
"operationId": "read_item_items__item_id__get",
"parameters": [
{
"required": True,
"schema": {"title": "Item_Id", "type": "string"},
"name": "item_id",
"in": "path",
}
],
}
}
},
"components": {
"schemas": {
"Item": {
"title": "Item",
"required": ["id", "value"],
"type": "object",
"properties": {
"id": {"title": "Id", "type": "string"},
"value": {"title": "Value", "type": "string"},
},
},
"Message": {
"title": "Message",
"required": ["message"],
"type": "object",
"properties": {"message": {"title": "Message", "type": "string"}},
},
"ValidationError": {
"title": "ValidationError",
"required": ["loc", "msg", "type"],
"type": "object",
"properties": {
"loc": {
"title": "Location",
"type": "array",
"items": {"type": "string"},
},
"msg": {"title": "Message", "type": "string"},
"type": {"title": "Error Type", "type": "string"},
},
},
"HTTPValidationError": {
"title": "HTTPValidationError",
"type": "object",
"properties": {
"detail": {
"title": "Detail",
"type": "array",
"items": {"$ref": "#/components/schemas/ValidationError"},
}
},
},
}
},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_path_operation():
response = client.get("/items/foo")
assert response.status_code == 200
assert response.json() == {"id": "foo", "value": "there goes my hero"}
def test_path_operation_not_found():
response = client.get("/items/bar")
assert response.status_code == 404
assert response.json() == {"message": "Item not found"}

View File

@@ -0,0 +1,115 @@
import os
import shutil
from starlette.testclient import TestClient
from additional_responses.tutorial002 import app
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {
"/items/{item_id}": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {
"image/png": {},
"application/json": {
"schema": {"$ref": "#/components/schemas/Item"}
},
},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
"summary": "Read Item Get",
"operationId": "read_item_items__item_id__get",
"parameters": [
{
"required": True,
"schema": {"title": "Item_Id", "type": "string"},
"name": "item_id",
"in": "path",
},
{
"required": False,
"schema": {"title": "Img", "type": "boolean"},
"name": "img",
"in": "query",
},
],
}
}
},
"components": {
"schemas": {
"Item": {
"title": "Item",
"required": ["id", "value"],
"type": "object",
"properties": {
"id": {"title": "Id", "type": "string"},
"value": {"title": "Value", "type": "string"},
},
},
"ValidationError": {
"title": "ValidationError",
"required": ["loc", "msg", "type"],
"type": "object",
"properties": {
"loc": {
"title": "Location",
"type": "array",
"items": {"type": "string"},
},
"msg": {"title": "Message", "type": "string"},
"type": {"title": "Error Type", "type": "string"},
},
},
"HTTPValidationError": {
"title": "HTTPValidationError",
"type": "object",
"properties": {
"detail": {
"title": "Detail",
"type": "array",
"items": {"$ref": "#/components/schemas/ValidationError"},
}
},
},
}
},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_path_operation():
response = client.get("/items/foo")
assert response.status_code == 200
assert response.json() == {"id": "foo", "value": "there goes my hero"}
def test_path_operation_img():
shutil.copy("./docs/img/favicon.png", "./image.png")
response = client.get("/items/foo?img=1")
assert response.status_code == 200
assert response.headers["Content-Type"] == "image/png"
assert len(response.content)
os.remove("./image.png")

View File

@@ -0,0 +1,117 @@
from starlette.testclient import TestClient
from additional_responses.tutorial003 import app
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {
"/items/{item_id}": {
"get": {
"responses": {
"404": {
"description": "The item was not found",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/Message"}
}
},
},
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/Item"},
"example": {"id": "bar", "value": "The bar tenders"},
}
},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
"summary": "Read Item Get",
"operationId": "read_item_items__item_id__get",
"parameters": [
{
"required": True,
"schema": {"title": "Item_Id", "type": "string"},
"name": "item_id",
"in": "path",
}
],
}
}
},
"components": {
"schemas": {
"Item": {
"title": "Item",
"required": ["id", "value"],
"type": "object",
"properties": {
"id": {"title": "Id", "type": "string"},
"value": {"title": "Value", "type": "string"},
},
},
"Message": {
"title": "Message",
"required": ["message"],
"type": "object",
"properties": {"message": {"title": "Message", "type": "string"}},
},
"ValidationError": {
"title": "ValidationError",
"required": ["loc", "msg", "type"],
"type": "object",
"properties": {
"loc": {
"title": "Location",
"type": "array",
"items": {"type": "string"},
},
"msg": {"title": "Message", "type": "string"},
"type": {"title": "Error Type", "type": "string"},
},
},
"HTTPValidationError": {
"title": "HTTPValidationError",
"type": "object",
"properties": {
"detail": {
"title": "Detail",
"type": "array",
"items": {"$ref": "#/components/schemas/ValidationError"},
}
},
},
}
},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_path_operation():
response = client.get("/items/foo")
assert response.status_code == 200
assert response.json() == {"id": "foo", "value": "there goes my hero"}
def test_path_operation_not_found():
response = client.get("/items/bar")
assert response.status_code == 404
assert response.json() == {"message": "Item not found"}

View File

@@ -0,0 +1,118 @@
import os
import shutil
from starlette.testclient import TestClient
from additional_responses.tutorial004 import app
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {
"/items/{item_id}": {
"get": {
"responses": {
"404": {"description": "Item not found"},
"302": {"description": "The item was moved"},
"403": {"description": "Not enough privileges"},
"200": {
"description": "Successful Response",
"content": {
"image/png": {},
"application/json": {
"schema": {"$ref": "#/components/schemas/Item"}
},
},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
"summary": "Read Item Get",
"operationId": "read_item_items__item_id__get",
"parameters": [
{
"required": True,
"schema": {"title": "Item_Id", "type": "string"},
"name": "item_id",
"in": "path",
},
{
"required": False,
"schema": {"title": "Img", "type": "boolean"},
"name": "img",
"in": "query",
},
],
}
}
},
"components": {
"schemas": {
"Item": {
"title": "Item",
"required": ["id", "value"],
"type": "object",
"properties": {
"id": {"title": "Id", "type": "string"},
"value": {"title": "Value", "type": "string"},
},
},
"ValidationError": {
"title": "ValidationError",
"required": ["loc", "msg", "type"],
"type": "object",
"properties": {
"loc": {
"title": "Location",
"type": "array",
"items": {"type": "string"},
},
"msg": {"title": "Message", "type": "string"},
"type": {"title": "Error Type", "type": "string"},
},
},
"HTTPValidationError": {
"title": "HTTPValidationError",
"type": "object",
"properties": {
"detail": {
"title": "Detail",
"type": "array",
"items": {"$ref": "#/components/schemas/ValidationError"},
}
},
},
}
},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_path_operation():
response = client.get("/items/foo")
assert response.status_code == 200
assert response.json() == {"id": "foo", "value": "there goes my hero"}
def test_path_operation_img():
shutil.copy("./docs/img/favicon.png", "./image.png")
response = client.get("/items/foo?img=1")
assert response.status_code == 200
assert response.headers["Content-Type"] == "image/png"
assert len(response.content)
os.remove("./image.png")

View File

@@ -69,10 +69,11 @@ openapi_schema = {
"/items/": {
"get": {
"responses": {
"404": {"description": "Not found"},
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
},
"tags": ["items"],
"summary": "Read Items Get",
@@ -82,6 +83,7 @@ openapi_schema = {
"/items/{item_id}": {
"get": {
"responses": {
"404": {"description": "Not found"},
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
@@ -108,7 +110,38 @@ openapi_schema = {
"in": "path",
}
],
}
},
"put": {
"responses": {
"404": {"description": "Not found"},
"403": {"description": "Operation forbidden"},
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
"tags": ["custom", "items"],
"summary": "Update Item Put",
"operationId": "update_item_items__item_id__put",
"parameters": [
{
"required": True,
"schema": {"title": "Item_Id", "type": "string"},
"name": "item_id",
"in": "path",
}
],
},
},
},
"components": {
@@ -158,3 +191,15 @@ def test_get_path(path, expected_status, expected_response):
response = client.get(path)
assert response.status_code == expected_status
assert response.json() == expected_response
def test_put():
response = client.put("/items/foo")
assert response.status_code == 200
assert response.json() == {"item_id": "foo", "name": "The Fighters"}
def test_put_forbidden():
response = client.put("/items/bar")
assert response.status_code == 403
assert response.json() == {"detail": "You can only update the item: foo"}

View File

View File

@@ -0,0 +1,44 @@
from starlette.testclient import TestClient
from extending_openapi.tutorial001 import app
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {
"title": "Custom title",
"version": "2.5.0",
"description": "This is a very custom OpenAPI schema",
"x-logo": {"url": "https://fastapi.tiangolo.com/img/logo-margin/logo-teal.png"},
},
"paths": {
"/items/": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"summary": "Read Items Get",
"operationId": "read_items_items__get",
}
}
},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test():
response = client.get("/items/")
assert response.status_code == 200
assert response.json() == [{"name": "Foo"}]