最新要闻
- 出庭当晚特朗普与“金主”举行晚宴,筹得200多万美元-世界微头条
- 欠2.4亿罚款又如何?贾跃亭国内“现身”:我不要烧太多钱也能快速成功 环球快讯
- 端午节放假3天自驾方便了!滴滴租车正式上线:跑车、房车全都有
- 全球即时:一机搞定全屋清洁!石头新品洗地机A10 Ultra今晚首销:到手价3399元
- 1799元咬死RTX 3060显卡 英特尔新驱动继续鸡血:性能猛增33%|当前快看
- 用上丰田氢燃料电池 海马7X-H来了:不到5分钟加注续航达800km-天天微速讯
- 汉字找茬王连歌曲不甘怎么过-世界热闻
- 石家庄太和网点电话_石家庄太和网
- 3个ChatGPT插件自动写书爆火!更多躺着赚钱玩法可以问AI自己_全球看热讯
- 3.1级地震发生时上海居民躲床下避难 监控视频显示:剧烈摇晃、余震可能性不大|今头条
- 足球服批发_足球装备批发|热推荐
- 男子赶集买毛蛋到家全变鸭子 后续会饲养:网友点赞赚大了
- 直言理想ONE被问界M7打残 学习华为要看什么书:李想整理分享
- 丰田手把手教经销商诋毁纯电|每日动态
- 矛盾的马斯克!特斯拉4D雷达首拆:千元成本、探测距离300米
- 女学霸边读研边兼职3年赚17万:为了明确自己毕业后适合什么工作
广告
手机
iphone11大小尺寸是多少?苹果iPhone11和iPhone13的区别是什么?
警方通报辅警执法直播中被撞飞:犯罪嫌疑人已投案
- iphone11大小尺寸是多少?苹果iPhone11和iPhone13的区别是什么?
- 警方通报辅警执法直播中被撞飞:犯罪嫌疑人已投案
- 男子被关545天申国赔:获赔18万多 驳回精神抚慰金
- 3天内26名本土感染者,辽宁确诊人数已超安徽
- 广西柳州一男子因纠纷杀害三人后自首
- 洱海坠机4名机组人员被批准为烈士 数千干部群众悼念
家电
flask自定义参数校验、序列化和反序列化_视讯
(相关资料图)
项目总体结构
我的工厂函数factory.py
from settings import settingfrom flask import Flaskfrom models.models import dbfrom flask_migrate import Migratefrom urls.router import bp_te, bp_lo# from flask_script import Managerfrom utils.log import set_log# from flask_limiter import Limiter# from flask_limiter.util import get_remote_address# https://www.cnblogs.com/Du704/p/13281032.htmlmysql_host = setting.MYSQL_HOSTmysql_port = setting.MYSQL_PORTmysql_user = setting.MYSQL_USERmysql_pwd = setting.MYSQL_PASSWORDmysql_database = setting.MYSQL_DATABASEenv_cnf = setting.ENV_CNFdef create_app(): set_log() application = Flask(__name__) DB_URI = f"mysql+pymysql://{mysql_user}:{mysql_pwd}@{mysql_host}:{mysql_port}/{mysql_database}" application.config["SQLALCHEMY_DATABASE_URI"] = DB_URI # 是否追踪数据库修改,一般不开启, 会影响性能 application.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False # 是否显示底层执行的SQL语句 application.config["SQLALCHEMY_ECHO"] = False # 初始化db,关联项目 db.app = application db.init_app(application) migrate = Migrate() migrate.init_app(application, db) # 注册蓝图 application.register_blueprint(bp_te) application.register_blueprint(bp_lo) # manager = Manager(application) return applicationapplication = create_app()
配置文件setting.py,读取数据库等配置信息
from configparser import ConfigParserfrom pathlib import Pathfrom utils.encryption import getDataAesBASE_DIR = Path(__file__).resolve().parent.parentconf = ConfigParser()conf.read("config.ini", encoding="utf-8")try: mysqlhost = conf.get("mysql", "host") mysqlport = conf.get("mysql", "port") mysqluser = conf.get("mysql", "user") mysqlpassword = conf.get("mysql", "password") mysqlname = conf.get("mysql", "name") secret = conf.get("serve", "secret") env_cnf = conf.get("serve", "env") redishost = conf.get("redis", "host") redisport = conf.get("redis", "port") redispwd = conf.get("redis", "password") redislibrary = conf.get("redis", "library") MINIOHOST = conf.get("minio", "clienthost") MINIOPORT = conf.get("minio", "clientport") miniopwd = conf.get("minio", "password") miniouser = conf.get("minio", "user") MINIOWEBHOST = conf.get("minio", "webhost") MINIOWEBPORT = conf.get("minio", "webport") MINIOHTTP = conf.get("minio", "http")except Exception as e: print(e) mysqlhost = "127.0.0.1" mysqlport = 3306 mysqluser = "root" mysqlpassword = "000000" mysqlname = "0" secret = "Wchime" env_cnf = "develop" redishost = "127.0.0.1" redisport = "6379" redispwd = "000000" redislibrary = "1" MINIOHOST = "127.0.0.1" MINIOPORT = "9000" MINIOWEBPORT = "9000" miniopwd = "000000" miniouser = "000000" MINIOHTTP = "http://" MINIOWEBHOST = "127.0.0.1"MYSQL_HOST = mysqlhostMYSQL_PORT = mysqlportMYSQL_USER = getDataAes(secret, mysqluser)MYSQL_PASSWORD = getDataAes(secret, mysqlpassword)MYSQL_DATABASE = mysqlnameREDIS_HOST = redishostREDIS_PORT = redisportREDIS_PASSWORD = getDataAes(secret, redispwd)REDIS_LIBRARY = redislibraryMINIOPWD = getDataAes(secret, miniopwd)MINIOUSER = getDataAes(secret, miniouser)ENV_CNF = env_cnfif __name__ == "__main__": print(mysqlpassword)
models.py数据库模型文件
import datetimefrom utils.core import dbfrom sqlalchemy_serializer import SerializerMixinclass Uu(db.Model, SerializerMixin): __tablename__ = "uu" id = db.Column(db.Integer, autoincrement=True, primary_key=True) name = db.Column(db.String(20), nullable=False) age = db.Column(db.Integer, nullable=False) ux_id = db.Column(db.Integer, db.ForeignKey("ux.id", ondelete="SET NULL"), nullable=True) ux = db.relationship("Ux", backref="uu") # , lazy="dynamic" des = db.Column(db.String(20), nullable=True) img = db.Column(db.String(128), nullable=True)class Ux(db.Model, SerializerMixin): serialize_rules = ("-uu",) __tablename__ = "ux" id = db.Column(db.Integer, autoincrement=True, primary_key=True) name = db.Column(db.String(20), nullable=False)
序列化文件serializes.py
from models import modelsfrom utils.base import Serialize, DeSerializeclass TestSerialize(Serialize): model = models.Uu fields = ["id", "name"] build_fiels = [ {"name": "ux_name", "source": "ux.name"}, {"name": "img", "method": True} ] def get_img(self, instance): return []class TestDeSerialize(DeSerialize): model = models.Ux required_fields = ["name"] ser_fields = ["id", "name"]
base.py自定义序列化和反序列化和参数解析文件
from models.models import dbfrom flask_restful import abortfrom sqlalchemy import inspectclass DeSerialize(object): """ 反序列化,增删改 """ model = None req_fields = None other_fields = [] req_data = {} insatance = None ser_fields = [] def __init__(self, insatance=None, data={}): self.insatance = insatance self.req_data = data @property def required_fields(self): return self._get_fileds(self.req_fields) @property def serializer_fields(self): return self.ser_fields if self.ser_fields else self.model().serializable_keys def _get_fileds(self, fileds): if fileds is None: values_valid = self.model().serializable_keys else: values_valid = fileds values_valid = list(values_valid) # 删除主键 try: primary_key = list(map(lambda x: x.name, inspect(self.model).primary_key))[0] if primary_key in values_valid: values_valid.remove(primary_key) except: if "id" in values_valid: values_valid.remove("id") return values_valid def _get_vaild_values(self): vaild_dict = {} err = "request data is empty" for key in self.required_fields: value = self.req_data.get(key) if value is None: err = f"{key} is not required" return False, err vaild_dict[key] = value for key in self.other_fields: value = self.req_data.get(key) if value is None: continue vaild_dict[key] = value return vaild_dict, err def _create(self): vaild_data, err = self._get_vaild_values() if vaild_data: try: instance = self.model(**vaild_data) db.session.add(instance) db.session.commit() self.insatance = instance return True, "success" except: err = "please correct fileds" return False, err else: return False, err def _update(self): vaild_data, err = self._get_vaild_values() if vaild_data: try: instance = self.insatance if instance is None: return False, "not find data" for key, value in vaild_data.items(): setattr(instance, key, value) db.session.commit() return True, "success" except: err = "please correct fileds" return False, err else: return False, err def save(self): if self.insatance is None: ret, msg = self._create() if ret is False: abort(400, msg=msg) else: ret, msg = self._update() if ret is False: abort(400, msg=msg) @property def data(self): if self.insatance is None: msg = "data is not save" abort(500, msg=msg) return self.insatance.to_dict(only=tuple(self.serializer_fields)) def delete(self): if self.insatance is None: msg = "not find data" abort(400, msg=msg) else: try: db.session.delete(self.insatance) db.session.commit() except Exception: msg = "delete exception" abort(400, msg=msg)class Serialize(object): """ 序列化 """ model = None fields = "__all__" modelsDatas = [] many = True date_format = "%Y-%m-%d" datetime_format = "%Y-%m-%d %H:%M:%S" time_format = "%H:%M:%S" build_fiels = [] def __init__(self, serializers, many=True): self.modelsDatas = serializers self.many = many @property def data(self): li = [] try: if self.many: for data in self.modelsDatas: da = data.to_dict(only=tuple(self.serializer_fields), date_format=self.date_format, datetime_format=self.datetime_format, time_format=self.time_format) da.update(self._get_build_files_values(data)) li.append(da) else: da = self.modelsDatas.to_dict(only=tuple(self.serializer_fields), date_format=self.date_format, datetime_format=self.datetime_format, time_format=self.time_format) da.update(self._get_build_files_values(self.modelsDatas)) li.append(da) except Exception as e: print(e) msg = "serialize error" li.append(msg) abort(500, msg=msg) return li @property def serializer_fields(self): return self.fields if self.fields != "__all__" else self.model().serializable_keys def _get_build_files_values(self, data): dit = {} for build in self.build_fiels: obj = data if build.get("method"): func = f"get_{build["name"]}" f = getattr(self, func) value = f(obj) dit[build["name"]] = value else: source_list = build["source"].split(".") value = None for source in source_list: value = getattr(obj, source, None) if value: obj = value else: break dit[build["name"]] = value return ditclass ParseQuery(object): """ 查询,排序 """ filer_query = frozenset(["gt", "ge", "lt", "le", "ne", "eq", "ic", "ni", "in"]) def __init__(self, model, req_data, filter_list=[], order_by=None): self.model = model self.req_data = req_data self.filter_list = filter_list self.order_by = order_by self._operator_funcs = { "gt": self.__gt_model, "ge": self.__ge_model, "lt": self.__lt_model, "le": self.__le_model, "ne": self.__ne_model, "eq": self.__eq_model, "ic": self.__ic_model, "ni": self.__ni_model, # "by": self.__by_model, "in": self.__in_model, } @property def _filter_data(self): search_dict = {} for fit in self.filter_list: val = self.req_data.get(fit) key = fit.split("__")[0] if val and hasattr(self.model, key): search_dict[fit] = val return search_dict def _parse_fields(self): li = [] for search_key, value in self._filter_data.items(): key, ope = search_key.split("__") if ope in self.filer_query: data = self._operator_funcs[ope](key=key, value=value) li.append(data) return li def _filter(self): data = tuple(self._parse_fields()) quety_data = self.model.query.filter(*data) if self.order_by: data = self._parse_order_by() quety_data = quety_data.order_by(*data) return quety_data @property def query(self): return self._filter() def pagination_class(self, page_num=1, page_size=10, max_page_size=50, error_out=False): pagin = self.query.paginate( page=page_num, per_page=page_size, error_out=error_out, max_per_page=max_page_size ) return pagin.items, pagin.total def _parse_order_by(self): """ 解析排序 :return: """ li = [] for ord in list(self.order_by): if ord.find("-") == -1: data = self.__by_model(ord) if data: li.append(data.asc()) else: ord = ord[1:] data = self.__by_model(ord) if data: li.append(data.desc()) return tuple(li) def __by_model(self, key): """ 排序时获取字段 :return: """ return getattr(self.model, key) def __gt_model(self, key, value): """ 大于 :param key: :param value: :return: """ return getattr(self.model, key) > value def __ge_model(self, key, value): """ 大于等于 :param key: :param value: :return: """ return getattr(self.model, key) >= value def __lt_model(self, key, value): """ 小于 :param key: :param value: :return: """ return getattr(self.model, key) < value def __le_model(self, key, value): """ 小于等于 :param key: :param value: :return: """ return getattr(self.model, key) <= value def __eq_model(self, key, value): """ 等于 :param key: :param value: :return: """ return getattr(self.model, key) == value def __ne_model(self, key, value): """ 不等于 :param key: :param value: :return: """ return getattr(self.model, key) != value def __ic_model(self, key, value): """ 包含 :param key: :param value: :return: """ return getattr(self.model, key).like("%{}%".format(value)) def __ni_model(self, key, value): """ 不包含 :param key: :param value: :return: """ return getattr(self.model, key).notlike("%{}%".format(value)) def __in_model(self, key, value): """ 查询多个相同字段的值 :param key: :param value: :return: """ return getattr(self.model, key).in_(value)
自定义序列化和反序列化后,接口将变得简单
上面的类视图将只有短短的几行代码
关键词:
flask自定义参数校验、序列化和反序列化_视讯
出庭当晚特朗普与“金主”举行晚宴,筹得200多万美元-世界微头条
微资讯!新华社权威快报|“一箭41星” 发射成功
欠2.4亿罚款又如何?贾跃亭国内“现身”:我不要烧太多钱也能快速成功 环球快讯
端午节放假3天自驾方便了!滴滴租车正式上线:跑车、房车全都有
全球即时:一机搞定全屋清洁!石头新品洗地机A10 Ultra今晚首销:到手价3399元
1799元咬死RTX 3060显卡 英特尔新驱动继续鸡血:性能猛增33%|当前快看
用上丰田氢燃料电池 海马7X-H来了:不到5分钟加注续航达800km-天天微速讯
汉字找茬王连歌曲不甘怎么过-世界热闻
全球热文:解析Spring内置作用域及其在实践中的应用
世界观点:pickle
RPG游戏开发日志: 世界观速讯
石家庄太和网点电话_石家庄太和网
3个ChatGPT插件自动写书爆火!更多躺着赚钱玩法可以问AI自己_全球看热讯
3.1级地震发生时上海居民躲床下避难 监控视频显示:剧烈摇晃、余震可能性不大|今头条
全球热点评!盘点C#最有价值的10个语法糖
Cannot Reference “XxxClass.xxx” Before Supertype Constructor Has Been Called
足球服批发_足球装备批发|热推荐
男子赶集买毛蛋到家全变鸭子 后续会饲养:网友点赞赚大了
直言理想ONE被问界M7打残 学习华为要看什么书:李想整理分享
丰田手把手教经销商诋毁纯电|每日动态
矛盾的马斯克!特斯拉4D雷达首拆:千元成本、探测距离300米
女学霸边读研边兼职3年赚17万:为了明确自己毕业后适合什么工作
开源即时通讯IM框架MobileIMSDK的H5端开发快速入门-世界关注
港股午评:恒指涨0.83% 恒生科技指数涨1.82% 天天看点
热点在线丨会定期删除 微软回应Edge私传图片问题:不含任何用户标识
天玑9200+加持 vivo X90s官宣:安卓最强悍的曲面屏旗舰
全球聚焦:李想最新发言引围观:我们谁都没兴趣搞、那点销量有啥可搞的
温州近视司机深夜高速上镜片突然脱落:两眼一抹黑
动态焦点:张雪峰称高考绝对是普通家庭改变命运的最好出路 回应建议报理科
2023年农业电商行业发展现状调查及行业未来趋势分析
用益-今日财经视点:美联储如期暂停加息但放鹰!
深度学习应用篇-元学习[15]:基于度量的元学习:SNAIL、RN、PN、MN
『题解』BZOJ2839 集合计数|天天热点评
热推荐:00-串口和SSH方式登录
戴德梁行亚洲REITs报告:C-REITs市场快速扩容 ESG将成为高质量发展新动力
因为读书值得
每日动态!六合一套装:八喜冰淇淋3.8元/杯抄底(商超8元)
巴黎治鼠患官宣失败:市民将与600万只老鼠共存
世界微速讯:全系降价3万后 蔚来ET5旅行版今日首发上市:这颜值30万你买吗?
每日热门:618最值得入手智能门锁非它莫属:华为智能门锁系列该怎么选?
业内第一!小鹏首宣开放北京城市智能辅助驾驶:不靠高精地图|天天热点
《长安三万里》曝角色海报 7月8日梦回诗意大唐 要闻
顶奢好文:3W字,穿透Spring事务原理、源码,至少读10遍|焦点速读
c++ mutex 每日消息
微头条丨月的暗面——戴冰选集_对于月的暗面——戴冰选集简单介绍
当前快报:大导演冯小刚多年的白癜风“消失”了引围观 这病真能治好?医生释疑
天天热资讯!高速错过出口 丰田亚洲龙缓行被大货车追尾致5死 官方:两司机均分心驾驶
全球聚焦:李想:理想MPV不会是埃尔法“私生子” 开上街比法拉利还瞩目
女子穿联名汉服进入迪士尼被拦 官方回应:网友吵翻 说好的穿衣自由呢
视讯!对象说冷怎么回
北斗GPS卫星授时服务器(NTP授时)应用于地铁自控系统
最新SMS-Activate短信验证码接收教程-观天下
陕西已累计收获小麦1213.63万亩占应收面积的八成以上
环球快资讯:女孩抓蛙一身泥巴被妈妈逮到 毫无畏惧:网友感慨这才是童年
热点评!《王者荣耀》诸葛亮新皮肤上线:最强大的仙尊 能御剑飞行
不花钱给电脑降温?有两种方法可行-全球聚焦
外卖起送价太高、凑单满减浪费?两部门发文
中国移动5G重大突破:无需安装APP 远程操控智能终端
这项重要工程动工,惠及众多荔湾街坊!|早安,荔湾-环球最资讯
ESP32物联网项目开发1-ESP32简介
美联储如市场预期“跳过”加息 2年期与10年期美债利差走阔 讯息
每日消息!雷军科普:金属背壳的手机曾经非常流行 为什么现在却很少有人用了?
曝驱逐舰07七月预售:搭载云辇-C 车长超比亚迪汉
苹果最强影像旗舰!iPhone 15 Pro Max相机曝光|环球即时
每隔俩烽火台设一个许愿栏又有何不可?
读发布!设计与部署稳定的分布式系统(第2版)笔记03_让系统稳定运行 每日精选
观热点:考古女生寄200斤洛阳铲装备吓坏父母 专业的标配
天天微动态丨用iPhone和用安卓的人 为什么谁也不服谁
快讯:药康生物(688046):6月14日北向资金增持1.11万股
焦点速讯:重新整理 .net core 实践篇——— 测试控制器[四十九]
云南一地多次发生巨响震动!官方回应:初判为浅源性地震-环球快资讯
天天观速讯丨英伟达太强 AMD最强显卡尚无客户采购:软件干不过CUDA
小米14设计定了:标准版小直屏 Pro版极窄微曲屏-环球热头条
快看点丨给全世界每个人发钱 OpenAI创始人这么飘了?
比尔·盖茨抵达北京:就健康和环境问题向中国取经|当前简讯
(上接第一版)
无敌时间_关于无敌时间介绍
创业板注册制IPO申请“开闸”满三年:千企竞逐上市梦 创业板加速构建创新资本中心
中央美术学院教授黄永玉去世 为具有重大影响力的艺术家
张柏芝发文呼吁大家关注糖尿病年轻化 减轻职场人群“高甜负担”
多地提前进入“烧烤模式” 新疆东部和南疆盆地有35℃以上高温
重金买下哈利波特老宅的屋主感到后悔 朝圣粉丝令其苦不堪言
歌手苏醒晒出自己和“球王”梅西的合影 兴奋之情溢于言表
腾讯音乐与原神达成深度战略合作 经典之声再度重逢
2023年“惠购湖北”家电消费券即将发放 总计投入财政资金3亿元
Set 接口及其常用方法
对标英伟达A100 国产GPU沐曦MXC500成功点亮:仅用时5个小时|世界消息
奶奶带孙子时邻居反馈“不像他爸” 去做亲子鉴定后崩溃:真不是
微软创始人比尔盖茨抵达北京!2019年以来首次访问
小米米家首款自发电动感单车采用南芯芯片:支持20W有线/无线快充
环球看热讯:买下《哈利波特》电影中房子的屋主后悔了 每天都有粉丝来打卡
澳琴企业首次亮相第九届上交会_热资讯
百度文心一格×京东618:打造电商行业首个AI线下广告,节省80%制作成本|焦点快看
环球视讯!人人有份!湖北将发放3亿元消费券:最大面额500元 京东可用
世界热点评!作别“战国七雄” 中国快递业变天了:进入三足鼎立
阿里巴巴集团公关部将独立成公关公司?知情人士:假的-环球关注
微控制器实时操作系统实践5选择IDE 全球短讯
文心一言 VS 讯飞星火 VS chatgpt (41)-- 算法导论5.4 5题
【环球播资讯】破“3”!30年期国债收益率创2006年以来新低