Cleanup stale ContentType content, Kuma

cleanup_stale_contenttypes.py

# Run in a shell_plus session

def scan_old_cts(dry_run=True):
  for ct in ContentType.objects.order_by('id'):
    if ct.model_class() is None:
      ct_relations(ct, dry_run)

def ct_relations(ct, dry_run):
  relations = [
    ('KeyAction', 'keyaction_set'),
    ('LogEntry', 'logentry_set'),
    ('OutdatedObject', 'outdatedobject_set'),
    ('Permission', 'permission_set'),
    ('TaggedItem', 'taggit_taggeditem_tagged_items'),
    ('Watch', 'watch_set'),
  ]
  has_stale = {}
  for model, rel in relations:
    count = getattr(ct, rel).count()
    if count:
      has_stale[model] = count
  if has_stale:
    print('ContentType %d "%s" has relations: %s' % (ct.id, ct, has_stale))
  else:
    print('ContentType %d "%s" has no relations, deleting.' % (ct.id, ct))
    if not dry_run:
      ct.delete()

def drop_old_permissions():
  for perm in Permission.objects.order_by('id'):
    if perm.content_type.model_class() is None:
      group_count = perm.group_set.count()
      user_count = perm.user_set.count()
      if group_count == 0 and user_count == 0:
        print('Dropping perm %d "%s"' % (perm.id, perm))
        perm.delete()
      else:
        print('perm %d "%s" needs cleanup: groups %d users %d' % (perm.id, perm, group_count, user_count))

def migrate_attachment_perms():
  perms = [
     ('add_attachment', 'attachment'),
     ('change_attachment', 'attachment'),
     ('delete_attachment', 'attachment'),
     ('add_attachmentrevision', 'attachmentrevision'),
     ('change_attachmentrevision', 'attachmentrevision'),
     ('delete_attachmentrevision', 'attachmentrevision'),
  ]
  for codename, modelname in perms:
    try:
      old_perm = Permission.objects.get(codename=codename, content_type__app_label='wiki', content_type__model=modelname)
      new_perm = Permission.objects.get(codename=codename, content_type__app_label='attachments', content_type__model=modelname)
    except Permission.DoesNotExist:
      pass
    else:
      for group in old_perm.group_set.all():
        group.permissions.add(new_perm)
        group.permissions.remove(old_perm)
      for user in old_perm.user_set.all():
        user.user_permissions.add(new_perm)
        user.user_permissions.remove(old_perm)

def drop_old_tags():
  tags_content_type_id = TaggedItem.objects.values_list('content_type_id', flat=True).distinct()
  for ct_id in tags_content_type_id:
    ct = ContentType.objects.get(id=ct_id)
    if ct.model_class() is None:
      TaggedItem.objects.filter(content_type_id=ct_id).delete()

def drop_old_logentries():
  ct_ids = LogEntry.objects.values_list('content_type_id', flat=True).distinct()
  for ct_id in ct_ids:
    try:
      ct = ContentType.objects.get(id=ct_id)
    except ContentType.DoesNotExist:
      LogEntry.objects.filter(content_type_id=ct_id).delete()
    else:
      if ct.model_class() is None:
        LogEntry.objects.filter(content_type_id=ct_id).delete()

scan_old_cts()
# migrate_attachment_perms()
# drop_old_permissions()
# drop_old_tags()
# drop_old_logentries()
# scan_old_cts(dry_run=False)