前面文章聊了工程目录，如果掌握了前面一些列文章就可以进行简单的接口开发了。
后续文章主要是唠接口的开发和fastapi框架的进阶阶段，那么这次是开发了注册和登录两个接口
?
redis
/sc_app/redispy.py
import?redis
__all__?=?[
????"redispy"
]
class?Redispy:
????def?__init__(self):
????????self._connect?=?redis.Redis(
????????????host="192.168.0.100",?port=6379,?db=0
????????)
????def?set_value(self,?name:?str,?value:?str,?is_data:?bool?=?False):
????????"""
????????往redis写入数据
????????:param?name:
????????:param?value:
????????:param?is_data:
????????:return:
????????"""
????????if?is_data:
????????????name?=?f"token:{name}"
????????self._connect.set(name,?value)
????def?get_value(self,?name:?str,?is_data:?bool?=?False):
????????"""
????????从redis获取value
????????:param?name:
????????:param?is_data:
????????:return:
????????"""
????????if?is_data:
????????????name?=?f"token:{name}"
????????return?self._connect.get(name).decode('utf-8')
????def?get_exists(self,?name,?is_data:?bool?=?False):
????????"""
????????判断redis中的key是否存在
????????:param?name:
????????:param?is_data:
????????:return:
????????"""
????????if?is_data:
????????????name?=?f"token:{name}"
????????return?self._connect.exists(name)
????def?__del__(self):
????????self._connect.close()
redispy?=?Redispy()
模型
/sc_app/schemas/users/register.py
from?pydantic?import?BaseModel
class?BaseUser(BaseModel):
????username:?str
????def?__repr__(self):
????????return?self.username
class?UserPwd(BaseUser):
????password:?str
????def?__repr__(self):
????????return?self.password
class?User(BaseUser):
????id:?int
????is_active:?bool
????def?__repr__(self):
????????return?self.id,?self.is_active
????class?Config:
????????orm_mode?=?True
/sc_app/schemas/users/login.py
from?pydantic?import?BaseModel
class?BaseUser(BaseModel):
????username:?str
????def?__repr__(self):
????????return?self.username
class?UserPwd(BaseUser):
????password:?str
????def?__repr__(self):
????????return?self.password
class?UserToken(BaseUser):
????token:?str
????def?__repr__(self):
????????return?self.token
????class?Config:
????????orm_mode?=?True
上面两段代码是注册和登录的模型，当然我是区分开了，你也可以把一样的代码设置为公用。
接口
/sc_app/routers/register.py
from?fastapi?import?Depends
from?fastapi?import?APIRouter
from?fastapi?import?HTTPException
from?sc_app.databases?import?get_db
from?sqlalchemy.orm?import?Session
from?sc_app.model?import?user_models
from?sc_app.schemas.users.register?import?User,?UserPwd
class?DatabaseUser:
????def?__init__(self,?db:?Session):
????????self.db?=?db
????def?get_user_username(self,?*,?username:?str):
????????return?self.db.query(
????????????user_models.Users
????????).filter(user_models.Users.username?==?username).first()
????def?register_user(self,?*,?user:?UserPwd):
????????fake_hashed_password?=?user.password?+?"notreallyhashed"
????????db_user?=?user_models.Users(
????????????username=user.username,?password=fake_hashed_password
????????)
????????self.db.add(db_user)
????????self.db.commit()
????????self.db.refresh(db_user)
????????return?db_user
????def?__del__(self):
????????self.db.close()
router?=?APIRouter(
????prefix="/user"
)
@router.post("/register/",?response_model=User)
def?api_register_user(user:?UserPwd,?db:?Session?=?Depends(get_db)):
????"""
????学员注册接口
????:param?user:
????:param?db:
????:return:
????"""
????connect?=?DatabaseUser(db)
????data?=?connect.get_user_username(username=user.username)
????if?data:
????????raise?HTTPException(
????????????status_code=400,
????????????detail="Data?does?not?exist?or?the?name?is?duplicate?!"
????????)
????get_user?=?connect.register_user(user=user)
????return?get_user
/sc_app/routers/login.py
import?time
from?fastapi?import?APIRouter
from?sc_app.databases?import?get_db
from?sqlalchemy.orm?import?Session
from?sc_app.model?import?user_models
from?sc_app.redispy?import?redispy
from?fastapi?import?HTTPException,?Depends,?status
from?sc_app.schemas.users.login?import?UserPwd,?UserToken
def?create_access_token(data:?dict):
????"""
????生成token
????:param?data:
????:return:
????"""
????new_data?=?data.copy()
????new_data["token"]?=?"".join(str(time.time()).split("."))
????return?new_data
def?check_user(user:?str,?pwd:?str,?db:?Session?=?Depends(get_db)):
????"""
????根据输入的用户信息，数据库查询比对账号和密码
????:param?user:
????:param?pwd:
????:param?db:
????:return:
????"""
????username,?password?=?db.query(
????????user_models.Users.username,?user_models.Users.password
????).filter(user_models.Users.username?==?user).first()
????return?True?if?username?==?user?and?password?==?pwd?else?False
router?=?APIRouter(
????prefix="/user"
)
@router.post("/login/",?response_model=UserToken)
def?api_login_user(user:?UserPwd,?db:?Session?=?Depends(get_db)):
????"""
????登录接口
????:param?user:
????:param?db:
????:return:
????"""
????#???如果返回False则抛出错误
????if?not?check_user(user.username,?user.password,?db):
????????raise?HTTPException(
????????????status_code=status.HTTP_401_UNAUTHORIZED,
????????????detail={
????????????????"status":?0,
????????????????"data":?{},
????????????????"error_msg":?"Incorrect?account?or?password",
????????????????"error_code":?1000
????????????}
????????)
????#???判断redis中key是否存在
????if?not?redispy.get_exists(user.username):
????????#???创建token
????????g_token?=?create_access_token(data={"user_info":?user.username})
????????token?=?g_token["token"]
????????#???往redis中写入token
????????redispy.set_value(user.username,?token,?is_data=True)
????????return?UserToken(token=token,?username=user.username)
????#???从redis中读取token
????token?=?redispy.get_value(user.username,?is_data=True)
????return?UserToken(token=token,?username=user.username)
启动并请求
/sc_app/main.py
import?uvicorn
from?fastapi?import?Depends
from?fastapi?import?FastAPI
from?sc_app.routers?import?register,?login
from?sc_app.dependencies?import?verify_token
from?sc_app.dependencies?import?verify_x_token
app?=?FastAPI(dependencies=[Depends(verify_token)])
app.include_router(router=register.router,?dependencies=[Depends(verify_x_token)])
app.include_router(router=login.router,?dependencies=[Depends(verify_x_token)])
@app.get("/")
def?index():
????"""
????首页
????:return:
????"""
????return?{"message":?"Welcome?to?the?home?page?!"}
if?__name__?==?'__main__':
????uvicorn.run(app="main:app",?reload=True,?debug=True)
上述代码是启动入口，运行后就可以在
postman中去请求，现在是有Bug的，因为没做token验证，所以在Header中传一个token字段就可以访问首页了，后面会利用jwt进行密码加密和token验证，但这个目前不影响请求注册和登录接口。
请求注册接口：
POST ：http://127.0.0.1:8000/user/register/
请求头参数：
x-token debugfeng
请求参数：
{
"username": "debugfeng11@qq.com",
"password": "123456"
}
请求结果：
{
"username": "debugfeng11@qq.com",
"id": 11,
"is_active": true
}
请求登录接口：
POST ：http://127.0.0.1:8000/user/login/
请求头参数：
x-token debugfeng
请求参数：
{
"username": "debugfeng05@qq.com",
"password": "123456notreallyhashed"
}
请求结果：
{
"username": "debugfeng05@qq.com",
"token": "1652006410799995"
}
今天先聊到这里吧，以上总结或许能帮助到你，或许帮助不到你，但还是希望能帮助到你，如有疑问、歧义，直接私信留言会及时修正发布；非常期待你的一键 3 连【 点赞、收藏、分享 】哟，谢谢！
未完成，待续……
一直在努力，希望你也是！
微信搜索公众号：就用python
