Python SPARQLWrapper class that works with VIVO's built-in SPARQL utility

vivosparql.py

"""
A subclass of SPARQLWrapper that will work with the built-in
SPARQL admin interface that comes with VIVO.

At the moment, will only return RS_JSON for SELECT and
N3 for CONSTRUCT queries.  

"""
from pprint import pprint
import urllib
import urllib2

import requests

from SPARQLWrapper import SPARQLWrapper, JSON, N3
from SPARQLWrapper.SPARQLExceptions import QueryBadFormed, EndPointNotFound, EndPointInternalError
from SPARQLWrapper.Wrapper import _SPARQL_JSON

#VIVO returns its RS_JSON as application/javascript
_SPARQL_JSON.append('application/javascript')

#VIVO web application username and password
username = 'user'
password = 'pass'
#VIVO web application URL
vivo_url = 'http://vivo.school.edu/'


class VIVOSparql(SPARQLWrapper):
    """
    Extension of SPARQLWrapper to work with the built-in VIVO 
    SPARQL query interface.  Eliminates the need to use Fuseki
    for SPARQL non-update queries.
    """

    def __init__(self,vivo_url, **kwargs):
        self.session = requests.session()
        self.vivo_url = vivo_url
        #Add the VIVO SPARQL end point path to the VIVO url.
        self.endpoint = vivo_url + 'admin/sparqlquery'
        SPARQLWrapper.__init__(self, self.endpoint, kwargs)
        #The resultFormat and rdfResultFormat are required 
        #parameters for the VIVO SPARQL interface.
        #This are set to RS_JSON for SELECT and 
        #N3 for CONSTRUCT.  These can be overridden when
        #called but have not been tested.
        self.addCustomParameter('resultFormat', 'RS_JSON')
        self.addCustomParameter('rdfResultFormat', 'N3')
        

    def login(self):
        payload = {
            'loginName': self.user,
            'loginPassword': self.passwd,
            'loginForm': 'Log in'
        }
        r = self.session.post(self.vivo_url + 'authenticate', 
            data=payload, 
            verify=False)
        self.cookies = urllib.urlencode(self.session.cookies)

    def logout(self):
        resp = self.session.get(self.vivo_url + 'logout')
        #Check response history for logout.
        logout_resp = resp.history[0]
        if logout_resp.status_code == 302:
            return True
        else:
            raise Exception('Logout failed.')

    def setQuery(self,query):
        #Let's handle the response format here by looking
        #at the response type.
        if 'construct' in query.lower():
            self.setReturnFormat(N3)
        else:
            self.setReturnFormat(JSON)
        SPARQLWrapper.setQuery(self, query)

    def _query(self):
        """
        Override _query method to use cookies acquired on login.
        """
        request = self._createRequest()
        try:
            opener = urllib2.build_opener()
            opener.addheaders.append(('Cookie', self.cookies))
            response = opener.open(request)
            return (response, self.returnFormat)
        except urllib2.HTTPError, e:
            if e.code == 400:
                raise QueryBadFormed()
            elif e.code == 404:
                raise EndPointNotFound()
            elif e.code == 500:
                raise EndPointInternalError(e.read())
            else:
                raise e
            return (None, self.returnFormat)

q = """
PREFIX vivo: <http://vivoweb.org/ontology/core#>
CONSTRUCT {?s a vivo:TempThing}
WHERE {?s a vivo:FacultyMember}
LIMIT 3
"""

q = """
PREFIX vivo: <http://vivoweb.org/ontology/core#> 
SELECT * 
WHERE {?s a vivo:FacultyMember}
LIMIT 3
""""

sparql = VIVOSparql(vivo_url)
sparql.setCredentials(username, password)
sparql.setQuery(q)
sparql.login()
results = sparql.queryAndConvert()
pprint(results)
sparql.logout()