Fastapi 简明教程

http://127.0.0.1:8000/docs

{
   "openapi": "3.0.2",
   "info": {
      "title": "FastAPI",
      "version": "0.1.0"
   },
   "paths": {
      "/": {
         "get": {
            "summary": "Index",
            "operationId": "index__get",
            "responses": {
               "200": {
                  "description": "Successful Response",
                  "content": {
                     "application/json": {
                        "schema": {}
                     }
                  }
               }
            }
         }
      }
   }
}

pip3 install uvicorn(standard)

uvicorn main:app –reload

>>> def division(a, b):
   return a/b
>>> division(10, 4)
2.5
>>> division(10, 2.5)
4.0

>>> division("Python",5)
TypeError: unsupported operand type(s) for /: 'str' and 'int'

pip3 install mypy

def division(x:int, y:int) -> int:
   return (x//y)
a=division(10,2)
print (a)
b=division(5,2.5)
print (b)
c=division("Hello",10)
print (c)

C:\python37>mypy typechk.py
typechk.py:7: error: Argument 2 to "division" has incompatible
type "float"; expected "int"
typechk.py:10: error: Argument 1 to "division" has
incompatible type "str"; expected "int"
Found 2 errors in 1 file (checked 1 source file)

x: int = 3
y: float = 3.14
nm: str = 'abc'
married: bool = False
names: list = ['a', 'b', 'c']
marks: tuple = (10, 20, 30)
marklist: dict = {'a': 10, 'b': 20, 'c': 30}

from typing import List, Tuple, Dict
# following line declares a List object of strings.
# If violated, mypy shows error
cities: List[str] = ['Mumbai', 'Delhi', 'Chennai']
# This is Tuple with three elements respectively
# of str, int and float type)
employee: Tuple[str, int, float] = ('Ravi', 25, 35000)
# Similarly in the following Dict, the object key should be str
# and value should be of int type, failing which
# static type checker throws error
marklist: Dict[str, int] = {'Ravi': 61, 'Anil': 72}

class rectangle:
   def __init__(self, w:int, h:int) ->None:
      self.width=w
      self.height=h

def area(r:rectangle)->int:
   return r.width*r.height
r1=rectangle(10,20)
print ("area = ", area(r1))

http://localhost:8000/hello/TutorialsPoint

/hello/TutorialsPoint

from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def index():
   return {"message": "Hello World"}

from fastapi import FastAPI, Path
app = FastAPI()
@app.get("/hello/{name}")
async def hello(name:str=Path(...,min_length=3,
max_length=10)):
   return {"name": name}

{
   "detail": [
      {
         "loc": [
            "path",
            "name"
         ],
         "msg": "ensure this value has at most 10 characters",
         "type": "value_error.any_str.max_length",
         "ctx": {
            "limit_value": 10
         }
      }
   ]
}

from fastapi import FastAPI, Path
app = FastAPI()
@app.get("/hello/{name}/{age}")
async def hello(*, name: str=Path(...,min_length=3 , max_length=10), age: int = Path(..., ge=1, le=100)):
   return {"name": name, "age":age}

{
   "detail": [
      {
         "loc": [
            "path",
            "name"
         ],
         "msg": "ensure this value has at least 3 characters",
         "type": "value_error.any_str.min_length",
         "ctx": {
            "limit_value": 3
         }
      },
      {
         "loc": [
            "path",
            "age"
         ],
         "msg": "ensure this value is less than or equal to 100",
         "type": "value_error.number.not_le",
         "ctx": {
            "limit_value": 100
         }
      }
   ]
}

from fastapi import FastAPI, Path, Query
@app.get("/hello/{name}/{age}")
async def hello(*, name: str=Path(...,min_length=3 ,
max_length=10), \
      age: int = Path(..., ge=1, le=100), \
      percent:float=Query(..., ge=0, le=100)):
   return {"name": name, "age":age}

{"name":"Ravi","age":20}

from typing import List
from pydantic import BaseModel
class Student(BaseModel):
   id: int
   name :str
   subjects: List[str] = []

>>> data = {
   'id': 1,
   'name': 'Ravikumar',
   'subjects': ["Eng", "Maths", "Sci"],
}
>>> s1=Student(**data)
>>> print (s1)
id=1 name='Ravikumar' subjects=['Eng', 'Maths', 'Sci']
>>> s1
Student(id=1, name='Ravikumar', subjects=['Eng', 'Maths', 'Sci'])
>>> s1.dict()
{'id': 1, 'name': 'Ravikumar', 'subjects': ['Eng', 'Maths', 'Sci']}

>>> data = {
   'id': [1,2],
   'name': 'Ravikumar',
   'subjects': ["Eng", "Maths", "Sci"],
}
>>> s1=Student(**data)
Traceback (most recent call last):
   File "<pyshell#13>", line 1, in <module>
      s1=Student(**data)
   File "pydantic\main.py", line 406, in
pydantic.main.BaseModel.__init__
pydantic.error_wrappers.ValidationError: 1 validation error
for Student
id
   value is not a valid integer (type=type_error.integer)

from typing import List
from pydantic import BaseModel, Field
class Student(BaseModel):
   id: int
   name :str = Field(None, title="The description of the item", max_length=10)
   subjects: List[str] = []

>>> data = {
   'id': 1,
   'name': 'Ravikumar Sharma',
   'subjects': ["Eng", "Maths", "Sci"],
}
>>> s1=Student(**data)
Traceback (most recent call last):
   File "<pyshell#28>", line 1, in <module>
      s1=Student(**data)
   File "pydantic\main.py", line 406, in
pydantic.main.BaseModel.__init__
pydantic.error_wrappers.ValidationError: 1 validation error for Student
name
   ensure this value has at most 10 characters
   (type=value_error.any_str.max_length; limit_value=10)

import uvicorn
from fastapi import FastAPI
from typing import List
from pydantic import BaseModel, Field
app = FastAPI()
class Student(BaseModel):
   id: int
   name :str = Field(None, title="name of student", max_length=10)
   subjects: List[str] = []
@app.post("/students/")
async def student_data(s1: Student):
   return s1

import uvicorn
from fastapi import FastAPI, Body
@app.post("/students")
async def student_data(name:str=Body(...),
marks:int=Body(...)):
   return {"name":name,"marks": marks}

@app.post("/students/{college}")
async def student_data(college:str, age:int, student:Student):
   retval={"college":college, "age":age, **student.dict()}
   return retval

from fastapi.responses import HTMLResponse
from fastapi import FastAPI
app = FastAPI()
@app.get("/hello/")
async def hello():
   ret='''
<html>
<body>
<h2>Hello World!</h2>
</body>
</html>
'''
   return HTMLResponse(content=ret)

pip3 install jinja2

from fastapi.templating import Jinja2Templates

templates = Jinja2Templates(directory="templates")

<html>
<body>
<h2>Hello World!</h2>
</body>
</html>

from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fastapi import FastAPI, Request
app = FastAPI()
templates = Jinja2Templates(directory="templates")
@app.get("/hello/", response_class=HTMLResponse)
async def hello(request: Request):
   return templates.TemplateResponse("hello.html", {"request": request})

<html>
<body>
<h2>Hello {{name}} Welcome to FastAPI</h2>
</body>
</html>

from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
from fastapi import FastAPI, Request
app = FastAPI()
templates = Jinja2Templates(directory="templates")
@app.get("/hello/{name}", response_class=HTMLResponse)
async def hello(request: Request, name:str):
   return templates.TemplateResponse("hello.html", {"request": request, "name":name})

pip3 install aiofiles

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

<html>
   <body>
      <form action="/submit" method="POST">
         <h3>Enter User name</h3>
         <p><input type='text' name='nm'/></p>
         <h3>Enter Password</h3>
         <p><input type='password' name='pwd'/></p>
         <p><input type='submit' value='Login'/></p>
      </form>
   </body>
</html>

@app.get("/login/", response_class=HTMLResponse)
async def login(request: Request):
   return templates.TemplateResponse("login.html", {"request": request})

pip3 install python-multipart

from fastapi import Form

@app.post("/submit/")
async def submit(nm: str = Form(...), pwd: str = Form(...)):
   return {"username": nm}

from pydantic import BaseModel
class User(BaseModel):
   username:str
   password:str
@app.post("/submit/", response_model=User)
async def submit(nm: str = Form(...), pwd: str = Form(...)):
   return User(username=nm, password=pwd)

<html>
   <body>
      <form action="http://localhost:8000/uploader" method="POST" enctype="multipart/form-data">
         <input type="file" name="file" />
         <input type="submit"/>
      </form>
   </body>
</html>

from fastapi import FastAPI, File, UploadFile, Request
import uvicorn
import shutil
from fastapi.responses import HTMLResponse
from fastapi.templating import Jinja2Templates
app = FastAPI()
templates = Jinja2Templates(directory="templates")
@app.get("/upload/", response_class=HTMLResponse)
async def upload(request: Request):
   return templates.TemplateResponse("uploadfile.html", {"request": request})

from fastapi import FastAPI, File, UploadFile
import shutil
@app.post("/uploader/")
async def create_upload_file(file: UploadFile = File(...)):
   with open("destination.png", "wb") as buffer:
      shutil.copyfileobj(file.file, buffer)
   return {"filename": file.filename}

response.set_cookie(key, value)

from typing import Optional
from fastapi import FastAPI, Header
app = FastAPI()
@app.get("/headers/")
async def read_header(accept_language: Optional[str] = Header(None)):
   return {"Accept-Language": accept_language}

from fastapi import FastAPI
from fastapi.responses import JSONResponse
app = FastAPI()
@app.get("/rspheader/")
def set_rsp_headers():
   content = {"message": "Hello World"}
   headers = {"X-Web-Framework": "FastAPI", "Content-Language": "en-US"}
   return JSONResponse(content=content, headers=headers)

from fastapi import FastAPI
app = FastAPI()
@app.get("/user/")
async def user(id: str, name: str, age: int):
   return {"id": id, "name": name, "age": age}
@app.get("/admin/")
async def admin(id: str, name: str, age: int):
   return {"id": id, "name": name, "age": age}

async def dependency(id: str, name: str, age: int):
   return {"id": id, "name": name, "age": age}

@app.get("/user/")
async def user(dep: dict = Depends(dependency)):
   return dep

class dependency:
   def __init__(self, id: str, name: str, age: int):
      self.id = id
      self.name = name
      self.age = age

@app.get("/user/")
async def user(dep: dependency = Depends(dependency)):
   return dep
@app.get("/admin/")
async def admin(dep: dependency = Depends(dependency)):
   return dep

async def validate(dep: dependency = Depends(dependency)):
   if dep.age > 18:
      raise HTTPException(status_code=400, detail="You are not eligible")
@app.get("/user/", dependencies=[Depends(validate)])
async def user():
   return {"message": "You are eligible"}

async def get_db():
   db = DBSession()
      try:
         yield db
      finally:
            db.close()

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
origins = [
   "http://192.168.211.:8000",
   "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"}

from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
data = []
class Book(BaseModel):
   id: int
   title: str
   author: str
   publisher: str

@app.post("/book")
def add_book(book: Book):
   data.append(book.dict())
   return data

@app.get("/list")
def get_books():
   return data

@app.get("/book/{id}")
def get_book(id: int):
   id = id - 1
   return data[id]

@app.put("/book/{id}")
def add_book(id: int, book: Book):
   data[id-1] = book
   return data

@app.delete("/book/{id}")
def delete_book(id: int):
   data.pop(id-1)
   return data

pip install sqlalchemy

from sqlalchemy import create_engine
from sqlalchemy.dialects.sqlite import *
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args = {"check_same_thread": False})

