Coverage for /usr/local/lib/python3.10/site-packages/opal_common-0.0.0-py3.10.egg/opal_common/corn_utils.py: 0%

1"""Utilities to run UVICORN / GUNICORN.""" 

2import multiprocessing 

3from typing import Dict 

4 

5import gunicorn.app.base 

6 

7 

8def calc_default_number_of_workers(): 

9 return (multiprocessing.cpu_count() * 2) + 1 

10 

11 

12class GunicornApp(gunicorn.app.base.BaseApplication): 

13 def __init__(self, app, options: Dict[str, str] = None): 

14 self.options = options or {} 

15 self.application = app 

16 super().__init__() 

17 

18 def load_config(self): 

19 config = { 

20 key: value 

21 for key, value in self.options.items() 

22 if key in self.cfg.settings and value is not None 

23 } 

24 for key, value in config.items(): 

25 self.cfg.set(key.lower(), value) 

26 

27 def load(self): 

28 return self.application 

29 

30 

31def run_gunicorn(app, number_of_workers=None, host=None, port=None, **kwargs): 

32 options = { 

33 "bind": "%s:%s" % (host or "127.0.0.1", port or "8080"), 

34 "workers": number_of_workers or calc_default_number_of_workers(), 

35 "worker_class": "uvicorn.workers.UvicornWorker", 

36 } 

37 options.update(kwargs) 

38 GunicornApp(app, options).run() 

39 

40 

41def run_uvicorn( 

42 app_path, number_of_workers=None, host=None, port=None, reload=False, **kwargs 

43): 

44 options = { 

45 "host": host or "127.0.0.1", 

46 "port": port or "8080", 

47 "reload": reload, 

48 "workers": number_of_workers or calc_default_number_of_workers(), 

49 } 

50 options.update(kwargs) 

51 import uvicorn 

52 

53 uvicorn.run(app_path, **options)