Viewing file:  Util.py (6.75 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
########################################################################
# $Header: /var/local/cvsroot/4Suite/Ft/Rdf/Util.py,v 1.19 2005/03/04 12:54:33 mbrown Exp $
"""
Some useful utilities for RDF processing
Copyright 2005 Fourthought, Inc. (USA).
Detailed license and copyright information: http://4suite.org/COPYRIGHT
Project home, documentation, distributions: http://4suite.org/
"""
__all__ = ['SerializeStatementList', 'SerializeModel',
'GetSubject', 'GetSubjects',
'GetPredicate', 'GetPredicates',
'GetObject', 'GetObjects',
'DeserializeFromUri', 'DeserializeFromString',
'DeserializeFromNode',
'VersaQuery', 'VersaDataToXml']
import cStringIO
from Ft.Rdf import Model
from Ft.Rdf.Drivers import Memory
from Ft.Rdf.Serializers.Dom import Serializer
def SerializeStatementList(slist):
db = Memory.CreateDb('')
m = Model.Model(db)
m.add(slist)
return SerializeModel(m)
def SerializeModel(m):
from Ft.Xml.Domlette import PrettyPrint
serializer = Serializer()
outdoc = serializer.serialize(m)
strstream = cStringIO.StringIO()
PrettyPrint(outdoc, stream=strstream)
return strstream.getvalue()
#Query convenience functions
def GetSubject(model, predicate, object, statementUri=None,
scope=None, **flags):
"""
Performs a complete on the given model, using the given object and
predicate, None for the subject, with any given optional parameters
passed through. The first resulting statement is returned, and the
subject extracted therefrom. If no statements result, None is returned
"""
kwargs = flags.copy()
if statementUri is not None:
kwargs.update({'statementUri': statementUri})
if scope is not None:
kwargs.update({'scope': scope})
stmts = apply(model.complete, (None, predicate, object), kwargs)
if not stmts: return None
return stmts[0].subject
def GetSubjects(model, predicate, object, statementUri=None,
scope=None, **flags):
"""
Performs a complete on the given model, using the given object and
predicate, None for the subject, with any given optional parameters
passed through. The list of subjects of the resulting statement is
returned. If no statements result, the empty list is returned
"""
kwargs = flags.copy()
if statementUri is not None:
kwargs.update({'statementUri': statementUri})
if scope is not None:
kwargs.update({'scope': scope})
stmts = apply(model.complete, (None, predicate, object), kwargs)
return map(lambda x: x.subject, stmts)
def GetPredicate(model, subject, object, statementUri=None,
scope=None, **flags):
"""
Performs a complete on the given model, using the given subject and
object, None for the predicate, with any given optional parameters
passed through. The first resulting statement is returned, and the
predicate extracted therefrom. If no statements result, None is returned
"""
kwargs = flags.copy()
if statementUri is not None:
kwargs.update({'statementUri': statementUri})
if scope is not None:
kwargs.update({'scope': scope})
stmts = apply(model.complete, (subject, object, None), kwargs)
if not stmts: return None
return stmts[0].predicate
def GetPredicates(model, subject, object, statementUri=None,
scope=None, **flags):
"""
Performs a complete on the given model, using the given subject and
predicate, None for the object, with any given optional parameters
passed through. The list of predicates of the resulting statement is
returned. If no statements result, the empty list is returned
"""
kwargs = flags.copy()
if statementUri is not None:
kwargs.update({'statementUri': statementUri})
if scope is not None:
kwargs.update({'scope': scope})
stmts = apply(model.complete, (subject, object, None), kwargs)
return map(lambda x: x.predicate, stmts)
def GetObject(model, subject, predicate, statementUri=None,
scope=None, **flags):
"""
Performs a complete on the given model, using the given subject and
predicate, None for the object, with any given optional parameters
passed through. The first resulting statement is returned, and the
object extracted therefrom. If no statements result, None is returned
"""
kwargs = flags.copy()
if statementUri is not None:
kwargs.update({'statementUri': statementUri})
if scope is not None:
kwargs.update({'scope': scope})
stmts = apply(model.complete, (subject, predicate, None), kwargs)
if not stmts: return None
return stmts[0].object
def GetObjects(model, subject, predicate, statementUri=None,
scope=None, **flags):
"""
Performs a complete on the given model, using the given subject and
predicate, None for the object, with any given optional parameters
passed through. The list of objects of the resulting statement is
returned. If no statements result, the empty list is returned
"""
kwargs = flags.copy()
if statementUri is not None:
kwargs.update({'statementUri': statementUri})
if scope is not None:
kwargs.update({'scope': scope})
stmts = apply(model.complete, (subject, predicate, None), kwargs)
return map(lambda x: x.object, stmts)
#Setting up models
def DeserializeFromUri(uri, driver=Memory, dbName='', create=False, scope=None):
from Ft.Xml.Domlette import NonvalidatingReader
doc = NonvalidatingReader.parseUri(uri)
model, db = DeserializeFromNode(doc, driver, dbName, create, scope)
return model, db
def DeserializeFromString(st, driver=Memory, dbName='', create=False, scope=None):
from Ft.Xml.Domlette import NonvalidatingReader
doc = NonvalidatingReader.parseString(st,scope)
model, db = DeserializeFromNode(doc, driver, dbName, create, scope)
return model, db
def DeserializeFromNode(node, driver=Memory, dbName='', create=False, scope=None,
modelName='default'):
driver.InitializeModule()
if create:
db = driver.CreateDb(dbName, modelName)
else:
db = driver.GetDb(dbName, modelName)
db.begin()
model = Model.Model(db)
serializer = Serializer()
serializer.deserialize(model, node, scope)
db.commit()
return model, db
#Querying
def VersaQuery(query, model, nsMap=None, scope=None):
from Ft.Rdf.Parsers import Versa
nsMap = nsMap or {}
con = Versa.CreateContext(model=model, nsMapping=nsMap, scope=scope)
exp = Versa.Compile(query)
results = exp.evaluate(con)
return results
def VersaDataToXml(data):
import Ft.Rdf.Parsers.Versa.Util
s = cStringIO.StringIO()
Ft.Rdf.Parsers.Versa.Util.ResultsToXml(data, s)
return s.getvalue()
|