使用Pydantic开发时,如何将返回数据由snake_case自动转为camel case
@[TOC]使用Pydantic开发时,如何将返回数据由snake_case自动转为camel case)
在 FastAPI 中,如果你希望自动将请求和响应中的字段名从 Python 的下划线命名(如 user_name)转换为 Java 的驼峰命名(如 userName),可以通过 Pydantic 的配置来实现。
下面的测试代码,使用的是pydantic 2.11.7
(topic_backend) c:\Code\topic-backend>pip show pydantic
Name: pydantic
Version: 2.11.7
Summary: Data validation using Python type hints
Home-page: https://github.com/pydantic/pydantic
Author:
Author-email: Samuel Colvin <s@muelcolvin.com>, Eric Jolibois <em.jolibois@gmail.com>, Hasan Ramezani <hasan.r67@gmail.com>, Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com>, Terrence Dorsey <terry@pydantic.dev>, David Montague <david@pydantic.dev>, Serge Matveenko <lig@countzero.co>, Marcelo Trylesinski <marcelotryle@gmail.com>, Sydney Runkle <sydneymarierunkle@gmail.com>, David Hewitt <mail@davidhewitt.io>, Alex Hall <alex.mojaki@gmail.com>, Victorien Plot <contact@vctrn.dev>
License-Expression: MIT
Location: c:\anaconda3\envs\topic_backend\lib\site-packages
Requires: annotated-types, pydantic-core, typing-extensions, typing-inspection
Required-by: fastapi(topic_backend) c:\Code\topic-backend>
FastAPI 使用 Pydantic 模型处理数据序列化和反序列化。你可以通过定义一个 alias_generator 函数将下划线命名转换为驼峰命名,并配置模型使用它。
定义类custom_style_model.py
from pydantic import BaseModel
import redef to_camel(string: str) -> str:return re.sub(r'_([a-zA-Z])', lambda m: m.group(1).upper(), string)class CamelModel(BaseModel):model_config = {"alias_generator": to_camel,"populate_by_name": True, }
Schema定义示例
将原来Schemal定义中的BaseModel,改为新的CamelModel
from pydantic import BaseModel
from typing import Optional
from app.schemas.custom_style_model import CamelModel, to_camel
import app.schemas.custom_style_model as custom_style_modelclass node_relationship_info(CamelModel):node_id_1: intnode_name_1: strnode_label_1: strnode_property_1: Optional[dict] = Nonenode_id_2: intnode_name_2: strnode_label_2: strnode_property_2: Optional[dict] = Nonerelationship_name: strrelationship_property: Optional[dict] = None
应用中的数据校验方式如下:
def list_node_relationship_info_by_page(skip: int = 0, limit: int = 10) -> List[node_relationship_info]:"""获取知识图谱节点关系信息(分页):param skip: 跳过数量:param limit: 限制数量:return: 知识图谱节点关系信息列表"""try:neo4j_handler = Neo4jHandler()# 查询所有节点和关系,包含分页query = """MATCH (n1)-[r]->(n2)RETURN id(n1) as node_id_1,n1.name as node_name_1, labels(n1)[0] as node_label_1, id(n2) as node_id_2,n2.name as node_name_2, labels(n2)[0] as node_label_2, properties(n1) as node_property_1, properties(n2) as node_property_2, type(r) as relationship_name,properties(r) as relationship_propertySKIP $skip LIMIT $limit"""result = neo4j_handler.execute_query(query, {"skip": skip, "limit": limit})# 转换结果为node_relationship_info对象列表node_relationship_list = []for record in result:print(type(record))print(record)try:node_id_1=record.get("node_id_1", ""),node_id_2=record.get("node_id_2", ""),node_name_1=record.get("node_name_1", ""),# print(type(node_name_1))node_name_2=record.get("node_name_2", ""),node_label_1=record.get("node_label_1", ""),node_label_2=record.get("node_label_2", ""),# node_property_1=record.get("node_property_1"),# node_property_2=record.get("node_property_2"),relationship_name=record.get("relationship_name", ""),node_id_1 = extract_data_value(node_id_1) node_id_2 = extract_data_value(node_id_2) node_name_1 = extract_data_value(node_name_1) node_name_2 = extract_data_value(node_name_2)node_label_1 = extract_data_value(node_label_1)node_label_2 = extract_data_value(node_label_2)relationship_name = extract_data_value(relationship_name)# relationship_property=record.get("relationship_property")record = {"node_id_1": node_id_1,"node_name_1": node_name_1,"node_label_1": node_label_1,"node_id_2": node_id_2,"node_name_2": node_name_2,"node_label_2": node_label_2,"relationship_name": relationship_name}node_info = node_relationship_info.model_validate(record)node_relationship_list.append(node_info)except Exception as e:logger.warning(f"转换记录失败: {e}, 记录: {record}")print(e)continueneo4j_handler.close()return node_relationship_listexcept Exception as e:logger.error(f"获取知识图谱节点关系信息失败: {e}")neo4j_handler.close()return []
测试结果
原返回结果
处理后的返回结果