(cls)
| 130 | |
| 131 | @classmethod |
| 132 | def _load_data_source_classes(cls): |
| 133 | data_sources: Dict[str, ClassInfo] = DynamicLoader.find_data_sources() |
| 134 | |
| 135 | with Session() as session: |
| 136 | for source_name in data_sources.keys(): |
| 137 | class_info = data_sources[source_name] |
| 138 | data_source_class = DynamicLoader.get_class(file_path=class_info.file_path, |
| 139 | class_name=class_info.name) |
| 140 | cls._data_source_classes[source_name] = data_source_class |
| 141 | |
| 142 | if session.query(DataSourceType).filter_by(name=source_name).first(): |
| 143 | continue |
| 144 | |
| 145 | config_fields = data_source_class.get_config_fields() |
| 146 | config_fields_str = json.dumps([config_field.dict() for config_field in config_fields]) |
| 147 | new_data_source = DataSourceType(name=source_name, |
| 148 | display_name=data_source_class.get_display_name(), |
| 149 | config_fields=config_fields_str) |
| 150 | session.add(new_data_source) |
| 151 | session.commit() |
no test coverage detected