Django异步查询并下载CSV文件

news2025/10/31 15:02:57

Django异步查询并下载CSV文件

通过循环遍历数据库,自动生成CSV文件的表头和内容.
Django V5.1

1. 视图

1.1 将同步的数据库查询转换为异步

async def get_blogs():
    # 使用sync_to_async包装Queryset方法
    blog_list = await sync_to_async(Blog.objects.all)()
    return blog_list

1.2 定义一个异步生成器逐行生成CSV数据

async def generate_csv():
    # 获取queryset list
    blogs = await get_blogs()
    # 生成csv文件头
    async for blog in blogs.values():
    	# 打印查看遍历的内容 
        await sync_to_async(print)(blog.keys())
        yield ','.join(blog.keys()) + '\n'
        break
    # 生成csv文件内容
    async for values in blogs.values_list():
    	# 打印查看遍历的内容 
        await sync_to_async(print)(values)
        yield ','.join(map(str, values)) + '\n'

1.3 异步下载

async def export_csv(request):
    response = StreamingHttpResponse(
        generate_csv(),
        content_type='text/csv'
    )
    response['Content-Disposition'] = 'attachment; filename="blogs.csv"'
    return response

2. 模板

    <div class="container-fluid mt-3">
        <form action="{% url 'blog:export_csv' %}">
            {% csrf_token %}
            <button type="submit" class="btn btn-primary form-control">异步下载测试</button>
        </form>
    </div>

3. url

path('export-csv', views.export_csv, name="export_csv")

4. model

class Blog(models.Model):
    title = models.CharField(max_length=200, verbose_name="标题")
    content = models.TextField(verbose_name="内容")
    pub_time = models.DateTimeField(auto_now_add=True, verbose_name="发布时间")
    category = models.ForeignKey(to=BlogCategory, on_delete=models.CASCADE, verbose_name="分类")
    author = models.ForeignKey(to=User, on_delete=models.CASCADE, verbose_name="作者")

    class Meta:
        verbose_name = "博客"
        verbose_name_plural = verbose_name
        ordering = ['-pub_time']

    def __str__(self):
        return self.title