28.FastAPI微服务应用示例

在本节内容中，我们以FastAPI框架为基础，开发一个简单的微服务应用，该应用由两部分微服务组成，作为示例，不考虑真实业务系统使用数据库来进行数据保存。总体规划如下：

微服务	说明	API	功能描述
认证微服务	负责系统用户的认证鉴权	/oth_api/login	实现用户登录功能并颁发令牌
业务微服务	模拟实现简单的业务系统功能	/business_api/user/mine	获取用户信息
		/business_api/business/business_info	获取Hello

28.1认证微服务

认证微服务主要负责用户登录系统并返回令牌。为简化开发，本例中用户数据采用字典列表来处理。

• settings.py 系统的基本配置项

# coding: utf-8
​
SECRET_KEY='5e9eb66688de11eca78070c94ec87656a649b0cc88de11eca8b470c94ec87656'
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_DAYS = 1

• users.py 用户列表数据

# coding: utf-8
​
# 用户列表
users = [
    {
        'usr_id': 'u000010001',
        'usr_acc': 'abc',
        'usr_pwd': '$pbkdf2-sha256$29000$MTAxMDEwMTg$fUI/40Zxj6.62GHgUX9PbDZ0SybccxZwn3Wl3qJ8U/M', # a1b2c3
        'salt': '10181010',
        'mobile': '15149768902',
        'email': 'abc@example.com'
    },
    {
        'usr_id': 'u000010002',
        'usr_acc': 'xyz',
        'usr_pwd': '$pbkdf2-sha256$29000$MTAxMDEwMTg$Eu7mZV80f.tu4RDXSsst9gDjV4fXJ.9S7t1hgcGMMVk', # x1y2z3
        'salt': '10101018',
        'mobile': '13801019029',
        'email': 'xyz@example.com'
    }
]

• logger.py 日志处理

# coding: utf-8
import os
from datetime import datetime
from loguru import logger
​
log_path = '/u01'
if not os.path.exists(log_path): os.mkdir(log_path)
​
log_file = '{0}/fa_28_{1}_log.log'.format(log_path, datetime.now().strftime('%Y-%m-%d'))
​
logger.add(log_file, rotation="12:00", retention="1 days", enqueue=True)

auth.py 系统登录认证

