代码如下
from fastapi import FastAPI, APIRouter, HTTPException, status
from pydantic import BaseModel
from fastapi.responses import JSONResponse
from typing import Optional
from fastapi.middleware.cors import CORSMiddleware
from utils.time import DateTimeEncoder, getRandomID
from utils.dbutils import insertDB, selectDB
from datetime import datetime
from pydantic import Field
#Pydantic 模型定义
class UserInfo(BaseModel):
id: int
user_id: str = Field(..., alias="userId")
user_name: str = Field(..., alias="userName")# 必须字段,没有默认值
real_name: Optional[str] = Field("", alias="realName") # 可选字段,默认值为空字符串
password: str
phone: str
person_avatar: Optional[str] = Field("", alias="personAvatar")
college: Optional[str] = Field("", alias="college")
major: Optional[str] = Field("", alias="major")
education: Optional[str] = Field("", alias="education")
person_desc: Optional[str] = Field(None, alias="personDesc")
create_time: datetime = Field(..., alias="createTime")
update_time: Optional[datetime] = Field(None, alias="updateTime")
is_deleted: int = Field(..., alias="isDeleted")
last_login_time: int = Field(..., alias="lastLoginTime")
class Config:
from_attributes = True
populate_by_name = True
class RegisterInfo(BaseModel):
userName: str
phone: str
password: str
# password: Optional[str] = None
class LoginInfo(BaseModel):
phone: str
password: str
# 登录
@router.post("/login")
async def login(loginInfo: LoginInfo):
selectSql = '''
select * from userinfo where phone = '{}' and password = '{}'
'''.format(loginInfo.phone, loginInfo.password)
userList = selectDB(selectSql)
if len(userList) == 1:
#UserInfo(**userList[0]) 的含义是将 userList[0] 字典中的所有键值对作为关键字参数传递给 UserInfo 模型的构造函数,从而创建一个 UserInfo 实例。
#例如,如果 userList[0] 是 {"id": 17, "user_id": "uid2024062618231266424", "user_name": "cc", ...},那么 UserInfo(**userList[0]) 相当于 UserInfo(id=17, user_id="uid2024062618231266424", user_name="cc", ...)。
userInfo = UserInfo(**userList[0])
#userInfo.dict(by_alias=True) 将模型转换为字典并使用别名,这样返回的数据就会使用驼峰命名法。
return {"code": 200, "msg": '登录', "data": userInfo.dict(by_alias=True)}
return {"code": 500, "msg": '登录失败', "data": {}}
#注册
@router.post("/register")
async def register(registerInfo: RegisterInfo):
selectSql = '''
SELECT * from userinfo WHERE phone='{}' or user_name='{}'
'''.format(registerInfo.phone, registerInfo.userName)
userList = selectDB(selectSql)
if len(userList) > 0:
return {"code": 500, "msg": '该用户已存在', "data": {}}
uid = 'uid' + getRandomID()
sql = '''
INSERT INTO userinfo(user_id, phone, user_name, password) VALUES('{}', '{}', '{}', '{}')
'''.format(uid, registerInfo.phone, registerInfo.userName,registerInfo.password)
effectRow = insertDB(sql=sql)
print(effectRow)
if effectRow != 1:
return {"code": 500, "msg": '注册失败', "data": {}}
return {"code": 200, "msg": '注册成功', "data": {"userId": uid}}
接口调用
postman调用登录接口,返回值