mirror of
https://github.com/fastapi-practices/fastapi-best-architecture.git
synced 2026-09-21 13:12:24 +00:00
clean up the utils code
This commit is contained in:
@@ -1,64 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
import json
|
||||
|
||||
|
||||
def list_to_tree(data_list, parent_id=0) -> list:
|
||||
"""
|
||||
递归获取树形结构数据
|
||||
|
||||
:param data_list: 数据列表
|
||||
:param parent_id: 父级id
|
||||
:return:
|
||||
"""
|
||||
tree = []
|
||||
for _data in data_list:
|
||||
if _data['parent_id'] == parent_id:
|
||||
tree.append(_data)
|
||||
_data['children'] = list_to_tree(data_list, _data['id'])
|
||||
return tree
|
||||
|
||||
|
||||
def ram_list_to_tree(data_list: list) -> list:
|
||||
"""
|
||||
利用对象内存共享生成树
|
||||
|
||||
:param data_list: 数据列表
|
||||
:return:
|
||||
"""
|
||||
res = {}
|
||||
for v in data_list:
|
||||
res.setdefault(v["id"], v)
|
||||
for v in data_list:
|
||||
res.setdefault(v["parent_id"], {}).setdefault("children", []).append(v)
|
||||
return res[0]["children"]
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
test_data1 = [
|
||||
{'id': 1, 'title': 'GGG', 'parent_id': 0},
|
||||
{'id': 2, 'title': 'AAA', 'parent_id': 0},
|
||||
{'id': 3, 'title': 'BBB', 'parent_id': 1},
|
||||
{'id': 4, 'title': 'CCC', 'parent_id': 1},
|
||||
{'id': 5, 'title': 'DDD', 'parent_id': 2},
|
||||
{'id': 6, 'title': 'EEE', 'parent_id': 3},
|
||||
{'id': 7, 'title': 'FFF', 'parent_id': 4},
|
||||
{'id': 3, 'title': 'BBB', 'parent_id': 1},
|
||||
]
|
||||
|
||||
print(json.dumps(list_to_tree(test_data1), indent=4))
|
||||
|
||||
test_data2 = [
|
||||
{'id': 10, 'parent_id': 8, 'name': "ACAB"},
|
||||
{'id': 9, 'parent_id': 8, 'name': "ACAA"},
|
||||
{'id': 8, 'parent_id': 7, 'name': "ACA"},
|
||||
{'id': 7, 'parent_id': 1, 'name': "AC"},
|
||||
{'id': 6, 'parent_id': 3, 'name': "ABC"},
|
||||
{'id': 5, 'parent_id': 3, 'name': "ABB"},
|
||||
{'id': 4, 'parent_id': 3, 'name': "ABA"},
|
||||
{'id': 3, 'parent_id': 1, 'name': "AB"},
|
||||
{'id': 2, 'parent_id': 0, 'name': "AA"},
|
||||
{'id': 1, 'parent_id': 0, 'name': "A"},
|
||||
]
|
||||
|
||||
print(json.dumps(ram_list_to_tree(test_data2), indent=4))
|
||||
@@ -1,16 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
from backend.app.core.path_conf import AvatarPath
|
||||
|
||||
|
||||
def cut_path(path: str = AvatarPath, split_point: str = 'app') -> list:
|
||||
"""
|
||||
切割路径
|
||||
|
||||
:param path:
|
||||
:param split_point:
|
||||
:return:
|
||||
"""
|
||||
after_path = path.split(split_point)
|
||||
return after_path
|
||||
@@ -1,25 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
from fastapi import Form
|
||||
|
||||
|
||||
def encode_as_form(cls):
|
||||
"""
|
||||
pydantic 类装饰器,将 pydantic 类转化为 form_data
|
||||
|
||||
示例::
|
||||
|
||||
@encode_as_form
|
||||
class Pydantic(BaseModel):
|
||||
...
|
||||
|
||||
:param cls:
|
||||
:return:
|
||||
"""
|
||||
cls.__signature__ = cls.__signature__.replace(
|
||||
parameters=[
|
||||
arg.replace(default=Form(...))
|
||||
for arg in cls.__signature__.parameters.values()
|
||||
]
|
||||
)
|
||||
return cls
|
||||
@@ -41,23 +41,3 @@ def is_mobile(text: str) -> bool:
|
||||
:return:
|
||||
"""
|
||||
return match_string(r"^1[3-9]\d{9}$", text)
|
||||
|
||||
|
||||
def is_wechat(text: str) -> bool:
|
||||
"""
|
||||
检查微信号
|
||||
|
||||
:param text:
|
||||
:return:
|
||||
"""
|
||||
return match_string(r"^[a-zA-Z]([-_a-zA-Z0-9]{5,19})+$", text)
|
||||
|
||||
|
||||
def is_qq(text: str) -> bool:
|
||||
"""
|
||||
检查QQ号
|
||||
|
||||
:param text:
|
||||
:return:
|
||||
"""
|
||||
return match_string(r"^[1-9][0-9]{4,10}$", text)
|
||||
|
||||
@@ -1,45 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
from email.mime.multipart import MIMEMultipart
|
||||
from email.mime.text import MIMEText
|
||||
|
||||
import aiosmtplib
|
||||
|
||||
from backend.app.common.log import log
|
||||
from backend.app.core.conf import settings
|
||||
from backend.app.utils.generate_string import get_uuid
|
||||
|
||||
__only_code = get_uuid()
|
||||
|
||||
SEND_RESET_PASSWORD_TEXT = f"您的重置密码验证码为:{__only_code}\n为了不影响您正常使用," \
|
||||
f"请在{int(settings.COOKIES_MAX_AGE / 60)}分钟内完成密码重置"
|
||||
|
||||
SEND_EMAIL_LOGIN_TEXT = f"您的登录验证码为:{__only_code}\n" \
|
||||
f"请在{int(settings.EMAIL_LOGIN_CODE_MAX_AGE / 60)}分钟内完成登录"
|
||||
|
||||
|
||||
async def send_verification_code_email(to: str, code: str, text: str = SEND_RESET_PASSWORD_TEXT):
|
||||
"""
|
||||
发送验证码电子邮件
|
||||
|
||||
:param to:
|
||||
:param code:
|
||||
:param text:
|
||||
:return:
|
||||
"""
|
||||
text = text.replace(__only_code, code)
|
||||
msg = MIMEMultipart()
|
||||
msg['Subject'] = settings.EMAIL_DESCRIPTION
|
||||
msg['From'] = settings.EMAIL_USER
|
||||
msg.attach(MIMEText(text, _charset='utf-8'))
|
||||
|
||||
# 登录smtp服务器并发送邮件
|
||||
try:
|
||||
smtp = aiosmtplib.SMTP(hostname=settings.EMAIL_SERVER, port=settings.EMAIL_PORT, use_tls=settings.EMAIL_SSL)
|
||||
async with smtp:
|
||||
await smtp.login(settings.EMAIL_USER, settings.EMAIL_PASSWORD)
|
||||
await smtp.sendmail(msg['From'], to, msg.as_string())
|
||||
await smtp.quit()
|
||||
except Exception as e:
|
||||
log.error('邮件发送失败 {}', e)
|
||||
raise Exception('邮件发送失败 {}'.format(e))
|
||||
Reference in New Issue
Block a user