在啟動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數據
簡單理解：對每條執行的代碼的前後都留了位置自定義操作，如：進行數據存儲的前後可以執行我們自定義的操作等等…
pre_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類，對於每一個類，自動觸發
pre_migrate # 執行migrate命令前，自動觸發
post_migrate # 執行migrate命令後，自動觸發
request_started # 請求到來前，自動觸發
request_finished # 請求結束後，自動觸發
got_request_exception # 請求異常後，自動觸發
setting_changed # 使用test測試修改配置文件時，自動觸發
template_rendered # 使用test測試渲染模板時，自動觸發
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)
