python：Fastapi - 初识接口开发

短命女 2023-09-29 14:09 9阅读 0赞

前面文章聊了工程目录，如果掌握了前面一些列文章就可以进行简单的接口开发了。

后续文章主要是唠接口的开发和fastapi框架的进阶阶段，那么这次是开发了注册和登录两个接口

![图片][68f96ce39e96423ab0b033ed2f0aa089.png]

### 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

上面两段代码是注册和登录的模型，当然我是区分开了，你也可以把一样的代码设置为公用。

> 如果对python自动化测试、web自动化、接口自动化、移动端自动化、面试经验交流等等感兴趣的测试人，可以 [**点我自行获取…**][Link 1]

### 接口 ###

/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/

请求头参数：