Coverage for /home/ubuntu/f8s/python/f8s/extension.py: 98%

40 statements  

« prev     ^ index     » next       coverage.py v7.4.3, created at 2024-02-28 23:26 +0000

1from typing import Any, Optional # noqa F401 

2 

3import os 

4import re 

5 

6import flask 

7import yaml 

8# ------------------------------------------------------------------------------ 

9 

10 

11API = flask.Blueprint('f8s', __name__, url_prefix='') 

12 

13 

14@API.route('/apidocs') 

15def apidocs(): 

16 # type: () -> Any 

17 ''' 

18 Route to F8s API documentation. 

19 

20 Returns: 

21 html: Flassger generated API page. 

22 ''' 

23 return flask.redirect(flask.url_for('flasgger.apidocs')) 

24# ------------------------------------------------------------------------------ 

25 

26 

27class F8s: 

28 api = API 

29 

30 def __init__(self, app=None): 

31 # type: (Optional[flask.Flask]) -> None 

32 ''' 

33 Initialize flask extension. 

34 

35 Args: 

36 app (flask.Flask, optional): Flask app. 

37 ''' 

38 if app is not None: 

39 self.init_app(app) 

40 

41 @property 

42 def name(self): 

43 # type: () -> str 

44 ''' 

45 Returns: 

46 str: Name of class. 

47 ''' 

48 return self.__class__.__name__.lower() 

49 

50 def init_app(self, app): 

51 # type: (flask.Flask) -> None 

52 ''' 

53 Add endpoints and error handlers to given app. 

54 

55 Args: 

56 app (Flask): Flask app. 

57 ''' 

58 app.extensions[self.name] = self 

59 app.register_blueprint(self.api) 

60 if not app.config['TESTING']: 

61 self.config = self.get_config(app) 

62 app.config[self.name] = self.config 

63 self.validate(self.config) 

64 

65 def get_config(self, app): 

66 # type: (flask.Flask) -> dict 

67 ''' 

68 Get config from envirnment variables or config file. 

69 

70 Args: 

71 app (flask.Flask): Flask app. 

72 

73 Returns: 

74 dict: Database config. 

75 ''' 

76 # get config variables from environment 

77 prefix = self.name.upper() + '_' 

78 env = {} 

79 for key, val in os.environ.items(): 

80 if key.startswith(prefix): 

81 k = re.sub(f'^{prefix}', '', key).lower() 

82 env[k] = val 

83 

84 # create config 

85 config_path = env.get('config_path', None) 

86 config = {} 

87 if config_path is not None: 

88 with open(config_path) as f: 

89 config = yaml.safe_load(f) 

90 

91 # update config with env vars 

92 config.update(env) 

93 return config 

94 

95 def validate(self, config): 

96 # type: (dict) -> None 

97 ''' 

98 Raises an error if given config is invalid. 

99 

100 Args: 

101 config (dict): Extension config. 

102 ''' 

103 pass