惯性聚合 高效追踪和阅读你感兴趣的博客、新闻、科技资讯
阅读原文 在惯性聚合中打开

推荐订阅源

F
Fortinet All Blogs
Microsoft Security Blog
Microsoft Security Blog
Cyber Security Advisories - MS-ISAC
Cyber Security Advisories - MS-ISAC
Threat Intelligence Blog | Flashpoint
Threat Intelligence Blog | Flashpoint
Vercel News
Vercel News
Application and Cybersecurity Blog
Application and Cybersecurity Blog
C
Check Point Blog
cs.AI updates on arXiv.org
cs.AI updates on arXiv.org
W
WeLiveSecurity
The Hacker News
The Hacker News
L
LINUX DO - 热门话题
T
Tenable Blog
Hugging Face - Blog
Hugging Face - Blog
Google Online Security Blog
Google Online Security Blog
博客园 - Franky
P
Proofpoint News Feed
H
Hacker News: Front Page
P
Privacy & Cybersecurity Law Blog
月光博客
月光博客
P
Proofpoint News Feed
Exploit-DB.com RSS Feed
Exploit-DB.com RSS Feed
The GitHub Blog
The GitHub Blog
云风的 BLOG
云风的 BLOG
博客园_首页
www.infosecurity-magazine.com
www.infosecurity-magazine.com
C
CERT Recently Published Vulnerability Notes
Forbes - Security
Forbes - Security
I
InfoQ
Stack Overflow Blog
Stack Overflow Blog
OSCHINA 社区最新新闻
OSCHINA 社区最新新闻
Attack and Defense Labs
Attack and Defense Labs
N
News and Events Feed by Topic
博客园 - 叶小钗
T
Threat Research - Cisco Blogs
aimingoo的专栏
aimingoo的专栏
D
Darknet – Hacking Tools, Hacker News & Cyber Security
小众软件
小众软件
大猫的无限游戏
大猫的无限游戏
MongoDB | Blog
MongoDB | Blog
CTFtime.org: upcoming CTF events
CTFtime.org: upcoming CTF events
Hacker News - Newest:
Hacker News - Newest: "LLM"
钛媒体:引领未来商业与生活新知
钛媒体:引领未来商业与生活新知
博客园 - 司徒正美
O
OpenAI News
G
Google Developers Blog
Martin Fowler
Martin Fowler
罗磊的独立博客
S
SegmentFault 最新的问题
T
Tor Project blog
量子位

博客园 - 干炸小黄鱼

timex 处理时间戳 gorm-gen go 雪花算法 golang每日一库--协程池库ants golang每日一库--json解析库gjson python高级编程-asyncio python高级编程-condition python高级编程-event python装饰器-自动重试 EAP系统 go实现实现 SECS/GEM 协议 设备通信协议 SECS go项目使用Jenkins进行CICD go操作ES mongo db聚合查询 go如何使用mongodb Apache ShardingSphere paxos and raft (分布式一致性算法) go使用zookeeper分布式锁以及和redis差异 go使用 seata 示例 Alibaba 分布式事务 Seata go中使用saga go中使用TCC示例 分布式事务TCC 熔断器 Hystrix OR Sentinel k8s下部署consul and etcd Consul OR Etcd 【力扣hot100】双指针-盛水最多的容器 【力扣hot100】滑动窗口-最小覆盖子串 shell脚本合集 分布式id生成器 springboot通用CURD Python PB级检索系统架构设计 rancher 在三台机器搭建k8s集群 python ssh clinet 数据库排序Null值字段靠后/靠前 常规web项目 docker-compose 例子 手搓一个验证码 使用itertools 中的groupby 对字典数组进行分组后排序 使用开源库 geoip2 获取某ip的经纬度地理信息 python中 apscheduler.schedulers.blocking.BlockingScheduler 定时执行任务 简单的python web项目的docker-compose.yml 示例 python和sliver交互 golang sliver二次开发自定义命令(格式乱后面再调) 基于rancher部署k8s 地理位置相关基础数据 flask migrate时报错 Can't locate revision identified by '3d80e4c025df'
pydantic做参数校验
干炸小黄鱼 · 2024-07-31 · via 博客园 - 干炸小黄鱼

定义一个统一的schema类对提交的业务参数进行格式和数据约束非常有必要,
下面使用 pydantic 来封装此工具;

