Source::
Building the docker image::
Routing example
Routing with the APIRouter class.
from fastapi import APIRouter
router - APIRouter()
@router.get("hello")
async def say_hello() -> dict:
return {"message": "Hello!"}
from fastapi import APIRouter
todo_router = APIRouter()
todo_list =[]
@todo_router.post("/todo")
async def add_todo(todo:dict) -> dict:
todo_list.append(todo)
return {"message": "Todo added successfully"}
@todo_router.get("/todo")
async def retrieve_todo() -> dict:
return {"todos": todo_list}
The APIRouter class works in the same way as the FastAPI class does. However, uvicorn cannot use the APIRouter insance to serve the application, unlike the FastAPIs. Routes define using the APIRouter class are added to the fastapi instance to enable their visibility.
This is an example of what the routes look like using the FastAPI class as oppose to the APIRouter class above::
from fastapi import FastAPI
app = FastAPI()
person_repository = InMemoryPersonRepository()
@app.post("/person", response_model=Person)
def person() -> Person:
return Person().save(person_repository)
@app.get("/people", response_model=int)
def people() -> int:
return person_repository.total()
INCLUDE_ROUTER()
from todo import todo_router
from fastapi import FastAPI
from todo import todo_router
app = FastAPI()
@app.get("/")
async def welcome()-> dict:
return{
"message": "Hello World"
}
app.include_router(todo_router)
uvicorn api:app --port 8000 --reload
# Test your FastAPI endpoints
GET <http://127.0.0.1:8000/>
Accept: application/json
###
#GET <http://0.0.0.0:8000/> #this notation of using 0.0.0.0 instead of 127.0.0.1 is a curl notation.
#Accept: application/json
### Send GET request to get all todo items - should return an empty list
GET <http://127.0.0.1:8000/todo>
Accept: application/json
### Send POST request with json body to create a new todo item
POST <http://127.0.0.1:8000/todo>
Content-Type: application/json
{
"item": "Finish this tutorial"
}
### Send GET request to get all todo items - should return a list with one item
GET <http://127.0.0.1:8000/todo>
Accept: application/json
###
Validating request bodies using Pydantic models
In FastAPI, request bodies can be validated to ensure only defined data is sent.
This is crucial, as it serves to sanitize requests data and reduce malicious attacks’ risks. This process i know as validation.
A model in FastAPI is structured class that dictates how data should be received or parsed.
Models are created by subclassing Pydantic’s Model class.
What is PYDANTIC?
Pydantic is a Python library that handles data validation using Python-type annotations.
Models, when defined, are used as type hints for request body objects and request-response objects. In this chapter, we will only look at using Pydantic models for request bodies.
Example model::
from pydantic import BaseModel
class PacktBook(BaseModel):
id: int
Name: str
Publishers: str
Isbn: str
Creating a model with the required request body structure and assigning it as a type to the request body ensures that only the data fields present in the model are passed.
To ensure the request body only contains fields in the the code below, create a new model.py file and add the following code to it::
from pydantic import BaseModel
class Todo(BaseModel):
id: int
item: str
In the code above, we have created a Pydantic model that accepts only fields::
Let’s go ahead and use the model in our POST route. In todo.py import the mode::
from model import Todo
from fastapi import APIRouter
from model import Todo
todo_router = APIRouter()
todo_list = []
@todo_router.post("/todo")
async def add_todo(todo: **Todo**) -> dict:
todo_list.append(todo)
return {"message": "Todo added successfully"}
@todo_router.get("/todo")
async def retrieve_todo() -> dict:
return {"todos": todo_list}
### Send POST request with json body to create a empy todo item - should return an error
POST <http://127.0.0.1:8000/todo>
Content-Type: application/json
{
}
### Send POST request with json body to create a new todo item
POST <http://127.0.0.1:8000/todo>
Content-Type: application/json
{
"id": 2,
"item": "Validation models help with input types."
}
POST <http://127.0.0.1:8000/todo>
HTTP/1.1 422 Unprocessable Entity
date: Sun, 19 Mar 2023 23:01:03 GMT
server: uvicorn
content-length: 162
content-type: application/json
{
"detail": [
{
"loc": [
"body",
"id"
],
"msg": "field required",
"type": "value_error.missing"
},
{
"loc": [
"body",
"item"
],
"msg": "field required",
"type": "value_error.missing"
}
]
}
Response file saved.
> 2023-03-19T170104.422.json
Response code: 422 (Unprocessable Entity); Time: 30ms (30 ms); Content length: 162 bytes (162 B)
POST <http://127.0.0.1:8000/todo>
HTTP/1.1 200 OK
date: Sun, 19 Mar 2023 23:04:30 GMT
server: uvicorn
content-length: 37
content-type: application/json
{
"message": "Todo added successfully"
}
Response file saved.
> 2023-03-19T170431.200.json
Response code: 200 (OK); Time: 3ms (3 ms); Content length: 37 bytes (37 B)
POST <http://127.0.0.1:8000/todo>
Content-Type: application/json
Content-Length: 67
Connection: Keep-Alive
User-Agent: Apache-HttpClient/4.5.13 (Java/17.0.6)
Accept-Encoding: br,deflate,gzip,x-gzip
{
"id": 2,
"item": "Validation models help with input types."
}
Nested models
Pydantic models can also be nested, such as the following::
class Item(BaseModel):
item: str
status: str
class Todo(BaseModel):
id: int
item: Item
### Send POST request with json body with nested models
POST <http://127.0.0.1:8000/todo>
Content-Type: application/json
{
"id": 4,
"item": {
"item": "Nested models",
"status": "completed"
}
}
POST <http://127.0.0.1:8000/todo>
HTTP/1.1 200 OK
date: Sun, 19 Mar 2023 23:24:21 GMT
server: uvicorn
content-length: 37
content-type: application/json
{
"message": "Todo added successfully"
}
Response file saved.
> 2023-03-19T172422.200.json
Response code: 200 (OK); Time: 3ms (3 ms); Content length: 37 bytes (37 B)
Path and query parameters
from fastapi import APIRouter, Path
@todo_router.get("/todo/{todo_id}")
async def get_single_todo(todo_id: int = Path(..., title="The ID of the todo to retrieve.")) -> dict:
for todo in todo_list:
if todo.id == todo_id:
return {
"todo": todo
}
return {
"message": "Todo with supplied ID doesn't exist."
}
### Send GET request to get a todo item by id
GET <http://127.0.0.1:8000/todo/10>
Content-Type: application/json
###
GET <http://127.0.0.1:8000/todo/10>
HTTP/1.1 200 OK
date: Sun, 19 Mar 2023 23:55:43 GMT
server: uvicorn
content-length: 50
content-type: application/json
{
"message": "Todo with supplied ID doesn't exist."
}
Response file saved.
> 2023-03-19T175544.200.json
Response code: 200 (OK); Time: 4ms (4 ms); Content length: 50 bytes (50 B)
<aside> 💡 TIP - PATH(…, KWARGS)
</aside>
Chapter 3 - Response Model & Error Handling.
class TodoItems(BaseModel): #response_model inherits from BaseModel like regular models.
todos: List[TodoItem]
class Config:
schema_extra = {
"example": {
"todos":[{
"item": "Example schema 1!"
},
{
"item": "Example schema 2!"
}
]
}
}
from model import Todo, TodoItem, TodoItems
@todo_router.get("/todo", response_model=TodoItems)#return a DTO
async def retrieve_todo()-> dict:
return {
"todos": todo_list
}
from fastapi import APIRouter, Path, HTTPException, status
from model import Todo, TodoItem, TodoItems
todo_router = APIRouter()
todo_list = []
@todo_router.post("/todo", status_code=201)
async def add_todo(todo: Todo) -> dict:
todo_list.append(todo)
return {
"message": "Todo added successfully."
}
@todo_router.get("/todo", response_model=TodoItems)
async def retrieve_todo() -> dict:
return {
"todos": todo_list
}
@todo_router.get("/todo/{todo_id}")
async def get_single_todo(todo_id: int = Path(..., title="The ID of the todo to retrieve.")) -> dict:
for todo in todo_list:
if todo.id == todo_id:
return {
"todo": todo
}
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Todo with supplied ID doesn't exist",
)
@todo_router.put("/todo/{todo_id}")
async def update_todo(todo_data: TodoItem, todo_id: int = Path(..., title="The ID of the todo to be updated.")) -> dict:
for todo in todo_list:
if todo.id == todo_id:
todo.item = todo_data.item
return {
"message": "Todo updated successfully."
}
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Todo with supplied ID doesn't exist",
)
@todo_router.delete("/todo/{todo_id}")
async def delete_single_todo(todo_id: int) -> dict:
for index in range(len(todo_list)):
todo = todo_list[index]
if todo.id == todo_id:
todo_list.pop(index)
return {
"message": "Todo deleted successfully."
}
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Todo with supplied ID doesn't exist",
)
@todo_router.delete("/todo")
async def delete_all_todo() -> dict:
todo_list.clear()
return {
"message": "Todos deleted successfully."
}
What is a middleware?
A middleware is a function that acts as an intermediary between an operation. In web APIs, and web applications, a middleware serves as a mediator in a request-response operation.
Configuring CORS
Cross-Origin Resource Sharing (CORS) serves as a rule that prevents unregistered clients access to a resource.
When our web API is consumed by a frontend application, the browser will not allow cross-origin HTTP requests. This means that resources can only be accessed from the exact origin as permitted by the API.
work on this - Dependency Injection