Python High-performance Web Framework – A comprehensive guide to FastApi

A list,

FastAPI is a modern, fast (and high-performance) Web framework for building apis, using Python 3.6+ and based on standard Python type hints.

It has the following advantages:

• Fast: High performance on par with NodeJS and Go (thanks to Starlette and Pydantic)

• Efficient coding: increase the speed of feature development by 200% to 300%

• Fewer bugs: Approximately 40% fewer human errors.

• Smart: Excellent editor support. Automatic completion is available everywhere, reducing debugging time

• Simplicity: Designed to be easy to use and learn, the document takes less time to read

• Brevity: Minimizes code duplication. Rich functionality through different parameter declarations. Fewer bugs

• Robust: Produce usable level code. There are also interactive documents that are automatically generated

• Standardization: Related open standards based on (and fully compatible with) API: OpenAPI (formerly known as Swagger) and JSON Schema.

Second, the installation

pip install fastapi
Copy the code

ASGI server can use Uvicorn:

pip install uvicorn[standard]
Copy the code

3. Simple examples

Create a main.py file and write the following:

from typing import Optional

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: Optional[str] = None):
    return {"item_id": item_id, "q": q}
Copy the code

Start the server:

uvicorn main:app --reload
Copy the code

To access the URL: http://127.0.0.1:8000/items/5? Q = someQuery, you will see the following JSON response:

{"item_id": 5, "q": "somequery"}
Copy the code

To access the URL: http://127.0.0.1:8000/docs, you will see the automatically generated interactive API documentation, generated by Swagger UI:

Access the URL: http://127.0.0.1:8000/redoc, you will see another automatically generated documentation (generated by ReDoc) :

Fourth, the request

Declare path “parameters” or “variables” using the same syntax as Python formatting strings:

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/{item_id}")

async def read_item(item_id):

    return {"item_id": item_id}
Copy the code

The value of the path argument item_id is passed to your function as the argument item_id. When declaring other function parameters that are not path parameters, they are automatically interpreted as “query string” parameters:

from fastapi import FastAPI

app = FastAPI()

fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


@app.get("/items/")

async def read_item(skip: int = 0, limit: int = 10):

    return fake_items_db[skip : skip + limit]
Copy the code

The query string is a collection of key-value pairs located in the URL’s? After that, it is separated by an ampersand.

You can use Query to perform additional validation of a Query:

from typing import Optional


from fastapi import FastAPI, Query

app = FastAPI()


@app.get("/items/")
async def read_items(q: Optional[str] = Query(None, max_length=50)):
    results = {"items": [{"item_id": "Foo"}, {"item_id": "Bar"}]}
    if q:
        results.update({"q": q})
    return results
Copy the code

Query has the following columns:

• Min_length Minimum length
• Max_length Indicates the maximum length
• Regex re match
• QueryThe first parameter is the default value,.Presentation is required

Path and Query are used to validate Query fields.

And you can also declare numerical verification:

from fastapi import FastAPI, Path, Query app = FastAPI() @app.get("/items/{item_id}") async def read_items( *, item_id: int = Path(... , title="The ID of the item to get", ge=0, le=1000), q: str, size: float = Query(... , gt=0, lt=10.5)): results = {"item_id": item_id} if q: results.update({"q": q}) return resultsCopy the code

• Gt: greater than

• Ge: greater than or equal to

• Lt: less than

• Le: less than or equal to

Similarly, cookies:

from typing import Optional


from fastapi import Cookie, FastAPI


app = FastAPI()


@app.get("/items/")
async def read_items(ads_id: Optional[str] = Cookie(None)):
    return {"ads_id": ads_id}
Copy the code

As well as the Header:

from typing import Optional


from fastapi import FastAPI, Header


app = FastAPI()


@app.get("/items/")
async def read_items(user_agent: Optional[str] = Header(None)):
    return {"User-Agent": user_agent}
Copy the code

You can also group paths with tags:

from typing import Optional, Set

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []



@app.post("/items/", response_model=Item, tags=["items"])

async def create_item(item: Item):
    return item



@app.get("/items/", tags=["items"])

async def read_items():
    return [{"name": "Foo", "price": 42}]



@app.get("/users/", tags=["users"])

async def read_users():
    return [{"username": "johndoe"}] 
Copy the code

You can also set summary and description:

from typing import Optional, Set

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []


