在 Django 中，如何在导入和验证期间对夹具数据运行自定义清理函数？答案

【问题标题】：In Django how can I run a custom clean function on fixture data during import and validation?在 Django 中，如何在导入和验证期间对夹具数据运行自定义清理函数？
【发布时间】：2012-07-26 20:39:09
【问题描述】：

在 ModelForm 中，我可以编写一个 clean_<field_name> 成员函数来自动验证和清理用户输入的数据，但是在 manage.py loaddata 期间我可以对脏 json 或 csv 文件（夹具）做些什么呢？

【问题讨论】：

标签： database django validation fixture

【解决方案1】：

假设加载了loaddata 的固件包含不需要验证的干净数据（通常作为先前dumpdata 的逆操作），所以简短的回答是，如果您不希望使用 loaddata 方法你需要清理你的输入。

但是，您可能可以在实现自定义数据清理代码时使用 loaddata 的一些基础——我相信您可以使用 Django serialization libs 轻松编写脚本来读取现有数据文件并保存通常在清理数据后生成的对象。

【讨论】：

是的，我的数据是用 dumpdata 生成的。模型和验证规则只是在加载数据之前发生了变化。就像你建议的那样，只会构建一个自定义的负载数据过滤器。似乎我是唯一一个使用他们的固定装置的 SO 用户。
啊，是的，那是一个棘手的情况。不过，听起来您确实走上了最佳路线。
不幸的是，我找不到 loaddata 的验证钩子，而且您不能只覆盖反序列化器以使 loaddata 以您的方式执行验证/clean_data，您必须复制所有 loaddata 命令并强制它使用你的解串器或只是在串行夹具领域工作以调整数据以适应模型......就像你建议的那样，但 simplejson 最适合我。

【解决方案2】：

如果其他人想做类似的事情，我定义了一个模型方法来进行清理（所以它可以从 ModelForms 中调用）

MAX_ZIPCODE_DIGITS = 9
MIN_ZIPCODE_DIGITS = 5

def clean_zip_code(self, s=None):
    #s = str(s or self.zip_code)
    if not s: return None
    s = re.sub("\D","",s)
    if len(s)>self.MAX_ZIPCODE_DIGITS: 
        s = s[:self.MAX_ZIPCODE_DIGITS]
    if len(s) in (self.MIN_ZIPCODE_DIGITS-1,self.MAX_ZIPCODE_DIGITS-1):
        s = '0'+s # FIXME: deal with other intermediate lengths
    if len(s)>=self.MAX_ZIPCODE_DIGITS:
        s = s[:self.MIN_ZIPCODE_DIGITS]+'-'+s[self.MIN_ZIPCODE_DIGITS:] 
    return s

然后编写了一个独立的 python 脚本，使用模型中找到的任何 clean_ 方法清理我的遗留 json 文件。

import os, json

def clean_json(app = 'XYZapp', model='Entity', fields='zip_code', cleaner_prefix='clean_'):
    # Set the DJANGO_SETTINGS_MODULE environment variable.
    os.environ['DJANGO_SETTINGS_MODULE'] = app+".settings"
    settings = __import__(app+'.settings').settings
    models   = __import__(app+'.models').models

    fpath    = os.path.join( settings.SITE_PROJECT_PATH, 'fixtures', model+'.json')
    if isinstance(fields,(str,unicode)):
        fields = [fields]
    Ns = []
    for field in fields:
        try:
            instance = getattr(models,model)()
        except AttributeError:
            print 'No model named %s could be found'%(model,)
            continue
        try:
            cleaner = getattr(instance, cleaner_prefix+field)
        except AttributeError:
            print 'No cleaner method named %s.%s could be found'%(model,cleaner_prefix+field)
            continue
        print 'Cleaning %s using %s.%s...'%(fpath,model,cleaner.__name__)
        fin = open(fpath,'r')
        if fin:
            l = json.load(fin)
            before = len(l)
            cleans = 0
            for i in range(len(l)):
                if 'fields' in l[i] and field in l[i]['fields']:
                    l[i]['fields'][field]=cleaner(l[i]['fields'][field]) # cleaner returns None to delete records
                    cleans += 1
            fin.close()
            after = len(l)
            assert after>.5*before
            Ns += [(before, after,cleans)]
            print 'Writing %d/%d (new/old) records after %d cleanups...'%Ns[-1]
            with open(fpath,'w') as fout:
                fout.write(json.dumps(l,indent=2,sort_keys=True))
    return Ns

if __name__ == '__main__':
    clean_json()

【讨论】：