zish-rob-crur / full-stack-template

full-stack-template

full-stack-template's Issues

关于 lzj 的一些问题 回复

1.读取配置文件

使用的是 pydantic 的 settings 解析，非标准配置都可以写在.env 文件里面
比如
class DbSettings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", env_prefix="DB_")
url: MySQLDsn = "mysql://user:password@localhost/db"
pool_size: int = 10

那么在 .env 里面就可以写
DB_URL=mysql://user:password@localhost/db
DB_POOL_SIZE=20
只要注意 prefix 即可

2. 增加任务

使用 celery 的 task 装饰器之后，就可以对这个函数进行异步调用，然后就是一个任务，会自动给发送到队列里面
具体看

@router.post("/users/")
def create_user_route(
    username: str,
    password: str,
):
    workflow = create_user.s(username, password) | send_user_hello_email.s()
    async_result = workflow.apply_async()
    logger.info("create_user_route task_id: {async_result.id}")
    return {"task_id": async_result.id}

当然 workflow 可以是一个链式的任务，也可以是一个 group 的任务，具体看 celery 的文档 或者 让 GPT4 解释一下

3.还有个问题，celery -A celery_app worker --loglevel=info -P eventlet 启动了中间件，然后具体执行任务，是python xxx.py

celery 是启动了 worker 之后，会自动去执行任务，不需要我们去执行任务，只需要把任务发送到队列里面即可，使用上面的 apply_async 或者 delay 即可

4. 不是一个进程的，python xxx.py是怎么获取到中间件的task并执行的

celery 会将任务的 Result 储存到 配置里面的 result_backend 里面，在每一次 apply_async 或者 delay 的时候，会拿到一个 async_result 然后里面有一个 id 可以将这个id 记录下来
然后可以通过这个 id 去查询这个任务的状态，结果等等
比如

async_result = workflow.apply_async()
logger.info("create_user_route task_id: {async_result.id}")

from celery.result import AsyncResult
async_result = AsyncResult(async_result.id)
print(async_result.status)
print(async_result.result)