@app.post(
    "/items/",
    response_model=Item,

    summary="Create an item",

    description="Create an item with all the information, name, description, price, tax and a set of unique tags",

)
async def create_item(item: Item):
    return item
Copy the code

Multi-line comments:

from typing import Optional, Set

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []


@app.post("/items/", response_model=Item, summary="Create an item")
async def create_item(item: Item):

    """

    Create an item with all the information:



    - **name**: each item must have a name

    - **description**: a long description

    - **price**: required

    - **tax**: if the item doesn't have tax, you can omit this

    - **tags**: a set of unique tag strings for this item

    """

    return item
Copy the code

Obsolete route:

from fastapi import FastAPI

app = FastAPI()


@app.get("/items/", tags=["items"])
async def read_items():
    return [{"name": "Foo", "price": 42}]


@app.get("/users/", tags=["users"])
async def read_users():
    return [{"username": "johndoe"}]



@app.get("/elements/", tags=["items"], deprecated=True)

async def read_elements():
    return [{"item_id": "Foo"}]
Copy the code

Five, response,

Use the response_model parameter to declare the model for the response:

from typing import List, Optional

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: List[str] = []



@app.post("/items/", response_model=Item)

async def create_item(item: Item):
    return item
Copy the code

• response_model_exclude_unset=TrueThe response will not contain those default values, but only the values actually set
• response_model_includeWhat attributes to include
• response_model_excludeOmit certain attributes

The status_code parameter declares the HTTP status code for the response:

from fastapi import FastAPI

app = FastAPI()



@app.post("/items/", status_code=201)

async def create_item(name: str):
    return {"name": name}
Copy the code

For Form fields, use Form:

from fastapi import FastAPI, Form app = FastAPI() @app.post("/login/") async def login(username: str = Form(...) , password: str = Form(...) ): return {"username": username}Copy the code

File is used to define uploaded files on the client (to receive uploaded files, pre-installed [python-multipart](HTTPS %3A//andrew-d.github. IO /python-multipart/)) :

from fastapi import FastAPI, File, UploadFile app = FastAPI() @app.post("/files/") async def create_file(file: bytes = File(...) ): return {"file_size": len(file)} @app.post("/uploadfile/") async def create_upload_file(file: UploadFile = File(...) ): return {"filename": file.filename}Copy the code

To return an HTTP error response to the client, you can use HTTPException.

from fastapi import FastAPI, HTTPException


app = FastAPI()

items = {"foo": "The Foo Wrestlers"}


@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}
Copy the code

Set the response description with response_description:

from typing import Optional, Set

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = []


@app.post(
    "/items/",
    response_model=Item,
    summary="Create an item",

    response_description="The created item",

)
async def create_item(item: Item):
    """
    Create an item with all the information:

    - **name**: each item must have a name
    - **description**: a long description
    - **price**: required
    - **tax**: if the item doesn't have tax, you can omit this
    - **tags**: a set of unique tag strings for this item
    """
    return item
Copy the code

Compatible with JSON

In some cases, you may need to convert data (such as the Pydantic model) to JSON form, such as storing it in a database. In this case, you need to use the JSONable_encoder () method.

from datetime import datetime
from typing import Optional

from fastapi import FastAPI

from fastapi.encoders import jsonable_encoder

from pydantic import BaseModel

fake_db = {}


class Item(BaseModel):
    title: str
    timestamp: datetime
    description: Optional[str] = None


app = FastAPI()


@app.put("/items/{id}")
def update_item(id: str, item: Item):

    json_compatible_item_data = jsonable_encoder(item)

    fake_db[id] = json_compatible_item_data 
Copy the code

Dependency injection

FastAPI provides an easy-to-use, yet powerful dependency injection system that allows developers to easily integrate components into FastAPI.

What is dependency injection?

Dependency injection is a design pattern that eliminates dependencies between classes. Dependency injection is the process of putting dependent classes into a container and parsing out instances of those classes. The goal is to decouple the class.

Example:

from typing import Optional

from fastapi import Depends, FastAPI

app = FastAPI()



async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):

    return {"q": q, "skip": skip, "limit": limit}



@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons


@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
    return commons
Copy the code

The dependency in this example is expected to receive the following parameters:

• A type ofstrIs an optional query parameterq
• A type ofintIs an optional query parameterskip, the default value is0
• A type ofintIs an optional query parameterlimit, the default value is100

The dependency function then returns a dict containing those values.

Using Class as a dependency:

from typing import Optional

from fastapi import Depends, FastAPI

app = FastAPI()


fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]


class CommonQueryParams:
    def __init__(self, q: Optional[str] = None, skip: int = 0, limit: int = 100):
        self.q = q
        self.skip = skip
        self.limit = limit


@app.get("/items/")

async def read_items(commons: CommonQueryParams = Depends(CommonQueryParams)):

    response = {}
    if commons.q:
        response.update({"q": commons.q})
    items = fake_items_db[commons.skip : commons.skip + commons.limit]
    response.update({"items": items})
    return response
Copy the code

Using nested child dependencies:

from typing import Optional

from fastapi import Cookie, Depends, FastAPI

app = FastAPI()


def query_extractor(q: Optional[str] = None):
    return q


def query_or_cookie_extractor(
    q: str = Depends(query_extractor), last_query: Optional[str] = Cookie(None)
):
    if not q:
        return last_query
    return q


@app.get("/items/")

async def read_query(query_or_default: str = Depends(query_or_cookie_extractor)):

    return {"q_or_cookie": query_or_default}
Copy the code

Using dependencies in paths:

from fastapi import Depends, FastAPI, Header, HTTPException app = FastAPI() async def verify_token(x_token: str = Header(...) ): if x_token ! = "fake-super-secret-token": raise HTTPException(status_code=400, detail="X-Token header invalid") async def verify_key(x_key: str = Header(...) ): if x_key ! = "fake-super-secret-key": raise HTTPException(status_code=400, detail="X-Key header invalid") return x_key @app.get("/items/", dependencies=[Depends(verify_token), Depends(verify_key)]) async def read_items(): return [{"item": "Foo"}, {"item": "Bar"}]Copy the code

Global dependencies that can be applied for all _ path operations _ :

from fastapi import Depends, FastAPI, Header, HTTPException async def verify_token(x_token: str = Header(...) ): if x_token ! = "fake-super-secret-token": raise HTTPException(status_code=400, detail="X-Token header invalid") async def verify_key(x_key: str = Header(...) ): if x_key ! = "fake-super-secret-key": raise HTTPException(status_code=400, detail="X-Key header invalid") return x_key app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)]) @app.get("/items/") async def read_items(): return [{"item": "Portal Gun"}, {"item": "Plumbus"}] @app.get("/users/") async def read_users(): return [{"username": "Rick"}, {"username": "Morty"}]Copy the code

Eight, security

In many frameworks and systems, just dealing with security and authentication takes a lot of effort and code (in many cases, it can be 50% or more of all code written).

FastAPI provides a variety of tools to help you easily and quickly handle security in a standard manner without having to research and learn all the security specifications.

JWT stands for JSON Web Tokens.

It is a standard for encoding JSON objects as dense, long strings with no Spaces. The string looks like this:

eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.SflKxwRJ SMeKKF2QT4fwpMeJf36POk6yJV_adQssw5cCopy the code

It is not encrypted, so anyone can restore data from the string contents.

But it was signed. Therefore, when you receive a token issued by you, you can verify that the token is actually issued by you.

In this way, you can create a token that is valid for 1 week. Then when the user accesses again the next day with the token, you know that the user is still logged in.

After a week the token will expire and the user will not be authenticated and must log in again to get a new token. And if a user (or a third party) tries to modify the token to tamper with the expiration date, you will notice because of the signature mismatch.

OAuth2

OAuth2 is a specification that defines several ways to handle authentication and authorization.

It is a fairly broad specification that covers some complex usage scenarios.

It includes methods of using “third parties” for authentication.

This is the mechanism behind all systems that say “Log in using Facebook, Google, Twitter, GitHub”.

The following demonstrates how to use OAuth2 and JWT for user authentication.

