134 lines
3.5 KiB
Python
134 lines
3.5 KiB
Python
from framework.main_framework import Response, main_function
|
|
from framework.requests_dataclass import SampleRequestsGET, SampleRequestsPOST
|
|
from dataclasses import dataclass
|
|
from dataclasses import asdict
|
|
from flask import Blueprint
|
|
from flask import request
|
|
from framework.m_mysql_cli import MysqlCli
|
|
|
|
|
|
@dataclass
|
|
class OverrideRequestsPOST(SampleRequestsPOST):
|
|
"""
|
|
创建POST请求参数
|
|
"""
|
|
|
|
def __init__(self):
|
|
super(OverrideRequestsPOST, self).__init__(json_data=request.get_data())
|
|
|
|
|
|
@dataclass
|
|
class OverrideRequestsGET(SampleRequestsGET):
|
|
"""
|
|
创建GET请求参数
|
|
"""
|
|
|
|
def __init__(self):
|
|
super(OverrideRequestsGET, self).__init__(json_data=request.args)
|
|
|
|
|
|
@dataclass
|
|
class ResponseInterface:
|
|
"""
|
|
返回的内容
|
|
"""
|
|
|
|
def convert_to_dict(self) -> dict:
|
|
"""
|
|
调用这个函数将ResponseInterface数据类转成字典类型的数据
|
|
:return: dict
|
|
"""
|
|
return asdict(self)
|
|
|
|
|
|
class OverrideMysqlCli(MysqlCli):
|
|
"""
|
|
创建Mysql操作类
|
|
"""
|
|
|
|
def __init__(self):
|
|
super(OverrideMysqlCli, self).__init__()
|
|
self.token_table = "accounts_info"
|
|
|
|
def handle_search_token_valid(self, username: str, user_token: str) -> None:
|
|
# 验证登录token
|
|
sql = f"""
|
|
SELECT
|
|
username
|
|
FROM {self.token_table}
|
|
WHERE username = '{username}' AND user_token = '{user_token} AND 1=1'
|
|
"""
|
|
self.create_sql(sql)
|
|
mysql_result: dict = self.fetchone()
|
|
self.check_mysql_exception(mysql_result=mysql_result)
|
|
self.check_mysql_response_null(mysql_result=mysql_result)
|
|
|
|
|
|
class OverriderMainResponse(Response):
|
|
"""
|
|
主要处理请求的类
|
|
"""
|
|
|
|
def __init__(self, connect_mysql: bool = True):
|
|
super(OverriderMainResponse, self).__init__()
|
|
self.is_connect_mysql = connect_mysql
|
|
# 返回的数据
|
|
|
|
# 是否连接数据库
|
|
if self.is_connect_mysql:
|
|
self.mysql_cli: OverrideMysqlCli = OverrideMysqlCli()
|
|
|
|
# 是否处理成功
|
|
self.handle_success: bool = False
|
|
|
|
def core(self) -> None:
|
|
"""
|
|
重写core函数处理核心内容 例如数据查询/数据处理
|
|
"""
|
|
params = OverrideRequestsGET()
|
|
params = OverrideRequestsPOST()
|
|
|
|
"""
|
|
处理完核心内容后将 handle_success 设置为True
|
|
self.handle_success = True
|
|
"""
|
|
# self.handle_success = True
|
|
if self.handle_success:
|
|
self.build_response_code(self.response_code_enum.RESPONSE_200_SUCCESS)
|
|
|
|
"""
|
|
请不要忘记将想要返回的内容赋予 self.build_response_data(object)
|
|
"""
|
|
|
|
def before_response(self) -> dict:
|
|
"""
|
|
重写before_response函数
|
|
例如关闭mysql/redis连接等
|
|
"""
|
|
if self.is_connect_mysql:
|
|
self.mysql_cli.close_connect()
|
|
|
|
return self.make_response()
|
|
|
|
|
|
"""
|
|
请不要忘记生成Flask蓝图
|
|
"""
|
|
sample_interface = Blueprint("sample_interface", __name__)
|
|
|
|
"""
|
|
请不要忘记指定URL路径
|
|
请不要忘记执行main_function函数
|
|
"""
|
|
|
|
|
|
@sample_interface.route("/sample/index", methods=["GET"])
|
|
def main():
|
|
"""
|
|
实例化OverrideMainResponse类 并传入main_function中
|
|
您只重写core()和before_response()
|
|
即可完成一个接口 它将返回一个dict直接发送给请求端
|
|
"""
|
|
override_main_response = OverriderMainResponse()
|
|
return main_function(override_main_response)
|