from sqlalchemy.orm import sessionmaker, Session
session = sessionmaker(autocommit=False, autoflush=False, bind=engine)

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

from sqlalchemy import Column, Integer, String
class Books(Base):
   __tablename__ = 'book'
   id = Column(Integer, primary_key=True, nullable=False)
   title = Column(String(50), unique=True)
   author = Column(String(50))
   publisher = Column(String(50))
   Base.metadata.create_all(bind=engine)

from typing import List
from pydantic import BaseModel, constr
class Book(BaseModel):
   id: int
   title: str
   author:str
   publisher: str
   class Config:
      orm_mode = True

from fastapi import FastAPI, Depends
app=FastAPI()
def get_db():
   db = session()
   try:
      yield db
   finally:
   db.close()
@app.post('/add_new', response_model=Book)
def add_book(b1: Book, db: Session = Depends(get_db)):
   bk=Books(id=b1.id, title=b1.title, author=b1.author,
publisher=b1.publisher)
   db.add(bk)
   db.commit()
   db.refresh(bk)
   return Books(**b1.dict())

@app.get('/list', response_model=List[Book])
def get_books(db: Session = Depends(get_db)):
   recs = db.query(Books).all()
   return recs

@app.get('/book/{id}', response_model=Book)
def get_book(id:int, db: Session = Depends(get_db)):
   return db.query(Books).filter(Books.id == id).first()