from datetime import datetime, timedelta from typing import Optional from fastapi import Depends, FastAPI, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from jose import JWTError, jwt from passlib.context import CryptContext from pydantic import BaseModel # to get a string like this run: # openssl rand -hex 32 SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7" ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 fake_users_db = { "johndoe": { "username": "johndoe", "full_name": "John Doe", "email": "[email protected]", "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW", "disabled": False, } } class Token(BaseModel): access_token: str token_type: str class TokenData(BaseModel): username: Optional[str] = None class User(BaseModel): username: str email: Optional[str] = None full_name: Optional[str] = None disabled: Optional[bool] = None class UserInDB(User): hashed_password: str pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") app = FastAPI() def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password) def get_user(db, username: str): if username in db: user_dict = db[username] return UserInDB(**user_dict) def authenticate_user(fake_db, username: str, password: str): user = get_user(fake_db, username) if not user: return False if not verify_password(password, user.hashed_password): return False return user def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt async def get_current_user(token: str = Depends(oauth2_scheme)): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) username: str = payload.get("sub") if username is None: raise credentials_exception token_data = TokenData(username=username) except JWTError: raise credentials_exception user = get_user(fake_users_db, username=token_data.username) if user is None: raise credentials_exception return user async def get_current_active_user(current_user: User = Depends(get_current_user)): if current_user.disabled: raise HTTPException(status_code=400, detail="Inactive user") return current_user @app.post("/token", response_model=Token) async def login_for_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=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"sub":  user.username}, expires_delta=access_token_expires ) return {"access_token": access_token, "token_type": "bearer"} @app.get("/users/me/", response_model=User) async def read_users_me(current_user: User = Depends(get_current_active_user)): return current_user @app.get("/users/me/items/") async def read_own_items(current_user: User = Depends(get_current_active_user)): return [{"item_id": "Foo", "owner": current_user.username}]Copy the code

• OAuth2PasswordBearer: Access tokenUrl addresses, retrieve tokens and return them

• OAuth2PasswordRequestForm is a kind of dependency, declare the request form as follows:

  • username

  • password

  • An optional scope field is a large string of space-delimited strings

  • An optional grant_type

  • An optional client_id

  • An optional client_secret

Ix. Middleware

“Middleware” is a function that works before each request is processed by a specific _ path operation _, and before each response is returned.

To create middleware you can use the decorator @app.Middleware (” HTTP “) at the top of functions.

Middleware parameters receive the following parameters:

• request

• A function, call_next, takes request as an argument

  • This function willrequestPass to the correspondingPath to the operation
  • It will then return the value generated by the corresponding _ path operation _response

• You can then modify the response further before returning it

import time

from fastapi import FastAPI, Request

app = FastAPI()



@app.middleware("http")

async def add_process_time_header(request: Request, call_next):

    start_time = time.time()

    response = await call_next(request)

    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)

    return response
Copy the code

Cross domain Settings

You can use CORSMiddleware in FastAPI applications to configure cross-domains:

from fastapi import FastAPI

from fastapi.middleware.cors import CORSMiddleware


app = FastAPI()


origins = [
    "http://localhost.tiangolo.com",
    "https://localhost.tiangolo.com",
    "http://localhost",
    "http://localhost:8080",

]



app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],

)



@app.get("/")
async def main():
    return {"message": "Hello World"}
Copy the code

It supports the following parameters:

• Allow_origins – A list of sources that allow cross-domain requests. For example, [‘https://example.org’, ‘https://www.example.org’]. You can use [‘*’] to allow any source.

• Allow_origin_regex – A regular expression string that matches the source to allow cross-domain requests. Such as’ https://. * \. Example\.org.

• Allow_methods – A list of HTTP methods that allow cross-domain requests. The default is [‘GET’]. You can use [‘*’] to allow all standard methods.

• Allow_headers – A list of HTTP request headers that allow cross-domain requests. The default value is []. You can use [‘*’] to allow all headers. Accept, Accept-language, Content-Language, and Content-Type request headers always allow CORS requests.

• Allow_credentials – Indicates that cookies are supported in cross-domain requests. The default is False. Allow_origins cannot be set to [‘*’], and must specify the source.

• Expose_headers – Indicates the response headers that can be accessed by the browser. The default value is [].

• Max_age – Sets the maximum time, in seconds, for the browser to cache CORS responses. The default value is 600.

Eleven,APIRouter

Routes can also be manipulated using APIRouter:

from fastapi import APIRouter

router = APIRouter()



@router.get("/users/", tags=["users"])

async def read_users():
    return [{"username": "Rick"}, {"username": "Morty"}]



@router.get("/users/me", tags=["users"])

async def read_user_me():
    return {"username": "fakecurrentuser"}



@router.get("/users/{username}", tags=["users"])

async def read_user(username: str):
    return {"username": username}
Copy the code

Do the same for all paths:

from fastapi import APIRouter, Depends, HTTPException from .. dependencies import get_token_header router = APIRouter( prefix="/items", tags=["items"], dependencies=[Depends(get_token_header)], responses={404: {"description": "Not found"}}, ) fake_items_db = {"plumbus": {"name": "Plumbus"}, "gun": {"name": "Portal Gun"}} @router.get("/") async def read_items(): return fake_items_db @router.get("/{item_id}") async def read_item(item_id: str): if item_id not in fake_items_db: raise HTTPException(status_code=404, detail="Item not found") return {"name": fake_items_db[item_id]["name"], "item_id": item_id} @router.put( "/{item_id}", tags=["custom"], responses={403: {"description": "Operation forbidden"}}, ) async def update_item(item_id: str): if item_id ! = "plumbus": raise HTTPException( status_code=403, detail="You can only update the item: plumbus" ) return {"item_id": item_id, "name": "The great Plumbus"}Copy the code

This example simplifies the code by adding prefixes, labels, dependencies, and returns to all paths rather than declaring them individually on each path.

12. Background Tasks

Background Tasks are tasks that run immediately after the response is returned.

from fastapi import BackgroundTasks, FastAPI


app = FastAPI()


def write_notification(email: str, message=""):
    with open("log.txt", mode="w") as email_file:
        content = f"notification for {email}: {message}"
        email_file.write(content)


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="some notification")
    return {"message": "Notification sent in the background"}
