query data source
(self)
| 181 | @expose('/info', methods=("GET",)) |
| 182 | @safe |
| 183 | def query_data_source(self) -> FlaskResponse: |
| 184 | """ |
| 185 | query data source |
| 186 | """ |
| 187 | try: |
| 188 | data_source_name = request.args.get('name', default=None, type=str) |
| 189 | data_source_type_id = request.args.get('typeId', default=None, type=int) |
| 190 | expire = request.args.get('expire', default=None, type=bool) |
| 191 | page_size = request.args.get('pageSize', default=10, type=int) |
| 192 | page_no = request.args.get('pageNo', default=1, type=int) |
| 193 | |
| 194 | # Ensure page size and page number are positive integers |
| 195 | page_size = max(1, page_size) |
| 196 | page_no = max(1, page_no) |
| 197 | |
| 198 | # Query and paginate data sources |
| 199 | page_info = DataSourceDAO.get_data_source_page( |
| 200 | page_no=page_no, |
| 201 | page_size=page_size, |
| 202 | name_filter=data_source_name, |
| 203 | type_filter=data_source_type_id, |
| 204 | expire_filter=expire |
| 205 | ) |
| 206 | |
| 207 | # Format the response |
| 208 | page_info_schema = DataSourcePageInfoSchema() |
| 209 | return self.response_format(data=page_info_schema.dump(page_info)) |
| 210 | |
| 211 | except DAONotFound as ex: |
| 212 | logger.exception(ex) |
| 213 | return self.handle_error(SolidUIErrorType.QUERY_DATASOURCE_ERROR) |
| 214 | |
| 215 | @expose('/connect/json', methods=("POST",)) |
| 216 | @safe |
nothing calls this directly
no test coverage detected