因项目需要,使用FastAPI框架编写了一个简单的HTTP API服务,用于生成短信发送记录和短信回复记录。
- is_valid_phone_numbers函数用于验证手机号码列表的格式是否正确。
- current_time_str函数返回当前时间的字符串表示,可选参数format用于指定时间格式。
- PhoneBaseModel是所有手机相关模型的基类,定义了mobile字段,并在validator中验证该字段是否由数字组成。
- SendRecord和DeliverRecord分别表示短信发送记录和短信回复记录的模型类,继承自PhoneBaseModel,定义了各自的字段。
- get_send_records和get_deliver_records是两个API接口的实现,分别用于生成短信发送记录和短信回复记录。它们都接收一个numbers参数,表示手机号码列表,以逗号分隔。如果输入号码格式错误,则返回400错误。如果发生其他异常,则返回500错误。接口返回相应的记录列表。
- 最后,在if __name__ == "__main__":中启动FastAPI服务,可以选择线上模式或debug模式运行
1、接口开发:
from fastapi import FastAPI
from fastapi import HTTPException, Query
from pydantic import BaseModel, Field, validator
import re
import random
from datetime import datetime
from typing import Optional
import uvicorn
app = FastAPI()
error_codes = ['DELIVRD', 'MK:0029', 'MK:0011', 'MK:0001']
def is_valid_phone_numbers(s):
pattern = r'^(1[3-9]\d{9})(,(1[3-9]\d{9}))*$'
return bool(re.match(pattern, s))
def current_time_str(format="%Y%m%d%H%M%S"):
now = datetime.now()
return now.strftime(format)
class PhoneBaseModel(BaseModel):
mobile: str = Field(..., min_length=11, max_length=11)
@validator('mobile')
def validate_mobile(cls, value):
# 检查mobile字段是否全部由数字组成
if not value.isdigit():
raise ValueError("Phone number must contain only digits.")
return value
class SendRecord(PhoneBaseModel):
errorCode: str
msgGroup: str = '0625161548000003527178'
receiveDate: Optional[str] = None
reportStatus: str = 'CM:0000'
submitDate: str
class DeliverRecord(PhoneBaseModel):
addSerial: str
sendTime: Optional[str] = None
smsContent: str = '2'
@app.get("/send_records", summary="sms", tags=['短信发送记录'])
async def get_send_records(numbers: str = Query(None, description="手机号码列表,以逗号分隔")):
# 验证文件格式
if not is_valid_phone_numbers(numbers):
raise HTTPException(status_code=400, detail="输入号码格式错误")
return {
"message": "输入号码格式错误"
}
try:
input_numbers = numbers.split(',') if numbers else numbers.split(',')
send_records = [
SendRecord(
errorCode=error_code,
mobile=mobile,
submitDate=current_time_str(),
receiveDate=current_time_str() if error_code == 'DELIVRD' else None
)
for mobile in input_numbers
for error_code in [random.choice(error_codes)]
]
return send_records
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/deliver_records", summary="sms", tags=['短信回复记录'])
async def get_deliver_records(numbers: str = Query(None, description="手机号码列表,以逗号分隔")):
if not is_valid_phone_numbers(numbers):
raise HTTPException(status_code=400, detail="输入号码格式错误")
return {
"message": "输入号码格式错误"
}
try:
input_numbers = numbers.split(',') if numbers else numbers.split(',')
deliver_records = [
DeliverRecord(
addSerial='106509712041021',
mobile=mobile,
sendTime=current_time_str(),
smsContent='2'
)
for mobile in input_numbers]
return deliver_records
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
## 线上模式
# uvicorn.run("sms_server:app", host="0.0.0.0", port=1315)
## debug 模式
uvicorn.run("sms_server:app", host="0.0.0.0", port=1315, reload=True, )
2、接口文档:
如果输入号码:XXXX,YYYY
request_url:http://xxx.xxx.x.xxx:1315/send_records?numbers=XXXX%2CYYYY
curl
命令:
两个curl
命令都用于向指定的HTTP服务器发送GET请求,用于获取短信发送记录。它们在功能上是等价的,只是书写风格有所不同。下面是对每个命令的具体解释:
第一个命令
1curl -X 'GET' \
2 'http://xxx.xxx.x.xxx:1315/send_records?numbers=XXXX%2CYYYY' \
3 -H 'accept: application/json'
-X 'GET'
明确指定了HTTP请求的方法为GET。虽然GET是curl发送HTTP请求的默认方法,显式写出是为了清晰或在需要时覆盖默认设置。'http://xxx.xxx.x.xxx:1315/send_records?numbers=XXXX%2CYYYY'
是目标URL,其中%2C
是逗号,
的URL编码形式,用于分隔多个电话号码。-H 'accept: application/json'
设置了请求头,告诉服务器客户端期望接收的内容类型为JSON。这有助于服务器返回适当格式的数据。
第二个命令
curl 'http://xxx.xxx.x.xxx:1315/send_records?numbers=15071207452%2C15071207453'
这个命令更简洁,没有明确指定请求方法和自定义头部。因为它省略了-X 'GET'
和-H 'accept: application/json'
,curl会默认使用GET方法,并且通常也会接受服务器返回的默认内容类型。在大多数情况下,如果服务器设计合理,它会根据请求的资源自动提供正确的响应内容类型,即使客户端没有明确指定Accept
头部。
注意:在URL中,特殊字符需要进行编码以便正确传输。"%2C" 是逗号(,)的URL编码形式。因此,"XXXX%2CYYYY" 解码后实际上是 "XXXX,YYYY"。这里的逗号用来分隔两个不同的电话号码,所以 "%2C" 在这个上下文中代表的是一个分隔符,用于区分多个数值或者项目。当你将此URL用于请求时,服务器会解码 "%2C" 返回其原始字符“,”,从而理解这些是两个独立的号码。
3、调用测试:
import requests
# FastAPI服务地址
base_url = "http://xxx.xxx.x.xxx:1315"
# 调用短信发送记录接口
numbers = "13800138000,1390013"
response = requests.get(f"{base_url}/send_records?numbers={numbers}")
if response.status_code == 200:
print("短信发送记录响应:", response.json())
else:
print(f"请求失败,状态码:{response.status_code}, 原因:{response.text}")
# 调用短信回复记录接口
response = requests.get(f"{base_url}/deliver_records?numbers={numbers}")
if response.status_code == 200:
print("短信回复记录响应:", response.json())
else:
print(f"请求失败,状态码:{response.status_code}, 原因:{response.text}")
输出结果:
请求失败,状态码:400, 原因:{"detail":"输入号码格式错误"}
请求失败,状态码:400, 原因:{"detail":"输入号码格式错误"}