(cls, name: str, config: dict)
| 59 | |
| 60 | @classmethod |
| 61 | async def create_data_source(cls, name: str, config: dict) -> BaseDataSource: |
| 62 | async with async_session() as session: |
| 63 | data_source_type = await session.execute( |
| 64 | select(DataSourceType).filter_by(name=name) |
| 65 | ) |
| 66 | data_source_type = data_source_type.scalar_one_or_none() |
| 67 | if data_source_type is None: |
| 68 | raise KnownException(message=f"Data source type {name} does not exist") |
| 69 | |
| 70 | data_source_class = DynamicLoader.get_data_source_class(name) |
| 71 | logger.info(f"validating config for data source {name}") |
| 72 | await data_source_class.validate_config(config) |
| 73 | config_str = json.dumps(config) |
| 74 | |
| 75 | data_source_row = DataSource(type_id=data_source_type.id, config=config_str, created_at=get_utc_time_now()) |
| 76 | session.add(data_source_row) |
| 77 | await session.commit() |
| 78 | |
| 79 | data_source = data_source_class(config=config, data_source_id=data_source_row.id) |
| 80 | cls._data_source_cache[data_source_row.id] = CachedDataSource(indexed_docs=0, failed_tasks=0, |
| 81 | instance=data_source) |
| 82 | |
| 83 | return data_source |
| 84 | |
| 85 | @classmethod |
| 86 | def delete_data_source(cls, data_source_id: int) -> str: |
no test coverage detected