Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Simple Djano cache system for function/class-based views with automatic key generation
#!/usr/bin/env python
# coding: utf-8
from asset.models import *
from common.models import *
from cost.models import *
from django.core.cache import cache
import inspect
from django.contrib.contenttypes.models import ContentType
from django.db.models.signals import pre_save, pre_delete
KeyRegistry = "KeyRegistry"
def generate_key(seperator, *args):
"""
:param seperator: key中的分隔符, 比如'-', '.'
:param args: 需要添加到key中的参数
:return: 生成的key
这个函数用来生成缓存的key, 有两种模式
1. class based view
会读取class name和method name
2. function view
会读取function name
最后用传入的seperator把所有东西, 包括args连接起来, 作为key返回
"""
key = []
frame_back = inspect.currentframe().f_back
func_name = frame_back.f_code.co_name
key.append(func_name)
if 'self' in frame_back.f_locals:
class_name = frame_back.f_locals['self'].__class__.__name__
key.insert(0, class_name)
key = seperator.join(key + [str(arg) for arg in args])
return key
def get_cachevalue(key):
"""
:param key: 缓存key值
:return: cached value
从缓存中获取结果, 结果可能为空
"""
cached_value = cache.get(key)
return cached_value
def set_cachevalue(key, value, models):
"""
:param key: 缓存key值
:param value: 和key对应的value值
:param models: model列表, 字符串类型, e.g. ["Switch", "Server"]
:return:
"""
if (isinstance(models, list) and all(isinstance(m, str) for m in models)):
try:
for modelname in models:
model_cls = ContentType.objects.get(model=modelname).model_class()
if KeyRegistry in dir(model_cls):
model_cls.KeyRegistry.append(key)
else:
model_cls.KeyRegistry = [key]
except KeyError:
print("必须使用Model定义时的名字")
cache.set(key, value)
else:
print("models必须是字符串构成的列表!")
def _clear_cache_when_save(sender, **kwargs):
'''
逻辑是, 只要某个Model的数据发生改变, 那么缓存中相关联的所有key-value都删除
因为没有办法在Model中知晓views中的逻辑, 无法【更新】缓存, 只能删除
'''
if KeyRegistry not in sender.__dict__:
pass
else:
for key in sender.__dict__[KeyRegistry]:
cache.delete(key)
def _clear_cache_when_delete(sender, **kwargs):
if KeyRegistry not in sender.__dict__:
pass
else:
for key in sender.__dict__[KeyRegistry]:
cache.delete(key)
pre_save.connect(_clear_cache_when_save)
pre_delete.connect(_clear_cache_when_delete)
@laike9m

This comment has been minimized.

Copy link
Owner Author

commented Aug 2, 2014

How to use it, an example (which is using djangorestframwork but you get the idea):

@api_view(['GET'])
def cost_month_detail(request):
    """  
    获取某个成本某月的详细采购记录
    """
    costs = Cost.objects.all()
    item = request.QUERY_PARAMS.get('item', 'undefined')
    cost_type = COST_TYPE_KEY_MAP.get(item, COST_TYPE_UNDEFINED)
    month = request.QUERY_PARAMS.get('month', None)
    key = generate_key('.' , *[item, month])
    data = get_cachevalue(key)
    if not data:
        req_ym_str = month.split('.')
        req_ym_int = [int(req_ym_str[0]), int(req_ym_str[1])]

        if cost_type and req_ym_int:
            costs = costs.filter( cost_type=cost_type,\
                    purchase_date__year=req_ym_int[0],\
                    purchase_date__month=req_ym_int[1])
        serializer = CostSerializer(costs)
        data = serializer.data
        set_cachevalue(key, data, [models[item]])
    return Response(data)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.