12345678910111213141516171819202122 |
- # -*- coding: utf-8 -*-
- # @Author: privacy
- # @Date: 2024-09-03 11:41:18
- # @Last Modified by: privacy
- # @Last Modified time: 2024-09-05 11:07:42
- import uvicorn
- original_callback = uvicorn.main.callback
- def callback(**kwargs):
- from celery.contrib.testing.worker import start_worker
- from celery_tasks import celery_app
- with start_worker(celery_app, concurrency=2, perform_ping_check=False, loglevel="info"):
- original_callback(**kwargs)
- uvicorn.main.callback = callback
- if __name__ == '__main__':
- uvicorn.main()
|