在启动Django框架的时候需要在数据库中生成初始化数据，可以使用post_migrate信号实现初始化数据库操作
python3 manage.py migrate –调用migrate之后，django会发起post_migrate信号，数据库基础数据的初始化可以写在里面具体实现
在app目录的apps.py中增加如下
import os
from django.apps import AppConfig
from django.db.models.signals import post_migrate
from django.conf import settings
import json
from datetime import datetime
def init_script(sender, **kwargs):
from deploy_management.models import Script
db_scripts = [s.name for s in Script.objects.all()]
script_list = []
with open(os.path.join(settings.CONFIG_ROOT, 'scripts.json'), 'r', encoding='utf-8') as f:
j = json.load(f)
for item in j:
if item.get('name') not in db_scripts:
item['create_time'] = datetime.now()
item['create_user'] = 'admin'
script_list.append(Script(**item))
Script.objects.bulk_create(script_list)
class DeployManagementConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'deploy_management'
def ready(self):
post_migrate.connect(init_script, sender=self)
scripts.json中是要插入的json数据
信号简介简单理解：对每条执行的代码的前后都留了位置自定义操作，如：进行数据存储的前后可以执行我们自定义的操作等等…
Django内置信号 Model signalspre_init # django的modal执行其构造方法前，自动触发 post_init # django的modal执行其构造方法后，自动触发 pre_save # django的modal对象保存前，自动触发 post_save # django的modal对象保存后，自动触发 pre_delete # django的modal对象删除前，自动触发 post_delete # django的modal对象删除后，自动触发 m2m_changed # django的modal中使用m2m字段操作第三张表（add,remove,clear）前后，自动触发 class_prepared # 程序启动时，检测已注册的app中modal类，对于每一个类，自动触发Management signals
pre_migrate # 执行migrate命令前，自动触发 post_migrate # 执行migrate命令后，自动触发Request/response signals
request_started # 请求到来前，自动触发 request_finished # 请求结束后，自动触发 got_request_exception # 请求异常后，自动触发Test signals
setting_changed # 使用test测试修改配置文件时，自动触发 template_rendered # 使用test测试渲染模板时，自动触发Database Wrappers
connection_created # 创建数据库连接时，自动触发使用多个信号
apps.py
from django.apps import AppConfig
class RbacConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'rbac'
def ready(self):
import rbac.signals
signals.py
# -*- coding: utf-8 -*-
"""
@author : Ma
@software : PyCharm
@file : users_view.py
@create : 2021/8/27 17:55
"""
import json
import logging
from django.db.models.signals import pre_save, pre_delete, post_save, post_migrate
from django.dispatch import receiver
from django_redis import get_redis_connection
from django.contrib.auth.models import Permission, User
@receiver(post_migrate)
def createSuperUser(sender, **kwargs):
if not User.objects.filter(username='admin', is_superuser=True):
user = User(
username='admin',
email='',
first_name='',
last_name='',
)
user.set_password('admin')
user.is_superuser = True
user.is_staff = True
user.save()
@receiver(pre_save, sender=Permission)
def update_permissions_to_redis(sender, instance, **kwargs):
"""
Permissions模型,更新时更新redis
:param sender:
:param instance:
:param kwargs:
:return:
"""
conn = get_redis_connection('user_info')
if instance.id:
if not instance.menu:
# 接口权限,判断权限path的变化,更新redis
try:
permission = Permission.objects.get(id=instance.id)
except Permission.DoesNotExist:
return
if permission.path != instance.path:
# 路径更改,删除原有记录并新增一条权限记录
if conn.hexists('user_permissions_manage', permission.path):
permissions = json.loads(conn.hget('user_permissions_manage', permission.path))
for index, value in enumerate(permissions):
if value.get('id') == instance.id:
del permissions[index]
if permissions:
conn.hset('user_permissions_manage', permission.path, json.dumps(permissions))
else:
conn.hdel('user_permissions_manage', permission.path)
if conn.hexists('user_permissions_manage', instance.path):
# 如存在路径记录, 添加
permissions = json.loads(conn.hget('user_permissions_manage', instance.path))
permissions.append({
'method': instance.method,
'sign': instance.sign,
'id': instance.id,
})
conn.hset('user_permissions_manage', instance.path, json.dumps(permissions))
else:
# 否则新增
conn.hset('user_permissions_manage', instance.path, json.dumps([{
'method': instance.method,
'sign': instance.sign,
'id': instance.id,
}]))
else:
# 路径未变,更改原有权限记录
if permission.method != instance.method or permission.sign != instance.sign:
permissions = json.loads(conn.hget('user_permissions_manage', instance.path))
for permission in permissions:
if permission.get('id') == instance.id:
permission['method'] = instance.method
permission['sign'] = instance.sign
conn.hset('user_permissions_manage', instance.path, json.dumps(permissions))
else:
# 菜单权限,判断是否由接口权限改为菜单权限,如果是则删除原有记录
try:
permission = Permission.objects.get(id=instance.id)
except Permission.DoesNotExist:
return
if not permission.menu and conn.hexists('user_permissions_manage', permission.path):
permissions = json.loads(conn.hget('user_permissions_manage', permission.path))
for index, value in enumerate(permissions):
if value.get('id') == instance.id:
del permissions[index]
if permissions:
conn.hset('user_permissions_manage', permission.path, json.dumps(permissions))
else:
conn.hdel('user_permissions_manage', permission.path)
@receiver(post_save, sender=Permission)
def create_permissions_to_redis(sender, instance, **kwargs):
"""
Permissions模型,创建时更新redis
:param sender:
:param instance:
:param kwargs:
:return:
"""
if not instance.menu and kwargs.get('created'):
conn = get_redis_connection('user_info')
if conn.exists('user_permissions_manage') and conn.hexists('user_permissions_manage', instance.path):
permissions = json.loads(conn.hget('user_permissions_manage', instance.path))
permissions.append({
'method': instance.method,
'sign': instance.sign,
'id': instance.id,
})
conn.hset('user_permissions_manage', instance.path, json.dumps(permissions))
else:
conn.hset('user_permissions_manage', instance.path, json.dumps([{
'method': instance.method,
'sign': instance.sign,
'id': instance.id,
}]))
@receiver(pre_delete, sender=Permission)
def delete_permissions_from_redis(sender, instance, **kwargs):
"""
Permissions模型,删除时更新redis
:param sender:
:param instance:
:param kwargs:
:return:
"""
if not instance.menu:
conn = get_redis_connection('user_info')
if conn.exists('user_permissions_manage') and conn.hexists('user_permissions_manage', instance.path):
permissions = json.loads(conn.hget('user_permissions_manage', instance.path))
for index, permission in enumerate(permissions):
if permission.get('id') == instance.id:
del permissions[index]
if permissions:
conn.hset('user_permissions_manage', instance.path, json.dumps(permissions))
else:
conn.hdel('user_permissions_manage', instance.path)
