diff options
Diffstat (limited to 'freebase-api')
| -rwxr-xr-x | freebase-api/freebase/__init__.py | 40 | ||||
| -rw-r--r-- | freebase-api/freebase/api/session.py | 141 | ||||
| -rwxr-xr-x | freebase-api/test/test_all.py | 124 |
3 files changed, 272 insertions, 33 deletions
diff --git a/freebase-api/freebase/__init__.py b/freebase-api/freebase/__init__.py index e69de29..3cca4ff 100755 --- a/freebase-api/freebase/__init__.py +++ b/freebase-api/freebase/__init__.py @@ -0,0 +1,40 @@ +import sys + +from freebase.api.session import HTTPMetawebSession +import sandbox + +__all__ = ["HTTPMetawebSession", "sandbox"] + +base = HTTPMetawebSession("freebase.com") + +# we want to add base's functions to __init__.py +# so that we can say freebase.func() and really +# just call base.func() + +# a little trick to refer to __init__ +# self isn't defined because __init__ is in +# a world in and of itself +self = sys.modules[__name__] + +for funcname in dir(base): + + # we only want the 'real' functions + if not funcname.startswith("_"): + func = getattr(base, funcname) + + # let's make sure we're getting functions + # instead of constants or whatever + if callable(func): + + # we're setting these functions + # so that they can be called like + # freebase.funcname -> base.func() + setattr(self, funcname, func) + + # make sure we import the base's + # functions if we import freebase + __all__.append(funcname) + +# we don't want any self-referencing +# business going. Plus, this is cleaner. +del self diff --git a/freebase-api/freebase/api/session.py b/freebase-api/freebase/api/session.py index e6b9c00..0d5159b 100644 --- a/freebase-api/freebase/api/session.py +++ b/freebase-api/freebase/api/session.py @@ -50,7 +50,7 @@ except ImportError: # appengine provides simplejson at django.utils.simplejson from django.utils import simplejson except ImportError: - raise Error("unable to import simplejson") + raise Exception("unable to import simplejson") try: from urllib import quote as urlquote except ImportError: @@ -249,11 +249,9 @@ class HTTPMetawebSession(MetawebSession): on whether urllib2 or httplib2 is in use? """ - if method == 'POST': - assert body is not None or form is not None - elif method == 'GET': + if method == 'GET': assert body is None - else: + if method != "GET" and method != "POST": assert 0, 'unknown method %s' % method url = self.service_url + service_path @@ -273,7 +271,7 @@ class HTTPMetawebSession(MetawebSession): assert ct is not None if form is not None: - qstr = '&'.join(['%s=%s' % (urlencode_weak(k), urlencode_weak(v)) + qstr = '&'.join(['%s=%s' % (urlencode_weak(unicode(k)), urlencode_weak(unicode(v))) for k,v in form.items()]) if method == 'POST': # put the args on the url if we're putting something else @@ -377,17 +375,23 @@ class HTTPMetawebSession(MetawebSession): - def login(self): - """sign in to the service""" + def login(self, username=None, password=None): + """sign in to the service. For a more complete description, + see http://www.freebase.com/view/en/api_account_login""" + + service = '/api/account/login' + + username = username or self.username + password = password or self.password - assert self.username is not None - assert self.password is not None + assert username is not None + assert password is not None - self.log.debug('LOGIN USERNAME: %s', self.username) + self.log.debug('LOGIN USERNAME: %s', username) - r = self._httpreq_json('/api/account/login', 'POST', - form=dict(username=self.username, - password=self.password)) + r = self._httpreq_json(service, 'POST', + form=dict(username=username, + password=password)) if r.code != '/api/status/ok': raise MetawebError(u'%s %r' % (r.get('code',''), r.messages)) @@ -395,9 +399,45 @@ class HTTPMetawebSession(MetawebSession): self.log.debug('LOGIN RESP: %r', r) self.log.debug('LOGIN COOKIES: %s', self.cookiejar) + def logout(self): + """logout of the service. For a more complete description, + see http://www.freebase.com/view/en/api_account_logout""" + + service = '/api/account/logout' + + self.log.debug("LOGOUT") + + r = self._httpreq_json(service, 'GET') + + if r.code != '/api/status/ok': + raise MetawebError(u'%s %r' % (r.get('code',''), r.messages)) #this should never happen + def user_info(self, mql_output=None): + """ get user_info. For a more complete description, + see http://www.freebase.com/view/en/api_service_user_info""" + + service = "/api/service/user_info" + + qstr = simplejson.dumps(mql_output) + + r = self._httpreq_json(service, 'POST', form=dict(mql_output=qstr)) + return r + + def loggedin(self): + """check to see whether a user is logged in or not. For a + more complete description, see http://www.freebase.com/view/en/api_account_loggedin""" + + service = "/api/account/loggedin" + try: + r = self._httpreq_json(service, 'GET') + if r.code == "/api/status/ok": + return True + + except MetawebError, me: + return False + def mqlreaditer(self, sq, asof=None): - """read a structure query""" + """read a structure query.""" cursor = True @@ -422,7 +462,8 @@ class HTTPMetawebSession(MetawebSession): return def mqlread(self, sq, asof=None): - """read a structure query""" + """read a structure query. For a more complete description, + see http://www.freebase.com/view/en/api_service_mqlread""" subq = dict(query=sq, escape=False) if asof: subq['as_of_time'] = asof @@ -470,20 +511,56 @@ class HTTPMetawebSession(MetawebSession): return [self._mqlresult(rs[key]) for key in keys] - def trans(self, guid): - """translate blob from guid """ - url = '/api/trans/raw' + urlquote(guid) + def raw(self, id): + """translate blob from id. For a more complete description, + see http://www.freebase.com/view/en/api_trans_raw""" + url = '/api/trans/raw' + urlquote(id) self.log.info(url) resp, body = self._httpreq(url) - self.log.info('%d bytes' % len(body)) + self.log.info('raw is %d bytes' % len(body)) return body + + def blurb(self, id, break_paragraphs=False, maxlength=200): + """translate only the text in blob from id. For a more + complete description, see http://www.freebase.com/view/en/api_trans_blurb""" + url = '/api/trans/blurb' + urlquote(id) + + self.log.info(url) + + resp, body = self._httpreq(url, form=dict(break_paragraphs=break_paragraphs, maxlength=maxlength)) + + self.log.info('blurb is %d bytes' % len(body)) + + return body + + def image_thumb(self, id, maxwidth=None, maxheight=None, mode="fit", onfail=None): + """ given the id of an image, this will return a URL of a thumbnail of the image. + The full details of how the image is cropped and finessed is detailed at + http://www.freebase.com/view/en/api_trans_image_thumb """ + + service = "/api/trans/image_thumb" + assert mode in ["fit", "fill", "fillcrop", "fillcropmid"] + + form = dict(mode=mode) + if maxwidth is not None: + form["maxwidth"] = maxwidth + if maxheight is not None: + form["maxheight"] = maxheight + if onfail is not None: + form["onfail"] = onfail + + resp, body = self._httpreq(service + urlquote(id), form=form) + self.log.info('image is %d bytes' % len(body)) + + return body def mqlwrite(self, sq): - """do a mql write""" + """do a mql write. For a more complete description, + see http://www.freebase.com/view/en/api_service_mqlwrite""" query = dict(query=sq, escape=False) qstr = simplejson.dumps(query) @@ -535,10 +612,14 @@ class HTTPMetawebSession(MetawebSession): return True def touch(self): + """ make sure you are accessing the most recent data. For a more + complete description, see http://www.freebase.com/view/en/api_service_touch""" return self.mqlflush() + def upload(self, body, content_type, document_id=False, permission_of=False): - """upload to the metaweb""" + """upload to the metaweb. For a more complete description, + see http://www.freebase.com/view/en/api_service_upload""" service = '/api/service/upload' @@ -568,8 +649,21 @@ class HTTPMetawebSession(MetawebSession): headers=headers, body=body, form=form) return self._mqlresult(r) + + def version(self): + """ get versions for various parts of freebase. For a more + complete description, see http://www.freebase.com/view/en/api_version""" + + service = "/api/version" + r = self._httpreq_json(service) + + self._check_mqlerror(r) + return r + + ### DEPRECATED IN API def reconcile(self, name, etype=['/common/topic']): - """reconcile name to guid""" + """DEPRECATED: reconcile name to guid. For a more complete description, + see http://www.freebase.com/view/en/dataserver_reconciliation""" service = '/dataserver/reconciliation' r = self._httpreq_json(service, 'GET', form={'name':name, 'types':','.join(etype)}) @@ -578,6 +672,7 @@ class HTTPMetawebSession(MetawebSession): # TODO non-conforming service, fix later #self._mqlresult(r) return r + if __name__ == '__main__': console = logging.StreamHandler() diff --git a/freebase-api/test/test_all.py b/freebase-api/test/test_all.py index f23dc14..1a9e124 100755 --- a/freebase-api/test/test_all.py +++ b/freebase-api/test/test_all.py @@ -28,8 +28,11 @@ import unittest -import sys -from freebase.api import HTTPMetawebSession +import sys, logging +import freebase +import random + +from freebase.api import HTTPMetawebSession, MetawebError USERNAME = 'username' @@ -39,28 +42,79 @@ TEST_QUERY = {'id': 'null', 'name': 'Sting'} class TestFreebase(unittest.TestCase): - def test_login(self): + def test_freebase_dot_login_logout(self): + freebase.login(username=USERNAME, password=PASSWORD) + self.assertNotEqual(freebase.user_info(), None) + self.assertEqual(freebase.loggedin(), True) + freebase.logout() + self.assertRaises(MetawebError, freebase.user_info) + self.assertEqual(freebase.loggedin(), False) + + def test_login_logout(self): mss = HTTPMetawebSession(API_HOST, username=USERNAME, password=PASSWORD) mss.login() + + user_info = mss.user_info() + self.assertNotEqual(None, user_info) + self.assertEqual(user_info.code, "/api/status/ok") + self.assertEqual(mss.loggedin(), True) + + mss.logout() + self.assertRaises(MetawebError, mss.user_info) + self.assertEqual(mss.loggedin(), False) + - # check mss._cookies? + def test_freebase_dot_read(self): + query = {'type':'/music/artist','guid':[{}],'name':'Sting', 'album':[{}]} + + result = freebase.mqlread(query) + + self.assertNotEqual(None, result) + self.assert_(result.has_key('guid')) + self.assert_(result.has_key('type')) + self.assert_(result.has_key('name')) + self.assert_(result.has_key('album')) + self.assertEqual(type([]), type(result['album'])) + self.assert_(len(result['album']) > 0) + self.assertEqual( 'Sting', result['name']) + self.assertEqual('#9202a8c04000641f8000000000092a01', result['guid'][0]['value']) + + def test_freebase_dot_write(self): + read_query = {'type':'/music/artist','name':'Yanni\'s Cousin Tom', 'id':{}} + + freebase.sandbox.login(username=USERNAME, password=PASSWORD) + result = freebase.sandbox.mqlread(read_query) + self.assertEqual(None, result) + + write_query = {'create':'unless_exists', 'type':'/music/artist','name':'Yanni'} + + write_result = freebase.sandbox.mqlwrite(write_query) + self.assertNotEqual(None, write_result) + self.assert_(write_result.has_key('create')) + self.assert_(write_result.has_key('type')) + self.assert_(write_result.has_key('name')) + self.assertEqual('existed', write_result['create']) + self.assertEqual('Yanni', write_result['name']) + self.assertEqual('/music/artist', write_result['type']) def test_read(self): - query = {'type':'/music/artist','id':[{}],'name':'Sting', 'album':[{}]} + query = {'type':'/music/artist','guid':[{}],'name':'Sting', 'album':[{}]} mss = HTTPMetawebSession(API_HOST) result = mss.mqlread(query) + self.assertNotEqual(None, result) - self.assert_(result.has_key('id')) + self.assert_(result.has_key('guid')) self.assert_(result.has_key('type')) self.assert_(result.has_key('name')) self.assert_(result.has_key('album')) - self.assertEqual([].__class__, result['album'].__class__) - self.assert_( result['album'].__len__() > 0) + self.assertEqual(type([]), type(result['album'])) + self.assert_(len(result['album']) > 0) self.assertEqual( 'Sting', result['name']) - self.assertEqual( '/guid/9202a8c04000641f8000000000092a01', result['id'][0]['value'], str(result['id'])) + self.assertEqual('#9202a8c04000641f8000000000092a01', result['guid'][0]['value']) + def test_write(self): read_query = {'type':'/music/artist','name':'Yanni\'s Cousin Tom', 'id':{}} @@ -81,6 +135,56 @@ class TestFreebase(unittest.TestCase): self.assertEqual('Yanni', write_result['name']) self.assertEqual('/music/artist', write_result['type']) + def test_trans_blurb(self): + kurt = "/en/kurt_vonnegut" + + blurb = freebase.blurb(kurt) + self.assert_(blurb.startswith("Kurt Vonnegut")) + self.assertNotEqual(len(blurb), 0) + + blurb14 = freebase.blurb(kurt, maxlength=14) + blurb57 = freebase.blurb(kurt, maxlength=57) + self.assertNotEqual(len(blurb14), len(blurb57)) + + blurbpar = freebase.blurb(kurt, break_paragraphs=True, maxlength=20000) + blurbnopar = freebase.blurb(kurt, break_paragraphs=False, maxlength=20000) + # self.assertNotEqual(blurbpar, blurbnopar) this doesn't work unless I get a good example + # of an article with paragraphs. + + def test_trans_raw(self): + kurt = "/en/kurt_vonnegut" + + self.assertRaises(MetawebError, lambda: freebase.raw(kurt)) + + r = freebase.mqlread({"id":kurt, "/common/topic/article":[{"id":None, "optional":True, "limit":1}]}) + raw = freebase.raw(r["/common/topic/article"][0].id) + self.assertNotEqual(len(raw), 0) + + def test_trans_image_thumb(self): + kurt = "/en/kurt_vonnegut" + + r = freebase.mqlread({"id":kurt, "/common/topic/image":[{"id":None, "optional":True, "limit":1}]}) + imageid = r["/common/topic/image"][0].id + rawimage = freebase.raw(imageid) + + thumbedimage = freebase.image_thumb(imageid, maxheight=99) + self.assertNotEqual(rawimage, thumbedimage) + + def test_upload(self): + my_text = "Kurt Vonnegut was an author! " + str(random.random()) + + freebase.sandbox.login(USERNAME, PASSWORD) + response = freebase.sandbox.upload(my_text, "text/plain") + + self.assertEqual(freebase.sandbox.raw(response.id), my_text) + # since it's text/plain, blurb should also be equal + self.assertEqual(freebase.sandbox.blurb(response.id), my_text) + + if __name__ == '__main__': - unittest.main() + if USERNAME == "username" and PASSWORD == "password": + print "In order to run the tests, we need to use a valid freebase username and password" + USERNAME = raw_input("Please enter your username: ") + PASSWORD = raw_input("Please enter your password (it'll appear in cleartext): ") + unittest.main() |