| 123456789101112131415161718192021222324252627 |
- # -*- coding:utf-8 -*-
- from configparser import ConfigParser
- from typing import Dict
- class DataSource(object):
- def __init__(self, ds_file: str):
- self.ds_file = ds_file
- self.config_parser = ConfigParser()
- self.config_parser.read(self.ds_file)
- self.ds_dict = {}
- self.source_type = None
- self.keys = []
- def get_datasource_dict(self) -> Dict[str, str]:
- for key in self.keys:
- try:
- self.ds_dict[key] = self.config_parser.get('base', key)
- except KeyError:
- raise KeyError('%s must specified at %s data source config.' % (key, self.source_type))
- return self.ds_dict
- def parse(self):
- if self.source_type is None:
- raise NotImplementedError('please use a specified class of data source.')
- return self.get_datasource_dict()
|
