def _get_query_text_from_execute_params()

in ossdbtoolsservice/query_execution/query_execution_service.py [0:0]

• 18 lines of code
• 11 McCabe index (conditional complexity)

    def _get_query_text_from_execute_params(self, params: ExecuteRequestParamsBase):
        if isinstance(params, ExecuteDocumentSelectionParams):
            workspace_service = self._service_provider[utils.constants.WORKSPACE_SERVICE_NAME]
            selection_range = params.query_selection.to_range() if params.query_selection is not None else None

            return workspace_service.get_text(params.owner_uri, selection_range)

        elif isinstance(params, ExecuteDocumentStatementParams):
            workspace_service = self._service_provider[utils.constants.WORKSPACE_SERVICE_NAME]
            query = workspace_service.get_text(params.owner_uri, None)
            selection_data_list: List[SelectionData] = compute_batches(sqlparse.split(query), query)

            for selection_data in selection_data_list:
                if selection_data.start_line <= params.line and selection_data.end_line >= params.line:
                    if (selection_data.end_line == params.line and selection_data.end_column < params.column or
                            selection_data.start_line == params.line and selection_data.start_column > params.column):
                        continue
                    return workspace_service.get_text(params.owner_uri, selection_data.to_range())

            return ''

        else:
            # Then params must be an instance of ExecuteStringParams, which has the query as an attribute
            return params.query