如何缓存Django/DRF中FileField的文件以避免重复读取

我之前在做类似的用户上传CSV数据的DRF项目时，也碰到过一模一样的性能问题——每次调用模型的列名查询、行数统计或者数据切片方法，都要重新从磁盘读取整个文件，高并发下服务器负载直接上去了！下面分享几个我亲测有效的缓存方案，从简单到进阶，你可以根据自己的场景选：

这个方案最适合同一个请求内多次调用模型方法的场景，比如序列化器里同时需要列名、行数和某段数据。我们可以在模型实例里加私有变量，把读取解析后的内容存起来，避免重复IO：

import csv
from django.db import models

class DataFile(models.Model):
    file = models.FileField(upload_to="data_files/")
    # 私有变量用于缓存，下划线开头表示内部使用
    _cached_content = None
    _cached_columns = None

    def get_columns(self):
        # 先检查缓存是否存在
        if self._cached_columns is not None:
            return self._cached_columns
        
        # 缓存不存在时读取文件解析
        with self.file.open('r', encoding='utf-8') as f:
            reader = csv.reader(f)
            self._cached_columns = next(reader)
        return self._cached_columns

    def get_row_count(self):
        if self._cached_content is None:
            with self.file.open('r', encoding='utf-8') as f:
                # 把整个文件内容转成列表缓存（注意大文件慎用）
                self._cached_content = list(csv.reader(f))
        # 减去表头行得到实际数据行数
        return len(self._cached_content) - 1

    def get_slice_data(self, column_idx, start=0, end=10):
        if self._cached_content is None:
            with self.file.open('r', encoding='utf-8') as f:
                self._cached_content = list(csv.reader(f))
        
        # 处理边界情况，避免索引越界
        max_data_row = len(self._cached_content) - 1
        end = min(end, max_data_row)
        # 跳过表头，取指定范围的列数据
        return [row[column_idx] for row in self._cached_content[start+1:end+1]]

优缺点：速度最快（内存直接读取），但缓存只在当前模型实例的生命周期内有效——如果重新查询实例或者重启服务，缓存就没了。适合单次请求内多次调用方法的场景，大文件不建议缓存整个内容，会占用过多内存。

如果需要跨请求、跨服务实例的持久化缓存，用Django自带的缓存框架就很方便。关键是要生成一个随文件更新而变化的缓存键，避免文件更新后还返回旧缓存：

import csv
from django.db import models
from django.core.cache import cache

class DataFile(models.Model):
    file = models.FileField(upload_to="data_files/")
    # 自动跟踪文件更新时间，用来生成缓存键
    updated_at = models.DateTimeField(auto_now=True)

    def _get_cache_key(self, suffix):
        # 结合实例ID、文件更新时间戳和后缀生成唯一键
        return f"data_file_{self.pk}_{int(self.updated_at.timestamp())}_{suffix}"

    def get_columns(self):
        cache_key = self._get_cache_key("columns")
        columns = cache.get(cache_key)
        if columns:
            return columns
        
        with self.file.open('r', encoding='utf-8') as f:
            reader = csv.reader(f)
            columns = next(reader)
            # 缓存1小时，可根据业务调整超时时间
            cache.set(cache_key, columns, timeout=3600)
        return columns

    def get_row_count(self):
        cache_key = self._get_cache_key("row_count")
        row_count = cache.get(cache_key)
        if row_count is not None:
            return row_count
        
        with self.file.open('r', encoding='utf-8') as f:
            # 用生成器统计行数，避免加载整个文件到内存
            row_count = sum(1 for _ in f) - 1
            cache.set(cache_key, row_count, timeout=3600)
        return row_count

注意点：

• 建议把Django的缓存后端换成Redis或者Memcached，默认的内存缓存（LocMemCache）不支持跨实例共享。
• updated_at字段会在每次保存模型时自动更新，所以当用户替换文件后，缓存键会变化，旧缓存自动失效，不会出现脏数据。
• 对于大文件，统计行数用生成器sum(1 for _ in f)比加载整个文件更省内存。

如果你的场景是文件上传后几乎不会修改，那直接把解析后的结果存到数据库里是最彻底的方案——完全不用再读取源文件：

import csv
from django.db import models

class DataFile(models.Model):
    file = models.FileField(upload_to="data_files/")
    # 用JSONField存列名，IntegerField存行数
    columns = models.JSONField(null=True, blank=True)
    row_count = models.IntegerField(null=True, blank=True)

    def save(self, *args, **kwargs):
        # 只有当文件存在且还没解析过时，才执行解析逻辑
        if self.file and not self.columns:
            with self.file.open('r', encoding='utf-8') as f:
                reader = csv.reader(f)
                self.columns = next(reader)
                # 统计行数（跳过表头）
                self.row_count = sum(1 for _ in f)
        super().save(*args, **kwargs)

    def get_columns(self):
        return self.columns or []

    def get_row_count(self):
        return self.row_count or 0

扩展优化：如果需要支持文件更新，可以在save方法里判断文件是否有变化，或者用post_save信号触发重新解析。如果常用特定切片数据，也可以把这些数据存到额外的JSONField里。

• 针对大文件：不要一次性缓存整个文件内容，可以只缓存表头和行数，切片数据按需读取但缓存切片结果（比如用cache.set(f"slice_{self.pk}_{column_idx}_{start}_{end}", data)）。
• DRF序列化器缓存：可以用django-cache-memoize库来缓存序列化器的方法，比如在序列化器的to_representation方法里缓存结果。
• 编码一致性：读取文件时一定要指定明确的编码（比如utf-8），避免不同环境下出现编码错误，也能保证解析结果的一致性。

内容的提问来源于stack exchange，提问作者ZeroSilver