@app.put('/update/{id}', response_model=Book)
def update_book(id:int, book:Book, db: Session = Depends(get_db)):
   b1 = db.query(Books).filter(Books.id == id).first()
   b1.id=book.id
   b1.title=book.title
   b1.author=book.author
   b1.publisher=book.publisher
   db.commit()
   return db.query(Books).filter(Books.id == id).first()
@app.delete('/delete/{id}')
def del_book(id:int, db: Session = Depends(get_db)):
   try:
      db.query(Books).filter(Books.id == id).delete()
      db.commit()
   except Exception as e:
      raise Exception(e)
   return {"delete status": "success"}

engine = create_engine('mysql+pymysql://user:password@localhost/test')

pip3 install pymongo

E:\mongodb\bin>mongod
..
waiting for connections on port 27017

from pymongo import MongoClient
client=MongoClient()

from pydantic import BaseModel
from typing import List
class Book(BaseModel):
   bookID: int
   title: str
   author:str
   publisher: str

from fastapi import FastAPI, status
app = FastAPI()

@app.post("/add_new", status_code=status.HTTP_201_CREATED)
def add_book(b1: Book):
   """Post a new message to the specified channel."""
   with MongoClient() as client:
      book_collection = client[DB][BOOK_COLLECTION]
      result = book_collection.insert_one(b1.dict())
      ack = result.acknowledged
      return {"insertion": ack}

