def get_instance()

in src/ab/utils/data_source.py [0:0]

• 13 lines of code
• 9 McCabe index (conditional complexity)

    def get_instance(data_source: dict = None, args: dict = None, cacheable=True, sampler=None, **kwargs):
        '''
        根据数据源配置生成db
        '''
        if not data_source:
            # some api may not use data_source
            return None

        table_name = data_source.get("table_name") or args.get('table_name') or kwargs.get("table_name")
        sql = data_source.get("sql") or args.get('sql') or kwargs.get("sql")
        partitions = data_source.get("partitions") or args.get('partitions') or kwargs.get("partitions")

        if cacheable:
            return CachedDataSource(data_source, table_name, partitions, sampler, sql=sql)
        else:
            return DataSource(data_source, table_name, partitions, sampler, sql=sql)