31 Commits
Author SHA1 Message Date
zhengxinonly 65e41ac9f1 style: 修改 fetch 请求为 ajax,统一请求风格 2024-04-21 13:01:48 +08:00
zhengxinonly b5057af31e fix(templates): 修复新增内容时还存留缓存数据 2024-01-31 23:51:10 +08:00
zhengxinonly e5a60da215 style: 修复错别字 2024-01-31 23:35:18 +08:00
zhengxinonly c6598df116 style: 修改返回信息格式 2024-01-31 22:52:41 +08:00
zhengxinonly 801eed010f refactor(all): 完成项目的前后端逻辑分离,静态页面可以单独部署 2024-01-28 17:43:25 +08:00
zhengxinonly 2b296c1c64 fix(templates): 修复引用样式错误 2023-12-23 00:51:31 +08:00
zhengxinonly b5625ecf1f refactor(templates): 移除 jinja2 includes 使用 2023-12-22 23:01:45 +08:00
zhengxinonly 7203bcfecc fix(passport): 修复初始化数据错误密码也可以登录问题 2023-12-22 22:46:54 +08:00
zhengxinonly 2f4f1faf82 fix: 完善新增权限操作 2023-12-05 16:50:21 +08:00
zhengxinonly 2ffddda02b style: pear admin layui 更新到 4.0 正式版 2023-11-26 00:51:02 +08:00
zhengxinonly 3086c1a1fa feature: 动态菜单生成 2023-11-07 20:53:20 +08:00
zhengxinonly eee1bda795 docs: 添加使用说明 2023-10-31 21:53:30 +08:00
zhengxinonly 7c07b6c94d feature: 用户角色授权 2023-10-31 21:45:10 +08:00
zhengxinonly fcf2ad232e feature: update flask to 3.0.0 2023-10-31 20:52:32 +08:00
zhengxinonly 537e21728b feature: 完成角色授权 2023-10-31 20:23:09 +08:00
zhengxinonly 6a8fb955f5 feature: 错误页面返回 2023-10-29 17:04:05 +08:00
zhengxinonly 261ac32825 feature: 用户管理 2023-10-29 15:19:14 +08:00
zhengxinonly 6daad895b4 feature: 部门增删改查 2023-10-27 20:51:10 +08:00
zhengxinonly 52e0737311 feature: 角色增删改 2023-10-14 20:20:46 +08:00
zhengxinonly 075b624522 feature: 权限列表-3 2023-10-14 19:17:01 +08:00
zhengxinonly 607cbc94ba feature: 权限列表-2 2023-10-10 22:55:15 +08:00
zhengxinonly f95ae54fd5 feature: 权限列表-01 2023-10-03 01:04:15 +08:00
zhengxinonly 20b611ac29 feature: 动态生成菜单栏 2023-10-02 00:57:20 +08:00
zhengxinonly 69a4955ce2 feature: 登录&权限拦截 2023-10-01 23:42:59 +08:00
zhengxinonly 8765b2c344 feature: 初始化 jwt 插件 2023-09-30 16:19:07 +08:00
zhengxinonly bdd766d2b2 feature: 初始化接口 api 2023-09-30 16:12:43 +08:00
zhengxinonly ba5e2cf67b feature: 添加静态文件 2023-09-29 14:57:03 +08:00
zhengxinonly 2c3c79b21f test: 添加测试数据 2023-09-29 01:48:46 +08:00
zhengxinonly 9bf89e5919 feature: 新增数据模型 2023-09-27 17:49:13 +08:00
zhengxinonly 4db050c93d feature: 配置文件&配置插件 2023-09-27 17:10:45 +08:00
zhengxinonly 49822daada feature: 项目初始化 2023-09-26 22:47:21 +08:00
1167 changed files with 156771 additions and 312224 deletions
+69
View File
@@ -0,0 +1,69 @@
[tool.commitizen]
name = "cz_customize"
tag_format = "$version"
version_scheme = "pep440"
version_provider = "poetry"
update_changelog_on_bump = true
major_version_zero = true
[tool.commitizen.customize]
message_template = "{{change_type}}{% if scope %}({{scope}}){% endif %}:{% if subject %} {{subject}}{% endif %} {% if footer %} {{footer}}{% endif %}"
example = "feature: this feature enable customize through config file"
schema = "<type>: <body>"
schema_pattern = "(feature|fix|docs|style|refactor|perf|test|revert|build).*?:(\\s.*)"
bump_pattern = "^(break|new|fix|hotfix)"
bump_map = { "break" = "MAJOR", "new" = "MINOR", "fix" = "PATCH", "hotfix" = "PATCH" }
change_type_order = ["BREAKING CHANGE", "feat", "fix", "refactor", "perf"]
info_path = "cz_customize_info.txt"
info = """
This is customized info
"""
commit_parser = "^(?P<change_type>feature|fix|docs|style|refactor|perf|test|revert|build).*?:\\s(?P<message>.*)?"
changelog_pattern = "^(feature|fix|docs|style|refactor|perf|test|revert|build)?(!)?"
change_type_map = { "feature" = "Feat", "bug fix" = "Fix" }
# 自定义提交指令
[[tool.commitizen.customize.questions]]
type = "list"
name = "change_type"
choices = [
{ value = "feature", name = "feature: 新功能" },
{ value = "fix", name = "fix: 修复 bug" },
{ value = "docs", name = "docs: 文档的修改" },
{ value = "style", name = "style: 格式化变动,不影响代码逻辑" },
{ value = "refactor", name = "refactor: 重构,即不是新增功能,也不是修改 bug 的代码变动" },
{ value = "perf", name = "perf: 性能优化" },
{ value = "test", name = "test: 增加测试" },
{ value = "revert", name = "revert: 回退" },
{ value = "build", name = "build: 打包" },
]
# choices = ["feature", "fix"] # short version
message = "请选择提交类型:"
# 修改内容的范围
[[tool.commitizen.customize.questions]]
type = "input"
name = "scope"
message = "请输入修改内容的范围(可选):"
# 自己的提交内容
[[tool.commitizen.customize.questions]]
type = "input"
name = "subject"
message = "请输入简要描述(必填):"
# 自己的提交内容
[[tool.commitizen.customize.questions]]
type = "input"
name = "body"
message = "请输入详细描述(可选):"
[[tool.commitizen.customize.questions]]
type = "input"
name = "footer"
message = "请输入需要关闭的 issue(可选):"
+4 -29
View File
@@ -1,29 +1,4 @@
# flask配置 FLASK_DEBUG=True # 开启调试模式
FLASK_APP=app.py FLASK_RUN_PORT=5000 # 设置运行的端口
FLASK_ENV=development FLASK_RUN_HOST=0.0.0.0 # 设置监听的 ip
FLASK_DEBUG=1 FLASK_APP="pear_admin:create_app()" # 运行程序
FLASK_RUN_HOST = 127.0.0.1
FLASK_RUN_PORT = 5000
# pear admin flask配置
SYSTEM_NAME = Pear Admin
# MySql配置信息
MYSQL_HOST=127.0.0.1
# MYSQL_HOST=dbserver
MYSQL_PORT=3306
MYSQL_DATABASE=PearAdminFlask
MYSQL_USERNAME=root
MYSQL_PASSWORD=123456
# Redis 配置
# REDIS_HOST=127.0.0.1
# REDIS_PORT=6379
# 密钥配置(记得改)
SECRET_KEY='pear-admin-flask'
# 邮箱配置
MAIL_SERVER='smtp.qq.com'
MAIL_USERNAME='123@qq.com'
MAIL_PASSWORD='XXXXX' # 生成的授权码
+4 -117
View File
@@ -1,118 +1,5 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
.hypothesis/
.pytest_cache/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
.python-version
# celery beat schedule file
celerybeat-schedule
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
#ide
.idea/ .idea/
.venv/
*.py[cod]
instance/
migrations/
+28
View File
@@ -0,0 +1,28 @@
repos:
- repo: https://github.com/python/black
rev: 23.9.1
hooks:
- id: black
language_version: python3.10
- repo: https://github.com/pycqa/isort
rev: 5.12.0
hooks:
- id: isort
args: [ "--profile", "black", "--filter-files" ]
- repo: https://github.com/pre-commit/mirrors-prettier
rev: "v3.0.3"
hooks:
- id: prettier
exclude: >
(?x)^(
.*.yaml|
package-lock.json|
yarn.lock|
^.+\.min\.(js|css)$
)$
stages: [ commit ]
- repo: https://github.com/commitizen-tools/commitizen
rev: 3.5.3
hooks:
- id: commitizen
stages: [ commit-msg ]
-25
View File
@@ -1,25 +0,0 @@
FROM python:3.7-alpine
RUN apk update \
&& apk --update add --no-cache gcc \
&& apk --update add --no-cache g++ \
&& apk --update add --no-cache tzdata \
&& apk --update add --no-cache libffi-dev \
&& apk --update add --no-cache libxslt-dev \
&& apk --update add --no-cache jpeg-dev
ENV TIME_ZONE Asia/Shanghai
ENV PIPURL "https://pypi.tuna.tsinghua.edu.cn/simple"
RUN echo "${TIME_ZONE}" > /etc/timezone \
&& ln -sf /usr/share/zoneinfo/${TIME_ZONE} /etc/localtime
COPY . /app
WORKDIR /app
RUN pip --no-cache-dir install -i ${PIPURL} --upgrade pip \
&& pip --no-cache-dir install -i ${PIPURL} -r requirement/requirement-dev.txt \
&& pip --no-cache-dir install -i ${PIPURL} gunicorn \
&& chmod +x start.sh
CMD ./start.sh
-21
View File
@@ -1,21 +0,0 @@
MIT License
Copyright (c) 2021 mkg
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
+42 -151
View File
@@ -1,164 +1,55 @@
<div align="center"> ## 本地启动
<br/>
<br/>
<h1 align="center">
Pear Admin Flask
</h1>
<h4 align="center">
开 箱 即 用 的 Flask 快 速 开 发 平 台
</h4>
[预 览](http://flask.pearadmin.com) | [官 网](http://www.pearadmin.com/) | [群聊](https://jq.qq.com/?_wv=1027&k=TkLAKUxC) | [社区](http://forum.pearadmin.com/) 环境要求:python 3.10 以上,安装了 poetry
初始化环境
<p align="center"> ```shell
<a href="#"> poetry install
<img src="https://img.shields.io/badge/pear%20admin%20flask-1.0.0-green" alt="Pear Admin Layui Version">
</a>
<a href="#">
<img src="https://img.shields.io/badge/Python-3.6+-green.svg" alt="Python Version">
</a>
<a href="#">
<img src="https://img.shields.io/badge/Mysql-5.3.2+-green.svg" alt="Mysql Version">
</a>
</p>
</div>
<div align="center">
<img width="92%" style="border-radius:10px;margin-top:20px;margin-bottom:20px;box-shadow: 2px 0 6px gray;" src="https://images.gitee.com/uploads/images/2020/1019/104805_042b888c_4835367.png" />
</div>
#### 项目简介
>Pear Admin Flask 基于 Flask 的后台管理系统,拥抱应用广泛的python语言,通过使用本系统,即可快速构建你的功能业务
>
>项目旨在为python开发者提供一个后台管理系统的模板,成为您构建信息管理系统,物联网后台....等等应用时灵活,简单的工具
>
>众人拾柴火焰高,欢迎pythoner参与项目~
Pear Admin Flask 有以下几个版本:
[master分支版本 ](https://gitee.com/pear-admin/pear-admin-flask/tree/master/)
> flask 2.0.1 + flask-sqlalchemy + mysql + 权限验证 + Flask-APScheduler 定时任务 + marshmallow 序列化与数据验证
[Mini 分支版本 ](https://gitee.com/pear-admin/pear-admin-flask/tree/mini/)
>flask 2.0.1 + flask-sqlalchemy + sqllite + 权限验证 + Flask-RESTful 序列化与数据验证
[v1 分支版本(不再更新,仅供参考) ](https://gitee.com/pear-admin/pear-admin-flask/tree/v1/)
>flask 1.12 + flask-sqlalchemy + mysql + 权限验证 + marshmallow 序列化与数据验证
[simple 分支版本(不再更新,仅供参考) ](https://gitee.com/pear-admin/pear-admin-flask/tree/simple/)
>flask 1.12 + flask-sqlalchemy + mysql + 极简权限 + marshmallow 序列化与数据验证
>
>项目最简版本
#### 内置功能
- [x] 用户管理:用户是系统操作者,该功能主要完成系统用户配置。
- [x] 权限管理:配置系统菜单,操作权限,按钮权限标识等。
- [x] 角色管理:角色菜单权限分配。
- [x] 操作日志:系统正常操作日志记录和查询;系统异常信息日志记录和查询。
- [x] 登录日志:系统登录日志记录查询包含登录异常。
- [x] 服务监控:监视当前系统CPU、内存、磁盘、python版本,运行时长等相关信息。
- [x] 文件上传: 图片上传示例
- [x] 定时任务: 简单的定时任务
- [ ] 代码生成: 构想中....
#### 项目结构
```
Pear Admin Flask
├─applications # 应用
│ ├─configs # 配置文件
│ │ ├─ common.py # 普通配置
│ │ └─ config.py # 配置文件对象
│ ├─extensions # 注册插件
│ ├─models # 数据模型
│ ├─static # 静态资源文件
│ ├─templates # 静态模板文件
│ └─views # 视图部分
│ ├─admin # 后台管理视图模块
│ └─index # 前台视图模块
├─docs # 文档说明(占坑)
├─migrations # 迁移文件记录
├─requirement # 依赖文件
├─test # 测试文件夹(占坑)
└─.env # 项目的配置文件
``` ```
#### 项目安装 初始化数据库
```bash
# 下 载
git clone https://gitee.com/pear-admin/pear-admin-flask
# 安 装
pip install -r requirement\requirement-dev.txt
# 配 置
.env
```
#### 修改配置
```python
.env
# MySql配置信息
MYSQL_HOST=127.0.0.1
MYSQL_PORT=3306
MYSQL_DATABASE=PearAdminFlask
MYSQL_USERNAME=root
MYSQL_PASSWORD=root
# Redis 配置
REDIS_HOST=127.0.0.1
REDIS_PORT=6379
# 密钥配置
SECRET_KEY='pear-admin-flask'
# 邮箱配置
MAIL_SERVER='smtp.qq.com'
MAIL_USERNAME='123@qq.com'
MAIL_PASSWORD='XXXXX' # 生成的授权码
```
#### Venv 安装
```bash
python -m venv venv
```
#### 运行项目
```bash
# 初 始 化 数 据 库
```shell
flask init flask init
``` ```
执行 flask run 命令启动项目 启动
#### 命令行创建视图 ```shell
flask run
```bash
# 示例
flask new --type view --name test/a
# 自动注册蓝图
# 访问http://127.0.0.1:5000/test/a/
``` ```
#### 预览项目 ## 贡献指南
| | | 如果想参与项目的贡献,提交代码之前需要启用 pre-commit、commitizen 对代码进行校验,运行以下指令即可。
|---------------------|---------------------|
| ![](docs/assets/1.jpg) | ![](docs/assets/2.jpg) | 初始化 pre-commit
| ![](docs/assets/3.jpg)| ![](docs/assets/4.jpg) |
| ![](docs/assets/5.jpg) | ![](docs/assets/6.jpg) | ```shell
pre-commit install
```
检查代码是否符合规范
```shell
git add .
```
```shell
pre-commit run --all-files
```
初始化 commitizen
```shell
pre-commit install -t commit-msg
```
使用
```shell
cz commit
```
代替 `git commit` 进行提交
-10
View File
@@ -1,10 +0,0 @@
from applications import create_app
from flask_migrate import Migrate
from applications.extensions import db
app = create_app()
migrate = Migrate(app, db)
if __name__ == '__main__':
app.run()
-44
View File
@@ -1,44 +0,0 @@
import os
from flask import Flask
from applications.common.script import init_script
from applications.extensions import init_plugs
from applications.view import init_view
from applications.configs import config
def create_app(config_name=None):
app = Flask(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
if not config_name:
# 尝试从本地环境中读取
config_name = os.getenv('FLASK_CONFIG', 'development')
# 引入数据库配置
app.config.from_object(config[config_name])
# 注册各种插件
init_plugs(app)
# 注册路由
init_view(app)
# 注册命令
init_script(app)
if os.environ.get('WERKZEUG_RUN_MAIN') == 'true':
logo()
return app
def logo():
print('''
_____ _ _ ______ _ _
| __ \ /\ | | (_) | ____| | | |
| |__) |__ __ _ _ __ / \ __| |_ __ ___ _ _ __ | |__ | | __ _ ___| | __
| ___/ _ \/ _` | '__| / /\ \ / _` | '_ ` _ \| | '_ \ | __| | |/ _` / __| |/ /
| | | __/ (_| | | / ____ \ (_| | | | | | | | | | | | | | | (_| \__ \ <
|_| \___|\__,_|_| /_/ \_\__,_|_| |_| |_|_|_| |_| |_| |_|\__,_|___/_|\_\\
''')
View File
-145
View File
@@ -1,145 +0,0 @@
import copy
from collections import OrderedDict
from io import BytesIO
from flask import session, make_response, current_app
from flask_login import current_user
from applications.common.utils.gen_captcha import gen_captcha
from applications.schemas import PowerOutSchema
# 授权路由存入session
def add_auth_session():
role = current_user.role
user_power = []
for i in role:
if i.enable == 0:
continue
for p in i.power:
if p.enable == 0:
continue
user_power.append(p.code)
session['permissions'] = user_power
# 生成菜单树
def make_menu_tree():
role = current_user.role
powers = []
for i in role:
# 如果角色没有被启用就直接跳过
if i.enable == 0:
continue
# 变量角色用户的权限
for p in i.power:
# 如果权限关闭了就直接跳过
if p.enable == 0:
continue
# 一二级菜单
if int(p.type) in [0,1] and p not in powers:
powers.append(p)
power_schema = PowerOutSchema(many=True) # 用已继承 ma.ModelSchema 类的自定制类生成序列化类
power_dict = power_schema.dump(powers) # 生成可序列化对象
power_dict.sort(key=lambda x: (x['parent_id'], x['id']), reverse=True)
menu_dict = OrderedDict()
for _dict in power_dict:
if _dict['id'] in menu_dict:
# 当前节点添加子节点
_dict['children'] = copy.deepcopy(menu_dict[_dict['id']])
_dict['children'].sort(key=lambda item: item['sort'])
# 删除子节点
del menu_dict[_dict['id']]
if _dict['parent_id'] not in menu_dict:
menu_dict[_dict['parent_id']] = [_dict]
else:
menu_dict[_dict['parent_id']].append(_dict)
return sorted(menu_dict.get(0), key=lambda item: item['sort'])
# 生成验证码
def get_captcha():
code, image = gen_captcha()
out = BytesIO()
session["code"] = code
image.save(out, 'png')
out.seek(0)
resp = make_response(out.read())
resp.content_type = 'image/png'
return resp, code
def get_render_config():
# 网站配置
config = dict(logo={
# 网站名称
"title": current_app.config.get("SYSTEM_NAME"),
# 网站图标
"image": "/static/admin/admin/images/logo.png"
# 菜单配置
}, menu={
# 菜单数据来源
"data": "/rights/menu",
"collaspe": False,
# 是否同时只打开一个菜单目录
"accordion": True,
"method": "GET",
# 是否开启多系统菜单模式
"control": False,
# 顶部菜单宽度 PX
"controlWidth": 500,
# 默认选中的菜单项
"select": "0",
# 是否开启异步菜单,false 时 data 属性设置为菜单数据,false 时为 json 文件或后端接口
"async": True
}, tab={
# 是否开启多选项卡
"enable": True,
# 切换选项卡时,是否刷新页面状态
"keepState": True,
# 是否开启 Tab 记忆
"session": True,
# 最大可打开的选项卡数量
"max": 30,
"index": {
# 标识 ID , 建议与菜单项中的 ID 一致
"id": "10",
# 页面地址
"href": "/admin/welcome",
# 标题
"title": "首页"
}
}, theme={
# 默认主题色,对应 colors 配置中的 ID 标识
"defaultColor": "2",
# 默认的菜单主题 dark-theme 黑 / light-theme 白
"defaultMenu": "dark-theme",
# 是否允许用户切换主题,false 时关闭自定义主题面板
"allowCustom": True
}, colors=[{
"id": "1",
"color": "#2d8cf0"
},
{
"id": "2",
"color": "#5FB878"
},
{
"id": "3",
"color": "#1E9FFF"
}, {
"id": "4",
"color": "#FFB800"
}, {
"id": "5",
"color": "darkgray"
}
], links=current_app.config.get("SYSTEM_PANEL_LINKS"), other={
# 主页动画时长
"keepLoad": 1200,
# 布局顶部主题
"autoHead": False
}, header=False)
return config
-57
View File
@@ -1,57 +0,0 @@
from flask_login import current_user
from applications.common.utils.validate import xss_escape
from applications.extensions import db
from applications.models import AdminLog
def login_log(request, uid, is_access):
info = {
'method': request.method,
'url': request.path,
'ip': request.remote_addr,
'user_agent': xss_escape(request.headers.get('User-Agent')),
'desc': xss_escape(request.form.get('username')),
'uid': uid,
'success': int(is_access)
}
log = AdminLog(
url=info.get('url'),
ip=info.get('ip'),
user_agent=info.get('user_agent'),
desc=info.get('desc'),
uid=info.get('uid'),
method=info.get('method'),
success=info.get('success')
)
db.session.add(log)
db.session.flush()
db.session.commit()
return log.id
def admin_log(request, is_access):
info = {
'method': request.method,
'url': request.path,
'ip': request.remote_addr,
'user_agent': xss_escape(request.headers.get('User-Agent')),
'desc': xss_escape(str(dict(request.values))),
'uid': current_user.id,
'success': int(is_access)
}
log = AdminLog(
url=info.get('url'),
ip=info.get('ip'),
user_agent=info.get('user_agent'),
desc=info.get('desc'),
uid=info.get('uid'),
method=info.get('method'),
success=info.get('success')
)
db.session.add(log)
db.session.commit()
return log.id
-94
View File
@@ -1,94 +0,0 @@
import datetime
from marshmallow import Schema
from marshmallow_sqlalchemy import SQLAlchemyAutoSchema
from applications.extensions import db, ma
class LogicalDeleteMixin(object):
"""
class Test(db.Model,LogicalDeleteMixin):
__tablename__ = 'admin_test'
id = db.Column(db.Integer, primary_key=True, comment='角色ID')
Test.query.filter_by(id=1).soft_delete()
Test.query.logic_all()
"""
create_at = db.Column(db.DateTime, default=datetime.datetime.now, comment='创建时间')
update_at = db.Column(db.DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now, comment='创建时间')
delete_at = db.Column(db.DateTime, comment='删除时间')
def auto_model_jsonify(data, model: db.Model):
"""
不需要建立schemas,直接使用orm的定义模型进行序列化
基本功能,待完善
示例
power_data = curd.auto_model_jsonify(model=Dept, data=dept)
"""
def get_model():
return model
class AutoSchema(SQLAlchemyAutoSchema):
class Meta(Schema):
model = get_model()
include_fk = True
include_relationships = True
load_instance = True
common_schema = AutoSchema(many=True) # 用已继承ma.ModelSchema类的自定制类生成序列化类
output = common_schema.dump(data)
return output
def model_to_dicts(schema: ma.Schema, data):
"""
:param schema: schema类
:param model: sqlalchemy查询结果
:return: 返回单个查询结果
"""
# 如果是分页器返回,需要传入model.items
common_schema = schema(many=True) # 用已继承ma.ModelSchema类的自定制类生成序列化类
output = common_schema.dump(data) # 生成可序列化对象
return output
def get_one_by_id(model: db.Model, id):
"""
:param model: 模型类
:param id: id
:return: 返回单个查询结果
"""
return model.query.filter_by(id=id).first()
def delete_one_by_id(model: db.Model, id):
"""
:param model: 模型类
:param id: id
:return: 返回单个查询结果
"""
r = model.query.filter_by(id=id).delete()
db.session.commit()
return r
# 启动状态
def enable_status(model: db.Model, id):
enable = 1
role = model.query.filter_by(id=id).update({"enable": enable})
if role:
db.session.commit()
return True
return False
# 停用状态
def disable_status(model: db.Model, id):
enable = 0
role = model.query.filter_by(id=id).update({"enable": enable})
if role:
db.session.commit()
return True
return False
-126
View File
@@ -1,126 +0,0 @@
from sqlalchemy import and_
from applications.extensions import db
class ModelFilter:
"""
orm多参数构造器
"""
filter_field = {}
filter_list = []
type_exact = "exact"
type_neq = "neq"
type_greater = "greater"
type_less = "less"
type_vague = "vague"
type_contains = "contains"
type_between = "between"
def __init__(self):
self.filter_field = {}
self.filter_list = []
def exact(self, field_name, value):
"""
准确查询字段
:param field_name: 模型字段名称
:param value: 值
"""
if value and value != '':
self.filter_field[field_name] = {"data": value, "type": self.type_exact}
def neq(self, field_name, value):
"""
不等于查询字段
:param field_name: 模型字段名称
:param value: 值
"""
if value and value != '':
self.filter_field[field_name] = {"data": value, "type": self.type_neq}
def greater(self, field_name, value):
"""
大于查询字段
:param field_name: 模型字段名称
:param value: 值
"""
if value and value != '':
self.filter_field[field_name] = {"data": value, "type": self.type_greater}
def less(self, field_name, value):
"""
大于查询字段
:param field_name: 模型字段名称
:param value: 值
"""
if value and value != '':
self.filter_field[field_name] = {"data": value, "type": self.type_less}
def vague(self, field_name, value: str):
"""
模糊查询字段
:param field_name: 模型字段名称
:param value: 值
"""
if value and value != '':
self.filter_field[field_name] = {"data": ('%' + value + '%'), "type": self.type_vague}
def left_vague(self, field_name, value: str):
"""
左模糊查询字段
:param field_name: 模型字段名称
:param value: 值
"""
if value and value != '':
self.filter_field[field_name] = {"data": ('%' + value), "type": self.type_vague}
def right_vague(self, field_name, value: str):
"""
左模糊查询字段
:param field_name: 模型字段名称
:param value: 值
"""
if value and value != '':
self.filter_field[field_name] = {"data": (value + '%'), "type": self.type_vague}
def contains(self, field_name, value: str):
"""
包含查询字段
:param field_name: 模型字段名称
:param value: 值
"""
if value and value != '':
self.filter_field[field_name] = {"data": value, "type": self.type_contains}
def between(self, field_name, value1, value2):
"""
范围查询字段
:param field_name: 模型字段名称
:param value: 值
"""
if value1 and value2 and value1 != '' and value2 != '':
self.filter_field[field_name] = {"data": [value1, value2], "type": self.type_between}
def get_filter(self, model: db.Model):
"""
获取过滤条件
:param model: 模型字段名称
"""
for k, v in self.filter_field.items():
if v.get("type") == self.type_vague:
self.filter_list.append(getattr(model, k).like(v.get("data")))
if v.get("type") == self.type_contains:
self.filter_list.append(getattr(model, k).contains(v.get("data")))
if v.get("type") == self.type_exact:
self.filter_list.append(getattr(model, k) == v.get("data"))
if v.get("type") == self.type_neq:
self.filter_list.append(getattr(model, k) != v.get("data"))
if v.get("type") == self.type_greater:
self.filter_list.append(getattr(model, k) > v.get("data"))
if v.get("type") == self.type_less:
self.filter_list.append(getattr(model, k) < v.get("data"))
if v.get("type") == self.type_between:
self.filter_list.append(getattr(model, k).between(v.get("data")[0], v.get("data")[1]))
return and_(*self.filter_list)
-26
View File
@@ -1,26 +0,0 @@
import os
import click
from applications.common.script.initdb import init_db
from applications.common.script.newmodular.new import NewViewModular
def init_script(app):
@app.cli.command()
def init():
init_db()
@app.cli.command()
@click.option('--type', prompt="请输入类型", help='新增的类型')
@click.option('--name', prompt="请输入新增的名称", help='新增的名称')
def new(type,name):
if type == 'view':
if name.count('/') > 1:
print("目前只支持二级目录,多级目录需要蓝图嵌套,本命令暂不支持,请手动创建")
quit()
if type == "view" and os.path.exists(f"applications/view/{name}.py"):
print(f'已经存在视图模块{name}.py')
quit()
NewViewModular(name=name).new_view()
-60
View File
@@ -1,60 +0,0 @@
from dotenv import dotenv_values
import sqlparse
import pymysql
config = dotenv_values('.flaskenv')
# MySql配置信息
HOST = config.get('MYSQL_HOST') or '127.0.0.1'
PORT = config.get('MYSQL_PORT') or 3306
DATABASE = config.get('MYSQL_DATABASE') or 'PearAdminFlask'
USERNAME = config.get('MYSQL_USERNAME') or 'root'
PASSWORD = config.get('MYSQL_PASSWORD') or '123456'
def is_exist_database():
db = pymysql.connect(host=HOST, port=int(PORT), user=USERNAME, password=PASSWORD, charset='utf8mb4')
cursor1 = db.cursor()
sql = "select * from information_schema.SCHEMATA WHERE SCHEMA_NAME = '%s' ; " % DATABASE
res = cursor1.execute(sql)
db.close()
return res
def init_database():
db = pymysql.connect(host=HOST, port=int(PORT), user=USERNAME, password=PASSWORD, charset='utf8mb4')
cursor1 = db.cursor()
sql = "CREATE DATABASE IF NOT EXISTS %s CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;" % DATABASE
res = cursor1.execute(sql)
db.close()
return res
def execute_fromfile(filename):
db = pymysql.connect(host=HOST, port=int(PORT), user=USERNAME, password=PASSWORD, database=DATABASE,
charset='utf8mb4')
fd = open(filename, 'r', encoding='utf-8')
cursor = db.cursor()
sqlfile = fd.read()
sqlfile = sqlparse.format(sqlfile, strip_comments=True).strip()
sqlcommamds = sqlfile.split(';')
for command in sqlcommamds:
try:
cursor.execute(command)
db.commit()
except Exception as msg:
db.rollback()
db.close()
def init_db():
if is_exist_database():
print('数据库已经存在,为防止误初始化,请手动删除 %s 数据库' % str(DATABASE))
return
if init_database():
print('数据库%s创建成功' % str(DATABASE))
execute_fromfile('test/pear.sql')
print('表创建成功')
print('欢迎使用pear-admin-flask,请使用 flask run 命令启动程序')
@@ -1,92 +0,0 @@
import os
import jinja2
class NewViewModular():
def __init__(self, name):
self.name = name
def path_is_exists(self):
return os.path.exists(f"applications/view/{self.name.split('/')[0]}")
def new_dirs(self):
if not self.path_is_exists():
# 如果不存在则创建目录
# 创建目录操作函数
os.makedirs(f"applications/view/{self.name.split('/')[0]}")
print(self.name + ' 创建成功')
return True
else:
return False
def add_view(self):
# jinja渲染
templateLoader = jinja2.FileSystemLoader(searchpath='applications/common/script/newmodular/template/')
templateEnv = jinja2.Environment(loader=templateLoader)
TEMPLATE_FILE = "view"
template = templateEnv.get_template(TEMPLATE_FILE, )
templateVars = {"name": self.name}
outputText = template.render(templateVars)
with open(f"applications/view/{self.name}.py", "a", encoding='utf-8') as f:
f.write(outputText)
f.close()
def add_init(self):
# 判断__init__.py
if os.path.exists(f"applications/view/{self.name.split('/')[0]}/__init__.py"):
with open(f"applications/view/{self.name.split('/')[0]}/__init__.py", "r") as file:
lines = file.readlines()
print(lines)
file.close()
import_line_num = 0
for line in lines:
if 'def' in line:
import_line_num = lines.index(line) - 2
with open(f"applications/view/{self.name.split('/')[0]}/__init__.py", "w") as file:
lines.insert(import_line_num,
f"from applications.view.{self.name.replace('/', '.')} import {self.name.replace('/', '_')}\n")
print(lines)
lines.insert(len(lines) - 1, f" app.register_blueprint({self.name.replace('/', '_')})\n")
print(lines)
file.writelines(lines)
file.close()
return True
else:
templateLoader = jinja2.FileSystemLoader(searchpath='applications/common/script/newmodular/template/')
templateEnv = jinja2.Environment(loader=templateLoader)
TEMPLATE_FILE = "__init__"
template = templateEnv.get_template(TEMPLATE_FILE, )
templateVars = {"name": self.name}
output = template.render(templateVars)
with open(f"applications/view/{self.name.split('/')[0]}/__init__.py", "a", encoding='utf-8') as f:
f.write(output)
f.close()
self.add_root_init()
return True
def add_root_init(self):
with open(f"applications/view/__init__.py", "r") as file:
lines = file.readlines()
file.close()
import_line_num = 0
for line in lines:
if 'def' in line:
import_line_num = lines.index(line) - 2
with open(f"applications/view/__init__.py", "w") as file:
lines.insert(
import_line_num,
f"from applications.view.{self.name.split('/')[0]} import register_{self.name.split('/')[0]}_views\n")
print(lines)
lines.insert(len(lines) - 1, f" register_{self.name.split('/')[0]}_views(app)\n")
print(lines)
file.writelines(lines)
file.close()
def new_view(self):
self.new_dirs()
self.add_view()
self.add_init()
@@ -1,6 +0,0 @@
from flask import Flask
from applications.view.{{name.replace('/','.')}} import {{name.replace('/','_')}}
def register_{{name.split('/')[0]}}_views(app: Flask):
app.register_blueprint({{name.replace('/','_')}})
@@ -1,8 +0,0 @@
from flask import Blueprint
{{name.replace('/','_')}} = Blueprint('{{name.replace('/','_')}}', __name__, url_prefix='/{{name}}')
@{{name.replace('/','_')}}.route('/')
def index():
return "这是{{name}}路由"
-54
View File
@@ -1,54 +0,0 @@
from apscheduler.events import (
EVENT_JOB_ADDED,
EVENT_JOB_ERROR,
EVENT_JOB_EXECUTED,
EVENT_JOB_MISSED,
EVENT_JOB_REMOVED,
EVENT_JOB_SUBMITTED,
)
from applications.extensions.init_apscheduler import scheduler
def job_missed(event):
"""Job missed event."""
with scheduler.app.app_context():
print(event) # noqa: T001
def job_error(event):
"""Job error event."""
with scheduler.app.app_context():
print(event) # noqa: T001
def job_executed(event):
"""Job executed event."""
with scheduler.app.app_context():
print(event) # noqa: T001
def job_added(event):
"""Job added event."""
with scheduler.app.app_context():
print(event) # noqa: T001
def job_removed(event):
"""Job removed event."""
with scheduler.app.app_context():
print(event) # noqa: T001
def job_submitted(event):
"""Job scheduled to run event."""
with scheduler.app.app_context():
print(event) # noqa: T001
scheduler.add_listener(job_missed, EVENT_JOB_MISSED)
scheduler.add_listener(job_error, EVENT_JOB_ERROR)
scheduler.add_listener(job_executed, EVENT_JOB_EXECUTED)
scheduler.add_listener(job_added, EVENT_JOB_ADDED)
scheduler.add_listener(job_removed, EVENT_JOB_REMOVED)
scheduler.add_listener(job_submitted, EVENT_JOB_SUBMITTED)
-15
View File
@@ -1,15 +0,0 @@
import datetime
task_list = ['task2', 'task3', 'task4']
def task2(a, b):
print(f'定时任务_1_{a},{b},{datetime.datetime.now()}')
def task3(a, b):
print(f'定时任务_2_{a}{b}{datetime.datetime.now()}')
def task4(a, b):
print(f'定时任务_4_{a}{b}{datetime.datetime.now()}')
-14
View File
@@ -1,14 +0,0 @@
from captcha.image import ImageCaptcha
from PIL import Image
from random import choices
def gen_captcha(content='2345689abcdefghijklmnpqrstuvwxyzABCDEFGHIJKLMNPQRSTUVWXYZ'):
""" 生成验证码 """
image = ImageCaptcha()
# 获取字符串
captcha_text = "".join(choices(content, k=4)).lower()
# 生成图像
captcha_image = Image.open(image.generate(captcha_text))
return captcha_text, captcha_image
-24
View File
@@ -1,24 +0,0 @@
from flask import jsonify
def success_api(msg: str = "成功"):
""" 成功响应 默认值”成功“ """
return jsonify(success=True, msg=msg)
def fail_api(msg: str = "失败"):
""" 失败响应 默认值“失败” """
return jsonify(success=False, msg=msg)
def table_api(msg: str = "", count=0, data=None, limit=10):
""" 动态表格渲染响应 """
res = {
'msg': msg,
'code': 0,
'data': data,
'count': count,
'limit': limit
}
return jsonify(res)
-13
View File
@@ -1,13 +0,0 @@
from flask_mail import Message
from applications.extensions.init_mail import mail
# send_mail(subject='title', recipients=['123@qq.com'], content='body')
def send_mail(subject, recipients, content):
try:
message = Message(subject=subject, recipients=recipients, body=content)
mail.send(message)
except Exception as e:
print('邮箱发送出错了')
raise
-25
View File
@@ -1,25 +0,0 @@
from functools import wraps
from flask import abort, request, jsonify, session
from flask_login import login_required
from applications.common.admin_log import admin_log
def authorize(power: str, log: bool = False):
def decorator(func):
@login_required
@wraps(func)
def wrapper(*args, **kwargs):
if not power in session.get('permissions'):
if log:
admin_log(request=request, is_access=False)
if request.method == 'GET':
abort(403)
else:
return jsonify(success=False, msg="权限不足!")
if log:
admin_log(request=request, is_access=True)
return func(*args, **kwargs)
return wrapper
return decorator
-36
View File
@@ -1,36 +0,0 @@
import os
from flask import current_app
from sqlalchemy import desc
from applications.extensions import db
from applications.extensions.init_upload import photos
from applications.models import Photo
from applications.schemas import PhotoOutSchema
from applications.common.curd import model_to_dicts
def get_photo(page, limit):
photo = Photo.query.order_by(desc(Photo.create_time)).paginate(page=page, per_page=limit, error_out=False)
count = Photo.query.count()
data = model_to_dicts(schema=PhotoOutSchema, data=photo.items)
return data, count
def upload_one(photo, mime):
filename = photos.save(photo)
file_url = '/_uploads/photos/'+filename
# file_url = photos.url(filename)
upload_url = current_app.config.get("UPLOADED_PHOTOS_DEST")
size = os.path.getsize(upload_url + '/' + filename)
photo = Photo(name=filename, href=file_url, mime=mime, size=size)
db.session.add(photo)
db.session.commit()
return file_url
def delete_photo_by_id(_id):
photo_name = Photo.query.filter_by(id=_id).first().name
photo = Photo.query.filter_by(id=_id).delete()
db.session.commit()
upload_url = current_app.config.get("UPLOADED_PHOTOS_DEST")
os.remove(upload_url + '/' + photo_name)
return photo
-20
View File
@@ -1,20 +0,0 @@
# xss过滤
from flask import abort, make_response, jsonify
def xss_escape(s: str):
if s is None:
return None
else:
return s.replace("&", "&amp;").replace(">", "&gt;").replace("<", "&lt;").replace("'", "&#39;").replace('"',
"&#34;")
def check_data(schema, data):
errors = schema.validate(data)
for k, v in errors.items():
for i in v:
# print("{}{}".format(k, i))
msg = "{}{}".format(k, i)
if errors:
abort(make_response(jsonify(result=False, msg=msg), 200))
-1
View File
@@ -1 +0,0 @@
from .config import config
-87
View File
@@ -1,87 +0,0 @@
import logging
import os
from urllib.parse import quote_plus as urlquote
class BaseConfig:
SYSTEM_NAME = os.getenv('SYSTEM_NAME', 'Pear Admin')
# 主题面板的链接列表配置
SYSTEM_PANEL_LINKS = [
{
"icon": "layui-icon layui-icon-auz",
"title": "官方网站",
"href": "http://www.pearadmin.com"
},
{
"icon": "layui-icon layui-icon-auz",
"title": "开发文档",
"href": "http://www.pearadmin.com"
},
{
"icon": "layui-icon layui-icon-auz",
"title": "开源地址",
"href": "https://gitee.com/Jmysy/Pear-Admin-Layui"
}
]
UPLOADED_PHOTOS_DEST = 'static/upload'
UPLOADED_FILES_ALLOW = ['gif', 'jpg']
# JSON配置
JSON_AS_ASCII = False
SECRET_KEY = os.getenv('SECRET_KEY', 'dev key')
# redis配置
REDIS_HOST = os.getenv('REDIS_HOST') or "127.0.0.1"
REDIS_PORT = int(os.getenv('REDIS_PORT') or 6379)
# mysql 配置
MYSQL_USERNAME = os.getenv('MYSQL_USERNAME') or "root"
MYSQL_PASSWORD = os.getenv('MYSQL_PASSWORD') or "123456"
MYSQL_HOST = os.getenv('MYSQL_HOST') or "127.0.0.1"
MYSQL_PORT = int(os.getenv('MYSQL_PORT') or 3306)
MYSQL_DATABASE = os.getenv('MYSQL_DATABASE') or "PearAdminFlask"
# mysql 数据库的配置信息
SQLALCHEMY_DATABASE_URI = f"mysql+pymysql://{MYSQL_USERNAME}:{urlquote(MYSQL_PASSWORD)}@{MYSQL_HOST}:{MYSQL_PORT}/{MYSQL_DATABASE}?charset=utf8mb4"
# 默认日志等级
LOG_LEVEL = logging.WARN
#
MAIL_SERVER = os.getenv('MAIL_SERVER') or 'smtp.qq.com'
MAIL_USE_TLS = False
MAIL_USE_SSL = True
MAIL_PORT = 465
MAIL_USERNAME = os.getenv('MAIL_USERNAME') or '123@qq.com'
MAIL_PASSWORD = os.getenv('MAIL_PASSWORD') or 'XXXXX' # 生成的授权码
# 默认发件人的邮箱,这里填写和MAIL_USERNAME一致即可
MAIL_DEFAULT_SENDER = ('pear admin', os.getenv('MAIL_USERNAME') or '123@qq.com')
class TestingConfig(BaseConfig):
""" 测试配置 """
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:' # 内存数据库
class DevelopmentConfig(BaseConfig):
""" 开发配置 """
SQLALCHEMY_TRACK_MODIFICATIONS = True
SQLALCHEMY_ECHO = False
class ProductionConfig(BaseConfig):
"""生成环境配置"""
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = False
SQLALCHEMY_POOL_RECYCLE = 8
LOG_LEVEL = logging.ERROR
config = {
'development': DevelopmentConfig,
'testing': TestingConfig,
'production': ProductionConfig
}
-23
View File
@@ -1,23 +0,0 @@
from flask import Flask
from .init_sqlalchemy import db, ma, init_databases
from .init_login import init_login_manager
from .init_debug_tool import init_debug_tool
from .init_template_directives import init_template_directives
from .init_error_views import init_error_views
from .init_mail import init_mail
from .init_apscheduler import init_scheduler
from .init_upload import init_upload
from .init_dotenv import init_dotenv
def init_plugs(app: Flask) -> None:
# init_debug_tool(app)
init_login_manager(app)
init_databases(app)
init_template_directives(app)
init_error_views(app)
init_mail(app)
init_scheduler(app)
init_upload(app)
init_dotenv()
@@ -1,12 +0,0 @@
from flask import Flask
from flask_apscheduler import APScheduler
scheduler = APScheduler()
def init_scheduler(app: Flask):
scheduler.init_app(app)
with app.app_context():
from applications.common.tasks import tasks
scheduler.start()
from applications.common.tasks import events
@@ -1,7 +0,0 @@
from flask import Flask
from flask_debugtoolbar import DebugToolbarExtension
def init_debug_tool(app: Flask):
toolbar = DebugToolbarExtension()
toolbar.init_app(app)
-14
View File
@@ -1,14 +0,0 @@
import os
from dotenv import load_dotenv
root_path = os.path.abspath(os.path.dirname(__file__)).split('applications')[0]
# dot_env_path = os.path.join(root_path, '.env')
flask_env_path = os.path.join(root_path, '.flaskenv')
# if os.path.exists(dot_env_path):
# load_dotenv(dot_env_path)
def init_dotenv():
if os.path.exists(flask_env_path):
load_dotenv(flask_env_path)
@@ -1,34 +0,0 @@
from flask import render_template, jsonify
def init_error_views(app):
@app.errorhandler(403)
def page_not_found(e):
return render_template('errors/403.html'), 403
@app.errorhandler(404)
def page_not_found(e):
return render_template('errors/404.html'), 404
@app.errorhandler(500)
def internal_server_error(e):
return render_template('errors/500.html'), 500
# Return validation errors as JSON
@app.errorhandler(422)
@app.errorhandler(400)
def handle_error(err):
headers = err.data.get("headers", None)
messages = err.data.get("messages", ["Invalid request."]).get('json')
print(err.data.get("messages"))
print(messages.items())
msg = ''
for i in messages.items():
msg = str(i[0]) + str(i[1][0])
break
if headers:
return jsonify({"success": False, "msg": msg})
else:
return jsonify({"success": False, "msg": msg})
-15
View File
@@ -1,15 +0,0 @@
from flask_login import LoginManager
def init_login_manager(app):
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'passport.login'
login_manager.login_message = u'请登录以访问此页面'
@login_manager.user_loader
def load_user(user_id):
from applications.models import User
user = User.query.get(int(user_id))
return user
-8
View File
@@ -1,8 +0,0 @@
from flask import Flask
from flask_mail import Mail
mail = Mail()
def init_mail(app: Flask):
mail.init_app(app)
@@ -1,93 +0,0 @@
import datetime
from flask import Flask, request
from flask_sqlalchemy import SQLAlchemy, BaseQuery
from flask_marshmallow import Marshmallow
from marshmallow import fields
from marshmallow.validate import (
URL, Email, Range, Length, Equal, Regexp,
Predicate, NoneOf, OneOf, ContainsOnly
)
URL.default_message = '无效的链接'
Email.default_message = '无效的邮箱地址'
Range.message_min = '不能小于{min}'
Range.message_max = '不能小于{max}'
Range.message_all = '不能超过{min}{max}这个范围'
Length.message_min = '长度不得小于{min}'
Length.message_max = '长度不得大于{max}'
Length.message_all = '长度不能超过{min}{max}这个范围'
Length.message_equal = '长度必须等于{equal}'
Equal.default_message = '必须等于{other}'
Regexp.default_message = '非法输入'
Predicate.default_message = '非法输入'
NoneOf.default_message = '非法输入'
OneOf.default_message = '无效的选择'
ContainsOnly.default_message = '一个或多个无效的选择'
fields.Field.default_error_messages = {
"required": "缺少必要数据",
"null": "数据不能为空",
"validator_failed": "非法数据",
}
fields.Str.default_error_messages = {
'invalid': "不是合法文本"
}
fields.Int.default_error_messages = {
"invalid": "不是合法整数"
}
fields.Number.default_error_messages = {
"invalid": "不是合法数字"
}
fields.Boolean.default_error_messages = {
"invalid": "不是合法布尔值"
}
class Query(BaseQuery):
def soft_delete(self):
return self.update({"delete_at": datetime.datetime.now()})
def logic_all(self):
return self.filter_by(delete_at=None).all()
def all_json(self, schema: Marshmallow().Schema):
return schema(many=True).dump(self.all())
def layui_paginate(self):
return self.paginate(page=request.args.get('page', type=int),
per_page=request.args.get('limit', type=int),
error_out=False)
def layui_paginate_json(self, schema: Marshmallow().Schema):
"""
返回dict
"""
_res = self.paginate(
page=request.args.get('page', type=int),
per_page=request.args.get('limit', type=int),
error_out=False
)
return schema(many=True).dump(_res.items), _res.total, _res.page, _res.per_page
def layui_paginate_db_json(self):
"""
db.query(A.name).layui_paginate_db_json()
"""
_res = self.paginate(page=request.args.get('page', type=int),
per_page=request.args.get('limit', type=int),
error_out=False)
return [dict(i) for i in _res.items], _res.total
db = SQLAlchemy(query_class=Query)
ma = Marshmallow()
def init_databases(app: Flask):
db.init_app(app)
ma.init_app(app)
@@ -1,7 +0,0 @@
from flask import session
def init_template_directives(app):
@app.template_global()
def authorize(power):
return bool(power in session.get('permissions'))
-9
View File
@@ -1,9 +0,0 @@
from flask import Flask
from flask_uploads import configure_uploads
from flask_uploads import UploadSet, IMAGES
photos = UploadSet('photos', IMAGES)
def init_upload(app: Flask):
configure_uploads(app, photos)
-9
View File
@@ -1,9 +0,0 @@
from marshmallow import EXCLUDE
from webargs.flaskparser import FlaskParser
class Parser(FlaskParser):
DEFAULT_UNKNOWN_BY_LOCATION = {"query": EXCLUDE}
parser = Parser()
-9
View File
@@ -1,9 +0,0 @@
from .admin_dept import Dept
from .admin_dict import DictType, DictData
from .admin_log import AdminLog
from .admin_photo import Photo
from .admin_power import Power
from .admin_role import Role
from .admin_role_power import role_power
from .admin_user import User
from .admin_user_role import user_role
-18
View File
@@ -1,18 +0,0 @@
import datetime
from applications.extensions import db
class Dept(db.Model):
__tablename__ = 'admin_dept'
id = db.Column(db.Integer, primary_key=True, comment="部门ID")
parent_id = db.Column(db.Integer, comment="父级编号")
dept_name = db.Column(db.String(50), comment="部门名称")
sort = db.Column(db.Integer, comment="排序")
leader = db.Column(db.String(50), comment="负责人")
phone = db.Column(db.String(20), comment="联系方式")
email = db.Column(db.String(50), comment="邮箱")
status = db.Column(db.Integer, comment='状态(1开启,0关闭)')
remark = db.Column(db.Text, comment="备注")
address = db.Column(db.String(255), comment="详细地址")
create_at = db.Column(db.DateTime, default=datetime.datetime.now, comment='创建时间')
update_at = db.Column(db.DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now, comment='创建时间')
-26
View File
@@ -1,26 +0,0 @@
import datetime
from applications.extensions import db
class DictType(db.Model):
__tablename__ = 'admin_dict_type'
id = db.Column(db.Integer, primary_key=True)
type_name = db.Column(db.String(255), comment='字典类型名称')
type_code = db.Column(db.String(255), comment='字典类型标识')
description = db.Column(db.String(255), comment='字典类型描述')
enable = db.Column(db.Integer, comment='是否开启')
create_time = db.Column(db.DateTime, default=datetime.datetime.now, comment='创建时间')
update_time = db.Column(db.DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now, comment='更新时间')
class DictData(db.Model):
__tablename__ = 'admin_dict_data'
id = db.Column(db.Integer, primary_key=True)
data_label = db.Column(db.String(255), comment='字典类型名称')
data_value = db.Column(db.String(255), comment='字典类型标识')
type_code = db.Column(db.String(255), comment='字典类型描述')
is_default = db.Column(db.Integer, comment='是否默认')
enable = db.Column(db.Integer, comment='是否开启')
remark = db.Column(db.String(255), comment='备注')
create_time = db.Column(db.DateTime, default=datetime.datetime.now, comment='创建时间')
update_time = db.Column(db.DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now, comment='更新时间')
-14
View File
@@ -1,14 +0,0 @@
import datetime
from applications.extensions import db
class AdminLog(db.Model):
__tablename__ = 'admin_admin_log'
id = db.Column(db.Integer, primary_key=True)
method = db.Column(db.String(10))
uid = db.Column(db.Integer)
url = db.Column(db.String(255))
desc = db.Column(db.Text)
ip = db.Column(db.String(255))
success = db.Column(db.Integer)
user_agent = db.Column(db.Text)
create_time = db.Column(db.DateTime, default=datetime.datetime.now)
-12
View File
@@ -1,12 +0,0 @@
import datetime
from applications.extensions import db
class Photo(db.Model):
__tablename__ = 'admin_photo'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(255), nullable=False)
href = db.Column(db.String(255))
mime = db.Column(db.CHAR(50), nullable=False)
size = db.Column(db.CHAR(30), nullable=False)
create_time = db.Column(db.DateTime, default=datetime.datetime.now)
-18
View File
@@ -1,18 +0,0 @@
import datetime
from applications.extensions import db
class Power(db.Model):
__tablename__ = 'admin_power'
id = db.Column(db.Integer, primary_key=True, comment='权限编号')
name = db.Column(db.String(255), comment='权限名称')
type = db.Column(db.String(1), comment='权限类型')
code = db.Column(db.String(30), comment='权限标识')
url = db.Column(db.String(255), comment='权限路径')
open_type = db.Column(db.String(10), comment='打开方式')
parent_id = db.Column(db.Integer, comment='父类编号')
icon = db.Column(db.String(128), comment='图标')
sort = db.Column(db.Integer, comment='排序')
create_time = db.Column(db.DateTime, default=datetime.datetime.now, comment='创建时间')
update_time = db.Column(db.DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now, comment='更新时间')
enable = db.Column(db.Integer, comment='是否开启')
-16
View File
@@ -1,16 +0,0 @@
import datetime
from applications.extensions import db
class Role(db.Model):
__tablename__ = 'admin_role'
id = db.Column(db.Integer, primary_key=True, comment='角色ID')
name = db.Column(db.String(255), comment='角色名称')
code = db.Column(db.String(255), comment='角色标识')
enable = db.Column(db.Integer, comment='是否启用')
remark = db.Column(db.String(255), comment='备注')
details = db.Column(db.String(255), comment='详情')
sort = db.Column(db.Integer, comment='排序')
create_time = db.Column(db.DateTime, default=datetime.datetime.now, comment='创建时间')
update_time = db.Column(db.DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now, comment='更新时间')
power = db.relationship('Power', secondary="admin_role_power", backref=db.backref('role'))
-9
View File
@@ -1,9 +0,0 @@
from applications.extensions import db
# 创建中间表
role_power = db.Table(
"admin_role_power", # 中间表名称
db.Column("id", db.Integer, primary_key=True, autoincrement=True, comment='标识'), # 主键
db.Column("power_id", db.Integer, db.ForeignKey("admin_power.id"), comment='用户编号'), # 属性 外键
db.Column("role_id", db.Integer, db.ForeignKey("admin_role.id"), comment='角色编号'), # 属性 外键
)
-25
View File
@@ -1,25 +0,0 @@
import datetime
from flask_login import UserMixin
from werkzeug.security import generate_password_hash, check_password_hash
from applications.extensions import db
class User(db.Model, UserMixin):
__tablename__ = 'admin_user'
id = db.Column(db.Integer, primary_key=True, autoincrement=True, comment='用户ID')
username = db.Column(db.String(20), comment='用户名')
realname = db.Column(db.String(20), comment='真实名字')
avatar = db.Column(db.String(255), comment='头像', default="/static/admin/admin/images/avatar.jpg")
remark = db.Column(db.String(255), comment='备注')
password_hash = db.Column(db.String(128), comment='哈希密码')
enable = db.Column(db.Integer, default=0, comment='启用')
dept_id = db.Column(db.Integer, comment='部门id')
create_at = db.Column(db.DateTime, default=datetime.datetime.now, comment='创建时间')
update_at = db.Column(db.DateTime, default=datetime.datetime.now, onupdate=datetime.datetime.now, comment='创建时间')
role = db.relationship('Role', secondary="admin_user_role", backref=db.backref('user'), lazy='dynamic')
def set_password(self, password):
self.password_hash = generate_password_hash(password)
def validate_password(self, password):
return check_password_hash(self.password_hash, password)
-9
View File
@@ -1,9 +0,0 @@
from applications.extensions import db
# 创建中间表
user_role = db.Table(
"admin_user_role", # 中间表名称
db.Column("id", db.Integer, primary_key=True, autoincrement=True, comment='标识'), # 主键
db.Column("user_id", db.Integer, db.ForeignKey("admin_user.id"), comment='用户编号'), # 属性 外键
db.Column("role_id", db.Integer, db.ForeignKey("admin_role.id"), comment='角色编号'), # 属性 外键
)
-7
View File
@@ -1,7 +0,0 @@
from .admin_user import UserOutSchema
from .admin_role import RoleOutSchema
from .admin_power import PowerOutSchema, PowerOutSchema2
from .admin_dict import DictDataOutSchema, DictTypeOutSchema
from .admin_dept import DeptOutSchema
from .admin_log import LogOutSchema
from .admin_photo import PhotoOutSchema
-25
View File
@@ -1,25 +0,0 @@
from applications.extensions import ma
from marshmallow import fields, validate
class DeptInSchema(ma.Schema):
parentId = fields.Integer(required=True)
deptName = fields.Str(required=True)
leader = fields.Str(required=True)
phone = fields.Str(required=True)
email = fields.Str(validate=validate.Email())
address = fields.Str()
status = fields.Str(validate=validate.OneOf(["0", "1"]))
sort = fields.Integer()
class DeptOutSchema(ma.Schema):
deptId = fields.Integer(attribute="id")
parentId = fields.Integer(attribute="parent_id")
deptName = fields.Str(attribute="dept_name")
leader = fields.Str()
phone = fields.Str()
email = fields.Str(validate=validate.Email())
address = fields.Str()
status = fields.Str(validate=validate.OneOf(["0", "1"]))
sort = fields.Integer()
-21
View File
@@ -1,21 +0,0 @@
from applications.extensions import ma
from marshmallow import fields
class DictTypeOutSchema(ma.Schema):
id = fields.Str(attribute="id")
typeName = fields.Str(attribute="type_name")
typeCode = fields.Str(attribute="type_code")
description = fields.Str(attribute="description")
createTime = fields.Str(attribute="create_time")
updateName = fields.Str(attribute="update_time")
remark = fields.Str()
enable = fields.Str()
class DictDataOutSchema(ma.Schema):
dataId = fields.Str(attribute="id")
dataLabel = fields.Str(attribute="data_label")
dataValue = fields.Str(attribute="data_value")
remark = fields.Str()
enable = fields.Str()
-14
View File
@@ -1,14 +0,0 @@
from applications.extensions import ma
from marshmallow import fields
class LogOutSchema(ma.Schema):
id = fields.Integer()
method = fields.Str()
uid = fields.Str()
url = fields.Str()
desc = fields.Str()
ip = fields.Str()
user_agent = fields.Str()
success = fields.Bool()
create_time = fields.DateTime()
-12
View File
@@ -1,12 +0,0 @@
from applications.extensions import ma
from marshmallow import fields
class PhotoOutSchema(ma.Schema):
id = fields.Integer()
name = fields.Str()
href = fields.Str()
mime = fields.Str()
size = fields.Str()
ext = fields.Str()
create_time = fields.DateTime()
-32
View File
@@ -1,32 +0,0 @@
from applications.extensions import ma
from marshmallow import fields
# 权限models序列化类
class PowerOutSchema(ma.Schema):
id = fields.Integer()
title = fields.Str(attribute="name")
type = fields.Str()
code = fields.Str()
href = fields.Str(attribute="url")
openType = fields.Str(attribute="open_type")
parent_id = fields.Integer()
icon = fields.Str()
sort = fields.Integer()
create_time = fields.DateTime()
update_time = fields.DateTime()
enable = fields.Integer()
class PowerOutSchema2(ma.Schema): # 序列化类
powerId = fields.Str(attribute="id")
powerName = fields.Str(attribute="name")
powerType = fields.Str(attribute="type")
powerUrl = fields.Str(attribute="url")
openType = fields.Str(attribute="open_type")
parentId = fields.Str(attribute="parent_id")
icon = fields.Str()
sort = fields.Integer()
create_time = fields.DateTime()
update_time = fields.DateTime()
enable = fields.Integer()
-14
View File
@@ -1,14 +0,0 @@
from applications.extensions import ma
from marshmallow import fields
class RoleOutSchema(ma.Schema):
id = fields.Integer()
roleName = fields.Str(attribute="name")
roleCode = fields.Str(attribute="code")
enable = fields.Str()
remark = fields.Str()
details = fields.Str()
sort = fields.Integer()
create_at = fields.DateTime()
update_at = fields.DateTime()
-20
View File
@@ -1,20 +0,0 @@
from applications.extensions import ma
from marshmallow import fields
from applications.models import Dept
# 用户models的序列化类
class UserOutSchema(ma.Schema):
id = fields.Integer()
username = fields.Str()
realname = fields.Str()
enable = fields.Integer()
create_at = fields.DateTime()
update_at = fields.DateTime()
dept = fields.Method("get_dept")
def get_dept(self, obj):
if obj.dept_id != None:
return Dept.query.filter_by(id=obj.dept_id).first().dept_name
else:
return None
-13
View File
@@ -1,13 +0,0 @@
from applications.view.admin import register_admin_views
from applications.view.index import register_index_views
from applications.view.passport import register_passport_views
from applications.view.rights import register_rights_view
from applications.view.department import register_dept_views
def init_view(app):
register_admin_views(app)
register_index_views(app)
register_rights_view(app)
register_passport_views(app)
register_dept_views(app)
-23
View File
@@ -1,23 +0,0 @@
from flask import Flask
from applications.view.admin.admin_log import admin_log
from applications.view.admin.dict import admin_dict
from applications.view.admin.index import admin_bp
from applications.view.admin.file import admin_file
from applications.view.admin.power import admin_power
from applications.view.admin.role import admin_role
from applications.view.admin.user import admin_user
from applications.view.admin.monitor import admin_monitor_bp
from applications.view.admin.task import admin_task
def register_admin_views(app: Flask):
app.register_blueprint(admin_bp)
app.register_blueprint(admin_user)
app.register_blueprint(admin_file)
app.register_blueprint(admin_monitor_bp)
app.register_blueprint(admin_log)
app.register_blueprint(admin_power)
app.register_blueprint(admin_role)
app.register_blueprint(admin_dict)
app.register_blueprint(admin_task)
-40
View File
@@ -1,40 +0,0 @@
from flask import Blueprint, request, render_template
from sqlalchemy import desc
from applications.common.utils.http import table_api
from applications.common.utils.rights import authorize
from applications.models import AdminLog
from applications.schemas import LogOutSchema
from applications.common.curd import model_to_dicts
admin_log = Blueprint('adminLog', __name__, url_prefix='/admin/log')
# 日志管理
@admin_log.get('/')
@authorize("admin:log:main")
def index():
return render_template('admin/admin_log/main.html')
# 登录日志
@admin_log.get('/loginLog')
@authorize("admin:log:main")
def login_log():
# orm查询
# 使用分页获取data需要.items
log = AdminLog.query.filter_by(url='/passport/login').order_by(desc(AdminLog.create_time)).layui_paginate()
count = log.total
return table_api(data= model_to_dicts(schema=LogOutSchema, data=log.items), count=count)
# 操作日志
@admin_log.get('/operateLog')
@authorize("admin:log:main")
def operate_log():
# orm查询
# 使用分页获取data需要.items
log = AdminLog.query.filter(
AdminLog.url != '/passport/login').order_by(
desc(AdminLog.create_time)).layui_paginate()
count = log.total
return table_api(data=model_to_dicts(schema=LogOutSchema, data=log.items), count=count)
-221
View File
@@ -1,221 +0,0 @@
from flask import Blueprint, render_template, request, jsonify
from applications.common import curd
from applications.common.helper import ModelFilter
from applications.common.utils.http import table_api, success_api, fail_api
from applications.common.utils.rights import authorize
from applications.common.utils.validate import xss_escape
from applications.extensions import db
from applications.models import DictType, DictData
from applications.schemas import DictTypeOutSchema, DictDataOutSchema
admin_dict = Blueprint('adminDict', __name__, url_prefix='/admin/dict')
# 数据字典
@admin_dict.get('/')
@authorize("admin:dict:main", log=True)
def main():
return render_template('admin/dict/main.html')
@admin_dict.get('/dictType/data')
@authorize("admin:dict:main", log=True)
def dict_type_data():
# 获取请求参数
type_name = xss_escape(request.args.get('typeName', type=str))
# 查询参数构造
mf = ModelFilter()
if type_name:
mf.vague(field_name="type_name", value=type_name)
# orm查询
# 使用分页获取data需要.items
dict_all = DictType.query.filter(mf.get_filter(DictType)).layui_paginate()
count = dict_all.total
data = curd.model_to_dicts(schema=DictTypeOutSchema, data=dict_all.items)
return table_api(data=data, count=count)
@admin_dict.get('/dictType/add')
@authorize("admin:dict:add", log=True)
def dict_type_add():
return render_template('admin/dict/add.html')
@admin_dict.post('/dictType/save')
@authorize("admin:dict:add", log=True)
def dict_type_save():
req_json = request.json
description = xss_escape(req_json.get("description"))
enable = xss_escape(req_json.get("enable"))
type_code = xss_escape(req_json.get("typeCode"))
type_name = xss_escape(req_json.get("typeName"))
d = DictType(type_name=type_name, type_code=type_code, enable=enable, description=description)
db.session.add(d)
db.session.commit()
if d.id is None:
return fail_api(msg="增加失败")
return success_api(msg="增加成功")
# 编辑字典类型
@admin_dict.get('/dictType/edit')
@authorize("admin:dict:edit", log=True)
def dict_type_edit():
_id = request.args.get('dictTypeId', type=int)
dict_type = DictType.query.filter_by(id=_id).first()
return render_template('admin/dict/edit.html', dict_type=dict_type)
# 编辑字典类型
@admin_dict.put('/dictType/update')
@authorize("admin:dict:edit", log=True)
def dict_type_update():
req_json = request.json
id = xss_escape(req_json.get("id"))
description = xss_escape(req_json.get("description"))
enable = xss_escape(req_json.get("enable"))
type_code = xss_escape(req_json.get("typeCode"))
type_name = xss_escape(req_json.get("typeName"))
DictType.query.filter_by(id=id).update({
"description": description,
"enable": enable,
"type_code": type_code,
"type_name": type_name
})
db.session.commit()
return success_api(msg="更新成功")
# 启用字典
@admin_dict.put('/dictType/enable')
@authorize("admin:dict:edit", log=True)
def dict_type_enable():
_id = request.json.get('id')
if id:
res = curd.enable_status(DictType,_id)
if not res:
return fail_api(msg="出错啦")
return success_api("启动成功")
return fail_api(msg="数据错误")
# 禁用字典
@admin_dict.put('/dictType/disable')
@authorize("admin:dict:edit", log=True)
def dict_type_dis_enable():
_id = request.json.get('id')
if id:
res = curd.disable_status(DictType,_id)
if not res:
return fail_api(msg="出错啦")
return success_api("禁用成功")
return fail_api(msg="数据错误")
# 删除字典类型
@admin_dict.delete('/dictType/remove/<int:_id>')
@authorize("admin:dict:remove", log=True)
def dict_type_delete(_id):
res = curd.delete_one_by_id(DictType,_id)
if not res:
return fail_api(msg="删除失败")
return success_api(msg="删除成功")
@admin_dict.get('/dictData/data')
@authorize("admin:dict:main", log=True)
def dict_code_data():
type_code = xss_escape(request.args.get('typeCode', type=str))
dict_data = DictData.query.filter_by(type_code=type_code).layui_paginate()
count = dict_data.total
data = curd.model_to_dicts(schema=DictDataOutSchema, data=dict_data.items)
return table_api(data=data, count=count)
# 增加字典数据
@admin_dict.get('/dictData/add')
@authorize("admin:dict:add", log=True)
def dict_data_add():
type_code = request.args.get('typeCode', type=str)
return render_template('admin/dict/data/add.html', type_code=type_code)
# 增加字典数据
@admin_dict.post('/dictData/save')
@authorize("admin:dict:add", log=True)
def dict_data_save():
req_json = request.json
data_label = xss_escape(req_json.get("dataLabel"))
data_value = xss_escape(req_json.get("dataValue"))
enable = xss_escape(req_json.get("enable"))
remark = xss_escape(req_json.get("remark"))
type_code = xss_escape(req_json.get("typeCode"))
d = DictData(data_label=data_label, data_value=data_value, enable=enable, remark=remark, type_code=type_code)
db.session.add(d)
db.session.commit()
if not d.id:
return jsonify(success=False, msg="增加失败")
return jsonify(success=True, msg="增加成功")
# 编辑字典数据
@admin_dict.get('/dictData/edit')
@authorize("admin:dict:edit", log=True)
def dict_data_edit():
_id = request.args.get('dataId', type=str)
dict_data = curd.get_one_by_id(DictData, _id)
return render_template('admin/dict/data/edit.html', dict_data=dict_data)
# 编辑字典数据
@admin_dict.put('/dictData/update')
@authorize("admin:dict:edit", log=True)
def dict_data_update():
req_json = request.json
id = req_json.get("dataId")
DictData.query.filter_by(id=id).update({
"data_label": xss_escape(req_json.get("dataLabel")),
"data_value": xss_escape(req_json.get("dataValue")),
"enable": xss_escape(req_json.get("enable")),
"remark": xss_escape(req_json.get("remark")),
"type_code": xss_escape(req_json.get("typeCode"))
})
db.session.commit()
return success_api(msg="更新成功")
# 启用字典数据
@admin_dict.put('/dictData/enable')
@authorize("admin:dict:edit", log=True)
def dict_data_enable():
_id = request.json.get('dataId')
if _id:
res = curd.enable_status(model=DictData, id=_id)
if not res:
return fail_api(msg="出错啦")
return success_api(msg="启动成功")
return fail_api(msg="数据错误")
# 禁用字典数据
@admin_dict.put('/dictData/disable')
@authorize("admin:dict:edit", log=True)
def dict_data_disenable():
_id = request.json.get('dataId')
if _id:
res = curd.disable_status(model=DictData, id=_id)
if not res:
return fail_api(msg="出错啦")
return success_api(msg="禁用成功")
return fail_api(msg="数据错误")
# 删除字典类型
@admin_dict.delete('dictData/remove/<int:id>')
@authorize("admin:dict:remove", log=True)
def dict_data_delete(id):
res = curd.delete_one_by_id(model=DictData, id=id)
if not res:
return fail_api(msg="删除失败")
return success_api(msg="删除成功")
-83
View File
@@ -1,83 +0,0 @@
import os
from flask import Blueprint, request, render_template, jsonify, current_app
from applications.common.utils.http import fail_api, success_api, table_api
from applications.common.utils.rights import authorize
from applications.extensions import db
from applications.models import Photo
from applications.common.utils import upload as upload_curd
admin_file = Blueprint('adminFile', __name__, url_prefix='/admin/file')
# 图片管理
@admin_file.get('/')
@authorize("admin:file:main", log=True)
def index():
return render_template('admin/photo/photo.html')
# 图片数据
@admin_file.get('/table')
@authorize("admin:file:main", log=True)
def table():
page = request.args.get('page', type=int)
limit = request.args.get('limit', type=int)
data, count = upload_curd.get_photo(page=page, limit=limit)
return table_api(data=data, count=count)
# 上传
@admin_file.get('/upload')
@authorize("admin:file:add", log=True)
def upload():
return render_template('admin/photo/photo_add.html')
# 上传接口
@admin_file.post('/upload')
@authorize("admin:file:add", log=True)
def upload_api():
if 'file' in request.files:
photo = request.files['file']
mime = request.files['file'].content_type
file_url = upload_curd.upload_one(photo=photo, mime=mime)
res = {
"msg": "上传成功",
"code": 0,
"success": True,
"data":
{"src": file_url}
}
return jsonify(res)
return fail_api()
# 图片删除
@admin_file.route('/delete', methods=['GET', 'POST'])
@authorize("admin:file:delete", log=True)
def delete():
_id = request.form.get('id')
res = upload_curd.delete_photo_by_id(_id)
if res:
return success_api(msg="删除成功")
else:
return fail_api(msg="删除失败")
# 图片批量删除
@admin_file.route('/batchRemove', methods=['GET', 'POST'])
@authorize("admin:file:delete", log=True)
def batch_remove():
ids = request.form.getlist('ids[]')
photo_name = Photo.query.filter(Photo.id.in_(ids)).all()
upload_url = current_app.config.get("UPLOADED_PHOTOS_DEST")
for p in photo_name:
os.remove(upload_url + '/' + p.name)
photo = Photo.query.filter(Photo.id.in_(ids)).delete(synchronize_session=False)
db.session.commit()
if photo:
return success_api(msg="删除成功")
else:
return fail_api(msg="删除失败")
-19
View File
@@ -1,19 +0,0 @@
from flask import Blueprint, render_template
from flask_login import login_required, current_user
admin_bp = Blueprint('admin', __name__, url_prefix='/admin')
# 首页
@admin_bp.get('/')
@login_required
def index():
user = current_user
return render_template('admin/index.html', user=user)
# 控制台页面
@admin_bp.get('/welcome')
@login_required
def welcome():
return render_template('admin/console/console.html')
-88
View File
@@ -1,88 +0,0 @@
import os
import platform
import re
from datetime import datetime
import time
import psutil
from flask import Blueprint, render_template, jsonify
from applications.common.utils.rights import authorize
admin_monitor_bp = Blueprint('adminMonitor', __name__, url_prefix='/admin/monitor')
# 系统监控
@admin_monitor_bp.get('/')
@authorize("admin:monitor:main", log=True)
def main():
# 主机名称
hostname = platform.node()
# 系统版本
system_version = platform.platform()
# python版本
python_version = platform.python_version()
# 逻辑cpu数量
cpu_count = psutil.cpu_count()
# cup使用率
cpus_percent = psutil.cpu_percent(interval=0.1)
# 内存
memory_information = psutil.virtual_memory()
# 内存使用率
memory_usage = memory_information.percent
memory_used = str(round(memory_information.used / 1024 / 1024))
memory_total = str(round(memory_information.total / 1024 / 1024))
memory_free = str(round(memory_information.free / 1024 / 1024))
# 磁盘信息
disk_partitions_list = []
# 判断是否在容器中
if not os.path.exists('/.dockerenv'):
disk_partitions = psutil.disk_partitions()
for i in disk_partitions:
a = psutil.disk_usage(i.device)
disk_partitions_dict = {
'device': i.device,
'fstype': i.fstype,
'total': str(round(a.total / 1024 / 1024)),
'used': str(round(a.used / 1024 / 1024)),
'free': str(round(a.free / 1024 / 1024)),
'percent': a.percent
}
disk_partitions_list.append(disk_partitions_dict)
# 开机时间
boot_time = datetime.fromtimestamp(psutil.boot_time()).replace(microsecond=0)
up_time = datetime.now().replace(microsecond=0) - boot_time
up_time_list = re.split(r':', str(up_time))
up_time_format = " {} 小时{} 分钟{}".format(up_time_list[0], up_time_list[1], up_time_list[2])
# 当前时间
time_now = time.strftime('%H:%M:%S ', time.localtime(time.time()))
return render_template('admin/monitor.html',
hostname=hostname,
system_version=system_version,
python_version=python_version,
cpus_percent=cpus_percent,
memory_usage=memory_usage,
cpu_count=cpu_count,
memory_used=memory_used,
memory_total=memory_total,
memory_free=memory_free,
boot_time=boot_time,
up_time_format=up_time_format,
disk_partitions_list=disk_partitions_list,
time_now=time_now
)
# 图表 api
@admin_monitor_bp.get('/polling')
@authorize("admin:monitor:main")
def ajax_polling():
# 获取cup使用率
cpus_percent = psutil.cpu_percent(interval=0.1)
# 获取内存使用率
memory_information = psutil.virtual_memory()
memory_usage = memory_information.percent
time_now = time.strftime('%H:%M:%S ', time.localtime(time.time()))
return jsonify(cups_percent=cpus_percent, memory_used=memory_usage, time_now=time_now)
-167
View File
@@ -1,167 +0,0 @@
from flask import Blueprint, render_template, request, jsonify
from applications.common import curd
from applications.common.utils.http import success_api, fail_api
from applications.common.utils.rights import authorize
from applications.common.utils.validate import xss_escape
from applications.extensions import db
from applications.models import Power
from applications.schemas import PowerOutSchema2
admin_power = Blueprint('adminPower', __name__, url_prefix='/admin/power')
@admin_power.get('/')
@authorize("admin:power:main", log=True)
def index():
return render_template('admin/power/main.html')
@admin_power.post('/data')
@authorize("admin:power:main", log=True)
def data():
power = Power.query.all()
res = {
"data": curd.model_to_dicts(schema=PowerOutSchema2, data=power)
}
return jsonify(res)
@admin_power.get('/add')
@authorize("admin:power:add", log=True)
def add():
return render_template('admin/power/add.html')
@admin_power.get('/selectParent')
@authorize("admin:power:main", log=True)
def select_parent():
power = Power.query.all()
res = curd.model_to_dicts(schema=PowerOutSchema2, data=power)
res.append({"powerId": 0, "powerName": "顶级权限", "parentId": -1})
res = {
"status": {"code": 200, "message": "默认"},
"data": res
}
return jsonify(res)
# 增加
@admin_power.post('/save')
@authorize("admin:power:add", log=True)
def save():
req = request.json
icon = xss_escape(req.get("icon"))
openType = xss_escape(req.get("openType"))
parentId = xss_escape(req.get("parentId"))
powerCode = xss_escape(req.get("powerCode"))
powerName = xss_escape(req.get("powerName"))
powerType = xss_escape(req.get("powerType"))
powerUrl = xss_escape(req.get("powerUrl"))
sort = xss_escape(req.get("sort"))
power = Power(
icon=icon,
open_type=openType,
parent_id=parentId,
code=powerCode,
name=powerName,
type=powerType,
url=powerUrl,
sort=sort,
enable=1
)
r = db.session.add(power)
db.session.commit()
return success_api(msg="成功")
# 权限编辑
@admin_power.get('/edit/<int:_id>')
@authorize("admin:power:edit", log=True)
def edit(_id):
power = curd.get_one_by_id(Power, _id)
icon = str(power.icon).split()
if len(icon) == 2:
icon = icon[1]
else:
icon = None
return render_template('admin/power/edit.html', power=power, icon=icon)
# 权限更新
@admin_power.put('/update')
@authorize("admin:power:edit", log=True)
def update():
req_json = request.json
id = request.json.get("powerId")
data = {
"icon": xss_escape(req_json.get("icon")),
"open_type": xss_escape(req_json.get("openType")),
"parent_id": xss_escape(req_json.get("parentId")),
"code": xss_escape(req_json.get("powerCode")),
"name": xss_escape(req_json.get("powerName")),
"type": xss_escape(req_json.get("powerType")),
"url": xss_escape(req_json.get("powerUrl")),
"sort": xss_escape(req_json.get("sort"))
}
res = Power.query.filter_by(id=id).update(data)
db.session.commit()
if not res:
return fail_api(msg="更新权限失败")
return success_api(msg="更新权限成功")
# 启用权限
@admin_power.put('/enable')
@authorize("admin:power:edit", log=True)
def enable():
_id = request.json.get('powerId')
if id:
res = curd.enable_status(Power, _id)
if not res:
return fail_api(msg="出错啦")
return success_api(msg="启用成功")
return fail_api(msg="数据错误")
# 禁用权限
@admin_power.put('/disable')
@authorize("admin:power:edit", log=True)
def dis_enable():
_id = request.json.get('powerId')
if id:
res = curd.disable_status(Power, _id)
if not res:
return fail_api(msg="出错啦")
return success_api(msg="禁用成功")
return fail_api(msg="数据错误")
# 权限删除
@admin_power.delete('/remove/<int:id>')
@authorize("admin:power:remove", log=True)
def remove(id):
power = Power.query.filter_by(id=id).first()
power.role = []
r = Power.query.filter_by(id=id).delete()
db.session.commit()
if r:
return success_api(msg="删除成功")
else:
return fail_api(msg="删除失败")
# 批量删除
@admin_power.delete('/batchRemove')
@authorize("admin:power:remove", log=True)
def batch_remove():
ids = request.form.getlist('ids[]')
for id in ids:
power = Power.query.filter_by(id=id).first()
power.role = []
r = Power.query.filter_by(id=id).delete()
db.session.commit()
return success_api(msg="批量删除成功")
-204
View File
@@ -1,204 +0,0 @@
from flask import Blueprint, render_template, request, jsonify
from flask_login import login_required
from applications.common.curd import model_to_dicts, enable_status, disable_status, get_one_by_id
from applications.common.helper import ModelFilter
from applications.common.utils.http import table_api, success_api, fail_api
from applications.common.utils.rights import authorize
from applications.common.utils.validate import xss_escape
from applications.extensions import db
from applications.models import Role, Power, User
from applications.schemas import RoleOutSchema, PowerOutSchema2
admin_role = Blueprint('adminRole', __name__, url_prefix='/admin/role')
# 用户管理
@admin_role.get('/')
@authorize("admin:role:main", log=True)
def main():
return render_template('admin/role/main.html')
# 表格数据
@admin_role.get('/data')
@authorize("admin:role:main", log=True)
def table():
# 获取请求参数
role_name = xss_escape(request.args.get('roleName', type=str))
role_code = xss_escape(request.args.get('roleCode', type=str))
# 查询参数构造
mf = ModelFilter()
if role_name:
mf.vague(field_name="name", value=role_name)
if role_code:
mf.vague(field_name="code", value=role_code)
# orm查询
# 使用分页获取data需要.items
role = Role.query.filter(mf.get_filter(Role)).layui_paginate()
count = role.total
# 返回api
return table_api(data=model_to_dicts(schema=RoleOutSchema, data=role.items), count=count)
# 角色增加
@admin_role.get('/add')
@authorize("admin:role:add", log=True)
def add():
return render_template('admin/role/add.html')
# 角色增加
@admin_role.post('/save')
@authorize("admin:role:add", log=True)
def save():
req = request.json
details = xss_escape(req.get("details"))
enable = xss_escape(req.get("enable"))
roleCode = xss_escape(req.get("roleCode"))
roleName = xss_escape(req.get("roleName"))
sort = xss_escape(req.get("sort"))
role = Role(
details=details,
enable=enable,
code=roleCode,
name=roleName,
sort=sort
)
db.session.add(role)
db.session.commit()
return success_api(msg="成功")
# 角色授权
@admin_role.get('/power/<int:_id>')
@authorize("admin:role:power", log=True)
def power(_id):
return render_template('admin/role/power.html', id=_id)
# 获取角色权限
@admin_role.get('/getRolePower/<int:id>')
@authorize("admin:role:main", log=True)
def get_role_power(id):
role = Role.query.filter_by(id=id).first()
check_powers = role.power
check_powers_list = []
for cp in check_powers:
check_powers_list.append(cp.id)
powers = Power.query.all()
power_schema = PowerOutSchema2(many=True) # 用已继承ma.ModelSchema类的自定制类生成序列化类
output = power_schema.dump(powers) # 生成可序列化对象
for i in output:
if int(i.get("powerId")) in check_powers_list:
i["checkArr"] = "1"
else:
i["checkArr"] = "0"
res = {
"data": output,
"status": {"code": 200, "message": "默认"}
}
return jsonify(res)
# 保存角色权限
@admin_role.put('/saveRolePower')
@authorize("admin:role:edit", log=True)
def save_role_power():
req_form = request.form
power_ids = req_form.get("powerIds")
power_list = power_ids.split(',')
role_id = req_form.get("roleId")
role = Role.query.filter_by(id=role_id).first()
powers = Power.query.filter(Power.id.in_(power_list)).all()
role.power = powers
db.session.commit()
return success_api(msg="授权成功")
# 角色编辑
@admin_role.get('/edit/<int:id>')
@authorize("admin:role:edit", log=True)
def edit(id):
r = get_one_by_id(model=Role, id=id)
return render_template('admin/role/edit.html', role=r)
# 更新角色
@admin_role.put('/update')
@authorize("admin:role:edit", log=True)
def update():
req_json = request.json
id = req_json.get("roleId")
data = {
"code": xss_escape(req_json.get("roleCode")),
"name": xss_escape(req_json.get("roleName")),
"sort": xss_escape(req_json.get("sort")),
"enable": xss_escape(req_json.get("enable")),
"details": xss_escape(req_json.get("details"))
}
role = Role.query.filter_by(id=id).update(data)
db.session.commit()
if not role:
return fail_api(msg="更新角色失败")
return success_api(msg="更新角色成功")
# 启用用户
@admin_role.put('/enable')
@authorize("admin:role:edit", log=True)
def enable():
id = request.json.get('roleId')
if id:
res = enable_status(Role, id)
if not res:
return fail_api(msg="出错啦")
return success_api(msg="启动成功")
return fail_api(msg="数据错误")
# 禁用用户
@admin_role.put('/disable')
@authorize("admin:role:edit", log=True)
def dis_enable():
_id = request.json.get('roleId')
if _id:
res = disable_status(Role, _id)
if not res:
return fail_api(msg="出错啦")
return success_api(msg="禁用成功")
return fail_api(msg="数据错误")
# 角色删除
@admin_role.delete('/remove/<int:id>')
@authorize("admin:role:remove", log=True)
def remove(id):
role = Role.query.filter_by(id=id).first()
# 删除该角色的权限和用户
role.power = []
role.user = []
r = Role.query.filter_by(id=id).delete()
db.session.commit()
if not r:
return fail_api(msg="角色删除失败")
return success_api(msg="角色删除成功")
# 批量删除
@admin_role.delete('/batchRemove')
@authorize("admin:role:remove", log=True)
@login_required
def batch_remove():
ids = request.form.getlist('ids[]')
for id in ids:
role = Role.query.filter_by(id=id).first()
# 删除该角色的权限和用户
role.power = []
role.user = []
r = Role.query.filter_by(id=id).delete()
db.session.commit()
return success_api(msg="批量删除成功")
-112
View File
@@ -1,112 +0,0 @@
from flask import Blueprint, request, render_template
from flask_apscheduler.utils import job_to_dict
from applications.common.tasks import tasks
from applications.common.tasks.tasks import task_list
from applications.common.utils.http import table_api, fail_api, success_api
from applications.extensions.init_apscheduler import scheduler
admin_task = Blueprint('adminTask', __name__, url_prefix='/admin/task')
@admin_task.route('/add_job', methods=['GET'])
def add_task():
scheduler.add_job(func=tasks.get(), id='4', args=(1, 1), trigger='interval', seconds=3,
replace_existing=True)
return '6'
@admin_task.get('/')
def main():
return render_template('admin/task/main.html')
@admin_task.route('/data', methods=['GET'])
def get_task(): # 获取
jobs = scheduler.get_jobs()
jobs_list = []
for job in jobs:
jobs_list.append(job_to_dict(job))
return table_api(data=jobs_list, count=len(jobs_list))
# 增加
@admin_task.get('/add')
def add():
return render_template('admin/task/add.html', task_list=task_list)
@admin_task.post('/save')
def save():
_id = request.json.get("id")
name = request.json.get("id")
type = request.json.get("type")
functions = request.json.get("functions")
datetime = request.json.get("datetime")
time = request.json.get("time")
if not hasattr(tasks, functions):
return fail_api()
if type == 'date':
scheduler.add_job(
func=getattr(tasks, functions),
id=_id,
name=name,
args=(1, 1),
trigger=type,
run_date=datetime,
replace_existing=True)
elif type == 'interval':
scheduler.add_job(
func=getattr(tasks, functions),
id=_id,
name=name,
args=(1, 1),
trigger=type,
replace_existing=True)
elif type == 'cron':
scheduler.add_job(
func=getattr(tasks, functions),
id=_id,
name=name,
args=(1, 1),
trigger=type,
replace_existing=True)
return success_api()
# 恢复
@admin_task.put('/enable')
def enable():
_id = request.json.get('id')
# print(id)
if _id:
scheduler.resume_job(str(_id))
return success_api(msg="启动成功")
return fail_api(msg="数据错误")
# 暂停
@admin_task.put('/disable')
def dis_enable():
_id = request.json.get('id')
if _id:
scheduler.pause_job(str(_id))
return success_api(msg="暂停成功")
return fail_api(msg="数据错误")
@admin_task.delete('/remove/<int:_id>')
def remove_job(_id): # 移除
scheduler.remove_job(str(_id))
return success_api(msg="删除成功")
# scheduler.add_job(func=task1, id='2', args=(1, 1), trigger='cron', day_of_week='0-6', hour=18, minute=24,
# second=10, replace_existing=True)
# scheduler.add_job(func=task4, id='4', args=(2, 2), trigger='interval', seconds=3,
# replace_existing=True, misfire_grace_time=3)
# # trigger='interval' 表示是一个循环任务,每隔多久执行一次
# scheduler.add_job(func=task2, id='3', args=(2, 2), trigger='interval', seconds=3,
# replace_existing=True)
-235
View File
@@ -1,235 +0,0 @@
from flask import Blueprint, render_template, request
from flask_login import login_required, current_user
from sqlalchemy import desc
from applications.common import curd
from applications.common.curd import model_to_dicts, enable_status, disable_status
from applications.common.helper import ModelFilter
from applications.common.utils.http import table_api, fail_api, success_api
from applications.common.utils.rights import authorize
from applications.common.utils.validate import xss_escape
from applications.extensions import db
from applications.models import Role
from applications.models import User, AdminLog
from applications.schemas import UserOutSchema
admin_user = Blueprint('adminUser', __name__, url_prefix='/admin/user')
# 用户管理
@admin_user.get('/')
@authorize("admin:user:main", log=True)
def main():
return render_template('admin/user/main.html')
# 用户分页查询
@admin_user.get('/data')
@authorize("admin:user:main", log=True)
def data():
# 获取请求参数
real_name = xss_escape(request.args.get('realName', type=str))
username = xss_escape(request.args.get('username', type=str))
dept_id = request.args.get('deptId', type=int)
# 查询参数构造
mf = ModelFilter()
if real_name:
mf.contains(field_name="realname", value=real_name)
if username:
mf.contains(field_name="username", value=username)
if dept_id:
mf.exact(field_name="dept_id", value=dept_id)
# orm查询
# 使用分页获取data需要.items
user = User.query.filter(mf.get_filter(model=User)).layui_paginate()
count = user.total
# 返回api
return table_api(data=model_to_dicts(schema=UserOutSchema, data=user.items), count=count)
# 用户增加
@admin_user.get('/add')
@authorize("admin:user:add", log=True)
def add():
roles = Role.query.all()
return render_template('admin/user/add.html', roles=roles)
@admin_user.post('/save')
@authorize("admin:user:add", log=True)
def save():
req_json = request.json
a = req_json.get("roleIds")
username = xss_escape(req_json.get('username'))
real_name = xss_escape(req_json.get('realName'))
password = xss_escape(req_json.get('password'))
role_ids = a.split(',')
if not username or not real_name or not password:
return fail_api(msg="账号姓名密码不得为空")
if bool(User.query.filter_by(username=username).count()):
return fail_api(msg="用户已经存在")
user = User(username=username, realname=real_name)
user.set_password(password)
db.session.add(user)
roles = Role.query.filter(Role.id.in_(role_ids)).all()
for r in roles:
user.role.append(r)
db.session.commit()
return success_api(msg="增加成功")
# 删除用户
@admin_user.delete('/remove/<int:id>')
@authorize("admin:user:remove", log=True)
def delete(id):
user = User.query.filter_by(id=id).first()
user.role = []
res = User.query.filter_by(id=id).delete()
db.session.commit()
if not res:
return fail_api(msg="删除失败")
return success_api(msg="删除成功")
# 编辑用户
@admin_user.get('/edit/<int:id>')
@authorize("admin:user:edit", log=True)
def edit(id):
user = curd.get_one_by_id(User,id)
roles = Role.query.all()
checked_roles = []
for r in user.role:
checked_roles.append(r.id)
return render_template('admin/user/edit.html', user=user, roles=roles, checked_roles=checked_roles)
# 编辑用户
@admin_user.put('/update')
@authorize("admin:user:edit", log=True)
def update():
req_json = request.json
a = xss_escape(req_json.get("roleIds"))
id = xss_escape(req_json.get("userId"))
username = xss_escape(req_json.get('username'))
real_name = xss_escape(req_json.get('realName'))
dept_id = xss_escape(req_json.get('deptId'))
role_ids = a.split(',')
User.query.filter_by(id=id).update({'username': username, 'realname': real_name, 'dept_id': dept_id})
u = User.query.filter_by(id=id).first()
roles = Role.query.filter(Role.id.in_(role_ids)).all()
u.role = roles
db.session.commit()
return success_api(msg="更新成功")
# 个人中心
@admin_user.get('/center')
@login_required
def center():
user_info = current_user
user_logs = AdminLog.query.filter_by(url='/passport/login').filter_by(uid=current_user.id).order_by(
desc(AdminLog.create_time)).limit(10)
return render_template('admin/user/center.html', user_info=user_info, user_logs=user_logs)
# 修改头像
@admin_user.get('/profile')
@login_required
def profile():
return render_template('admin/user/profile.html')
# 修改头像
@admin_user.put('/updateAvatar')
@login_required
def update_avatar():
url = request.json.get("avatar").get("src")
r = User.query.filter_by(id=current_user.id).update({"avatar": url})
db.session.commit()
if not r:
return fail_api(msg="出错啦")
return success_api(msg="修改成功")
# 修改当前用户信息
@admin_user.put('/updateInfo')
@login_required
def update_info():
req_json = request.json
r = User.query.filter_by(id=current_user.id).update(
{"realname": req_json.get("realName"), "remark": req_json.get("details")})
db.session.commit()
if not r:
return fail_api(msg="出错啦")
return success_api(msg="更新成功")
# 修改当前用户密码
@admin_user.get('/editPassword')
@login_required
def edit_password():
return render_template('admin/user/edit_password.html')
# 修改当前用户密码
@admin_user.put('/editPassword')
@login_required
def edit_password_put():
res_json = request.json
if res_json.get("newPassword") == '':
return fail_api("新密码不得为空")
if res_json.get("newPassword") != res_json.get("confirmPassword"):
return fail_api("俩次密码不一样")
user = current_user
is_right = user.validate_password(res_json.get("oldPassword"))
if not is_right:
return fail_api("旧密码错误")
user.set_password(res_json.get("newPassword"))
db.session.add(user)
db.session.commit()
return success_api("更改成功")
# 启用用户
@admin_user.put('/enable')
@authorize("admin:user:edit", log=True)
def enable():
_id = request.json.get('userId')
if _id:
res = enable_status(model=User, id=_id)
if not res:
return fail_api(msg="出错啦")
return success_api(msg="启动成功")
return fail_api(msg="数据错误")
# 禁用用户
@admin_user.put('/disable')
@authorize("admin:user:edit", log=True)
def dis_enable():
_id = request.json.get('userId')
if _id:
res = disable_status(model=User,id=_id)
if not res:
return fail_api(msg="出错啦")
return success_api(msg="禁用成功")
return fail_api(msg="数据错误")
# 批量删除
@admin_user.delete('/batchRemove')
@authorize("admin:user:remove", log=True)
def batch_remove():
ids = request.form.getlist('ids[]')
for id in ids:
user = User.query.filter_by(id=id).first()
user.role = []
res = User.query.filter_by(id=id).delete()
db.session.commit()
return success_api(msg="批量删除成功")
-144
View File
@@ -1,144 +0,0 @@
from flask import Blueprint, render_template, request, jsonify
from applications.common import curd
from applications.common.utils import validate
from applications.common.utils.http import success_api, fail_api
from applications.common.utils.rights import authorize
from applications.common.utils.validate import xss_escape
from applications.extensions import db
from applications.models import Dept, User
from applications.schemas import DeptOutSchema
dept_bp = Blueprint('dept', __name__, url_prefix='/dept')
def register_dept_views(app):
app.register_blueprint(dept_bp)
@dept_bp.get('/')
@authorize("admin:dept:main", log=True)
def main():
return render_template('admin/dept/main.html')
@dept_bp.post('/data')
@authorize("admin:dept:main", log=True)
def data():
dept = Dept.query.order_by(Dept.sort).all()
power_data = curd.model_to_dicts(schema=DeptOutSchema, data=dept)
res = {
"data": power_data
}
return jsonify(res)
@dept_bp.get('/add')
@authorize("admin:dept:add", log=True)
def add():
return render_template('admin/dept/add.html')
@dept_bp.get('/tree')
@authorize("admin:dept:main", log=True)
def tree():
dept = Dept.query.order_by(Dept.sort).all()
power_data = curd.model_to_dicts(schema=DeptOutSchema, data=dept)
res = {
"status": {"code": 200, "message": "默认"},
"data": power_data
}
return jsonify(res)
@dept_bp.post('/save')
@authorize("admin:dept:add", log=True)
def save():
req_json = request.json
dept = Dept(
parent_id=req_json.get('parentId'),
dept_name=xss_escape(req_json.get('deptName')),
sort=xss_escape(req_json.get('sort')),
leader=xss_escape(req_json.get('leader')),
phone=xss_escape(req_json.get('phone')),
email=xss_escape(req_json.get('email')),
status=xss_escape(req_json.get('status')),
address=xss_escape(req_json.get('address'))
)
r = db.session.add(dept)
db.session.commit()
return success_api(msg="成功")
@dept_bp.get('/edit')
@authorize("admin:dept:edit", log=True)
def edit():
_id = request.args.get("deptId")
dept = curd.get_one_by_id(model=Dept, id=_id)
return render_template('admin/dept/edit.html', dept=dept)
# 启用
@dept_bp.put('/enable')
@authorize("admin:dept:edit", log=True)
def enable():
id = request.json.get('deptId')
if id:
enable = 1
d = Dept.query.filter_by(id=id).update({"status": enable})
if d:
db.session.commit()
return success_api(msg="启用成功")
return fail_api(msg="出错啦")
return fail_api(msg="数据错误")
# 禁用
@dept_bp.put('/disable')
@authorize("admin:dept:edit", log=True)
def dis_enable():
id = request.json.get('deptId')
if id:
enable = 0
d = Dept.query.filter_by(id=id).update({"status": enable})
if d:
db.session.commit()
return success_api(msg="禁用成功")
return fail_api(msg="出错啦")
return fail_api(msg="数据错误")
@dept_bp.put('/update')
@authorize("admin:dept:edit", log=True)
def update():
json = request.json
id = json.get("deptId"),
data = {
"dept_name": validate.xss_escape(json.get("deptName")),
"sort": validate.xss_escape(json.get("sort")),
"leader": validate.xss_escape(json.get("leader")),
"phone": validate.xss_escape(json.get("phone")),
"email": validate.xss_escape(json.get("email")),
"status": validate.xss_escape(json.get("status")),
"address": validate.xss_escape(json.get("address"))
}
d = Dept.query.filter_by(id=id).update(data)
if not d:
return fail_api(msg="更新失败")
db.session.commit()
return success_api(msg="更新成功")
@dept_bp.delete('/remove/<int:_id>')
@authorize("admin:dept:remove", log=True)
def remove(_id):
d = Dept.query.filter_by(id=_id).delete()
if not d:
return fail_api(msg="删除失败")
res = User.query.filter_by(dept_id=_id).update({"dept_id": None})
db.session.commit()
if res:
return success_api(msg="删除成功")
else:
return fail_api(msg="删除失败")
-12
View File
@@ -1,12 +0,0 @@
from applications.view.index.index import index_bp
from . import index
def register_index_views(app):
"""
初始化蓝图
"""
app.register_blueprint(index_bp)
-8
View File
@@ -1,8 +0,0 @@
from flask import render_template, Blueprint
index_bp = Blueprint('Index', __name__, url_prefix='/')
@index_bp.route('/')
def index():
return render_template('index/index.html')
-76
View File
@@ -1,76 +0,0 @@
from flask import Blueprint, session, redirect, url_for, render_template, request
from flask_login import current_user, login_user, login_required, logout_user
from applications.common import admin as index_curd
from applications.common.admin_log import login_log
from applications.common.utils.http import fail_api, success_api
from applications.models import User
passport_bp = Blueprint('passport', __name__, url_prefix='/passport')
def register_passport_views(app):
app.register_blueprint(passport_bp)
# 获取验证码
@passport_bp.get('/getCaptcha')
def get_captcha():
resp, code = index_curd.get_captcha()
session["code"] = code
return resp
# 登录
@passport_bp.get('/login')
def login():
if current_user.is_authenticated:
return redirect(url_for('admin.index'))
return render_template('admin/login.html')
# 登录
@passport_bp.post('/login')
def login_post():
req = request.form
username = req.get('username')
password = req.get('password')
code = req.get('captcha').__str__().lower()
if not username or not password or not code:
return fail_api(msg="用户名或密码没有输入")
s_code = session.get("code", None)
session["code"] = None
if not all([code, s_code]):
return fail_api(msg="参数错误")
if code != s_code:
return fail_api(msg="验证码错误")
user = User.query.filter_by(username=username).first()
if not user:
return fail_api(msg="不存在的用户")
if user.enable == 0:
return fail_api(msg="用户被暂停使用")
if username == user.username and user.validate_password(password):
# 登录
login_user(user)
# 记录登录日志
login_log(request, uid=user.id, is_access=True)
# 存入权限
index_curd.add_auth_session()
return success_api(msg="登录成功")
login_log(request, uid=user.id, is_access=False)
return fail_api(msg="用户名或密码错误")
# 退出登录
@passport_bp.post('/logout')
@login_required
def logout():
logout_user()
session.pop('permissions')
return success_api(msg="注销成功")
-9
View File
@@ -1,9 +0,0 @@
from flask import Blueprint, Flask
rights_bp = Blueprint('rights', __name__, url_prefix='/rights')
from . import routes
def register_rights_view(app: Flask):
app.register_blueprint(rights_bp)
-20
View File
@@ -1,20 +0,0 @@
# 渲染配置
from flask import jsonify
from flask_login import login_required
from . import rights_bp
from ...common import admin
@rights_bp.get('/configs')
@login_required
def configs():
return admin.get_render_config()
# 菜单
@rights_bp.get('/menu')
@login_required
def menu():
menu_tree = admin.make_menu_tree()
return jsonify(menu_tree)
+36
View File
@@ -0,0 +1,36 @@
import os
from datetime import timedelta
class BaseConfig:
SECRET_KEY = os.getenv("SECRET_KEY", "pear-admin-flask")
SQLALCHEMY_DATABASE_URI = ""
ROOT_PATH = os.path.dirname(os.path.abspath(__file__))
JWT_TOKEN_LOCATION = ["headers"]
JWT_ACCESS_TOKEN_EXPIRES = timedelta(days=7)
class DevelopmentConfig(BaseConfig):
"""开发配置"""
SQLALCHEMY_DATABASE_URI = "sqlite:///pear_admin.db"
SQLALCHEMY_TRACK_MODIFICATIONS = False
class TestingConfig(BaseConfig):
"""测试配置"""
SQLALCHEMY_DATABASE_URI = "sqlite:///:memory:" # 内存数据库
class ProductionConfig(BaseConfig):
"""生成环境配置"""
SQLALCHEMY_DATABASE_URI = "mysql://root:root@127.0.0.1:3306/pear_admin"
SQLALCHEMY_TRACK_MODIFICATIONS = False
config = {"dev": DevelopmentConfig, "test": TestingConfig, "prod": ProductionConfig}
-14
View File
@@ -1,14 +0,0 @@
version: '3'
services:
flask:
build: .
ports:
- "8000:8000"
restart: always
links:
- mysql:dbserver
mysql:
image: "mysql:5.7"
environment:
- MYSQL_ROOT_PASSWORD=root
restart: always
BIN
View File
Binary file not shown.

Before

Width:  |  Height:  |  Size: 58 KiB

BIN
View File
Binary file not shown.

Before

Width:  |  Height:  |  Size: 44 KiB

BIN
View File
Binary file not shown.

Before

Width:  |  Height:  |  Size: 89 KiB

BIN
View File
Binary file not shown.

Before

Width:  |  Height:  |  Size: 130 KiB

BIN
View File
Binary file not shown.

Before

Width:  |  Height:  |  Size: 63 KiB

BIN
View File
Binary file not shown.

Before

Width:  |  Height:  |  Size: 86 KiB

-19
View File
@@ -1,19 +0,0 @@
import os
import multiprocessing
bind = '0.0.0.0:8000'
backlog = 512
chdir = os.path.dirname(os.path.abspath(__file__))
timeout = 30
worker_class = 'sync'
workers = multiprocessing.cpu_count() * 2 + 1
threads = 2
loglevel = 'info'
access_log_format = '%(t)s %(p)s %(h)s "%(r)s" %(s)s %(L)s %(b)s %(f)s" "%(a)s"'
if not os.path.exists('logs'):
os.mkdir('logs')
accesslog = os.path.join(chdir, "logs/gunicorn_access.log")
errorlog = os.path.join(chdir, "logs/gunicorn_error.log")
-1
View File
@@ -1 +0,0 @@
Generic single-database configuration.
-45
View File
@@ -1,45 +0,0 @@
# A generic, single database configuration.
[alembic]
# template used to generate migration files
# file_template = %%(rev)s_%%(slug)s
# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false
# Logging configuration
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARN
handlers = console
qualname =
[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S
-96
View File
@@ -1,96 +0,0 @@
from __future__ import with_statement
import logging
from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from flask import current_app
from alembic import context
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
# Interpret the configs file for Python logging.
# This line sets up loggers basically.
fileConfig(config.config_file_name)
logger = logging.getLogger('alembic.env')
# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
config.set_main_option(
'sqlalchemy.url',
str(current_app.extensions['migrate'].db.engine.url).replace('%', '%%'))
target_metadata = current_app.extensions['migrate'].db.metadata
# other values from the configs, defined by the needs of env.py,
# can be acquired:
# my_important_option = configs.get_main_option("my_important_option")
# ... etc.
def run_migrations_offline():
"""Run migrations in 'offline' mode.
This configures the context with just a URL
and not an Engine, though an Engine is acceptable
here as well. By skipping the Engine creation
we don't even need a DBAPI to be available.
Calls to context.execute() here emit the given string to the
script output.
"""
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url, target_metadata=target_metadata, literal_binds=True
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online():
"""Run migrations in 'online' mode.
In this scenario we need to create an Engine
and associate a connection with the context.
"""
# this callback is used to prevent an auto-migration from being generated
# when there are no changes to the schema
# reference: http://alembic.zzzcomputing.com/en/latest/cookbook.html
def process_revision_directives(context, revision, directives):
if getattr(config.cmd_opts, 'autogenerate', False):
script = directives[0]
if script.upgrade_ops.is_empty():
directives[:] = []
logger.info('No changes in schema detected.')
connectable = engine_from_config(
config.get_section(config.config_ini_section),
prefix='sqlalchemy.',
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(
connection=connection,
target_metadata=target_metadata,
process_revision_directives=process_revision_directives,
**current_app.extensions['migrate'].configure_args
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()
-24
View File
@@ -1,24 +0,0 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
# revision identifiers, used by Alembic.
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}
def upgrade():
${upgrades if upgrades else "pass"}
def downgrade():
${downgrades if downgrades else "pass"}
-28
View File
@@ -1,28 +0,0 @@
"""empty message
Revision ID: 7634e028e338
Revises: 8b664608a7c7
Create Date: 2021-06-01 16:43:50.853692
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = '7634e028e338'
down_revision = '8b664608a7c7'
branch_labels = None
depends_on = None
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('admin_user', sa.Column('dept_id', sa.Integer(), nullable=True, comment='部门id'))
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_column('admin_user', 'dept_id')
# ### end Alembic commands ###
-42
View File
@@ -1,42 +0,0 @@
"""empty message
Revision ID: 8b664608a7c7
Revises: ec21e19825ff
Create Date: 2021-06-01 14:37:20.327189
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = '8b664608a7c7'
down_revision = 'ec21e19825ff'
branch_labels = None
depends_on = None
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('admin_dept',
sa.Column('id', sa.Integer(), nullable=False, comment='部门ID'),
sa.Column('parent_id', sa.Integer(), nullable=True, comment='父级编号'),
sa.Column('dept_name', sa.String(length=50), nullable=True, comment='部门名称'),
sa.Column('sort', sa.Integer(), nullable=True, comment='排序'),
sa.Column('leader', sa.String(length=50), nullable=True, comment='负责人'),
sa.Column('phone', sa.String(length=20), nullable=True, comment='联系方式'),
sa.Column('email', sa.String(length=50), nullable=True, comment='邮箱'),
sa.Column('status', sa.Integer(), nullable=True, comment='状态(1开启,0关闭)'),
sa.Column('remark', sa.Text(), nullable=True, comment='备注'),
sa.Column('address', sa.String(length=255), nullable=True, comment='详细地址'),
sa.Column('create_at', sa.DateTime(), nullable=True, comment='创建时间'),
sa.Column('update_at', sa.DateTime(), nullable=True, comment='创建时间'),
sa.PrimaryKeyConstraint('id')
)
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table('admin_dept')
# ### end Alembic commands ###
-133
View File
@@ -1,133 +0,0 @@
"""empty message
Revision ID: ec21e19825ff
Revises:
Create Date: 2021-04-27 11:54:00.453274
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = 'ec21e19825ff'
down_revision = None
branch_labels = None
depends_on = None
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('admin_admin_log',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('method', sa.String(length=10), nullable=True),
sa.Column('uid', sa.Integer(), nullable=True),
sa.Column('url', sa.String(length=255), nullable=True),
sa.Column('desc', sa.Text(), nullable=True),
sa.Column('ip', sa.String(length=255), nullable=True),
sa.Column('success', sa.Integer(), nullable=True),
sa.Column('user_agent', sa.Text(), nullable=True),
sa.Column('create_time', sa.DateTime(), nullable=True),
sa.PrimaryKeyConstraint('id')
)
op.create_table('admin_dict_data',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('data_label', sa.String(length=255), nullable=True, comment='字典类型名称'),
sa.Column('data_value', sa.String(length=255), nullable=True, comment='字典类型标识'),
sa.Column('type_code', sa.String(length=255), nullable=True, comment='字典类型描述'),
sa.Column('is_default', sa.Integer(), nullable=True, comment='是否默认'),
sa.Column('enable', sa.Integer(), nullable=True, comment='是否开启'),
sa.Column('remark', sa.String(length=255), nullable=True, comment='备注'),
sa.Column('create_time', sa.DateTime(), nullable=True, comment='创建时间'),
sa.Column('update_time', sa.DateTime(), nullable=True, comment='更新时间'),
sa.PrimaryKeyConstraint('id')
)
op.create_table('admin_dict_type',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('type_name', sa.String(length=255), nullable=True, comment='字典类型名称'),
sa.Column('type_code', sa.String(length=255), nullable=True, comment='字典类型标识'),
sa.Column('description', sa.String(length=255), nullable=True, comment='字典类型描述'),
sa.Column('enable', sa.Integer(), nullable=True, comment='是否开启'),
sa.Column('create_time', sa.DateTime(), nullable=True, comment='创建时间'),
sa.Column('update_time', sa.DateTime(), nullable=True, comment='更新时间'),
sa.PrimaryKeyConstraint('id')
)
op.create_table('admin_photo',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(length=255), nullable=False),
sa.Column('href', sa.String(length=255), nullable=True),
sa.Column('mime', sa.CHAR(length=50), nullable=False),
sa.Column('size', sa.CHAR(length=30), nullable=False),
sa.Column('create_time', sa.DateTime(), nullable=True),
sa.PrimaryKeyConstraint('id')
)
op.create_table('admin_power',
sa.Column('id', sa.Integer(), nullable=False, comment='权限编号'),
sa.Column('name', sa.String(length=255), nullable=True, comment='权限名称'),
sa.Column('type', sa.String(length=1), nullable=True, comment='权限类型'),
sa.Column('code', sa.String(length=30), nullable=True, comment='权限标识'),
sa.Column('url', sa.String(length=255), nullable=True, comment='权限路径'),
sa.Column('open_type', sa.String(length=10), nullable=True, comment='打开方式'),
sa.Column('parent_id', sa.Integer(), nullable=True, comment='父类编号'),
sa.Column('icon', sa.String(length=128), nullable=True, comment='图标'),
sa.Column('sort', sa.Integer(), nullable=True, comment='排序'),
sa.Column('create_time', sa.DateTime(), nullable=True, comment='创建时间'),
sa.Column('update_time', sa.DateTime(), nullable=True, comment='更新时间'),
sa.Column('enable', sa.Integer(), nullable=True, comment='是否开启'),
sa.PrimaryKeyConstraint('id')
)
op.create_table('admin_role',
sa.Column('id', sa.Integer(), nullable=False, comment='角色ID'),
sa.Column('name', sa.String(length=255), nullable=True, comment='角色名称'),
sa.Column('code', sa.String(length=255), nullable=True, comment='角色标识'),
sa.Column('enable', sa.Integer(), nullable=True, comment='是否启用'),
sa.Column('remark', sa.String(length=255), nullable=True, comment='备注'),
sa.Column('details', sa.String(length=255), nullable=True, comment='详情'),
sa.Column('sort', sa.Integer(), nullable=True, comment='排序'),
sa.Column('create_time', sa.DateTime(), nullable=True, comment='创建时间'),
sa.Column('update_time', sa.DateTime(), nullable=True, comment='更新时间'),
sa.PrimaryKeyConstraint('id')
)
op.create_table('admin_user',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False, comment='用户ID'),
sa.Column('username', sa.String(length=20), nullable=True, comment='用户名'),
sa.Column('realname', sa.String(length=20), nullable=True, comment='真实名字'),
sa.Column('avatar', sa.String(length=255), nullable=True, comment='头像'),
sa.Column('remark', sa.String(length=255), nullable=True, comment='备注'),
sa.Column('password_hash', sa.String(length=128), nullable=True, comment='哈希密码'),
sa.Column('enable', sa.Integer(), nullable=True, comment='启用'),
sa.Column('create_at', sa.DateTime(), nullable=True, comment='创建时间'),
sa.Column('update_at', sa.DateTime(), nullable=True, comment='创建时间'),
sa.PrimaryKeyConstraint('id')
)
op.create_table('admin_role_power',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False, comment='标识'),
sa.Column('power_id', sa.Integer(), nullable=True, comment='用户编号'),
sa.Column('role_id', sa.Integer(), nullable=True, comment='角色编号'),
sa.ForeignKeyConstraint(['power_id'], ['admin_power.id'], ),
sa.ForeignKeyConstraint(['role_id'], ['admin_role.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_table('admin_user_role',
sa.Column('id', sa.Integer(), autoincrement=True, nullable=False, comment='标识'),
sa.Column('user_id', sa.Integer(), nullable=True, comment='用户编号'),
sa.Column('role_id', sa.Integer(), nullable=True, comment='角色编号'),
sa.ForeignKeyConstraint(['role_id'], ['admin_role.id'], ),
sa.ForeignKeyConstraint(['user_id'], ['admin_user.id'], ),
sa.PrimaryKeyConstraint('id')
)
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table('admin_user_role')
op.drop_table('admin_role_power')
op.drop_table('admin_user')
op.drop_table('admin_role')
op.drop_table('admin_power')
op.drop_table('admin_photo')
op.drop_table('admin_dict_type')
op.drop_table('admin_dict_data')
op.drop_table('admin_admin_log')
# ### end Alembic commands ###
+32
View File
@@ -0,0 +1,32 @@
from flask import Flask, render_template
from configs import config
from pear_admin.apis import register_apis
from pear_admin.extensions import register_extensions
from pear_admin.orms import UserORM
from pear_admin.views import register_views
def create_app(config_name="dev"):
app = Flask("pear-admin-flask")
app.config.from_object(config[config_name])
register_extensions(app)
register_apis(app)
register_views(app)
# @app.errorhandler(403)
# def handle_404(e):
# return render_template("error/403.html")
#
# @app.errorhandler(404)
# def handle_403(e):
# return render_template("error/404.html")
#
# @app.errorhandler(500)
# def handle_500(e):
# return render_template("error/500.html")
return app
+19
View File
@@ -0,0 +1,19 @@
from flask import Blueprint, Flask
from .department import department_api
from .passport import passport_api
from .rights import rights_api
from .role import role_api
from .user import user_api
def register_apis(app: Flask):
apis = Blueprint("api", __name__, url_prefix="/api/v1")
apis.register_blueprint(passport_api)
apis.register_blueprint(rights_api)
apis.register_blueprint(role_api)
apis.register_blueprint(department_api)
apis.register_blueprint(user_api)
app.register_blueprint(apis)
+68
View File
@@ -0,0 +1,68 @@
from flask import Blueprint, request
from flask_sqlalchemy.pagination import Pagination
from pear_admin.extensions import db
from pear_admin.orms import DepartmentORM
department_api = Blueprint("department", __name__)
@department_api.get("/department")
def department_list():
page = request.args.get("page", default=1, type=int)
per_page = request.args.get("limit", default=10, type=int)
_type = request.args.get("type")
ret = []
if _type == "tree":
dept_obj: DepartmentORM = DepartmentORM.query.get(1)
for child in dept_obj.children:
child_data = child.json()
child_data["children"] = []
if child.children:
child_data["isParent"] = True
for son in child.children:
son_data = son.json()
# son_data["isParent"] = True
child_data["children"].append(son_data)
ret.append(child_data)
return {"code": 0, "message": "请求权限数据成功", "data": ret}
q = db.select(DepartmentORM)
pages: Pagination = db.paginate(q, page=page, per_page=per_page)
return {
"code": 0,
"msg": "获取部门数据成功",
"data": [item.json() for item in pages.items],
"count": pages.total,
}
@department_api.post("/department")
def create_department():
data = request.get_json()
if data["id"]:
del data["id"]
department = DepartmentORM(**data)
department.save()
return {"code": 0, "msg": "新增部门成功"}
@department_api.put("/department/<int:rid>")
def change_department(rid):
data = request.get_json()
del data["id"]
department_obj = DepartmentORM.query.get(rid)
for key, value in data.items():
setattr(department_obj, key, value)
department_obj.save()
return {"code": 0, "msg": "修改部门成功"}
@department_api.delete("/department/<int:rid>")
def del_department(rid):
department_obj = DepartmentORM.query.get(rid)
department_obj.delete()
return {"code": 0, "msg": "删除删除成功"}
+79
View File
@@ -0,0 +1,79 @@
from collections import OrderedDict
from copy import deepcopy
from flask import Blueprint, make_response, request
from flask_jwt_extended import (
create_access_token,
create_refresh_token,
get_current_user,
jwt_required,
)
from pear_admin.extensions import db
from pear_admin.orms import RightsORM, RoleORM, UserORM
passport_api = Blueprint("passport", __name__)
@passport_api.post("/login")
def login_in():
data = request.get_json()
user: UserORM = db.session.execute(
db.select(UserORM).where(UserORM.username == data["username"])
).scalar()
if not user:
return {"message": "用户不存在", "code": -1}, 401
if not user.check_password(data["password"]):
return {"message": "用户密码错误", "code": -1}, 401
access_token = create_access_token(user)
refresh_token = create_refresh_token(user)
response = make_response(
{
"code": 0,
"msg": "登录成功",
"access_token": access_token,
"refresh_token": refresh_token,
}
)
return response
@passport_api.route("/logout", methods=["GET", "POST"])
@jwt_required()
def logout():
return {"msg": "退出登录成功", "code": 0}
@passport_api.get("/menu")
@jwt_required()
def menus_api():
rights_orm_list = set()
current_user: UserORM = get_current_user()
for role in current_user.role_list:
for rights_orm in role.rights_list:
if rights_orm.type != "auth":
rights_orm_list.add(rights_orm)
rights_list = [rights_orm.menu_json() for rights_orm in rights_orm_list]
rights_list.sort(key=lambda x: (x["pid"], x["id"]), reverse=True)
menu_dict_list = OrderedDict()
for menu_dict in rights_list:
if menu_dict["id"] in menu_dict_list.keys(): # 如果当前节点已经存在与字典中
# 当前节点添加子节点
menu_dict["children"] = deepcopy(menu_dict_list[menu_dict["id"]])
menu_dict["children"].sort(key=lambda item: item["sort"])
# 删除子节点
del menu_dict_list[menu_dict["id"]]
if menu_dict["pid"] not in menu_dict_list:
menu_dict_list[menu_dict["pid"]] = [menu_dict]
else:
menu_dict_list[menu_dict["pid"]].append(menu_dict)
return sorted(menu_dict_list.get(0), key=lambda item: item["sort"])
+101
View File
@@ -0,0 +1,101 @@
from copy import deepcopy
from flask import Blueprint, request
from flask_jwt_extended import jwt_required
from pear_admin.extensions import db
from pear_admin.orms import RightsORM
rights_api = Blueprint("rights", __name__)
@rights_api.get("/rights")
@jwt_required()
def rights_list():
t = request.args.get("type", default="")
if t == "tree": # tree 组件数据,用于权限赋值
# 1. 获取所有的权限
rights_all = db.session.execute(db.select(RightsORM)).scalars()
rights_list = [
{"id": r.id, "pid": r.pid, "title": r.name, "sort": r.sort}
for r in rights_all
]
# 2. 获取已有权限 id 集合
# 3. 列表转属性组件
rights_list.sort(key=lambda item: (item["pid"], item["id"]), reverse=True)
tree_dict = {}
for rights_dict in rights_list: # 遍历子节点
# 2. 如果当前阶段已经存在于树状表格字典,则是父节点
if rights_dict["id"] in tree_dict.keys():
# 将之前的节点添加到父节点之下
rights_dict["children"] = deepcopy(tree_dict[rights_dict["id"]])
rights_dict["children"].sort(key=lambda item: item["sort"])
del tree_dict[rights_dict["id"]]
# 1. 如果父节点未出现在树状字典里面,就新增子节点列表,否则就追加
if rights_dict["pid"] not in tree_dict.keys():
tree_dict[rights_dict["pid"]] = [rights_dict]
else:
tree_dict[rights_dict["pid"]].append(rights_dict)
return {"code": 0, "data": tree_dict.get(0)}
if t == "treetable": # tree table 数据
page = request.args.get("page", type=int, default=1)
per_page = request.args.get("per_page", type=int, default=10)
q = db.select(RightsORM).where(RightsORM.type == "menu")
pages = db.paginate(q, page=page, per_page=per_page, error_out=False)
ret = []
# 构建树状表格的数据
for page in pages.items:
data = page.json()
data["children"] = []
for child in page.children:
child_data = child.json()
if child.children:
child_data["children"] = [
sub_child.json() for sub_child in child.children
]
child_data["isParent"] = True
data["children"].append(child_data)
data["isParent"] = True
ret.append(data)
return {"code": 0, "msg": "请求权限数据成功", "count": pages.total, "data": ret}
return {"code": 0, "msg": "请求权限数据成功,"}
@rights_api.post("/rights")
@jwt_required()
def create_rights():
data = request.get_json()
if not data["pid"]:
data["pid"] = 0
if not data["sort"]:
data["sort"] = int(data["sort"])
rights_obj = RightsORM(**data)
rights_obj.save()
return {"code": 0, "msg": "新增权限数据成功"}
@rights_api.put("/rights/<int:rid>")
@jwt_required()
def change_rights(rid):
data = request.get_json()
del data["id"]
rights_obj = RightsORM.query.get(rid)
for key, value in data.items():
setattr(rights_obj, key, value)
rights_obj.save()
return {"code": 0, "msg": "修改权限数据成功"}
@rights_api.delete("/rights/<int:rid>")
@jwt_required()
def delete_rights(rid):
rights_obj = RightsORM.query.get(rid)
rights_obj.delete()
return {"code": 0, "msg": "删除权限数据成功"}
+82
View File
@@ -0,0 +1,82 @@
from flask import Blueprint, request
from flask_sqlalchemy.pagination import Pagination
from pear_admin.extensions import db
from pear_admin.orms import RightsORM, RoleORM
role_api = Blueprint("role", __name__)
@role_api.get("/role")
def role_list():
page = request.args.get("page", default=1, type=int)
per_page = request.args.get("limit", default=10, type=int)
q = db.select(RoleORM)
pages: Pagination = db.paginate(q, page=page, per_page=per_page)
return {
"code": 0,
"msg": "获取角色数据成功",
"data": [item.json() for item in pages.items],
"count": pages.total,
}
@role_api.post("/role")
def create_role():
data = request.get_json()
if data["id"]:
del data["id"]
role = RoleORM(**data)
role.save()
return {"code": 0, "msg": "新增角色成功"}
@role_api.put("/role/<int:rid>")
def change_role(rid):
data = request.get_json()
del data["id"]
role_obj = RoleORM.query.get(rid)
for key, value in data.items():
setattr(role_obj, key, value)
role_obj.save()
return {"code": 0, "msg": "修改角色权限成功"}
@role_api.delete("/role/<int:rid>")
def del_role(rid):
role_obj = RoleORM.query.get(rid)
role_obj.delete()
return {"code": 0, "msg": "删除角色成功"}
@role_api.get("/role/role_rights/<int:rid>")
def role_rights(rid):
role: RoleORM = db.session.execute(
db.select(RoleORM).where(RoleORM.id == rid)
).scalar()
own_rights_list = [r.id for r in role.rights_list]
return {
"code": 0,
"msg": "返回角色权限数据成功",
"data": own_rights_list,
}
@role_api.put("/role/role_rights/<int:rid>")
def change_role_rights(rid):
rights_ids = request.json.get("rights_ids", "")
rights_list = rights_ids.split(",")
role: RoleORM = db.session.execute(
db.select(RoleORM).where(RoleORM.id == rid)
).scalar()
rights_obj_list = db.session.execute(
db.select(RightsORM).where(RightsORM.id.in_(rights_list))
).all()
role.rights_list = [r[0] for r in rights_obj_list]
role.save()
return {"code": 0, "msg": "授权成功"}
+102
View File
@@ -0,0 +1,102 @@
from datetime import datetime
from flask import Blueprint, request
from flask_jwt_extended import current_user, jwt_required
from flask_sqlalchemy.pagination import Pagination
from pear_admin.extensions import db
from pear_admin.orms import RoleORM, UserORM
user_api = Blueprint("user", __name__)
@user_api.get("/user")
def user_list():
page = request.args.get("page", default=1, type=int)
per_page = request.args.get("limit", default=10, type=int)
q = db.select(UserORM)
pages: Pagination = db.paginate(q, page=page, per_page=per_page)
return {
"code": 0,
"msg": "获取用户数据成功",
"data": [item.json() for item in pages.items],
"count": pages.total,
}
@user_api.post("/user")
def create_user():
data = request.get_json()
if data["id"]:
del data["id"]
role = UserORM(**data)
create_at = data["create_at"]
if create_at:
role.create_at = datetime.strptime(create_at, "%Y-%m-%d %H:%M:%S")
role.password = "123456"
role.save()
return {"code": 0, "msg": "新增用户成功"}
@user_api.put("/user/<int:uid>")
def change_user(uid):
data = request.get_json()
del data["id"]
user_obj = UserORM.query.get(uid)
for key, value in data.items():
if key == "create_at":
value = datetime.strptime(value, "%Y-%m-%d %H:%M:%S")
setattr(user_obj, key, value)
user_obj.save()
return {"code": 0, "msg": "修改用户信息成功"}
@user_api.delete("/user/<int:rid>")
def del_user(rid):
user_obj = UserORM.query.get(rid)
user_obj.delete()
return {"code": 0, "msg": "删除用户成功"}
@user_api.get("/user/user_role/<int:uid>")
def get_user_role(uid):
user: UserORM = db.session.execute(
db.select(UserORM).where(UserORM.id == uid)
).scalar()
wn_role_list = [r.id for r in user.role_list]
return {
"code": 0,
"msg": "返回角色权限数据成功",
"data": wn_role_list,
}
@user_api.put("/user/user_role/<int:rid>")
def change_user_role(rid):
role_ids = request.json.get("rights_ids", "")
role_list = role_ids.split(",")
user: UserORM = db.session.execute(
db.select(UserORM).where(UserORM.id == rid)
).scalar()
role_obj_list = db.session.execute(
db.select(RoleORM).where(RoleORM.id.in_(role_list))
).all()
user.role_list = [r[0] for r in role_obj_list]
user.save()
return {"code": 0, "msg": "授权成功"}
@user_api.get("/user/profile")
@jwt_required()
def user_profile():
return {
"code": 0,
"msg": "获取个人数据成功",
"data": current_user.json(),
}
+13
View File
@@ -0,0 +1,13 @@
from flask import Flask
from .init_db import db, migrate
from .init_jwt import jwt
from .init_script import register_script
def register_extensions(app: Flask):
db.init_app(app)
migrate.init_app(app, db)
jwt.init_app(app)
register_script(app)

Some files were not shown because too many files have changed in this diff Show More