Copy the code

Static files

First you need to install aiofiles:

pip install aiofiles 
Copy the code

Use:

from fastapi import FastAPI

from fastapi.staticfiles import StaticFiles


app = FastAPI()


app.mount("/static", StaticFiles(directory="static"), name="static")
Copy the code

14. Sub-applications

If you have two separate FastAPI apps, you can set one as the main app and one as a child app:

from fastapi import FastAPI


app = FastAPI()




@app.get("/app")

def read_main():
    return {"message": "Hello World from main app"}



subapi = FastAPI()


@subapi.get("/sub")
def read_sub():
    return {"message": "Hello World from sub API"}


app.mount("/subapi", subapi)
Copy the code

Xv. Agency

Root_path can be used to set up the proxy.

Using the command line:

uvicorn main:app --root-path /api/v1 
Copy the code

Or set it in code:

from fastapi import FastAPI, Request


app = FastAPI(root_path="/api/v1")



@app.get("/app")
def read_main(request: Request):
    return {"message": "Hello World", "root_path": request.scope.get("root_path")}
Copy the code

Use templates

You can use any template in FastAPI; a common choice is Jinja2.

pip install jinja2
Copy the code

Use:

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles

from fastapi.templating import Jinja2Templates


app = FastAPI()

app.mount("/static", StaticFiles(directory="static"), name="static")



templates = Jinja2Templates(directory="templates")



@app.get("/items/{id}", response_class=HTMLResponse)

async def read_item(request: Request, id: str):
    return templates.TemplateResponse("item.html", {"request": request, "id": id})
Copy the code

Templates /item.html:

<html>
<head>
    <title>Item Details</title>
    <link href="{{ url_for('static', path='/styles.css') }}" rel="stylesheet">
</head>
<body>

    <h1>Item ID: {{ id }}</h1>

</body>
</html>
Copy the code

17: web sockets

from fastapi import FastAPI, WebSocket from fastapi.responses import HTMLResponse app = FastAPI() html = """ <! DOCTYPE html> <html> <head> <title>Chat</title> </head> <body> <h1>WebSocket Chat</h1> <form action="" onsubmit="sendMessage(event)"> <input type="text" id="messageText" autocomplete="off"/> <button>Send</button> </form> <ul id='messages'> </ul> <script> var ws = new WebSocket("ws://localhost:8000/ws"); ws.onmessage = function(event) { var messages = document.getElementById('messages') var message = document.createElement('li') var content = document.createTextNode(event.data) message.appendChild(content) messages.appendChild(message) }; function sendMessage(event) { var input = document.getElementById("messageText") ws.send(input.value) input.value = '' event.preventDefault() } </script> </body> </html> """ @app.get("/") async def get(): return HTMLResponse(html) @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() while True: data = await websocket.receive_text() await websocket.send_text(f"Message text was: {data}")Copy the code

Startup-shutdown event

You can set the app’s start and close event callbacks:

from fastapi import FastAPI

app = FastAPI()

items = {}

@app.on_event("shutdown")
def shutdown_event():
    with open("log.txt", mode="a") as log:
        log.write("Application shutdown")


@app.on_event("startup")
async def startup_event():
    items["foo"] = {"name": "Fighters"}
    items["bar"] = {"name": "Tenders"}


@app.get("/items/{item_id}")
async def read_items(item_id: str):
    return items[item_id]
Copy the code