fastapi 安全性 / APIRouter / BackgroundTasks / 元数据 / 测试调试

# 安全性 main.py
from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# oauth2_scheme(some, parameters) 是课调用的，可以被Depends使用

@app.get("/items/")
async def read_items(token: str = Depends(oauth2_scheme)):
    return {"token": token}

# 安全性
from fastapi import FastAPI, Depends
from typing import Optional
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# oauth2_scheme(some, parameters) 是课调用的，可以被Depends使用

class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None

def fake_decode_token(token):
    return User(username=token+"fakedecoded", email="abc.mail", full_name="michael")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    return fake_decode_token(token)

@app.get("/users/me/")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

# 安全性
from fastapi import FastAPI, Depends, HTTPException, status
from typing import Optional
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel

fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "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,
    },
}

def fake_hash_password(password: str):
    return "fakehashed" + password

app = FastAPI()
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
# oauth2_scheme(some, parameters) 是课调用的，可以被Depends使用

class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None

class UserInDB(User):
    hashed_password: str


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict) # 过滤，获取hasded_password

def fake_decode_token(token):
    return get_user(fake_users_db, token)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = fake_decode_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    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")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    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(form_data.password)
    if not hashed_password == user.hashed_password:
        raise HTTPException(status_code=400, detail="Incorrect username or password")

    return {"access_token": user.username, "token_type": "bearer"}


@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

# dependencies.py
# 我们了解到我们将需要一些在应用程序的好几个地方所使用的依赖项。
# 因此，我们将它们放在它们自己的 dependencies 模块

from fastapi import Header, HTTPException

async def get_token_header(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
       raise HTTPException(status_code=400, detail="X-Token header invalid")

async def get_query_token(token: str):
    if token != "jessica":
        raise HTTPException(status_code=400, detail="No Jessica token provided")

# main.py 你的应用程序中将所有内容联结在一起的主文件
# 你的大部分逻辑现在都存在于其自己的特定模块中
# 因此主文件的内容将非常简单
from fastapi import Depends, FastAPI
from dependencies import get_query_token, get_token_header
from internal import admin
from routers import items, users
# from .routers.items import router # 以下两行名称重叠，注意避免
# from .routers.users import router

app = FastAPI(dependencies=[Depends(get_query_token)])

app.include_router(users.router)
app.include_router(items.router)
# users.router 包含了 app/routers/users.py 文件中的 APIRouter
# items.router 包含了 app/routers/items.py 文件中的 APIRouter
app.include_router(
    admin.router,
    prefix="/admin", # 添加路径前缀，而不必修改admin.router
    tags=["admin"],
    dependencies=[Depends(get_token_header)],
    responses={418: {"description": "I'm a teapot"}},
    # 但这只会影响我们应用中的 APIRouter，
    # 而不会影响使用admin.router的任何其他代码
)
#  app.include_router()，可以将每个 APIRouter 添加到主 FastAPI 应用程序中

# 多次使用不同的 prefix 包含同一个路由器
app.include_router(
    admin.router,
    prefix="/admin_test", # 添加路径前缀，而不必修改admin.router
    tags=["admin"],
    dependencies=[Depends(get_token_header)],
    responses={418: {"description": "I'm a teapot, diff "}},
    # 但这只会影响我们应用中的 APIRouter，
    # 而不会影响使用admin.router的任何其他代码
)

# 也可以在另一个 APIRouter 中包含一个 APIRouter
# router.include_router(other_router)

@app.get("/")
async def root():
    return {"message": "Hello Bigger Applications!"}

# internal/admin.py
from fastapi import APIRouter

router = APIRouter()

@router.post("/")
async def update_admin():
    return {"message": "Admin getting schwifty"}

# routers/items.py
# 此模块中的所有路径操作都有相同的：
#   路径 prefix：/items
#   tags：（仅有一个 items 标签）
#   额外的 responses
#   dependencies：它们都需要我们创建的 X-Token 依赖项

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"}},
) # 这个路径操作将包含标签的组合：["items"，"custom"]
# 也会有两个响应，一个用于 404，一个用于 403
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"}

# routers/users.py
# 将与用户相关的路径操作与其他代码分开，以使其井井有条
from fastapi import APIRouter
router = APIRouter()
# 你可以将 APIRouter 视为一个「迷你 FastAPI」类
# 在此示例中，该变量被命名为 router，但你可以根据你的想法自由命名

