如何用Django,Celery,AlpineJS还有HTMX SSE实现ProgressBar

Open Table of contents

开发环境
安装配置SSE & Celery
- 配置SSE
- 配置Celery
如何实现progressbar
Demo
总结

开发环境

基本的环境可以参考如何在Django中配置Alpine.js，HTMX和tailwindcss。

唯二的区别有htmx插件sse，以及celery，请参考以下步骤进行安装配置。

安装配置SSE & Celery

配置SSE

目前安装htmx的插件有两种方式：CDN 以及 npm包管理器。

如果是想快速安装，建议可以使用CDN。把下面链接直接添加到标签中即可：

<head>
    ...
    <script src="https://cdn.jsdelivr.net/npm/htmx.org@2.0.10/dist/htmx.min.js" integrity="sha384-H5SrcfygHmAuTDZphMHqBJLc3FhssKjG7w/CeCpFReSfwBWDTKpkzPP8c+cLsK+V" crossorigin="anonymous"></script>
    <script src="https://cdn.jsdelivr.net/npm/htmx-ext-sse@2.2.4" integrity="sha384-A986SAtodyH8eg8x8irJnYUk7i9inVQqYigD6qZ9evobksGNIXfeFvDwLSHcp31N" crossorigin="anonymous"></script>
    ...
</head>

如果习惯用npm管理器，也可以执行下面命令安装到本地：

npm install htmx-ext-sse
cp ../node_modules/htmx-ext-sse/dist/sse.min.js ./app/static/app/js/
# ./app/static/app/js/是存放django静态数据的目录，每个人的命名规则或喜好会有些许不同，请根据实际情况修改该目标路径。

接着，在django template中引用该插件：

{% load static %}
<head>
    ...
    <script src="{% static 'app/sse.min.js' %}"></script>
    ...
</head>

配置Celery

关于Celery的安装配置，Celery的官网已经讲得非常具体了，这边会概括并解释一些步骤。

celery.py - 这个是配置Celery的关键文件，通常是新建在settings.py所在目录的文件夹，即项目文件夹下。
在celery.py中, 有两个地方需要特别注意下。

...
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()
...

如果修改了app.config_from_object中的namespace，那么在settings.py中对于Celery的每个配置项的前缀也要相应地更改，否则启动worker的时候可能会因为找不到配置项而报错。比如我们把namespace改成了TEST，那么相应的BROKER和BACKEND需要改成如下：

# settings.py
...
# Celery Configuration
TEST_BROKER_URL = "redis://localhost:6379/0"
TEST_RESULT_BACKEND = "redis://localhost:6379/0"
...

从设置我们看到，我们用了Redis分别作为broker和result backend。如果我们有docker，可以直接运行docker run -d --name redis -p 6379:6379 redis 来快速让Redis跑起来。
在写Task的时候，一般都是会在各个app目录下新建tasks.py文件，然后在app.autodiscover_tasks()帮助下，启动worker的时候会去扫描各个app里的tasks.py文件注册到任务池中。否则，需要手动在CELERY_IMPORTS的配置项中逐一列举。

运行celery -A demo_progressbar worker -l INFO --pool=solo来启动celery worker等活来吧！

如何实现progressbar

新建tasks.py

之前说过，task一般都会跟着app，所以我们可以在app目录下面新建一个tasks.py，这样在启动worker的时候会自动扫描所有task，可以从输出日志中找到那些task被识别到了。 tasks

因为我们打算实现一个progressbar，可以用time.sleep来模拟。

@shared_task(bind=True)
def dosomething(self):
    progress = 0
    n = 100
    for i in range(n):
        time.sleep(0.05)
        progress += 1
        self.update_state(state="PROGRESS", meta={"current": progress, "total": n})

self.update_state: 更新存储在backend中对于task的状态，Celery默认的状态有PENDING, STARTED, SUCCESS, FAILURE, RETRY, REVOKED。而使用方法update_state就可以创建自定义的状态，比如这里的PROGRESS，meta的值代表progressbar的最新状态。

调用task

由于tasks.py文件直接新建在app目录中，我们可以在视图函数中引入后直接调用。

...
from .tasks import dosomething

def start(request):
    task = dosomething.delay()
    return render(request, "demo/progressbar.html", {"task_id": task.id})

当这个dosomething通过delay()方法被调用的时候, 会返回一个AsyncResult，它有一个id属性，把它发送到客户端以便后续查询task的状态。

前端更新progressbar

sse的具体实现，可以参考如何在Django中实现SSE。 progressbar.html

<div x-data="{
currentVal: 0 ,
minVal: 0 ,
maxVal: 100,
calcPercentage(min, max, val){return ((val-min)/(max-min))*100}
}" class="progress-bar" @htmx:sse-before-message="currentVal = $event.detail.data;$event.preventDefault()">
    <div hx-ext="sse" sse-connect="{% url 'progress' %}?task={{ task_id }}" sse-swap="status" sse-close="done">
        Contents of this box will be updated in real time.
        <div class="mt-5 mb-5 flex h-2.5 w-full overflow-hidden rounded-md bg-neutral-50 dark:bg-neutral-900"
             role="progressbar" aria-label="default progress bar" :aria-valuenow="currentVal" :aria-valuemin="minVal"
             :aria-valuemax="maxVal">
            <div class="h-full rounded-md bg-black dark:bg-white"
                 :style="`width: ${calcPercentage(minVal, maxVal, currentVal)}%`">
            </div>
        </div>
    </div>
</div>

sse-connect: 通过连接progress的EventSource并且查询task_id来持续返回状态。

views.py

...
async def progress(request):
    task_id = request.GET.get("task")
    async def progress_generator():
        while True:
            result = AsyncResult(task_id)
            if result.state == "PROGRESS":
                yield f"event: status\ndata: {result.info['current']}\n\n"
            elif result.state == "SUCCESS":
                yield f"event: status\ndata: 100\n\n"
                yield f"event: done\ndata: {None}\n\n"
                break
            await asyncio.sleep(1)
    return StreamingHttpResponse(progress_generator(), content_type="text/event-stream")
...

@htmx:sse-before-message: 这个事件发生在接收到sse的返回结果后，但是还未更新DOM内容前，所以可以用来更新currentVal的值并且放弃后续的操作($event.preventDefault)。至于为什么事件的写法不一样和官方文档里不一样，主要是因为⌈@⌋这种写法是Alpine.js支持的x-on的简化模式，而它只支持Kebab Case的写法即not-camel-case来代表notCamelCase，所以我们要做一个转换，详细可参考Event Naming。
sse-swap="status": 监听status事件，配合上面的sse-before-message事件使用
sse-close="done": 监听done事件来关闭sse通道，否则会一直发送sse-connect指定的url请求服务器。如果AsyncResult的状态是SUCCESS的话，那么就一直返回”event: status\ndata: 100\n\n”,接着”event: done\ndata: {None}\n\n”

Demo

demo

总结

以上就是整个用htmx sse extension和celery一起实现进度条的实现过程，虽然celery的依赖项（broker+backend）我们用Docker版的redis简化了，但还是感觉有些麻烦，截止发文Django 6.0也发布很久了，里面自带的task框架会简化一部分内容，虽然也挺抽象的（没有官方的backend），但聊胜于无，后面有机会的话再聊一聊这个自带的框架。