Compare commits

...

14 Commits

Author SHA1 Message Date
Sebastián Ramírez
5278314f2f 🔖 Bump version, new security features and bug fixes 2018-12-28 20:40:40 +04:00
Sebastián Ramírez
4a0316bcfe 🎨 Add missing type definition 2018-12-28 20:39:04 +04:00
Sebastián Ramírez
0393a093d3 Improve security utilities and add tests 2018-12-28 20:35:48 +04:00
Sebastián Ramírez
27f530a7ff 📝 Update docs, clarify what's a schema 2018-12-28 16:32:03 +04:00
Sebastián Ramírez
c3e5e65093 🎨 Fix missing format 2018-12-28 16:11:45 +04:00
Sebastián Ramírez
804ec460fc ⬆️ Add tests, fix issues and update Pydantic 2018-12-28 16:10:29 +04:00
Sebastián Ramírez
0125ea4f83 📝 Update tutorials 2018-12-28 16:03:54 +04:00
Sebastián Ramírez
216770118a ✏️ Fix typos 2018-12-27 17:25:39 +04:00
Sebastián Ramírez
a935d66b10 📝 Update docs about alternatives, inspiration and benchmarks 2018-12-27 17:14:46 +04:00
Sebastián Ramírez
dd2541bc97 📝 Improve explanation of request bodies 2018-12-26 19:01:15 +04:00
Sebastián Ramírez
098e629344 🔖 Bump version, after changes in OAuth2 utils 2018-12-24 20:21:28 +04:00
Sebastián Ramírez
bbe5f28b77 📝 Add docs for OAuth2 security 2018-12-24 20:20:48 +04:00
Sebastián Ramírez
4a0922ebab ♻️ Update OAuth2 class utilities to be dependencies 2018-12-24 20:20:21 +04:00
Sebastián Ramírez
8f16868c6a Add passlib and pyjwt to development dependencies 2018-12-24 20:19:05 +04:00
62 changed files with 2186 additions and 269 deletions

View File

