77from openpyxl .utils import get_column_letter
88from openpyxl .worksheet .datavalidation import DataValidation
99from sqlalchemy .engine .row import Row
10- from typing import List
10+ from typing import Any , Dict , List , Literal , Union
11+ from config .database import Base
1112from config .env import CachePathConfig
1213
1314
@@ -38,13 +39,71 @@ def worship():
3839 """ )
3940
4041
42+ class SqlalchemyUtil :
43+ """
44+ sqlalchemy工具类
45+ """
46+
47+ @classmethod
48+ def base_to_dict (
49+ cls , obj : Union [Base , Dict ], transform_case : Literal ['no_case' , 'snake_to_camel' , 'camel_to_snake' ] = 'no_case'
50+ ):
51+ """
52+ 将sqlalchemy模型对象转换为字典
53+
54+ :param obj: sqlalchemy模型对象或普通字典
55+ :param transform_case: 转换得到的结果形式,可选的有'no_case'(不转换)、'snake_to_camel'(下划线转小驼峰)、'camel_to_snake'(小驼峰转下划线),默认为'no_case'
56+ :return: 字典结果
57+ """
58+ if isinstance (obj , Base ):
59+ base_dict = obj .__dict__ .copy ()
60+ base_dict .pop ('_sa_instance_state' , None )
61+ elif isinstance (obj , dict ):
62+ base_dict = obj .copy ()
63+ if transform_case == 'snake_to_camel' :
64+ return {CamelCaseUtil .snake_to_camel (k ): v for k , v in base_dict .items ()}
65+ elif transform_case == 'camel_to_snake' :
66+ return {SnakeCaseUtil .camel_to_snake (k ): v for k , v in base_dict .items ()}
67+
68+ return base_dict
69+
70+ @classmethod
71+ def serialize_result (
72+ cls , result : Any , transform_case : Literal ['no_case' , 'snake_to_camel' , 'camel_to_snake' ] = 'no_case'
73+ ):
74+ """
75+ 将sqlalchemy查询结果序列化
76+
77+ :param result: sqlalchemy查询结果
78+ :param transform_case: 转换得到的结果形式,可选的有'no_case'(不转换)、'snake_to_camel'(下划线转小驼峰)、'camel_to_snake'(小驼峰转下划线),默认为'no_case'
79+ :return: 序列化结果
80+ """
81+ if isinstance (result , (Base , dict )):
82+ return cls .base_to_dict (result , transform_case )
83+ elif isinstance (result , list ):
84+ return [cls .serialize_result (row , transform_case ) for row in result ]
85+ elif isinstance (result , Row ):
86+ if all ([isinstance (row , Base ) for row in result ]):
87+ return [cls .base_to_dict (row , transform_case ) for row in result ]
88+ elif any ([isinstance (row , Base ) for row in result ]):
89+ return [cls .serialize_result (row , transform_case ) for row in result ]
90+ else :
91+ result_dict = result ._asdict ()
92+ if transform_case == 'snake_to_camel' :
93+ return {CamelCaseUtil .snake_to_camel (k ): v for k , v in result_dict .items ()}
94+ elif transform_case == 'camel_to_snake' :
95+ return {SnakeCaseUtil .camel_to_snake (k ): v for k , v in result_dict .items ()}
96+ return result_dict
97+ return result
98+
99+
41100class CamelCaseUtil :
42101 """
43102 下划线形式(snake_case)转小驼峰形式(camelCase)工具方法
44103 """
45104
46105 @classmethod
47- def snake_to_camel (cls , snake_str ):
106+ def snake_to_camel (cls , snake_str : str ):
48107 """
49108 下划线形式字符串(snake_case)转换为小驼峰形式字符串(camelCase)
50109
@@ -57,41 +116,14 @@ def snake_to_camel(cls, snake_str):
57116 return words [0 ] + '' .join (word .capitalize () for word in words [1 :])
58117
59118 @classmethod
60- def transform_result (cls , result ):
119+ def transform_result (cls , result : Any ):
61120 """
62121 针对不同类型将下划线形式(snake_case)批量转换为小驼峰形式(camelCase)方法
63122
64123 :param result: 输入数据
65124 :return: 小驼峰形式结果
66125 """
67- if result is None :
68- return result
69- # 如果是字典,直接转换键
70- elif isinstance (result , dict ):
71- return {cls .snake_to_camel (k ): v for k , v in result .items ()}
72- # 如果是一组字典或其他类型的列表,遍历列表进行转换
73- elif isinstance (result , list ):
74- return [
75- cls .transform_result (row )
76- if isinstance (row , (dict , Row ))
77- else (
78- cls .transform_result ({c .name : getattr (row , c .name ) for c in row .__table__ .columns }) if row else row
79- )
80- for row in result
81- ]
82- # 如果是sqlalchemy的Row实例,遍历Row进行转换
83- elif isinstance (result , Row ):
84- return [
85- cls .transform_result (row )
86- if isinstance (row , dict )
87- else (
88- cls .transform_result ({c .name : getattr (row , c .name ) for c in row .__table__ .columns }) if row else row
89- )
90- for row in result
91- ]
92- # 如果是其他类型,如模型实例,先转换为字典
93- else :
94- return cls .transform_result ({c .name : getattr (result , c .name ) for c in result .__table__ .columns })
126+ return SqlalchemyUtil .serialize_result (result = result , transform_case = 'snake_to_camel' )
95127
96128
97129class SnakeCaseUtil :
@@ -100,7 +132,7 @@ class SnakeCaseUtil:
100132 """
101133
102134 @classmethod
103- def camel_to_snake (cls , camel_str ):
135+ def camel_to_snake (cls , camel_str : str ):
104136 """
105137 小驼峰形式字符串(camelCase)转换为下划线形式字符串(snake_case)
106138
@@ -112,41 +144,14 @@ def camel_to_snake(cls, camel_str):
112144 return re .sub ('([a-z0-9])([A-Z])' , r'\1_\2' , words ).lower ()
113145
114146 @classmethod
115- def transform_result (cls , result ):
147+ def transform_result (cls , result : Any ):
116148 """
117149 针对不同类型将下划线形式(snake_case)批量转换为小驼峰形式(camelCase)方法
118150
119151 :param result: 输入数据
120152 :return: 小驼峰形式结果
121153 """
122- if result is None :
123- return result
124- # 如果是字典,直接转换键
125- elif isinstance (result , dict ):
126- return {cls .camel_to_snake (k ): v for k , v in result .items ()}
127- # 如果是一组字典或其他类型的列表,遍历列表进行转换
128- elif isinstance (result , list ):
129- return [
130- cls .transform_result (row )
131- if isinstance (row , (dict , Row ))
132- else (
133- cls .transform_result ({c .name : getattr (row , c .name ) for c in row .__table__ .columns }) if row else row
134- )
135- for row in result
136- ]
137- # 如果是sqlalchemy的Row实例,遍历Row进行转换
138- elif isinstance (result , Row ):
139- return [
140- cls .transform_result (row )
141- if isinstance (row , dict )
142- else (
143- cls .transform_result ({c .name : getattr (row , c .name ) for c in row .__table__ .columns }) if row else row
144- )
145- for row in result
146- ]
147- # 如果是其他类型,如模型实例,先转换为字典
148- else :
149- return cls .transform_result ({c .name : getattr (result , c .name ) for c in result .__table__ .columns })
154+ return SqlalchemyUtil .serialize_result (result = result , transform_case = 'camel_to_snake' )
150155
151156
152157def bytes2human (n , format_str = '%(value).1f%(symbol)s' ):
0 commit comments