(cls, data_source_id: int)
| 84 | |
| 85 | @classmethod |
| 86 | def delete_data_source(cls, data_source_id: int) -> str: |
| 87 | with Session() as session: |
| 88 | data_source = session.query(DataSource).filter_by(id=data_source_id).first() |
| 89 | if data_source is None: |
| 90 | raise KnownException(message=f"Data source {data_source_id} does not exist") |
| 91 | |
| 92 | data_source_name = data_source.type.name |
| 93 | logger.info(f"Deleting data source {data_source_id} ({data_source_name})...") |
| 94 | session.delete(data_source) |
| 95 | session.commit() |
| 96 | |
| 97 | del cls._data_source_cache[data_source_id] |
| 98 | |
| 99 | return data_source_name |
| 100 | |
| 101 | @classmethod |
| 102 | def init(cls): |
no test coverage detected