@app.get("/books", response_model=List[str])
def get_books():
   """Get all books in list form."""
   with MongoClient() as client:
      book_collection = client[DB][BOOK_COLLECTION]
      booklist = book_collection.distinct("title")
      return booklist

[
   "Computer Fundamentals",
   "Python Cookbook",
   "Let Us Python"
]

@app.get("/books/{id}", response_model=Book)
def get_book(id: int):
   """Get all messages for the specified channel."""
   with MongoClient() as client:
      book_collection = client[DB][BOOK_COLLECTION]
      b1 = book_collection.find_one({"bookID": id})
      return b1

@strawberry.type
class Book:
   title: str
   author: str
   price: int

@strawberry.type
class Query:
   @strawberry.field
   def book(self) -> Book:
   return Book(title="Computer Fundamentals", author="Sinha", price=300)

schema = strawberry.Schema(query=Query)

graphql_app = GraphQL(schema)
app = FastAPI()

app.add_route("/book", graphql_app)
app.add_websocket_route("/book", graphql_app)

from fastapi import FastAPI
app = FastAPI()
@app.get("/app")
def mainindex():
   return {"message": "Hello World from Top level app"}

subapp = FastAPI()
@subapp.get("/sub")
def subindex():
   return {"message": "Hello World from sub app"}

app.mount("/subapp", subapp)

import time
from fastapi import FastAPI, Request
app = FastAPI()
@app.middleware("http")
async def addmiddleware(request: Request, call_next):
   print("Middleware works!")
   response = await call_next(request)
   return response
@app.get("/")
async def index():
   return {"message":"Hello World"}
@app.get("/{name}")
async def hello(name:str):
   return {"message":"Hello "+name}

pip3 install flask

from flask import Flask
flask_app = Flask(__name__)
@flask_app.route("/")
def index_flask():
   return "Hello World from Flask!"

from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def index():
   return {"message": "Hello World from FastAPI!"}

from fastapi.middleware.wsgi import WSGIMiddleware
app.mount("/flask", WSGIMiddleware(flask_app))

uvicorn flaskapp:app –reload

{"message":"Hello World from FastAPI!"}

Hello World from Flask!

iwr https://get.deta.dev/cli.ps1 -useb | iex

PS C:\fastapi_deta_app> iwr https://get.deta.dev/cli.ps1 -useb | iex
Deta was installed successfully to
C:\Users\User\.deta\bin\deta.exe
Run 'deta --help' to get started

PS C:\fastapi_deta_app> deta login
Please, log in from the web page. Waiting...
https://web.deta.sh/cli/60836
Logged in successfully.

# main.py
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):
   return {"item_id": item_id}

PS C:\fastapi_deta_app> deta new
Successfully created a new micro
{
   "name": "fastapi_deta_app",
   "id": "2b236e8f-da6a-409b-8d51-7c3952157d3c",
   "project": "c03xflte",
   "runtime": "python3.9",
   "endpoint": "https://vfrjgd.deta.dev",
   "region": "ap-southeast-1",
   "visor": "enabled",
   "http_auth": "disabled"
}
Adding dependencies...
…..
Installing collected packages: typing-extensions, pydantic, idna, sniffio, anyio, starlette, fastapi
Successfully installed anyio-3.4.0 fastapi-0.70.0 idna-3.3 pydantic-1.8.2 sniffio-1.2.0 starlette-0.16.0 typingextensions-4.0.0