import logging
from pydantic import BaseModel, ValidationError, root_validator

class PydanticValidationError(Exception):
    def __init__(self, msg):
        self.message = msg


class BaseSchema(BaseModel):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    class Config:
        anystr_strip_whitespace = True
        use_enum_values = True
        arbitrary_types_allowed = True

    @root_validator(pre=True)
    def _pre_empty_data(cls, values: dict):
        """将空字符串或null字符串转换为None"""
        for k, v in values.items():
            if v == "" or v == "null":
                values[k] = None
        return values

    @classmethod
    def data_validation(cls, data):
        """参数校验,并自定义返回信息"""
        # TODO 这里并不一定是全部的,后面如果碰到其他的,再添加
        try:
            res = cls.parse_obj(data)
            errs = []
        except ValidationError as e:
            res = None
            errs = e.errors()
        for err in errs:
            logging.exception(err)
            fields = list(err["loc"])
            field = fields[0]
            field_info = cls.__fields__.get(field).field_info  # type: ignore
            type_info = err["type"].split(".")
            fields[0] = field_info.title if field_info.title else field
            title = ">>".join([str(tmp) for tmp in fields])
            if len(type_info) == 2:
                if type_info[0] == "type_error" and type_info[1] == "enum":
                    raise PydanticValidationError(f"{title}数据错误,请传指定可选值")
                if type_info[0] == "type_error":
                    raise PydanticValidationError(f"{title}数据类型错误,需要是{type_info[1]}类型")
                if type_info[0] == "value_error" and type_info[1] == "missing":
                    raise PydanticValidationError(f"{title}不能为空")
                if type_info[0] == "value_error" and type_info[1] == "const":
                    raise PydanticValidationError(f"{title}数据错误,请传指定值")
                if type_info[0] == "value_error" and type_info[1] == "ipv4address":
                    raise PydanticValidationError(f"{title}数据错误,需要是ipv4地址")
            if len(type_info) == 3:
                if type_info[2] == "not_gt":
                    raise PydanticValidationError(f"{title}的数值必须大于{field_info.gt}")
                if type_info[2] == "not_lt":
                    raise PydanticValidationError(f"{title}的数值必须小于{field_info.lt}")
                if type_info[2] == "not_ge":
                    raise PydanticValidationError(f"{title}的数值必须大于或等于{field_info.ge}")
                if type_info[2] == "not_le":
                    raise PydanticValidationError(f"{title}的数值必须小于或等于{field_info.le}")
                if type_info[2] == "min_length":
                    raise PydanticValidationError(f"{title}的最小字符长度为{field_info.min_length}")
                if type_info[2] == "max_length":
                    raise PydanticValidationError(f"{title}的最大字符长度为{field_info.max_length}")
                if type_info[2] == "min_items":
                    raise PydanticValidationError(f"{title}中的最少需要{field_info.min_items}个元素")
                if type_info[2] == "max_items":
                    raise PydanticValidationError(f"{title}中的最多只能{field_info.min_items}个元素")
                if type_info[1] == "none" and type_info[2] == "not_allowed":
                    raise PydanticValidationError(f"{title}不能为空")
            raise PydanticValidationError(f"{title}参数错误")
        return res

定义一个schema来接收参数, 它继承BaseSchema

class VictimOrgSchema(BaseSchema):
    title: str = Field(max_length=50, title="组织名称", description="组织名称")
    street: Optional[str] = Field(max_length=255, title="街道具体地址", description="街道具体地址", default="")
    latitude: str = Field(description="经度")
    longitude: str = Field(description="纬度")
    first_id: Optional[int] = Field(description="一级总指挥id")
    second_id: Optional[int] = Field(description="二级总指挥id")
    third_id: Optional[int] = Field(description="三级总指挥id")
    department_id: Optional[int] = Field(title="行业性质",description="行业性质")
    org_nature: int = Field(title="组织性质",description="组织性质")
    contacts: Optional[str] = Field(max_length=255, title="联系人", description="联系人")
    phone: Optional[str] = Field(max_length=11, title="联系电话", description="联系电话")
    logo: Union[FileStorage, str, None] = Field(description="logo")
    province: int = Field(description="省份code")
    city: int = Field(description="城市code")
    team_target: int = Field(description="是否靶标单位", default=0)

view里面进行参数校验

args = dict(request.form) if request.form else {}
data_obj = VictimOrgSchema.data_validation(args)