run.py 543 B

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