@@ -28,7 +28,7 @@ FastAPI is a modern, fast (high-performance), web framework for building APIs wi
The key features are:
* **Fast**: Very high performance, on par with **NodeJS** and **Go** (thanks to Starlette and Pydantic).
* **Fast**: Very high performance, on par with **NodeJS** and **Go** (thanks to Starlette and Pydantic). [One of the fastest Python frameworks available](#performance).
* **Fast to code**: Increase the speed to develop features by about 200% to 300% *.
* **Less bugs**: Reduce about 40% of human (developer) induced errors. *
@@ -36,7 +36,7 @@ The key features are:
* **Easy**: Designed to be easy to use and learn. Less time reading docs.
* **Short**: Minimize code duplication. Multiple features from each parameter declaration. Less bugs.
* **Robust**: Get production-ready code. With automatic interactive documentation.
* **Standards-based**: Based on (and fully compatible with) the open standards for APIs: <a href="https://github.com/OAI/OpenAPI-Specification" target="_blank">OpenAPI</a> and <a href="http://json-schema.org/" target="_blank">JSON Schema</a>.
* **Standards-based**: Based on (and fully compatible with) the open standards for APIs: <a href="https://github.com/OAI/OpenAPI-Specification" target="_blank">OpenAPI</a> (previously known as Swagger) and <a href="http://json-schema.org/" target="_blank">JSON Schema</a>.
<small>* estimation based on tests on an internal development team, building production applications.</small>
@@ -291,7 +291,7 @@ Coming back to the previous code example, **FastAPI** will:
* Check that it has an optional attribute `is_offer`, that should be a `bool`, if present.
* All this would also work for deeply nested JSON objects.
* Convert from and to JSON automatically.
* Document everything as an OpenAPI schema, that can be used by:
* Document everything with OpenAPI, that can be used by:
* Interactive documentation sytems.
* Automatic client code generation systems, for many languages.
* Provide 2 interactive documentation web interfaces directly.
@@ -342,6 +342,11 @@ For a more complete example including more features, see the <a href="https://fa
* ...and more.
## Performance
Independent TechEmpower benchmarks show **FastAPI** applications running under Uvicorn as <a href="https://www.techempower.com/benchmarks/#section=test&runid=a979de55-980d-4721-a46f-77298b3f3923&hw=ph&test=fortune&l=zijzen-7" target="_blank">one of the fastest Python frameworks available</a>, only below Starlette and Uvicorn themselves (used internally by FastAPI). (*)
To understand more about it, see the section <a href="https://fastapi.tiangolo.com/benchmarks/" target="_blank">Benchmarks</a>.
## Optional Dependencies
@@ -366,7 +371,7 @@ Used by FastAPI / Starlette:
* <a href="http://www.uvicorn.org" target="_blank"><code>uvicorn</code></a> - for the server that loads and serves your application.
You can install all of these with `pip3 install fastapi[full]`.
You can install all of these with `pip3 install fastapi[all]`.
## License

333
docs/alternatives.md Normal file
View File

@@ -0,0 +1,333 @@
What inspired **FastAPI**, how it compares to other alternatives and what it learned from them.
## Intro
**FastAPI** wouldn't exist if not for the previous work of others.
There have been many tools created before that have helped inspire its creation.
I have been avoiding the creation of a new framework for several years. First I tried to solve all the features covered by **FastAPI** using many different frameworks, plug-ins and tools.
But at some point, there was no other option than creating something that provided all these features, taking the best ideas from previous tools, and combining them in the best way possible, using language features that weren't even available before (Python 3.6+ type hints).
## Previous tools
### <a href="https://www.djangoproject.com/" target="_blank">Django</a>
It's the most popular Python framework and is widely trusted. It is used to build systems like Instagram.
It's relatively tightly coupled with relational databases (like MySQL or PostgreSQL), so, having a NoSQL database (like Couchbase, MongoDB, Cassandra, etc) as the main store engine is not very easy.
It was created to generate the HTML in the backend, not to create APIs used by a modern frontend (like React, Vue.js and Angular) or by other systems (like <abbr title="Internet of Things">IoT</abbr> devices) communicating with it.
### <a href="https://www.django-rest-framework.org/" target="_blank">Django REST Framework</a>
Django REST framework was created to be a flexible toolkit for building Web APIs using Django underneath, to improve its API capabilities.
It is used by many companies including Mozilla, Red Hat and Eventbrite.
It was one of the first examples of **automatic API documentation**, and this was specifically one of the first ideas that inspired "the search for" **FastAPI**.
!!! note
Django REST Framework was created by Tom Christie. The same creator of Starlette and Uvicorn, on which **FastAPI** is based.
!!! check "Inspired **FastAPI** to"
Have an automatic API documentation web user interface.
### <a href="http://flask.pocoo.org/" target="_blank">Flask</a>
Flask is a "microframework", it doesn't include database integrations nor many of the things that come by default in Django.
This simplicity and flexibility allow doing things like using NoSQL databases as the main data storage system.
As it is very simple, it's relatively intuitive to learn, although the documentation gets somewhat technical at some points.
It is also commonly used for other applications that don't necessarily need a database, user management, or any of the many features that come pre-built in Django. Although many of these features can be added with plug-ins.
This decoupling of parts, and being a "microframework" that could be extended to cover exactly what is needed was a key feature that I wanted to keep.
Given the simplicity of Flask, it seemed like a good match for building APIs. The next thing to find was a "Django REST Framework" for Flask.
!!! check "Inspired **FastAPI** to"
Be a micro-framework. Making it easy to mix and match the tools and parts needed.
Have a simple and easy to use routing system.
### <a href="https://swagger.io/" target="_blank">Swagger</a> / <a href="https://github.com/OAI/OpenAPI-Specification/" target="_blank">OpenAPI</a>
The main feature I wanted from Django REST Framework was the automatic API documentation.
Then I found that there was a standard to document APIs, using JSON (or YAML, an extension of JSON) called Swagger.
And there was a web user interface for Swagger APIs already created. So, being able to generate Swagger documentation for an API would allow using this web user interface automatically.
At some point, Swagger was given to the Linux Foundation, to be renamed OpenAPI.
That's why when talking about version 2.0 it's common to say "Swagger", and for version 3+ "OpenAPI".
!!! check "Inspired **FastAPI** to"
Adopt and use an open standard for API specifications, instead of a custom schema.
And integrate standards-based user interface tools:
* <a href="https://github.com/swagger-api/swagger-ui" target="_blank">Swagger UI</a>
* <a href="https://github.com/Rebilly/ReDoc" target="_blank">ReDoc</a>
These two were chosen for being fairly popular and stable, but doing a quick search, you could find dozens of additional alternative user interfaces for OpenAPI (that you can use with **FastAPI**).
### Flask REST frameworks
There are several Flask REST frameworks, but after investing the time and work into investigating them, I found that many are discontinued or abandoned, with several standing issues that made them unfit.
## <a href="https://marshmallow.readthedocs.io/en/3.0/" target="_blank">Marshmallow</a>
One of the main features needed by API systems is data "<abbr title="also called marshalling, convertion">serialization</abbr>" which is taking data from the code (Python) and converting it into something that can be sent through the network. For example, converting an object containing data from a database into a JSON object. Converting `datetime` objects into strings, etc.
Another big feature needed by APIs is data validation, making sure that the data is valid, given certain parameters. For example, that some field is an `int`, and not some random string. This is especially useful for incoming data.
Without a data validation system, you would have to do all the checks by hand, in code.
These features are what Marshmallow was built to provide. It is a great library, and I have used it a lot before.
But it was created before there existed Python type hints. So, to define every <abbr title="the definition of how data should be formed">schema</abbr> you need to use specific utils and classes provided by Marshmallow.
!!! check "Inspired **FastAPI** to"
Use code to define "schemas" that provide data types and validation, automatically.
### <a href="https://webargs.readthedocs.io/en/latest/" target="_blank">Webargs</a>
Another big feature required by APIs is <abbr title="reading and converting to Python data">parsing</abbr> data from incoming requests.
Webargs is a tool that was made to provide that on top of several frameworks, including Flask.
It uses Marshmallow underneath to do the data validation. And it was created by the same guys.
It's a great tool and I have used it a lot too, before having **FastAPI**.
!!! info
Webargs was created by the same Marshmallow guys.
!!! check "Inspired **FastAPI** to"
Have automatic validation of incoming request data.
### <a href="https://apispec.readthedocs.io/en/stable/" target="_blank">APISpec</a>
Marshmallow and Webargs provide validation, parsing and serialization as plug-ins.
But documentation is still missing. Then APISpec was created.
It is a plug-in for many frameworks (and there's a plug-in for Starlette too).
The way it works is that you write the definition of the schema using YAML format inside the docstring of each function handling a route.
And it generates Swagger 2.0 schemas (OpenAPI 2.0).
That's how it works in Flask, Starlette, Responder, etc.
But then, we have again the problem of having a micro-syntax, inside of a Python string (a big YAML).
The editor can't help much with that. And if we modify parameters or Marshmallow schemas and forget to also modify that YAML docstring, the generated schema would be obsolete.
!!! info
APISpec was created by the same Marshmallow guys.
!!! check "Inspired **FastAPI** to"
Support the open standard for APIs, OpenAPI.
### <a href="https://flask-apispec.readthedocs.io/en/latest/" target="_blank">Flask-apispec</a>
It's a Flask plug-in, that ties together Webargs, Marshmallow and APISpec.
It uses the information from Webargs and Marshmallow to automatically generate Swagger 2.0 schemas, using APISpec.
It's a great tool, very under-rated. It should be way more popular than many Flask plug-ins out there. It might be due to its documentation being too concise and abstract.
This solved having to write YAML (another syntax) inside of Python docstrings.
This combination of Flask, Flask-apispec with Marshmallow and Webargs was my favorite backend stack until building **FastAPI**.
Using it led to the creation of several Flask full-stack generators. These are the main stack I (and several external teams) have been using up to now:
* <a href="https://github.com/tiangolo/full-stack" target="_blank">https://github.com/tiangolo/full-stack</a>
* <a href="https://github.com/tiangolo/full-stack-flask-couchbase" target="_blank">https://github.com/tiangolo/full-stack-flask-couchbase</a>
* <a href="https://github.com/tiangolo/full-stack-flask-couchdb" target="_blank">https://github.com/tiangolo/full-stack-flask-couchdb</a>
And these same full-stack generators were the base of the <a href="/project-generation/" target="_blank">**FastAPI** project generator</a>.
!!! info
Flask-apispec was created by the same Marshmallow guys.
!!! check "Inspired **FastAPI** to"
Generate the OpenAPI schema automatically, from the same code that defines serialization and validation.
### <a href="https://nestjs.com/" target="_blank">NestJS</a> (and <a href="https://angular.io/" target="_blank">Angular</a>)
This isn't even Python, NestJS is a JavaScript (TypeScript) NodeJS framework inspired by Angular.
It achieves something somewhat similar to what can be done with Flask-apispec.
It has an integrated dependency injection system, inspired by Angular two. It requires pre-registering the "injectables" (like all the other dependency injection systems I know), so, it adds to the verbosity and code repetition.
As the parameters are described with TypeScript types (similar to Python type hints), editor support is quite good.
But as TypeScript data is not preserved after compilation to JavaScript, it cannot rely on the types to define validation, serialization and documentation at the same time. Due to this and some design decisions, to get validation, serialization and automatic schema generation, it's needed to add decorators in many places. So, it becomes quite verbose.
It can't handle nested models very well. So, if the JSON body in the request is a JSON object that has inner fields that in turn are nested JSON objects, it cannot be properly documented and validated.
!!! check "Inspired **FastAPI** to"
Use Python types to have great editor support.
Have a powerful dependency injection system. Find a way to minimize code repetition.
### <a href="https://sanic.readthedocs.io/en/latest/" target="_blank">Sanic</a>
It was one of the first extremely fast Python frameworks based on `asyncio`. It was made to be very similar to Flask.
!!! note "Technical Details"
It used <a href="https://github.com/MagicStack/uvloop" target="_blank">`uvloop`</a> instead of the default Python `asyncio` loop. That's what made it so fast.
It <a href="https://github.com/huge-success/sanic/issues/761" target="_blank">still doesn't implement the ASGI spec for Python asynchronous web development</a>, but it clearly inspired Uvicorn and Starlette, that are currently faster than Sanic in open benchmarks.
!!! check "Inspired **FastAPI** to"
Find a way to have a crazy performance.
That's why **FastAPI** is based on Starlette, as it is the fastest framework available (tested by third-party benchmarks).
### <a href="https://moltenframework.com/" target="_blank">Molten</a>
I discovered Molten in the first stages of building **FastAPI**. And it has quite similar ideas:
* Based on Python type hints.
* Validation and documentation from these types.
* Dependency Injection system.
It doesn't use a data validation, serialization and documentation third-party library like Pydantic, it has its own. So, these data type definitions would not be reusable as easily.
It requires a little bit more verbose configurations. And as it is based on WSGI (instead of ASGI), it is not designed to take advantage of the high-performance provided by tools like Uvicorn, Starlette and Sanic.
The dependency injection system requires pre-registration of the dependencies and the dependencies are solved based on the declared types. So, it's not possible to declare more than one "component" that provides a certain type.
Routes are declared in a single place, using functions declared in other places (instead of using decorators that can be placed right on top of the function that handles the endpoint). This is closer to how Django does it than to how Flask (and Starlette) does it. It separates in the code things that are relatively tightly coupled.
!!! check "Inspired **FastAPI** to"
Define extra validations for data types using the "default" value of model attributes. This improves editor support, and it was not available in Pydantic before.
This actually inspired updating parts of Pydantic, to support the same validation declaration style (all this functionality is now already available in Pydantic).
### <a href="https://github.com/encode/apistar" target="_blank">APIStar</a> (<= 0.5)
Right before deciding to build **FastAPI** I found **APIStar** server. It had almost everything I was looking for and had a great design.
It was actually the first implementation of a framework using Python type hints to declare parameters and requests that I ever saw (before NestJS and Molten).
It had automatic data validation, data serialization and OpenAPI schema generation based on the same type hints in several places.
Body schema definitions didn't use the same Python type hints like Pydantic, it was a bit more similar to Marshmallow, so, editor support wouldn't be as good, but still, APIStar was the best available option.
It had the best performance benchmarks at the time (only surpassed by Starlette).
At first, it didn't have an automatic API documentation web UI, but I knew I could add Swagger UI to it.
It had a dependency injection system. It required pre-registration of components, as other tools discussed above. But still, it was a great feature.
I was never able to use it in a full project, as it didn't have security integration, so, I couldn't replace all the features I was having with the full-stack generators based on Flask-apispec. I had in my backlog of projects to create a pull request adding that functionality.
But then, the project's focus shifted.
It was no longer an API web framework, as the creator needed to focus on Starlette.
Now APIStar is a set of tools to validate OpenAPI specifications, not a web framework.
!!! info
APIStar was created by Tom Christie. The same guy that created:
* Django REST Framework
* Starlette (in which **FastAPI** is based)
* Uvicorn (used by Starlette and **FastAPI**)
!!! check "Inspired **FastAPI** to"
Exist.
The idea of declaring multiple things (data validation, serialization and documentation) with the same Python types, that at the same time provided great editor support, was something I considered a brilliant idea.
And after searching for a long time for a similar framework and testing many different alternatives, APIStar was the best option available.
Then APIStar stopped to exist as a server and Starlette was created, and was a new better foundation for such a system. That was the final inspiration to build **FastAPI**.
I consider **FastAPI** a "spiritual successor" to APIStar, while improving and increasing the features, typing system, and other parts, based on the learnings from all these previous tools.
## Used by **FastAPI**
### <a href="https://pydantic-docs.helpmanual.io/" target="_blank">Pydantic</a>
Pydantic is a library to define data validation, serialization and documentation (using JSON Schema) based on Python type hints.
That makes it extremely intuitive.
It is comparable to Marshmallow. Although it's faster than Marshmallow in benchmarks. And as it is based on the same Python type hints, the editor support is great.
!!! check "**FastAPI** uses it to"
Handle all the data validation, data serialization and automatic model documentation (based on JSON Schema).
**FastAPI** then takes that JSON Schema data and puts it in OpenAPI, apart from all the other things it does.
### <a href="https://www.starlette.io/" target="_blank">Starlette</a>
Starlette is a lightweight <abbr title="The new standard for building asynchronous Python web">ASGI</abbr> framework/toolkit, which is ideal for building high-performance asyncio services.
It is very simple and intuitive. It's designed to be easily extensible, and have modular components.
It has:
* Seriously impressive performance.
* WebSocket support.
* GraphQL support.
* In-process background tasks.
* Startup and shutdown events.
* Test client built on requests.
* CORS, GZip, Static Files, Streaming responses.
* Session and Cookie support.
* 100% test coverage.
* 100% type annotated codebase.
* Zero hard dependencies.
Starlette is currently the fastest Python framework tested. Only surpassed by Uvicorn, which is not a framework, but a server.
Starlette provides all the basic web microframework functionality.
But it doesn't provide automatic data validation, serialization or documentation.
That's one of the main things that **FastAPI** adds on top, all based on Python type hints (using Pydantic). That, plus the dependency injection system, security utilities, OpenAPI schema generation, etc.
!!! note "Technical Details"
ASGI is a new "standard" being developed by Django core team members. It is still not a "Python standard" (a PEP), although they are in the process of doing that.
Nevertheless, it is already being used as a "standard" by several tools. This greatly improves interoperability, as you could switch Uvicorn for any other ASGI server (like Daphne or Hypercorn), or you could add ASGI compatible tools, like `python-socketio`.
!!! check "**FastAPI** uses it to"
Handle all the core web parts. Adding features on top.
The class `FastAPI` itself inherits directly from the class `Starlette`.
So, anything that you can do with Starlette, you can do it directly with **FastAPI**, as it is basically Starlette on steroids.
### <a href="https://www.uvicorn.org/" target="_blank">Uvicorn</a>
Uvicorn is a lightning-fast ASGI server, built on uvloop and httptools.
It is not a web framework, but a server. For example, it doesn't provide tools for routing by paths. That's something that a framework like Starlette (or **FastAPI**) would provide on top.
It is the recommended server for Starlette and **FastAPI**.
!!! check "**FastAPI** recommends it as"
The main web server to run **FastAPI** applications.
You can combine it with Gunicorn, to have an asynchronous multiprocess server.
Check more details in the <a href="/deployment/" target="_blank">Deployment</a> section.
## Benchmarks and speed
To understand, compare, and see the difference between Uvicorn, Starlette and FastAPI, check the section about [Benchmarks](/benchmarks/).

32
docs/benchmarks.md Normal file
View File

@@ -0,0 +1,32 @@
Independent TechEmpower benchmarks show **FastAPI** applications running under Uvicorn as <a href="https://www.techempower.com/benchmarks/#section=test&runid=a979de55-980d-4721-a46f-77298b3f3923&hw=ph&test=fortune&l=zijzen-7" target="_blank">one of the fastest Python frameworks available</a>, only below Starlette and Uvicorn themselves (used internally by FastAPI). (*)
But when checking benchmarks and comparisons you should have the following in mind.
## Benchmarks and speed
When you check the benchmarks, it is common to see several tools of different types compared as equivalent.
Specifically, to see Uvicorn, Starlette and FastAPI compared together (among many other tools).
The simplest the problem solved by the tool, the better performance it will get. And most of the benchmarks don't test the additional features provided by the tool.
The hierarchy is like:
* **Uvicorn**: an ASGI server
* **Starlette**: (uses Uvicorn) a web microframework
* **FastAPI**: (uses Starlette) an API microframework with several additional features for building APIs, with data validation, etc.
* **Uvicorn**:
* Will have the best performance, as it doesn't have much extra code apart from the server itself.
* You wouldn't write an application in Uvicorn directly. That would mean that your code would have to include more or less, at least, all the code provided by Starlette (or **FastAPI**). And if you did that, your final application would have the same overhead as having used a framework and minimizing your app code and bugs.
* If you are comparing Uvicorn, compare it against Daphne, Hypercorn, uWSGI, etc. Application servers.
* **Starlette**:
* Will have the next best performance, after Uvicorn. In fact, Starlette uses Uvicorn to run. So, it probably can only get "slower" than Uvicorn by having to execute more code.
* But it provides you the tools to build simple web applications, with routing based on paths, etc.
* If you are comparing Starlette, compare it against Sanic, Flask, Django, etc. Web frameworks (or microframeworks).
* **FastAPI**:
* The same way that Starlette uses Uvicorn and cannot be faster than it, **FastAPI** uses Starlette, so it cannot be faster than it.
* FastAPI provides more features on top of Starlette. Features that you almost always need when building APIs, like data validation and serialization. And by using it, you get automatic documentation for free (the automatic documentation doesn't even add overhead to running applications, it is generated on startup).
* If you didn't use FastAPI and used Starlette directly (or another tool, like Sanic, Flask, Responder, etc) you would have to implement all the data validation and serialization yourself. So, your final application would still have the same overhead as if it was built using FastAPI. And in many cases, this data validation and serialization is the biggest amount of code written in applications.
* So, by using FastAPI you are saving development time, bugs, lines of code, and you would probably get the same performance (or better) you would if you didn't use it (as you would have to implement it all in your code).
* If you are comparing FastAPI, compare it against a web application framework (or set of tools) that provides data validation, serialization and documentation, like Flask-apispec, NestJS, Molten, etc. Frameworks with integrated automatic data validation, serialization and documentation.

View File

Binary file not shown.

After

Width:  |  Height:  |  Size: 60 KiB

View File

Binary file not shown.

After

Width:  |  Height:  |  Size: 86 KiB

View File

Binary file not shown.

After

Width:  |  Height:  |  Size: 108 KiB

View File

Binary file not shown.

After

Width:  |  Height:  |  Size: 155 KiB

View File

@@ -28,7 +28,7 @@ FastAPI is a modern, fast (high-performance), web framework for building APIs wi
The key features are:
* **Fast**: Very high performance, on par with **NodeJS** and **Go** (thanks to Starlette and Pydantic).
* **Fast**: Very high performance, on par with **NodeJS** and **Go** (thanks to Starlette and Pydantic). [One of the fastest Python frameworks available](#performance).
* **Fast to code**: Increase the speed to develop features by about 200% to 300% *.
* **Less bugs**: Reduce about 40% of human (developer) induced errors. *
@@ -291,7 +291,7 @@ Coming back to the previous code example, **FastAPI** will:
* Check that it has an optional attribute `is_offer`, that should be a `bool`, if present.
* All this would also work for deeply nested JSON objects.
* Convert from and to JSON automatically.
* Document everything as an OpenAPI schema, that can be used by:
* Document everything with OpenAPI, that can be used by:
* Interactive documentation sytems.
* Automatic client code generation systems, for many languages.
* Provide 2 interactive documentation web interfaces directly.
@@ -342,6 +342,11 @@ For a more complete example including more features, see the <a href="https://fa
* ...and more.
## Performance
Independent TechEmpower benchmarks show **FastAPI** applications running under Uvicorn as <a href="https://www.techempower.com/benchmarks/#section=test&runid=a979de55-980d-4721-a46f-77298b3f3923&hw=ph&test=fortune&l=zijzen-7" target="_blank">one of the fastest Python frameworks available</a>, only below Starlette and Uvicorn themselves (used internally by FastAPI). (*)
To understand more about it, see the section <a href="https://fastapi.tiangolo.com/benchmarks/" target="_blank">Benchmarks</a>.
## Optional Dependencies
@@ -366,7 +371,7 @@ Used by FastAPI / Starlette:
* <a href="http://www.uvicorn.org" target="_blank"><code>uvicorn</code></a> - for the server that loads and serves your application.
You can install all of these with `pip3 install fastapi[full]`.
You can install all of these with `pip3 install fastapi[all]`.
## License

View File

@@ -18,6 +18,6 @@ async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
response = {}
if commons.q:
response.update({"q": commons.q})
items = fake_items_db[commons.skip : commons.limit]
items = fake_items_db[commons.skip : commons.skip + commons.limit]
response.update({"items": items})
return response

View File

@@ -18,6 +18,6 @@ async def read_items(commons=Depends(CommonQueryParams)):
response = {}
if commons.q:
response.update({"q": commons.q})
items = fake_items_db[commons.skip : commons.limit]
items = fake_items_db[commons.skip : commons.skip + commons.limit]
response.update({"items": items})
return response

View File

@@ -18,6 +18,6 @@ async def read_items(commons: CommonQueryParams = Depends()):
response = {}
if commons.q:
response.update({"q": commons.q})
items = fake_items_db[commons.skip : commons.limit]
items = fake_items_db[commons.skip : commons.skip + commons.limit]
response.update({"items": items})
return response

View File

@@ -7,4 +7,4 @@ fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"
@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 100):
return fake_items_db[skip:limit]
return fake_items_db[skip : skip + limit]

View File

@@ -10,7 +10,14 @@ fake_users_db = {
"email": "johndoe@example.com",
"hashed_password": "fakehashedsecret",
"disabled": False,
}
},
"alice": {
"username": "alice",
"full_name": "Alice Wonderson",
"email": "alice@example.com",
"hashed_password": "fakehashedsecret2",
"disabled": True,
},
}
app = FastAPI()
@@ -64,12 +71,11 @@ async def get_current_active_user(current_user: User = Depends(get_current_user)
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
data = form_data.parse()
user_dict = fake_users_db.get(data.username)
user_dict = fake_users_db.get(form_data.username)
if not user_dict:
raise HTTPException(status_code=400, detail="Incorrect username or password")
user = UserInDB(**user_dict)
hashed_password = fake_hash_password(data.password)
hashed_password = fake_hash_password(form_data.password)
if not hashed_password == user.hashed_password:
raise HTTPException(status_code=400, detail="Incorrect username or password")

View File

@@ -1,5 +1,4 @@
from datetime import datetime, timedelta
from typing import Optional
import jwt
from fastapi import Depends, FastAPI, Security
@@ -23,7 +22,8 @@ fake_users_db = {
"username": "johndoe",
"full_name": "John Doe",
"email": "johndoe@example.com",
"password_hash": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
"hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
"disabled": False,
}
}
@@ -39,9 +39,9 @@ class TokenPayload(BaseModel):
class User(BaseModel):
username: str
email: Optional[str] = None
full_name: Optional[str] = None
disabled: Optional[bool] = None
email: str = None
full_name: str = None
disabled: bool = None
class UserInDB(User):
@@ -102,24 +102,21 @@ async def get_current_user(token: str = Security(oauth2_scheme)):
async def get_current_active_user(current_user: User = Depends(get_current_user)):
if not current_user.disabled:
if current_user.disabled:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
@app.post("/token", response_model=Token)
async def route_login_access_token(form_data: OAuth2PasswordRequestForm):
data = form_data.parse()
user = authenticate_user(fake_users_db, data.username, data.password)
async def route_login_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
user = authenticate_user(fake_users_db, form_data.username, form_data.password)
if not user:
raise HTTPException(status_code=400, detail="Incorrect email or password")
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
return {
"access_token": create_access_token(
data={"username": data.username}, expires_delta=access_token_expires
),
"token_type": "bearer",
}
access_token = create_access_token(
data={"username": form_data.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
@app.get("/users/me", response_model=User)

View File

@@ -1,4 +1,15 @@
To declare a request body, you use <a href="https://pydantic-docs.helpmanual.io/" target="_blank">Pydantic</a> models with all their power and benefits.
When you need to send data from a client (let's say, a browser) to your API, you send it as a **request body**.
A **request** body is data sent by the client to your API. A **response** body is the data your API sends to the client.
Your API almost always has to send a **response** body. But clients don't necessarily need to send **request** bodies all the time.
To declare a **request** body, you use <a href="https://pydantic-docs.helpmanual.io/" target="_blank">Pydantic</a> models with all their power and benefits.
!!! info
You cannot send a request body using a `GET` operation (HTTP method).
To send data, you have to use one of: `POST` (the more common), `PUT`, `DELETE` or `PATCH`.
## Import Pydantic's `BaseModel`

View File

@@ -7,7 +7,9 @@ This is especially the case for user models, because:
* The **database model** would probably need to have a hashed password.
!!! danger
Never store user's plaintext passwords. Always store a secure hash that you can then verify.
Never store user's plaintext passwords. Always store a "secure hash" that you can then verify.
If you don't know, you will learn what a "password hash" is in the <a href="/tutorial/security/simple-oauth2/#password-hashing" target="_blank">security chapters</a>.
## Multiple models
@@ -17,6 +19,39 @@ Here's a general idea of how the models could look like with their password fiel
{!./src/extra_models/tutorial001.py!}
```
#### About `**user_dict`
`UserInDB(**user_dict)` means:
Pass the keys and values of the `user_dict` directly as key-value arguments, equivalent to:
```Python
UserInDB(
username = user_dict["username"],
password = user_dict["password"],
email = user_dict["email"],
full_name = user_dict["full_name"],
)
```
And then adding the extra `hashed_password=hashed_password`, like in:
```Python
UserInDB(**user_in.dict(), hashed_password=hashed_password)
```
...ends up being like:
```Python
UserInDB(
username = user_dict["username"],
password = user_dict["password"],
email = user_dict["email"],
full_name = user_dict["full_name"],
hashed_password = hashed_password,
)
```
!!! warning
The supporting additional functions are just to demo a possible flow of the data, but they of course are not providing any real security.

View File

@@ -57,6 +57,32 @@ You will see the alternative automatic documentation (provided by <a href="https
![ReDoc](https://fastapi.tiangolo.com/img/index/index-02-redoc-simple.png)
### OpenAPI
**FastAPI** generates a "schema" with all your API using the **OpenAPI** standard for defining APIs.
#### "Schema"
A "schema" is a definition or description of something. Not the code that implements it, but just the abstract description.
#### API "schema"
In this case, OpenAPI is a specification that dictates how to define a schema of your API.
This OpenAPI schema would include your API paths, the posible parameters they take, etc.
#### Data "schema"
The term "schema" might also refer to the shape of some data, like a JSON content.
In that case, it would mean the JSON attributes, and data types they have, etc.
#### OpenAPI and JSON Schema
OpenAPI defines an API schema for your API. And that schema includes definitions (or "schemas") of the data sent and received by your API using **JSON Schema**, the standard for JSON data schemas.
#### Check it
If you are curious about how the raw OpenAPI schema looks like, it is just an automatically generated JSON with the descriptions of all your API.
You can see it directly at: <a href="http://127.0.0.1:8000/openapi.json" target="_blank">http://127.0.0.1:8000/openapi.json</a>.
@@ -84,6 +110,14 @@ It will show a JSON starting with something like:
...
```
#### What for?
This OpenAPI schema is what powers the 2 interactive documentation systems included.
And there are dozens of alternatives, all based on OpenAPI. You could easily add any of those alternatives to your application built with **FastAPI**.
You could also use it to generate code automatically, for clients that communicate with your API. For example, frontend, mobile or IoT applications.
## Recap, step by step
### Step 1: import `FastAPI`

View File

@@ -39,13 +39,13 @@ pip install fastapi[all]
This is what you would probably do once you want to deploy your application to production:
```bash
```
pip install fastapi
```
Also install `uvicorn` to work as the server:
```bash
```
pip install uvicorn
```

View File

@@ -1,5 +1,207 @@
Coming soon...
Now that we have all the security flow, let's make the application actually secure, using JWT tokens and secure password hashing.
```Python
This code is something you can actually use in your application, save the password hashes in your database, etc.
We are going to start from where we left in the previous chapter and increment it.
## About JWT
JWT means "JSON Web Tokens".
It's a standard to codify a JSON object in a long string.
It is not encrypted, so, anyone could recover the information from the contents.
But it's signed. So, when you receive a token that you emitted, you can verify that you actually emitted it.
That way, you can create a token with an expiration of, let's say, 1 week, and then, after a week, when the user comes back with the token, you know he's still signed into your system.
And after a week, the token will be expired. And if the user (or a third party) tried to modify the token to change the expiration, you would be able to discover it, because the signature would not match.
If you want to play with JWT tokens and see how they work, check <a href="https://jwt.io/" target="_blank">https://jwt.io</a>.
## Install `PyJWT`
We need to install `PyJWT` to generate and verity the JWT tokens in Python:
```bash
pip install pyjwt
```
## Password hashing
"Hashing" means converting some content (a password in this case) into a sequence of bytes (just a string) that look like gibberish.
Whenever you pass exactly the same content (exactly the same password) you get exactly the same gibberish.
But you cannot convert from the gibberish back to the password.
### What for?
If your database is stolen, the thief won't have your users' plaintext passwords, only the hashes.
So, the thief won't be able to try to use that password in another system (as many users use the same password everywhere, this would be dangerous).
## Install `passlib`
PassLib is a great Python package to handle password hashes.
It supports many secure hashing algorithms, and utilities to work with them.
The recommended algorithm is "Bcrypt".
So, install PassLib with Bcrypt:
```bash
pip install passlib[bcrypt]
```
!!! tip
With `passlib`, you could even configure it to be able to read passwords created by **Django** (among many others).
So, you would be able to, for example, share the same data from a Django application in a database with a FastAPI application. Or gradually migrate a Django application using the same database.
## Hash and verify the passwords
Import the tools we need from `passlib`.
Create a PassLib "context". This is what will be used to hash and verify passwords.
!!! tip
The PassLib context also has functionality to use different hashing algorithms, deprecate old ones, but allow verifying them, etc.
For example, you could use it to read and verify passwords generated by another system (like Django) but hash any new passwords with a different algorithm like Bcrypt.
And be compatible with all of them at the same time.
Create a utility function to hash a password coming from the user.
And another utility to verify if a received password matches the hash stored.
And another one to authenticate and return a user.
```Python hl_lines="7 51 58 59 62 63 72 73 74 75 76 77 78"
{!./src/security/tutorial004.py!}
```
!!! note
If you check the new (fake) database `fake_users_db`, you will see how the hashed password looks like now: `"$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW"`.
## Handle JWT tokens
Import the modules installed.
Create a random secret key that will be used to sign the JWT tokens.
To generate a secure random secret, key use the command:
```bash
openssl rand -hex 32
```
And copy the output to the variable `SECRET_KEY` (don't use the one in the example).
Create a variable `ALGORITHM` with the algorithm used to sign the JWT token and set it to `"HS256"`.
And another one for the `TOKEN_SUBJECT`, and set it to, for example, `"access"`.
Create a variable for the expiration of the token.
Define a Pydantic Model that will be used in the token endpoint for the response.
Create a utility function to generate a new access token.
```Python hl_lines="3 6 14 15 16 17 31 32 33 81 82 83 84 85 86 87 88 89"
{!./src/security/tutorial004.py!}
```
## Update the dependencies
Update `get_current_user` to receive the same token as before, but this time, using JWT tokens.
Decode the received token, verify it, and return the current user.
If the token is invalid, return an HTTP error right away.
```Python hl_lines="92 93 94 95 96 97 98 99 100 101"
{!./src/security/tutorial004.py!}
```
## Update the `/token` path operation
Create a `timedelta` with the expiration time of the token.
Create a real JWT access token and return it.
```Python hl_lines="115 116 117 118 119"
{!./src/security/tutorial004.py!}
```
## Check it
Run the server and go to the docs: <a href="http://127.0.0.1:8000/docs" target="_blank">http://127.0.0.1:8000/docs</a>.
You'll see the user interface like:
<img src="/img/tutorial/security/image07.png">
Authorize the application the same way as before.
Using the credentials:
Username: `johndoe`
Password: `secret`
!!! check
Notice that nowhere in the code is the plaintext password "`secret`", we only have the hashed version.
<img src="/img/tutorial/security/image08.png">
Call the endpoint `/users/me`, you will get the response as:
```JSON
{
"username": "johndoe",
"email": "johndoe@example.com",
"full_name": "John Doe",
"disabled": false
}
```
<img src="/img/tutorial/security/image09.png">
If you open the developer tools, you could see how the data sent and received is just the token, the password is only sent in the first request to authenticate the user:
<img src="/img/tutorial/security/image10.png">
!!! note
Notice the header `Authorization`, with a value that starts with `Bearer `.
## Advanced usage with `scopes`
We didn't use it in this example, but `Security` can receive a parameter `scopes`, as a list of strings.
It would describe the scopes required for a specific path operation, as different path operations might require different security scopes, even while using the same `OAuth2PasswordBearer` (or any of the other tools).
This only applies to OAuth2, and it might be, more or less, an advanced feature, but it is there, if you need to use it.
## Recap
This concludes our tour for the security features of **FastAPI**.
In almost any framework handling the security becomes a rather complex subject quite quickly.
Many packages that simplify it a lot have to make many compromises with the data model, database, and available features. And some of these packages that simplify things too much actually have security flaws underneath.
---
**FastAPI** doesn't make any compromise with any database, data model or tool.
It gives you all the flexibility to chose the ones that fit your project the best.
And you can use directly many well maintained and widely used packages like `passlib` and `pyjwt`, because **FastAPI** doesn't require any complex mechanisms to integrate external packages.
But it provides you the tools to simplify the process as much as possible without compromising flexibility, robustness or security.
And you can use secure, standard protocols like OAuth2 in a relatively simple way.

View File

@@ -16,13 +16,13 @@ But for the login path operation, we need to use these names to be compatible wi
The spec also states that the `username` and `password` must be sent as form data (so, no JSON here).
### `scopes`
### `scope`
The spec also says that the client can send another field of "`scopes`".
The spec also says that the client can send another form field "`scope`".
As a long string with all these "scopes" separated by spaces.
The form field name is `scope` (in singular), but it is actually a long string with "scopes" separated by spaces.
Each "scope" is just a string.
Each "scope" is just a string (without spaces).
They are normally used to declare specific security permissions, for exampe:
@@ -39,8 +39,6 @@ They are normally used to declare specific security permissions, for exampe:
For OAuth2 they are just strings.
And when using `scopes` it normally referes to a long string of "scopes" separated by spaces.
## Code to get the `username` and `password`
@@ -48,17 +46,17 @@ Now let's use the utilities provided by **FastAPI** to handle this.
### `OAuth2PasswordRequestForm`
First, import `OAuth2PasswordRequestForm`, and use it as the body declaration of the path `/token`:
First, import `OAuth2PasswordRequestForm`, and use it as a dependency with `Depends` for the path `/token`:
```Python hl_lines="2 63"
```Python hl_lines="2 73"
{!./src/security/tutorial003.py!}
```
`OAuth2PasswordRequestForm` declares a form body with:
`OAuth2PasswordRequestForm` is a class dependency that declares a form body with:
* The `username`.
* The `password`.
* An optional `scopes` field as a big string, composed of strings separated by spaces.
* An optional `scope` field as a big string, composed of strings separated by spaces.
* An optional `grant_type`.
!!! tip
@@ -69,24 +67,20 @@ First, import `OAuth2PasswordRequestForm`, and use it as the body declaration of
* An optional `client_id` (we don't need it for our example).
* An optional `client_secret` (we don't need it for our example).
### Parse and use the form data
`OAuth2PasswordRequestForm` provides a `.parse()` method that converts the `scopes` string into an actual list of strings.
We are not using `scopes` in this example, but the functionality is there if you need it.
### Use the form data
!!! tip
The `.parse()` method returns a Pydantic model `OAuth2PasswordRequestData`.
The instance of the dependency class `OAuth2PasswordRequestForm` won't have an attribute `scope` with the long string separated by spaces, instead, it will have a `scopes` attribute with the actual list of strings for each scope sent.
But you don't need to import it, your editor will know its type and provide you with completion and type checks automatically.
We are not using `scopes` in this example, but the functionality is there if you need it.
Now, get the user data from the (fake) database, using this `username`.
Now, get the user data from the (fake) database, using the `username` from the form field.
If there is no such user, we return an error saying "incorrect username or password".
For the error, we use the exception `HTTPException` provided by Starlette directly:
```Python hl_lines="4 64 65 66 67"
```Python hl_lines="4 74 75 76"
{!./src/security/tutorial003.py!}
```
@@ -98,9 +92,23 @@ Let's put that data in the Pydantic `UserInDB` model first.
You should never save plaintext passwords, so, we'll use the (fake) password hashing system.
If the password doesn't match, we return the same error.
If the passwords don't match, we return the same error.
```Python hl_lines="68 69 70 71"
#### Password hashing
"Hashing" means: converting some content (a password in this case) into a sequence of bytes (just a string) that look like gibberish.
Whenever you pass exactly the same content (exactly the same password) you get exactly the same gibberish.
But you cannot convert from the gibberish back to the password.
##### What for?
If your database is stolen, the thief won't have your users' plaintext passwords, only the hashes.
So, the thief won't be able to try to use that password in another system (as many users use the same password everywhere, this would be dangerous).
```Python hl_lines="77 78 79 80"
{!./src/security/tutorial003.py!}
```
@@ -112,11 +120,11 @@ Pass the keys and values of the `user_dict` directly as key-value arguments, equ
```Python
UserInDB(
username=user_dict["username"],
email=user_dict["email"],
full_name=user_dict["full_name"],
disabled=user_dict["disabled"],
hashed_password=user_dict["hashed_password"],
username = user_dict["username"],
email = user_dict["email"],
full_name = user_dict["full_name"],
disabled = user_dict["disabled"],
hashed_password = user_dict["hashed_password"],
)
```
@@ -124,18 +132,18 @@ UserInDB(
The response of the `token` endpoint must be a JSON object.
It should have a `token_type`. In our case, as we are using "Bearer" tokens, the token type should be `bearer`.
It should have a `token_type`. In our case, as we are using "Bearer" tokens, the token type should be "`bearer`".
And it should have an `access_token`, with a string containing our access token.
For this simple example, we are going to just be completely insecure and return the same `username` as the token.
!!! tip
In the next chapter, you will see a real secure implementation, with password hasing and JWT tokens.
In the next chapter, you will see a real secure implementation, with password hashing and JWT tokens.
But for now, let's focus on the specific details we need.
```Python hl_lines="73"
```Python hl_lines="82"
{!./src/security/tutorial003.py!}
```
@@ -151,7 +159,7 @@ Both of these dependencies will just return an HTTP error if the user doesn't ex
So, in our endpoint, we will only get a user if the user exists, was correctly authenticated, and is active:
```Python hl_lines="49 50 51 52 53 56 57 58 59 77"
```Python hl_lines="57 58 59 60 61 62 63 66 67 68 69 86"
{!./src/security/tutorial003.py!}
```
@@ -166,6 +174,7 @@ Click the "Authorize" button.
Use the credentials:
User: `johndoe`
Password: `secret`
<img src="/img/tutorial/security/image04.png">
@@ -200,6 +209,24 @@ If you click the lock icon and logout, and then try the same operation again, yo
}
```
### Inactive user
Now try with an inactive user, authenticate with:
User: `alice`
Password: `secret2`
And try to use the operation `GET` with the path `/users/me`.
You will get an "inactive user" error, like:
```JSON
{
"detail": "Inactive user"
}
```
## Recap
You now have the tools to implement a complete security system based on `username` and `password` for your API.

View File

@@ -1,6 +1,6 @@
"""FastAPI framework, high performance, easy to learn, fast to code, ready for production"""
__version__ = "0.1.13"
__version__ = "0.1.15"
from .applications import FastAPI
from .routing import APIRouter

View File

@@ -58,8 +58,6 @@ def get_flat_dependant(dependant: Dependant) -> Dependant:
security_schemes=dependant.security_requirements.copy(),
)
for sub_dependant in dependant.dependencies:
if sub_dependant is dependant:
raise ValueError("recursion", dependant.dependencies)
flat_sub = get_flat_dependant(sub_dependant)
flat_dependant.path_params.extend(flat_sub.path_params)
flat_dependant.query_params.extend(flat_sub.query_params)
@@ -99,7 +97,7 @@ def get_dependant(*, path: str, call: Callable, name: str = None) -> Dependant:
elif (
param.default == param.empty
or param.default is None
or type(param.default) in param_supported_types
or isinstance(param.default, param_supported_types)
) and (
param.annotation == param.empty
or lenient_issubclass(param.annotation, param_supported_types)
@@ -197,16 +195,12 @@ def add_param_to_body_fields(*, param: inspect.Parameter, dependant: Dependant)
dependant.body_params.append(field)
def is_coroutine_callable(call: Callable = None) -> bool:
if not call:
return False
def is_coroutine_callable(call: Callable) -> bool:
if inspect.isfunction(call):
return asyncio.iscoroutinefunction(call)
if inspect.isclass(call):
return False
call = getattr(call, "__call__", None)
if not call:
return False
return asyncio.iscoroutinefunction(call)
@@ -220,7 +214,8 @@ async def solve_dependencies(
request=request, dependant=sub_dependant, body=body
)
if sub_errors:
return {}, errors
errors.extend(sub_errors)
continue
assert sub_dependant.call is not None, "sub_dependant.call must be a function"
if is_coroutine_callable(sub_dependant.call):
solved = await sub_dependant.call(**sub_values)
@@ -244,7 +239,7 @@ async def solve_dependencies(
values.update(query_values)
values.update(header_values)
values.update(cookie_values)
errors = path_errors + query_errors + header_errors + cookie_errors
errors += path_errors + query_errors + header_errors + cookie_errors
if dependant.body_params:
body_values, body_errors = await request_body_to_args( # type: ignore # body_params checked above
dependant.body_params, body
@@ -301,7 +296,7 @@ async def request_body_to_args(
received_body = {}
for field in required_params:
value = received_body.get(field.alias)
if value is None:
if value is None or (isinstance(field.schema, params.Form) and value == ""):
if field.required:
errors.append(
ErrorWrapper(

View File

@@ -147,61 +147,65 @@ def get_openapi_path(
security_schemes: Dict[str, Any] = {}
definitions: Dict[str, Any] = {}
assert route.methods is not None, "Methods must be a list"
for method in route.methods:
operation = get_openapi_operation_metadata(route=route, method=method)
parameters: List[Dict] = []
flat_dependant = get_flat_dependant(route.dependant)
security_definitions, operation_security = get_openapi_security_definitions(
flat_dependant=flat_dependant
)
if operation_security:
operation.setdefault("security", []).extend(operation_security)
if security_definitions:
security_schemes.update(security_definitions)
all_route_params = get_openapi_params(route.dependant)
validation_definitions, operation_parameters = get_openapi_operation_parameters(
all_route_params=all_route_params
)
definitions.update(validation_definitions)
parameters.extend(operation_parameters)
if parameters:
operation["parameters"] = parameters
if method in METHODS_WITH_BODY:
request_body_oai = get_openapi_operation_request_body(
body_field=route.body_field, model_name_map=model_name_map
if route.include_in_schema:
for method in route.methods:
operation = get_openapi_operation_metadata(route=route, method=method)
parameters: List[Dict] = []
flat_dependant = get_flat_dependant(route.dependant)
security_definitions, operation_security = get_openapi_security_definitions(
flat_dependant=flat_dependant
)
if request_body_oai:
operation["requestBody"] = request_body_oai
if "ValidationError" not in definitions:
definitions["ValidationError"] = validation_error_definition
definitions[
"HTTPValidationError"
] = validation_error_response_definition
status_code = str(route.status_code)
response_schema = {"type": "string"}
if lenient_issubclass(route.content_type, JSONResponse):
if route.response_field:
response_schema, _ = field_schema(
route.response_field,
model_name_map=model_name_map,
ref_prefix=REF_PREFIX,
if operation_security:
operation.setdefault("security", []).extend(operation_security)
if security_definitions:
security_schemes.update(security_definitions)
all_route_params = get_openapi_params(route.dependant)
validation_definitions, operation_parameters = get_openapi_operation_parameters(
all_route_params=all_route_params
)
definitions.update(validation_definitions)
parameters.extend(operation_parameters)
if parameters:
operation["parameters"] = parameters
if method in METHODS_WITH_BODY:
request_body_oai = get_openapi_operation_request_body(
body_field=route.body_field, model_name_map=model_name_map
)
else:
response_schema = {}
content = {route.content_type.media_type: {"schema": response_schema}}
operation["responses"] = {
status_code: {"description": route.response_description, "content": content}
}
if all_route_params or route.body_field:
operation["responses"][str(HTTP_422_UNPROCESSABLE_ENTITY)] = {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {"$ref": REF_PREFIX + "HTTPValidationError"}
}
},
if request_body_oai:
operation["requestBody"] = request_body_oai
if "ValidationError" not in definitions:
definitions["ValidationError"] = validation_error_definition
definitions[
"HTTPValidationError"
] = validation_error_response_definition
status_code = str(route.status_code)
response_schema = {"type": "string"}
if lenient_issubclass(route.content_type, JSONResponse):
if route.response_field:
response_schema, _ = field_schema(
route.response_field,
model_name_map=model_name_map,
ref_prefix=REF_PREFIX,
)
else:
response_schema = {}
content = {route.content_type.media_type: {"schema": response_schema}}
operation["responses"] = {
status_code: {
"description": route.response_description,
"content": content,
}
}
path[method.lower()] = operation
if all_route_params or route.body_field:
operation["responses"][str(HTTP_422_UNPROCESSABLE_ENTITY)] = {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {"$ref": REF_PREFIX + "HTTPValidationError"}
}
},
}
path[method.lower()] = operation
return path, security_schemes, definitions

View File

@@ -1,6 +1,8 @@
from fastapi.openapi.models import APIKey, APIKeyIn
from fastapi.security.base import SecurityBase
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.status import HTTP_403_FORBIDDEN
class APIKeyBase(SecurityBase):
@@ -9,26 +11,41 @@ class APIKeyBase(SecurityBase):
class APIKeyQuery(APIKeyBase):
def __init__(self, *, name: str, scheme_name: str = None):
self.model = APIKey(in_=APIKeyIn.query, name=name)
self.model = APIKey(**{"in": APIKeyIn.query}, name=name)
self.scheme_name = scheme_name or self.__class__.__name__
async def __call__(self, requests: Request) -> str:
return requests.query_params.get(self.model.name)
async def __call__(self, request: Request) -> 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"
)
return api_key
class APIKeyHeader(APIKeyBase):
def __init__(self, *, name: str, scheme_name: str = None):
self.model = APIKey(in_=APIKeyIn.header, name=name)
self.model = APIKey(**{"in": APIKeyIn.header}, name=name)
self.scheme_name = scheme_name or self.__class__.__name__
async def __call__(self, requests: Request) -> str:
return requests.headers.get(self.model.name)
async def __call__(self, request: Request) -> str:
api_key: str = request.headers.get(self.model.name)
if not api_key:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
return api_key
class APIKeyCookie(APIKeyBase):
def __init__(self, *, name: str, scheme_name: str = None):
self.model = APIKey(in_=APIKeyIn.cookie, name=name)
self.model = APIKey(**{"in": APIKeyIn.cookie}, name=name)
self.scheme_name = scheme_name or self.__class__.__name__
async def __call__(self, requests: Request) -> str:
return requests.cookies.get(self.model.name)
async def __call__(self, request: Request) -> str:
api_key: str = request.cookies.get(self.model.name)
if not api_key:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
return api_key

View File

@@ -1,9 +1,26 @@
import binascii
from base64 import b64decode
from fastapi.openapi.models import (
HTTPBase as HTTPBaseModel,
HTTPBearer as HTTPBearerModel,
)
from fastapi.security.base import SecurityBase
from fastapi.security.utils import get_authorization_scheme_param
from pydantic import BaseModel
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.status import HTTP_403_FORBIDDEN
class HTTPBasicCredentials(BaseModel):
username: str
password: str
class HTTPAuthorizationCredentials(BaseModel):
scheme: str
credentials: str
class HTTPBase(SecurityBase):
@@ -12,16 +29,41 @@ class HTTPBase(SecurityBase):
self.scheme_name = scheme_name or self.__class__.__name__
async def __call__(self, request: Request) -> str:
return request.headers.get("Authorization")
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"
)
return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)
class HTTPBasic(HTTPBase):
def __init__(self, *, scheme_name: str = None):
def __init__(self, *, scheme_name: str = None, realm: str = None):
self.model = HTTPBaseModel(scheme="basic")
self.scheme_name = scheme_name or self.__class__.__name__
self.realm = realm
async def __call__(self, request: Request) -> str:
return request.headers.get("Authorization")
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
# unauthorized_headers = {"WWW-Authenticate": "Basic"}
invalid_user_credentials_exc = HTTPException(
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"
)
try:
data = b64decode(param).decode("ascii")
except (ValueError, UnicodeDecodeError, binascii.Error):
raise invalid_user_credentials_exc
username, separator, password = data.partition(":")
if not (separator):
raise invalid_user_credentials_exc
return HTTPBasicCredentials(username=username, password=password)
class HTTPBearer(HTTPBase):
@@ -30,7 +72,13 @@ class HTTPBearer(HTTPBase):
self.scheme_name = scheme_name or self.__class__.__name__
async def __call__(self, request: Request) -> str:
return request.headers.get("Authorization")
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"
)
return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)
class HTTPDigest(HTTPBase):
@@ -39,4 +87,10 @@ class HTTPDigest(HTTPBase):
self.scheme_name = scheme_name or self.__class__.__name__
async def __call__(self, request: Request) -> str:
return request.headers.get("Authorization")
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"
)
return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)

View File

@@ -1,24 +1,14 @@
from typing import List, Optional
from typing import Optional
from fastapi.openapi.models import OAuth2 as OAuth2Model, OAuthFlows as OAuthFlowsModel
from fastapi.params import Form
from fastapi.security.base import SecurityBase
from pydantic import BaseModel, Schema
from fastapi.security.utils import get_authorization_scheme_param
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.status import HTTP_403_FORBIDDEN
class OAuth2PasswordRequestData(BaseModel):
grant_type: str = "password"
username: str
password: str
scope: Optional[List[str]] = None
# Client ID and secret might come from headers
client_id: Optional[str] = None
client_secret: Optional[str] = None
class OAuth2PasswordRequestForm:
"""
This is a dependency class, use it like:
@@ -28,7 +18,7 @@ class OAuth2PasswordRequestForm:
data = form_data.parse()
print(data.username)
print(data.password)
for scope in data.scope:
for scope in data.scopes:
print(scope)
if data.client_id:
print(data.client_id)
@@ -40,8 +30,8 @@ class OAuth2PasswordRequestForm:
It creates the following Form request parameters in your endpoint:
grant_type: the OAuth2 spec says it is required and MUST be the fixed string "password".
Nevertheless, this model is permissive and allows not passing it. If you want to enforce it,
use instead the OAuth2PasswordRequestFormStrict model.
Nevertheless, this dependency class is permissive and allows not passing it. If you want to enforce it,
use instead the OAuth2PasswordRequestFormStrict dependency.
username: username string. The OAuth2 spec requires the exact field name "username".
password: password string. The OAuth2 spec requires the exact field name "password".
scope: Optional string. Several scopes (each one a string) separated by spaces. E.g.
@@ -50,9 +40,6 @@ class OAuth2PasswordRequestForm:
using HTTP Basic auth, as: client_id:client_secret
client_secret: optional string. OAuth2 recommends sending the client_id and client_secret (if any)
using HTTP Basic auth, as: client_id:client_secret
It has the method parse() that returns a model with all the same data and the scopes extracted as a list of strings.
"""
def __init__(
@@ -67,24 +54,61 @@ class OAuth2PasswordRequestForm:
self.grant_type = grant_type
self.username = username
self.password = password
self.scope = scope
self.scopes = scope.split()
self.client_id = client_id
self.client_secret = client_secret
def parse(self) -> OAuth2PasswordRequestData:
return OAuth2PasswordRequestData(
grant_type=self.grant_type,
username=self.username,
password=self.password,
scope=self.scope.split(),
client_id=self.client_id,
client_secret=self.client_secret,
)
class OAuth2PasswordRequestFormStrict(OAuth2PasswordRequestForm):
# The OAuth2 spec says it MUST have the value "password"
grant_type: str = Schema(..., regex="password")
"""
This is a dependency class, use it like:
@app.post("/login")
def login(form_data: Oauth2PasswordRequestFormStrict = Depends()):
data = form_data.parse()
print(data.username)
print(data.password)
for scope in data.scopes:
print(scope)
if data.client_id:
print(data.client_id)
if data.client_secret:
print(data.client_secret)
return data
It creates the following Form request parameters in your endpoint:
grant_type: the OAuth2 spec says it is required and MUST be the fixed string "password".
This dependency is strict about it. If you want to be permissive, use instead the
OAuth2PasswordRequestFormStrict dependency class.
username: username string. The OAuth2 spec requires the exact field name "username".
password: password string. The OAuth2 spec requires the exact field name "password".
scope: Optional string. Several scopes (each one a string) separated by spaces. E.g.
"items:read items:write users:read profile openid"
client_id: optional string. OAuth2 recommends sending the client_id and client_secret (if any)
using HTTP Basic auth, as: client_id:client_secret
client_secret: optional string. OAuth2 recommends sending the client_id and client_secret (if any)
using HTTP Basic auth, as: client_id:client_secret
"""
def __init__(
self,
grant_type: str = Form(..., regex="password"),
username: str = Form(...),
password: str = Form(...),
scope: str = Form(""),
client_id: Optional[str] = Form(None),
client_secret: Optional[str] = Form(None),
):
super().__init__(
grant_type=grant_type,
username=username,
password=password,
scope=scope,
client_id=client_id,
client_secret=client_secret,
)
class OAuth2(SecurityBase):
@@ -95,7 +119,12 @@ class OAuth2(SecurityBase):
self.scheme_name = scheme_name or self.__class__.__name__
async def __call__(self, request: Request) -> str:
return request.headers.get("Authorization")
authorization: str = request.headers.get("Authorization")
if not authorization:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
return authorization
class OAuth2PasswordBearer(OAuth2):
@@ -107,9 +136,9 @@ class OAuth2PasswordBearer(OAuth2):
async def __call__(self, request: Request) -> str:
authorization: str = request.headers.get("Authorization")
if not authorization or "Bearer " not in 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"
)
token = authorization.replace("Bearer ", "")
return token
return param

View File

@@ -1,6 +1,8 @@
from fastapi.openapi.models import OpenIdConnect as OpenIdConnectModel
from fastapi.security.base import SecurityBase
from starlette.exceptions import HTTPException
from starlette.requests import Request
from starlette.status import HTTP_403_FORBIDDEN
class OpenIdConnect(SecurityBase):
@@ -9,4 +11,9 @@ class OpenIdConnect(SecurityBase):
self.scheme_name = scheme_name or self.__class__.__name__
async def __call__(self, request: Request) -> str:
return request.headers.get("Authorization")
authorization: str = request.headers.get("Authorization")
if not authorization:
raise HTTPException(
status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
)
return authorization

View File

@@ -0,0 +1,8 @@
from typing import Tuple
def get_authorization_scheme_param(authorization_header_value: str) -> Tuple[str, str]:
if not authorization_header_value:
return "", ""
scheme, _, param = authorization_header_value.partition(" ")
return scheme, param

View File

@@ -59,6 +59,8 @@ nav:
- Concurrency and async / await: 'async.md'
- Deployment: 'deployment.md'
- Project Generation - Template: 'project-generation.md'
- Alternatives, Inspiration and Comparisons: 'alternatives.md'
- Benchmarks: 'benchmarks.md'
markdown_extensions:
- markdown.extensions.codehilite:

View File

@@ -20,7 +20,7 @@ classifiers = [
]
requires = [
"starlette >=0.9.7",
"pydantic >=0.16"
"pydantic >=0.17"
]
description-file = "README.md"
requires-python = ">=3.6"
@@ -44,8 +44,8 @@ doc = [
"markdown-include"
]
dev = [
"prospector",
"rope"
"pyjwt",
"passlib[bcrypt]"
]
all = [
"requests",

View File

@@ -1,6 +1,4 @@
from fastapi import Depends, FastAPI, Path, Query, Security
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel
from fastapi import FastAPI, Path, Query
app = FastAPI()
@@ -144,8 +142,6 @@ def get_path_param_le_ge_int(item_id: int = Path(..., le=3, ge=1)):
@app.get("/query")
def get_query(query):
if query is None:
return "foo bar"
return f"foo bar {query}"
@@ -158,8 +154,6 @@ def get_query_optional(query=None):
@app.get("/query/int")
def get_query_type(query: int):
if query is None:
return "foo bar"
return f"foo bar {query}"
@@ -184,30 +178,9 @@ def get_query_param(query=Query(None)):
@app.get("/query/param-required")
def get_query_param_required(query=Query(...)):
if query is None:
return "foo bar"
return f"foo bar {query}"
@app.get("/query/param-required/int")
def get_query_param_required_type(query: int = Query(...)):
if query is None:
return "foo bar"
return f"foo bar {query}"
reusable_oauth2b = OAuth2PasswordBearer(tokenUrl="/token")
class User(BaseModel):
username: str
def get_current_user(oauth_header: str = Security(reusable_oauth2b)):
user = User(username=oauth_header)
return user
@app.get("/security/oauth2b")
def read_current_user(current_user: User = Depends(get_current_user)):
return current_user

View File

@@ -1078,19 +1078,6 @@ openapi_schema = {
],
}
},
"/security/oauth2b": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"summary": "Read Current User Get",
"operationId": "read_current_user_security_oauth2b_get",
"security": [{"OAuth2PasswordBearer": []}],
}
},
},
"components": {
"schemas": {
@@ -1119,13 +1106,7 @@ openapi_schema = {
}
},
},
},
"securitySchemes": {
"OAuth2PasswordBearer": {
"type": "oauth2",
"flows": {"password": {"scopes": {}, "tokenUrl": "/token"}},
}
},
}
},
}
@@ -1134,6 +1115,7 @@ openapi_schema = {
"path,expected_status,expected_response",
[
("/api_route", 200, {"message": "Hello World"}),
("/non_decorated_route", 200, {"message": "Hello World"}),
("/nonexistent", 404, {"detail": "Not Found"}),
("/openapi.json", 200, openapi_schema),
],

View File

@@ -343,7 +343,7 @@ def test_head():
def test_options():
response = client.head("/items/foo")
response = client.options("/items/foo")
assert response.status_code == 200
assert response.headers["x-fastapi-item-id"] == "foo"

View File

@@ -0,0 +1,23 @@
from fastapi import APIRouter, FastAPI
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.testclient import TestClient
app = FastAPI()
router = APIRouter()
@router.route("/items/")
def read_items(request: Request):
return JSONResponse({"hello": "world"})
app.include_router(router)
client = TestClient(app)
def test_sub_router():
response = client.get("/items/")
assert response.status_code == 200
assert response.json() == {"hello": "world"}

25
tests/test_param_class.py Normal file
View File

@@ -0,0 +1,25 @@
from fastapi import FastAPI
from fastapi.params import Param
from starlette.testclient import TestClient
app = FastAPI()
@app.get("/items/")
def read_items(q: str = Param(None)):
return {"q": q}
client = TestClient(app)
def test_default_param_query_none():
response = client.get("/items/")
assert response.status_code == 200
assert response.json() == {"q": None}
def test_default_param_query():
response = client.get("/items/?q=foo")
assert response.status_code == 200
assert response.json() == {"q": "foo"}

View File

@@ -40,9 +40,19 @@ response_not_valid_int = {
("/query/int?query=42.5", 422, response_not_valid_int),
("/query/int?query=baz", 422, response_not_valid_int),
("/query/int?not_declared=baz", 422, response_missing),
("/query/int/optional", 200, "foo bar"),
("/query/int/optional?query=50", 200, "foo bar 50"),
("/query/int/optional?query=foo", 422, response_not_valid_int),
("/query/int/default", 200, "foo bar 10"),
("/query/int/default?query=50", 200, "foo bar 50"),
("/query/int/default?query=foo", 422, response_not_valid_int),
("/query/param", 200, "foo bar"),
("/query/param?query=50", 200, "foo bar 50"),
("/query/param-required", 422, response_missing),
("/query/param-required?query=50", 200, "foo bar 50"),
("/query/param-required/int", 422, response_missing),
("/query/param-required/int?query=50", 200, "foo bar 50"),
("/query/param-required/int?query=foo", 422, response_not_valid_int),
],
)
def test_get_path(path, expected_status, expected_response):

View File

@@ -1,25 +0,0 @@
from starlette.testclient import TestClient
from .main import app
client = TestClient(app)
def test_security_oauth2_password_bearer():
response = client.get(
"/security/oauth2b", headers={"Authorization": "Bearer footokenbar"}
)
assert response.status_code == 200
assert response.json() == {"username": "footokenbar"}
def test_security_oauth2_password_bearer_wrong_header():
response = client.get("/security/oauth2b", headers={"Authorization": "footokenbar"})
assert response.status_code == 403
assert response.json() == {"detail": "Not authenticated"}
def test_security_oauth2_password_bearer_no_header():
response = client.get("/security/oauth2b")
assert response.status_code == 403
assert response.json() == {"detail": "Not authenticated"}

View File

@@ -0,0 +1,68 @@
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")
class User(BaseModel):
username: str
def get_current_user(oauth_header: str = Security(api_key)):
user = User(username=oauth_header)
return user
@app.get("/users/me")
def read_current_user(current_user: User = Depends(get_current_user)):
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 == 403
assert response.json() == {"detail": "Not authenticated"}

View File

@@ -0,0 +1,68 @@
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")
class User(BaseModel):
username: str
def get_current_user(oauth_header: str = Security(api_key)):
user = User(username=oauth_header)
return user
@app.get("/users/me")
def read_current_user(current_user: User = Depends(get_current_user)):
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 == 403
assert response.json() == {"detail": "Not authenticated"}

View File

@@ -0,0 +1,68 @@
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")
class User(BaseModel):
username: str
def get_current_user(oauth_header: str = Security(api_key)):
user = User(username=oauth_header)
return user
@app.get("/users/me")
def read_current_user(current_user: User = Depends(get_current_user)):
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 == 403
assert response.json() == {"detail": "Not authenticated"}

View File

@@ -0,0 +1,247 @@
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"},
}
}
)
class User(BaseModel):
username: str
def get_current_user(oauth_header: str = Security(reusable_oauth2)):
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: User = Depends(get_current_user)):
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 == 403
assert response.json() == {"detail": "Not authenticated"}
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,74 @@
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")
class User(BaseModel):
username: str
def get_current_user(oauth_header: str = Security(oid)):
user = User(username=oauth_header)
return user
@app.get("/users/me")
def read_current_user(current_user: User = Depends(get_current_user)):
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 == 403
assert response.json() == {"detail": "Not authenticated"}

View File

@@ -28,7 +28,13 @@ openapi_schema = {
}
def test_openapi_scheme():
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_items():
response = client.get("/items/")
assert response.status_code == 200
assert response.json() == [{"name": "Foo"}]

View File

@@ -83,7 +83,7 @@ openapi_schema = {
}
def test_openapi_scheme():
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema

View File

@@ -101,7 +101,7 @@ openapi_schema = {
}
def test_openapi_scheme():
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema

View File

@@ -24,7 +24,7 @@ openapi_schema = {
}
def test_openapi_scheme():
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema

View File

@@ -35,7 +35,7 @@ html_contents = """
"""
def test_openapi_scheme():
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema

View File

@@ -0,0 +1,144 @@
import pytest
from starlette.testclient import TestClient
from dependencies.tutorial004 import app
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": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
"summary": "Read Items Get",
"operationId": "read_items_items__get",
"parameters": [
{
"required": False,
"schema": {"title": "Q", "type": "string"},
"name": "q",
"in": "query",
},
{
"required": False,
"schema": {"title": "Skip", "type": "integer", "default": 0},
"name": "skip",
"in": "query",
},
{
"required": False,
"schema": {"title": "Limit", "type": "integer", "default": 100},
"name": "limit",
"in": "query",
},
],
}
}
},
"components": {
"schemas": {
"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
@pytest.mark.parametrize(
"path,expected_status,expected_response",
[
(
"/items",
200,
{
"items": [
{"item_name": "Foo"},
{"item_name": "Bar"},
{"item_name": "Baz"},
]
},
),
(
"/items?q=foo",
200,
{
"items": [
{"item_name": "Foo"},
{"item_name": "Bar"},
{"item_name": "Baz"},
],
"q": "foo",
},
),
(
"/items?q=foo&skip=1",
200,
{"items": [{"item_name": "Bar"}, {"item_name": "Baz"}], "q": "foo"},
),
(
"/items?q=bar&limit=2",
200,
{"items": [{"item_name": "Foo"}, {"item_name": "Bar"}], "q": "bar"},
),
(
"/items?q=bar&skip=1&limit=1",
200,
{"items": [{"item_name": "Bar"}], "q": "bar"},
),
(
"/items?limit=1&q=bar&skip=1",
200,
{"items": [{"item_name": "Bar"}], "q": "bar"},
),
],
)
def test_get(path, expected_status, expected_response):
response = client.get(path)
assert response.status_code == expected_status
assert response.json() == expected_response

View File

@@ -74,7 +74,7 @@ openapi_schema = {
},
"process_after": {
"title": "Process_After",
"type": "string",
"type": "number",
"format": "time-delta",
},
},

View File

@@ -0,0 +1,36 @@
from starlette.testclient import TestClient
from path_operation_advanced_configuration.tutorial001 import app
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": "some_specific_id_you_define",
}
}
},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_get():
response = client.get("/items/")
assert response.status_code == 200
assert response.json() == [{"item_id": "Foo"}]

View File

@@ -0,0 +1,23 @@
from starlette.testclient import TestClient
from path_operation_advanced_configuration.tutorial002 import app
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
def test_get():
response = client.get("/items/")
assert response.status_code == 200
assert response.json() == [{"item_id": "Foo"}]

View File

@@ -0,0 +1,112 @@
from starlette.testclient import TestClient
from path_operation_configuration.tutorial005 import app
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {
"/items/": {
"post": {
"responses": {
"200": {
"description": "The created item",
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/Item"}
}
},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
"summary": "Create an item",
"description": "\n Create an item with all the information:\n \n * name: each item must have a name\n * description: a long description\n * price: required\n * tax: if the item doesn't have tax, you can omit this\n * tags: a set of unique tag strings for this item\n ",
"operationId": "create_item_items__post",
"requestBody": {
"content": {
"application/json": {
"schema": {"$ref": "#/components/schemas/Item"}
}
},
"required": True,
},
}
}
},
"components": {
"schemas": {
"Item": {
"title": "Item",
"required": ["name", "price"],
"type": "object",
"properties": {
"name": {"title": "Name", "type": "string"},
"price": {"title": "Price", "type": "number"},
"description": {"title": "Description", "type": "string"},
"tax": {"title": "Tax", "type": "number"},
"tags": {
"title": "Tags",
"uniqueItems": True,
"type": "array",
"items": {"type": "string"},
"default": [],
},
},
},
"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_query_params_str_validations():
response = client.post("/items/", json={"name": "Foo", "price": 42})
assert response.status_code == 200
assert response.json() == {
"name": "Foo",
"price": 42,
"description": None,
"tax": None,
"tags": [],
}

View File

@@ -0,0 +1,73 @@
import pytest
from starlette.testclient import TestClient
from path_operation_configuration.tutorial006 import app
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": {}}},
}
},
"tags": ["items"],
"summary": "Read Items Get",
"operationId": "read_items_items__get",
}
},
"/users/": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"tags": ["users"],
"summary": "Read Users Get",
"operationId": "read_users_users__get",
}
},
"/elements/": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"tags": ["items"],
"summary": "Read Elements Get",
"operationId": "read_elements_elements__get",
"deprecated": True,
}
},
},
}
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
@pytest.mark.parametrize(
"path,expected_status,expected_response",
[
("/items/", 200, [{"name": "Foo", "price": 42}]),
("/users/", 200, [{"username": "johndoe"}]),
("/elements/", 200, [{"item_id": "Foo"}]),
],
)
def test_query_params_str_validations(path, expected_status, expected_response):
response = client.get(path)
assert response.status_code == expected_status
assert response.json() == expected_response

View File

@@ -1,3 +1,4 @@
import pytest
from starlette.testclient import TestClient
from query_params_str_validations.tutorial010 import app
@@ -80,7 +81,42 @@ openapi_schema = {
}
def test_openapi_scheme():
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema
regex_error = {
"detail": [
{
"ctx": {"pattern": "^fixedquery$"},
"loc": ["query", "item-query"],
"msg": 'string does not match regex "^fixedquery$"',
"type": "value_error.str.regex",
}
]
}
@pytest.mark.parametrize(
"q_name,q,expected_status,expected_response",
[
(None, None, 200, {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}),
(
"item-query",
"fixedquery",
200,
{"items": [{"item_id": "Foo"}, {"item_id": "Bar"}], "q": "fixedquery"},
),
("q", "fixedquery", 200, {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}),
("item-query", "nonregexquery", 422, regex_error),
],
)
def test_query_params_str_validations(q_name, q, expected_status, expected_response):
url = "/items/"
if q_name and q:
url = f"{url}?{q_name}={q}"
response = client.get(url)
assert response.status_code == expected_status
assert response.json() == expected_response

View File

@@ -81,7 +81,7 @@ openapi_schema = {
}
def test_openapi_scheme():
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema

View File

@@ -81,7 +81,7 @@ openapi_schema = {
}
def test_openapi_scheme():
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema

View File

@@ -82,7 +82,7 @@ openapi_schema = {
}
def test_openapi_scheme():
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema

View File

@@ -96,7 +96,7 @@ openapi_schema = {
}
def test_openapi_scheme():
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema

View File

@@ -33,7 +33,7 @@ openapi_schema = {
}
def test_openapi_scheme():
def test_openapi_schema():
response = client.get("/openapi.json")
assert response.status_code == 200
assert response.json() == openapi_schema

View File

@@ -0,0 +1,171 @@
from starlette.testclient import TestClient
from security.tutorial003 import app
client = TestClient(app)
openapi_schema = {
"openapi": "3.0.2",
"info": {"title": "Fast API", "version": "0.1.0"},
"paths": {
"/token": {
"post": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
},
},
},
"summary": "Login Post",
"operationId": "login_token_post",
"requestBody": {
"content": {
"application/x-www-form-urlencoded": {
"schema": {"$ref": "#/components/schemas/Body_login"}
}
},
"required": True,
},
}
},
"/users/me": {
"get": {
"responses": {
"200": {
"description": "Successful Response",
"content": {"application/json": {"schema": {}}},
}
},
"summary": "Read Users Me Get",
"operationId": "read_users_me_users_me_get",
"security": [{"OAuth2PasswordBearer": []}],
}
},
},
"components": {
"schemas": {
"Body_login": {
"title": "Body_login",
"required": ["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": {
"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_login():
response = client.post("/token", data={"username": "johndoe", "password": "secret"})
assert response.status_code == 200
assert response.json() == {"access_token": "johndoe", "token_type": "bearer"}
def test_login_incorrect_password():
response = client.post(
"/token", data={"username": "johndoe", "password": "incorrect"}
)
assert response.status_code == 400
assert response.json() == {"detail": "Incorrect username or password"}
def test_login_incorrect_username():
response = client.post("/token", data={"username": "foo", "password": "secret"})
assert response.status_code == 400
assert response.json() == {"detail": "Incorrect username or password"}
def test_no_token():
response = client.get("/users/me")
assert response.status_code == 403
assert response.json() == {"detail": "Not authenticated"}
def test_token():
response = client.get("/users/me", headers={"Authorization": "Bearer johndoe"})
assert response.status_code == 200
assert response.json() == {
"username": "johndoe",
"full_name": "John Doe",
"email": "johndoe@example.com",
"hashed_password": "fakehashedsecret",
"disabled": False,
}
def test_incorrect_token():
response = client.get("/users/me", headers={"Authorization": "Bearer nonexistent"})
assert response.status_code == 400
assert response.json() == {"detail": "Invalid authentication credentials"}
def test_incorrect_token_type():
response = client.get(
"/users/me", headers={"Authorization": "Notexistent testtoken"}
)
assert response.status_code == 403
assert response.json() == {"detail": "Not authenticated"}
def test_inactive_user():
response = client.get("/users/me", headers={"Authorization": "Bearer alice"})
assert response.status_code == 400
assert response.json() == {"detail": "Inactive user"}