R/utils.R

In utubun/keggr: KEGG API

# Constants

BASEURL <- 'http://rest.kegg.jp'


# functions

kegg <- function(verb, ...) {

    # find db/database argument

    switch(
        verb,
        list = .kegg_list(...)
    )
}


# Utilities functions ----------------------------------------------------------

.kegg_check_response <- function(resp) {

    if (httr::http_error(resp)) {

        message(
            sprintf(
                '\u001b[31m\u26A0 %s %s: Database \'%s\' does not exist!',
                httr::http_status(resp)$reason,
                resp$url,
                basename(resp$url)
            )
        )


        return(invisible(FALSE))
    }

    message(
        sprintf('\u001b[32m\u2713\u001b[30m Query %s is Ok!', resp$url)
    )

    return(invisible(TRUE))
}

.kegg_resp_content_type <- function(resp) {

    cont_type <- httr::headers(resp)$`content-type`
    cont_type <- unlist(strsplit(cont_type, split = '; '))
    cont_type <- gsub('charset=', '', cont_type)
    cont_type[2] <- toupper(cont_type[2])
    message(paste(cont_type))
    return(cont_type)

}

.kegg_verify_db <- function(verb, db, silent = FALSE) {

    msg <- ''

    # check if the `verb` is character vector of length 1

    if(length(verb) != 1) {
        msg <- paste(msg, '\U2757 Argument `verb` must be a single character.\n')
    }

    # check if the `verb` is correctly specified

    if(!verb %in% c('info', 'list', 'get', 'conv', 'link', 'ddi')) {
        msg <- paste(msg, '\U2757 Argument `verb` must be one of: info, list, get, conv, link, ddi\n')
    }

    # exit if message exists

    if(msg != '') {
        message(msg)
        return(FALSE)
    }

    # connect to database

    tempdb    <- DBI::dbConnect(RSQLite::SQLite(), './data/kegger.db')

    # query databases available for this verb

    query <- DBI::dbSendQuery(
        tempdb,
        sprintf('SELECT database FROM verbs WHERE %s = 1', verb)
    )

    # list databases available for this verb

    dblist <- DBI::fetch(query)$database
    dblist <- dblist[dblist == db]

    # if list is empty, check if database is kidorg or taxid

    if(!length(dblist) & (grepl('[a-z]{3}', db) | grepl('^T\\d{5}$', db))) {

        dbClearResult(query)

        query <- DBI::dbSendQuery(
            tempdb,
            sprintf('SELECT * FROM org WHERE kid = "%s" OR taxid = "%s"', db, db)
        )

        # list databases available for this verb

        dblist <- DBI::fetch(query)$kid

    }

    # clear query and close database connection

    dbClearResult(query)
    dbDisconnect(tempdb)

    # check if the database argument specified is correct

    if(!length(dblist)) {
        msg <- paste(
            msg,
            sprintf(
                '\U2757 Database name for query %s must be one of: %s',
                verb,
                paste(dblist, collapse = ', ')
                )
            )
    }

    if(msg != '' & !silent) {
        message(msg)
        return(FALSE)
    }

    if(!silent) {
        message(sprintf('\U2705 Database `%s` is ok.', db))
    }

    return(TRUE)
}



.kegg_req_parse_header <- function(req) {
    return(NULL)
}

.kegg_verify_operation <- function(operation) {
    operation <- match.arg(
        operation,
        c('info', 'list', 'find', 'get', 'conv', 'link', 'ddi')
    )

    return(operation)
}

.kegg_rest <- function(query) {

    url <- httr::modify_url('http://rest.kegg.jp', path = query)

    res <- httr::GET(url)

    if (status_code(res) != 200) {
        stop(http_status(res)$reason, call. = FALSE)
    }

    return(res)
}