@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}

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)
# 任务函数，可以是普通函数 或 async def函数


@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_notification, email, message="hello, sending email to me")
    # add_task 参数： 函数名， 顺序参数， 关键字参数
    return {"message": "Notification sent in the background"}

from fastapi import BackgroundTasks, FastAPI, Depends
from typing import Optional

app = FastAPI()

def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(message)

def get_query(background_tasks: BackgroundTasks, q: Optional[str] = None):
    if q:
        message = f"found query: {q}\n"
        background_tasks.add_task(write_log, message)
    return q

@app.post("/send-notification/{email}")
async def send_notification(email: str, background_tasks: BackgroundTasks, q: str = Depends(get_query)):
    message = f"message to {email}\n"
    background_tasks.add_task(write_log, message)
    return {"message": "Message sent"}

from fastapi import BackgroundTasks, FastAPI, Depends
from typing import Optional

description = """
goodApp API helps you do awesome stuff. 🚀🚀🚀

## Items

You can **read items**.

## Users

You will be able to:

* **Create users** (_not implemented_).
* **Read users** (_not implemented_).
"""

app = FastAPI(
    title="good_App",
    description=description,
    version="0.0.100",
    terms_of_service="http://example.com/terms/",
    contact={
        "name": "Deadpoolio the Amazing",
        "url": "http://x-force.example.com/contact/",
        "email": "dp@x-force.example.com",
    },
    license_info={
        "name": "Apache 2.0",
        "url": "https://www.apache.org/licenses/LICENSE-2.0.html",
    },
)

from fastapi import FastAPI
tags_metadata = [
    {
        "name": "users", # 必须的，跟相应的 tags 参数有相同的名
        "description": "Operations with users. The **login** logic is also here.",
    },
    {
        "name": "items",
        "description": "Manage items. So _fancy_ they have their own docs.",
        "externalDocs": {
            "description": "Items external docs",
            "url": "https://fastapi.tiangolo.com/", # url 必须的
        },
    },
]

app = FastAPI(openapi_tags=tags_metadata)

@app.get("/users/", tags=["users"])
async def get_users():
    return [{"name": "Harry"}, {"name": "Ron"}]

@app.get("/items/", tags=["items"])
async def get_items():
    return [{"name": "wand"}, {"name": "flying broom"}]

app = FastAPI(openapi_tags=tags_metadata, openapi_url="/api/v100/michael.json")

from fastapi import FastAPI
from fastapi.testclient import TestClient

app = FastAPI()

@app.get("/")
async def read_main():
    return {"msg": "Hello World"}

client = TestClient(app)

def test_read_main(): 
    # 测试函数是普通 def， 这样你可以用 pytest 来执行
    # 否则的话，需要使用 @pytest.mark.anyio装饰函数
    # 且 使用 from httpx import AsyncClient 定义 client
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"msg": "Hello World"}

(pt19) D:\gitcode\Python_learning\fastapi\9> pytest main13.py
=================================================================== test session starts ====================================================================
platform win32 -- Python 3.8.10, pytest-6.2.5, py-1.10.0, pluggy-1.0.0
rootdir: D:\gitcode\Python_learning\fastapi\9
collected 1 item

main13.py .                                                                                                                                           [100%]

===================================================================== warnings summary ===================================================================== 
..\..\..\..\programdata\anaconda3\envs\pt19\lib\site-packages\aiofiles\os.py:10
..\..\..\..\programdata\anaconda3\envs\pt19\lib\site-packages\aiofiles\os.py:10
..\..\..\..\programdata\anaconda3\envs\pt19\lib\site-packages\aiofiles\os.py:10
..\..\..\..\programdata\anaconda3\envs\pt19\lib\site-packages\aiofiles\os.py:10
..\..\..\..\programdata\anaconda3\envs\pt19\lib\site-packages\aiofiles\os.py:10
  d:\programdata\anaconda3\envs\pt19\lib\site-packages\aiofiles\os.py:10: DeprecationWarning: "@coroutine" decorator is deprecated since Python 3.8, use "async def" instead
    def run(*args, loop=None, executor=None, **kwargs):

-- Docs: https://docs.pytest.org/en/stable/warnings.html
============================================================== 1 passed, 5 warnings in 0.45s =============================================================== 

# main14.py
import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    a = "a"
    b = "b" + a
    return {"hello world": b}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)