(self)
| 33 | @expose('/queryDatabases', methods=("GET",)) |
| 34 | @safe |
| 35 | def get_databases(self) -> FlaskResponse: |
| 36 | data_source_name = request.args.get('dataSourceName') |
| 37 | # Call the service method to get databases. You need to implement this logic |
| 38 | |
| 39 | data_source = DataSourceDAO.get_data_source_name(data_source_name) |
| 40 | if data_source is None: |
| 41 | return self.handle_error(SolidUIErrorType.QUERY_DATASOURCE_ERROR) |
| 42 | |
| 43 | data_source_type = DataSourceTypeDAO.get_id(data_source.datasource_type_id) |
| 44 | if data_source_type is None: |
| 45 | return self.handle_error(SolidUIErrorType.QUERY_DATASOURCE_TYPE_ERROR) |
| 46 | |
| 47 | # Parse the JSON parameters |
| 48 | params = json.loads(data_source.parameter) |
| 49 | |
| 50 | # Create a JDBC client |
| 51 | jdbc_client = JdbcClientFactory.create_client( |
| 52 | db_type=data_source_type.name.lower(), # Assuming MySQL for example |
| 53 | host=params.get("host"), |
| 54 | port=params.get("port"), |
| 55 | username=params.get("username"), |
| 56 | password=params.get("password"), |
| 57 | database=params.get("database"), |
| 58 | extra_params=params.get("params", {}) |
| 59 | ) |
| 60 | |
| 61 | try: |
| 62 | # Retrieve and return the list of databases |
| 63 | databases = JdbcClientFactory.get_databases(jdbc_client) |
| 64 | return self.response_format(data=databases) |
| 65 | |
| 66 | except DAONotFound as ex: |
| 67 | logger.exception(ex) |
| 68 | return self.handle_error(SolidUIErrorType.QUERY_METADATA_DB_ERROR) |
| 69 | |
| 70 | @expose('/queryTables', methods=("GET",)) |
| 71 | @safe |
no test coverage detected