public
Last active

Cache State Backend

  • Download Gist
async.py
Python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29
# -*- coding: utf-8 -*-
from celery.task import task
 
 
@task
def generate(model, pk, attr):
instance = model._default_manager.get(pk=pk)
file = getattr(instance, attr)
file.generate(save=True)
return attr
 
 
class CeleryCacheStateBackend(object):
def is_invalid(self, file):
if not getattr(file, '_file', None):
# No file on object. Have to check storage.
return not file.storage.exists(file.name)
return False
 
def validate(self, file):
if self.is_invalid(file):
generate.delay(file.instance.__class__, file.instance.pk, file.attname)
 
def invalidate(self, file):
file.delete(save=False)
generate.delay(file.instance.__class__, file.instance.pk, file.attname) # pre_cache
 
def clear(self, file):
file.detele(save=False)
filename.py
Python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36
# -*- coding: utf-8 -*-
from django.db.models.signals import post_init
 
 
def _post_init_handler(sender, instance, **kwargs):
value = getattr(instance, _post_init_handler.field)
setattr(instance, "_%s_state" % _post_init_handler.field, value)
 
 
class FilenameCacheStateBackend(object):
def has_changed(self, file):
field_name = getattr(file.field, 'image_field', None)
return getattr(file.instance, "_%s_state" % field_name) != getattr(file.instance, field_name)
 
def is_invalid(self, file):
if not getattr(file, '_file', None):
# No file on object. Have to check storage.
return not file.storage.exists(file.name)
return False
 
def validate(self, file):
if self.is_invalid(file):
file.generate(save=True)
 
def invalidate(self, file):
if self.has_changed(file):
file.delete(save=False)
file.generate(save=True)
 
def clear(self, file):
file.delete(save=False)
 
def register_field(self, model_class, field, attname):
uid = "%s.%s.%s" % (model_class.__module__, model_class.__name__, field.image_field)
_post_init_handler.field = field.image_field
post_init.connect(_post_init_handler, sender=model_class, dispatch_uid="%s_init" % uid)

Please sign in to comment on this gist.

Something went wrong with that request. Please try again.