112 lines
3.1 KiB
Python
112 lines
3.1 KiB
Python
|
from framework.main_framework import Response, main_function
|
||
|
|
from framework.requests_dataclass import SampleRequests
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from dataclasses import asdict
|
||
|
|
from flask import Blueprint
|
||
|
|
from flask import request
|
||
|
|
from framework.mysql_cli import MysqlCliV3
|
||
|
|
from framework.response_code import RequestMethods
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class OverrideSampleRequestParams(SampleRequests):
|
||
|
|
"""
|
||
|
|
创建POST请求参数
|
||
|
|
"""
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
if request.method == "GET":
|
||
|
|
super(OverrideSampleRequestParams, self).__init__(json_data=request.args, methods=RequestMethods.GET)
|
||
|
|
if request.method == "POST":
|
||
|
|
super(OverrideSampleRequestParams, self).__init__(json_data=request.get_data(), methods=RequestMethods.POST)
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class ResponseInterface:
|
||
|
|
"""
|
||
|
|
返回的内容
|
||
|
|
"""
|
||
|
|
|
||
|
|
def convert_to_dict(self) -> dict:
|
||
|
|
"""
|
||
|
|
调用这个函数将ResponseInterface数据类转成字典类型的数据
|
||
|
|
:return: dict
|
||
|
|
"""
|
||
|
|
return asdict(self)
|
||
|
|
|
||
|
|
|
||
|
|
class OverrideMysqlCli(MysqlCliV3):
|
||
|
|
"""
|
||
|
|
创建Mysql操作类
|
||
|
|
"""
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
super().__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):
|
||
|
|
super(OverriderMainResponse, self).__init__()
|
||
|
|
# self.is_connect_mysql = connect_mysql
|
||
|
|
self.mysql_cli: MysqlCliV3 = MysqlCliV3()
|
||
|
|
self.set_connect_mysql(mysql_cli=self.mysql_cli)
|
||
|
|
|
||
|
|
def core(self) -> None:
|
||
|
|
"""
|
||
|
|
重写core函数处理核心内容 例如数据查询/数据处理
|
||
|
|
"""
|
||
|
|
params = OverrideSampleRequestParams()
|
||
|
|
|
||
|
|
"""
|
||
|
|
处理完核心内容后将 handle_success 设置为True
|
||
|
|
self.handle_success = True
|
||
|
|
"""
|
||
|
|
self.set_handle_success(handle_success=True)
|
||
|
|
|
||
|
|
def before_response(self) -> dict:
|
||
|
|
"""
|
||
|
|
重写before_response函数
|
||
|
|
例如关闭mysql/redis连接等
|
||
|
|
"""
|
||
|
|
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)
|