<s> from xml . sax . saxutils import XMLGenerator <EOL> from xml . dom import XML_NAMESPACE <EOL> from xml . sax . xmlreader import AttributesNSImpl <EOL> from xml . etree import cElementTree as ElementTree <EOL> from rdflib import Literal , URIRef , BNode , Graph , Variable <EOL> from rdflib . query import ( <EOL> Result , <EOL> ResultParser , <EOL> ResultSerializer , <EOL> ResultException <EOL> ) <EOL> SPARQL_XML_NAMESPACE = u'<STR_LIT>' <EOL> RESULTS_NS_ET = '<STR_LIT>' % SPARQL_XML_NAMESPACE <EOL> """<STR_LIT>""" <EOL> class XMLResultParser ( ResultParser ) : <EOL> def parse ( self , source ) : <EOL> return XMLResult ( source ) <EOL> class XMLResult ( Result ) : <EOL> def __init__ ( self , source ) : <EOL> xmlstring = source . read ( ) <EOL> if isinstance ( xmlstring , unicode ) : <EOL> xmlstring = xmlstring . encode ( '<STR_LIT:utf-8>' ) <EOL> try : <EOL> tree = ElementTree . fromstring ( xmlstring ) <EOL> except Exception , e : <EOL> try : <EOL> raise e . __class__ ( "<STR_LIT>" % ( xmlstring , e ) ) <EOL> except : <EOL> raise e <EOL> boolean = tree . find ( RESULTS_NS_ET + '<STR_LIT>' ) <EOL> results = tree . find ( RESULTS_NS_ET + '<STR_LIT>' ) <EOL> if boolean is not None : <EOL> type_ = '<STR_LIT>' <EOL> elif results is not None : <EOL> type_ = '<STR_LIT>' <EOL> else : <EOL> g = Graph ( ) <EOL> try : <EOL> g . parse ( data = xmlstring ) <EOL> if len ( g ) == <NUM_LIT:0> : <EOL> raise <EOL> type_ = '<STR_LIT>' <EOL> except : <EOL> raise ResultException ( <EOL> "<STR_LIT>" ) <EOL> Result . __init__ ( self , type_ ) <EOL> if type_ == '<STR_LIT>' : <EOL> self . bindings = [ ] <EOL> for result in results : <EOL> r = { } <EOL> for binding in result : <EOL> r [ Variable ( binding . get ( '<STR_LIT:name>' ) ) ] = parseTerm ( binding [ <NUM_LIT:0> ] ) <EOL> self . bindings . append ( r ) <EOL> self . vars = [ Variable ( x . get ( "<STR_LIT:name>" ) ) <EOL> for x in tree . findall ( <EOL> '<STR_LIT>' % ( <EOL> RESULTS_NS_ET , RESULTS_NS_ET ) ) ] <EOL> elif type_ == '<STR_LIT>' : <EOL> self . askAnswer = boolean . text . lower ( ) . strip ( ) == "<STR_LIT:true>" <EOL> elif type_ == '<STR_LIT>' : <EOL> self . graph = g <EOL> def parseTerm ( element ) : <EOL> """<STR_LIT>""" <EOL> tag , text = element . tag , element . text <EOL> if tag == RESULTS_NS_ET + '<STR_LIT>' : <EOL> if text is None : <EOL> text = '<STR_LIT>' <EOL> datatype = None <EOL> lang = None <EOL> if element . get ( '<STR_LIT>' , None ) : <EOL> datatype = URIRef ( element . get ( '<STR_LIT>' ) ) <EOL> elif element . get ( "<STR_LIT>" % XML_NAMESPACE , None ) : <EOL> lang = element . get ( "<STR_LIT>" % XML_NAMESPACE ) <EOL> ret = Literal ( text , datatype = datatype , lang = lang ) <EOL> return ret <EOL> elif tag == RESULTS_NS_ET + '<STR_LIT>' : <EOL> return URIRef ( text ) <EOL> elif tag == RESULTS_NS_ET + '<STR_LIT>' : <EOL> return BNode ( text ) <EOL> else : <EOL> raise TypeError ( "<STR_LIT>" % element ) <EOL> class XMLResultSerializer ( ResultSerializer ) : <EOL> def __init__ ( self , result ) : <EOL> ResultSerializer . __init__ ( self , result ) <EOL> def serialize ( self , stream , encoding = "<STR_LIT:utf-8>" ) : <EOL> writer = SPARQLXMLWriter ( stream , encoding ) <EOL> if self . result . type == '<STR_LIT>' : <EOL> writer . write_header ( [ ] ) <EOL> writer . write_ask ( self . result . askAnswer ) <EOL> else : <EOL> writer . write_header ( self . result . vars ) <EOL> writer . write_results_header ( ) <EOL> for b in self . result . bindings : <EOL> writer . write_start_result ( ) <EOL> for key , val in b . iteritems ( ) : <EOL> writer . write_binding ( key , val ) <EOL> writer . write_end_result ( ) <EOL> writer . close ( ) <EOL> class SPARQLXMLWriter : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , output , encoding = '<STR_LIT:utf-8>' ) : <EOL> writer = XMLGenerator ( output , encoding ) <EOL> writer . startDocument ( ) <EOL> writer . startPrefixMapping ( u'<STR_LIT>' , SPARQL_XML_NAMESPACE ) <EOL> writer . startPrefixMapping ( u'<STR_LIT>' , XML_NAMESPACE ) <EOL> writer . startElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT>' ) , <EOL> u'<STR_LIT>' , AttributesNSImpl ( { } , { } ) ) <EOL> self . writer = writer <EOL> self . _output = output <EOL> self . _encoding = encoding <EOL> self . _results = False <EOL> def write_header ( self , allvarsL ) : <EOL> self . writer . startElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT>' ) , <EOL> u'<STR_LIT>' , AttributesNSImpl ( { } , { } ) ) <EOL> for i in xrange ( <NUM_LIT:0> , len ( allvarsL ) ) : <EOL> attr_vals = { <EOL> ( None , u'<STR_LIT:name>' ) : unicode ( allvarsL [ i ] ) , <EOL> } <EOL> attr_qnames = { <EOL> ( None , u'<STR_LIT:name>' ) : u'<STR_LIT:name>' , <EOL> } <EOL> self . writer . startElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT>' ) , <EOL> u'<STR_LIT>' , AttributesNSImpl ( attr_vals , attr_qnames ) ) <EOL> self . writer . endElementNS ( ( SPARQL_XML_NAMESPACE , <EOL> u'<STR_LIT>' ) , u'<STR_LIT>' ) <EOL> self . writer . endElementNS ( ( SPARQL_XML_NAMESPACE , u'<STR_LIT>' ) , u'<STR_LIT>' ) <EOL> def write_ask ( self , val ) : <EOL> self . writer . startElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT>' ) , <EOL> u'<STR_LIT>' , AttributesNSImpl ( { } , { } ) ) <EOL> self . writer . characters ( str ( val ) . lower ( ) ) <EOL> self . writer . endElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT>' ) , u'<STR_LIT>' ) <EOL> def write_results_header ( self ) : <EOL> self . writer . startElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT>' ) , <EOL> u'<STR_LIT>' , AttributesNSImpl ( { } , { } ) ) <EOL> self . _results = True <EOL> def write_start_result ( self ) : <EOL> self . writer . startElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT:result>' ) , <EOL> u'<STR_LIT:result>' , AttributesNSImpl ( { } , { } ) ) <EOL> self . _resultStarted = True <EOL> def write_end_result ( self ) : <EOL> assert self . _resultStarted <EOL> self . writer . endElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT:result>' ) , u'<STR_LIT:result>' ) <EOL> self . _resultStarted = False <EOL> def write_binding ( self , name , val ) : <EOL> assert self . _resultStarted <EOL> attr_vals = { <EOL> ( None , u'<STR_LIT:name>' ) : unicode ( name ) , <EOL> } <EOL> attr_qnames = { <EOL> ( None , u'<STR_LIT:name>' ) : u'<STR_LIT:name>' , <EOL> } <EOL> self . writer . startElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT>' ) , <EOL> u'<STR_LIT>' , AttributesNSImpl ( attr_vals , attr_qnames ) ) <EOL> if isinstance ( val , URIRef ) : <EOL> self . writer . startElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT>' ) , <EOL> u'<STR_LIT>' , AttributesNSImpl ( { } , { } ) ) <EOL> self . writer . characters ( val ) <EOL> self . writer . endElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT>' ) , u'<STR_LIT>' ) <EOL> elif isinstance ( val , BNode ) : <EOL> self . writer . startElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT>' ) , <EOL> u'<STR_LIT>' , AttributesNSImpl ( { } , { } ) ) <EOL> self . writer . characters ( val ) <EOL> self . writer . endElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT>' ) , u'<STR_LIT>' ) <EOL> elif isinstance ( val , Literal ) : <EOL> attr_vals = { } <EOL> attr_qnames = { } <EOL> if val . language : <EOL> attr_vals [ ( XML_NAMESPACE , u'<STR_LIT>' ) ] = val . language <EOL> attr_qnames [ ( XML_NAMESPACE , u'<STR_LIT>' ) ] = u"<STR_LIT>" <EOL> elif val . datatype : <EOL> attr_vals [ ( None , u'<STR_LIT>' ) ] = val . datatype <EOL> attr_qnames [ ( None , u'<STR_LIT>' ) ] = u'<STR_LIT>' <EOL> self . writer . startElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT>' ) , <EOL> u'<STR_LIT>' , AttributesNSImpl ( attr_vals , attr_qnames ) ) <EOL> self . writer . characters ( val ) <EOL> self . writer . endElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT>' ) , u'<STR_LIT>' ) <EOL> else : <EOL> raise Exception ( "<STR_LIT>" % val ) <EOL> self . writer . endElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT>' ) , u'<STR_LIT>' ) <EOL> def close ( self ) : <EOL> if self . _results : <EOL> self . writer . endElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT>' ) , u'<STR_LIT>' ) <EOL> self . writer . endElementNS ( <EOL> ( SPARQL_XML_NAMESPACE , u'<STR_LIT>' ) , u'<STR_LIT>' ) <EOL> self . writer . endDocument ( ) </s>
