def get_primary_keys()

in python-phoenixdb/phoenixdb/meta.py [0:0]

• 31 lines of code
• 4 McCabe index (conditional complexity)

    def get_primary_keys(self, catalog=None, schema=None, table=None):
        if self._connection._closed:
            raise ProgrammingError('The cursor is already closed.')

        state = common_pb2.QueryState()
        state.type = common_pb2.StateType.METADATA
        state.op = common_pb2.MetaDataOperation.GET_PRIMARY_KEYS
        state.has_args = True
        state.has_op = True

        catalog_arg = self._moa_string_arg_factory(catalog)
        schema_arg = self._moa_string_arg_factory(schema)
        table_arg = self._moa_string_arg_factory(table)
        state.args.extend([catalog_arg, schema_arg, table_arg])

        with DictCursor(self._connection) as cursor:
            syncResultResponse = cursor.get_sync_results(state)
            if not syncResultResponse.more_results:
                return []

            signature = common_pb2.Signature()
            signature.columns.append(self._column_meta_data_factory(1, 'TABLE_CAT', 12))
            signature.columns.append(self._column_meta_data_factory(2, 'TABLE_SCHEM', 12))
            signature.columns.append(self._column_meta_data_factory(3, 'TABLE_NAME', 12))
            signature.columns.append(self._column_meta_data_factory(4, 'COLUMN_NAME', 12))
            signature.columns.append(self._column_meta_data_factory(5, 'KEY_SEQ', 5))
            signature.columns.append(self._column_meta_data_factory(6, 'PK_NAME', 12))
            # The following are non-standard Phoenix extensions
            # This returns '\x00\x00\x00A' or '\x00\x00\x00D' , but that's consistent with Java
            signature.columns.append(self._column_meta_data_factory(7, 'ASC_OR_DESC', 12))
            signature.columns.append(self._column_meta_data_factory(8, 'DATA_TYPE', 5))
            signature.columns.append(self._column_meta_data_factory(9, 'TYPE_NAME', 12))
            signature.columns.append(self._column_meta_data_factory(10, 'COLUMN_SIZE', 5))
            signature.columns.append(self._column_meta_data_factory(11, 'TYPE_ID', 5))
            signature.columns.append(self._column_meta_data_factory(12, 'VIEW_CONSTANT', 12))

            cursor.fetch(signature)
            return cursor.fetchall()