# coding: utf-8
from datetime import datetime
from datetime import timedelta
from jose import jwt
import json
import settings
​
# 生成Token
def build_access_token(data: dict):
    for_encode = {'sub': json.dumps(data)}
    expire = datetime.utcnow() + timedelta(days=settings.ACCESS_TOKEN_EXPIRE_DAYS)
    for_encode.update({"exp": expire})
    jwt_code = jwt.encode(for_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return jwt_code

• main.py 主程序

# coding: utf-8
from fastapi import FastAPI
from fastapi import Request
from fastapi import Depends
from fastapi import HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from passlib.hash import pbkdf2_sha256
from users import users
from auth import build_access_token
from logger import logger
​
app = FastAPI()
​
# 登录并返回Token
@app.post(path='/oth_api/login')
async def login(request: Request, form: OAuth2PasswordRequestForm = Depends()):
    found_usr = None
    for usr in users:
        if usr['usr_acc'] == form.username:
            found_usr = usr
​
    if found_usr is None or pbkdf2_sha256.hash(form.password, salt=found_usr['salt'].encode()) != found_usr['usr_pwd']:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    else:
        logger.info(request.client.host)
        found_usr.update({'client_host': request.client.host})
        return {"access_token": build_access_token(found_usr), "token_type": "Bearer"}

在本例的main.py中，生成令牌时，添加了客户端的IP地址。这样就可以在令牌校验时对客户端IP进行校验。

启动应用并进行测试请求：

curl -d "username=xyz&password=x1y2z3" -X POST http://127.0.0.1:8000/oth_api/login -i
HTTP/1.1 200 OK
date: Wed, 09 Feb 2022 12:05:24 GMT
server: uvicorn
content-length: 508
content-type: application/json
​
{"access_token":"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ7XCJ1c3JfaWRcIjogXCJ1MDAwMDEwMDAyXCIsIFwidXNyX2FjY1wiOiBcInh5elwiLCBcInVzcl9wd2RcIjogXCIkcGJrZGYyLXNoYTI1NiQyOTAwMCRNVEF4TURFd01UZyRFdTdtWlY4MGYudHU0UkRYU3NzdDlnRGpWNGZYSi45Uzd0MWhnY0dNTVZrXCIsIFwic2FsdFwiOiBcIjEwMTAxMDE4XCIsIFwibW9iaWxlXCI6IFwiMTM4MDEwMTkwMjlcIiwgXCJlbWFpbFwiOiBcInh5ekBleGFtcGxlLmNvbVwiLCBcImNsaWVudF9ob3N0XCI6IFwiMTI3LjAuMC4xXCJ9IiwiZXhwIjoxNjQ0NDk0NzI1fQ.p2ABXCt9RgaU2fKNo83P53z9L2YRXMLXtZ51IgIF53A","token_type":"Bearer"}

将上面命令返回的令牌记下来，在后面的请求中使用。

28.2业务微服务

我们另外在创建一个工程项目，目录结构如下：

api
    __init__.py
    business
        __init__.py
        business.py
    user
        __init__.py
        user_info.py
common
    __init__.py
    common.py
logger.py
main.py
settings.py

文件及代码如下：

• settings.py

# coding: utf-8
from fastapi.security import OAuth2PasswordBearer
​
SECRET_KEY='5e9eb66688de11eca78070c94ec87656a649b0cc88de11eca8b470c94ec87656'
ALGORITHM = "HS256"
​
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/oth_api/login")

• logger.py

# coding: utf-8
import os
from datetime import datetime
from loguru import logger
​
log_path = '/u01'
if not os.path.exists(log_path): os.mkdir(log_path)
​
log_file = '{0}/fa_28_{1}_log.log'.format(log_path, datetime.now().strftime('%Y-%m-%d'))
​
logger.add(log_file, rotation="12:00", retention="1 days", enqueue=True)

• common\common.py

# coding: utf-8
from fastapi import Depends
from fastapi import HTTPException
from logger import logger
from jose import jwt
import json
import settings
​
# Token校验
def verify_token(token: str, host: str):
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        logger.info(payload)
        sub = json.loads(payload['sub'])
        if sub['client_host'] != host:
            return False
    except Exception as ex:
        logger.info(str(ex))
        return False
    return True
​
#获取当前用户
def find_current_usr(token: str = Depends(settings.oauth2_scheme)):
    current_usr = None
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        current_usr = json.loads(payload['sub'])
    except Exception as ex:
        logger.info(str(ex))
        raise HTTPException(status_code=401)
    if current_usr is None: raise HTTPException(status_code=401)
    return current_usr

以上代码中的令牌校验部分没有使用 raise 抛出异常，原因是我们本例的令牌校验不放在每个业务方法中实现，而是在主程序中采用中间件来实现。另外，在令牌校验中增加了对客户端IP地址的校验，从而使用户必须使用登录时的浏览器进行访问，也可将客户端类型user-agent等其他信息包含在令牌中。

• main.py

# coding: utf-8
from fastapi import FastAPI
from fastapi import Request
from fastapi.responses import JSONResponse
from logger import logger
from common.common import verify_token
from api import user
from api import business
​
app = FastAPI()
​
​
@app.middleware("http")
async def middle(request: Request, call_next):
    json_response = JSONResponse(status_code=401, content={'detail': 'Unauthorized'})
​
    # 若不存在authorization则返回鉴权失败
    if not 'authorization' in request.headers:
        return json_response
​
    authorization = request.headers['authorization']
    logger.info(authorization)
    # 若authorization类型不是Bearer则返回鉴权失败
    if not authorization.startswith('Bearer'):
        return json_response
​
    token = authorization[7:]
    logger.info(token)
​
    # 若令牌校验失败则返回鉴权失败
    if not verify_token(token, request.client.host):
        return json_response
​
    response = await call_next(request)
    return response
​
​
app.include_router(user.router, prefix='/business_api')
app.include_router(business.router, prefix='/business_api')

以上代码中，通过中间件进行令牌校验。

• api\user\__init__.py

# coding: utf-8
from fastapi import APIRouter
​
router = APIRouter(prefix='/user')
​
import api.user.user_info

• api\user\user_info.py

# coding: utf-8
from fastapi import Depends
from api.user import router
from common.common import find_current_usr
from pydantic import BaseModel
​
class UserOut(BaseModel):
    usr_id: str
    usr_acc: str
    mobile: str
    email: str
​
@router.get('/mine', response_model=UserOut)
async def mine(current_usr: dict = Depends(find_current_usr)):
    return current_usr

以上代码通过响应体输出用户信息，这样就可以限制用户的其他信息的输出。

• api\business\__init__.py

# coding: utf-8
from fastapi import APIRouter
​
router = APIRouter(prefix='/business')
​
import api.business.business

• api\business\business.py

# coding: utf-8
from fastapi import Depends
from api.business import router
​
@router.get('/business_info')
async def business_info():
    return {'business': 'Hello Business!'}

启动应用，由于认证微服务启动时未指定端口号，所以业务微服务在启动时必须指定端口号：

uvicorn main:app --reload --port 8080

使用上面登录后的令牌进行请求测试：

curl -H "Authorization:Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ7XCJ1c3JfaWRcIjogXCJ1MDAwMDEwMDAyXCIsIFwidXNyX2FjY1wiOiBcInh5elwiLCBcInVzcl9wd2RcIjogXCIkcGJrZGYyLXNoYTI1NiQyOTAwMCRNVEF4TURFd01UZyRFdTdtWlY4MGYudHU0UkRYU3NzdDlnRGpWNGZYSi45Uzd0MWhnY0dNTVZrXCIsIFwic2FsdFwiOiBcIjEwMTAxMDE4XCIsIFwibW9iaWxlXCI6IFwiMTM4MDEwMTkwMjlcIiwgXCJlbWFpbFwiOiBcInh5ekBleGFtcGxlLmNvbVwiLCBcImNsaWVudF9ob3N0XCI6IFwiMTI3LjAuMC4xXCJ9IiwiZXhwIjoxNjQ0NDk0NzI1fQ.p2ABXCt9RgaU2fKNo83P53z9L2YRXMLXtZ51IgIF53A" http://127.0.0.1:8080/business_api/user/mine
{"usr_id":"u000010002","usr_acc":"xyz","mobile":"13801019029","email":"xyz@example.com"}

curl -H "Authorization:Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ7XCJ1c3JfaWRcIjogXCJ1MDAwMDEwMDAyXCIsIFwidXNyX2FjY1wiOiBcInh5elwiLCBcInVzcl9wd2RcIjogXCIkcGJrZGYyLXNoYTI1NiQyOTAwMCRNVEF4TURFd01UZyRFdTdtWlY4MGYudHU0UkRYU3NzdDlnRGpWNGZYSi45Uzd0MWhnY0dNTVZrXCIsIFwic2FsdFwiOiBcIjEwMTAxMDE4XCIsIFwibW9iaWxlXCI6IFwiMTM4MDEwMTkwMjlcIiwgXCJlbWFpbFwiOiBcInh5ekBleGFtcGxlLmNvbVwiLCBcImNsaWVudF9ob3N0XCI6IFwiMTI3LjAuMC4xXCJ9IiwiZXhwIjoxNjQ0NDk0NzI1fQ.p2ABXCt9RgaU2fKNo83P53z9L2YRXMLXtZ51IgIF53A" http://127.0.0.1:8080/business_api/business/business_info
{"business":"Hello Business!"}

不携带令牌直接访问：

curl http://127.0.0.1:8080/business_api/business/business_info
{"detail":"Unauthorized"}

以上就是一个简单的示例，在实际开发中可以基于以上程序设计思路进行深化，希望我整理并编写的关于FastAPI的应用文档对大家有所帮助。