Ubuntu One APIs by Example (part 1)
One of the nice things about working at Canonical is that we produce open source software. I, specifically, work in the team that does the desktop clients for Ubuntu One which is a really cool job, and a really cool piece of software. However, one thing not enough people know, is that we offer damn nice APIs for developers. We have to, since all our client code is open source, so we need those APIs for ourselves.
So, here is a small tutorial about using some of those APIs. I did it using Python and PyQt for several reasons:
- Both are great tools for prototyping
- Both have good support for the required stuff (DBus, HTTP, OAuth)
- It's what I know and enjoy. Since I did this code on a sunday, I am not going to use other things.
Having said that, there is nothing python-specific or Qt-specific in the code. Where I do a HTTP request using QtNetwork, you are free to use libsoup, or whatever.
So, on to the nuts and bolts. The main pieces of Ubuntu One, from a infrastructure perspective, are Ubuntu SSO Client, that handles user registration and login, and SyncDaemon, which handles file synchronization.
To interact with them, on Linux, they offer DBus interfaces. So, for example, this is a fragment of code showing a way to get the Ubuntu One credentials (this would normally be part of an object's __init__):
# Get the session bus bus = dbus.SessionBus() : : : # Get the credentials proxy and interface self.creds_proxy = bus.get_object("com.ubuntuone.Credentials", "/credentials", follow_name_owner_changes=True) # Connect to signals so you get a call when something # credential-related happens self.creds_iface = dbus.Interface(self.creds_proxy, "com.ubuntuone.CredentialsManagement") self.creds_proxy.connect_to_signal('CredentialsFound', self.creds_found) self.creds_proxy.connect_to_signal('CredentialsNotFound', self.creds_not_found) self.creds_proxy.connect_to_signal('CredentialsError', self.creds_error) # Call for credentials self._credentials = None self.get_credentials()
You may have noticed that get_credentials doesn't actually return the credentials. What it does is, it tells SyncDaemon to fetch the credentials, and then, when/if they are there, one of the signals will be emitted, and one of the connected methods will be called. This is nice, because it means you don't have to worry about your app blocking while SyncDaemon is doing all this.
But what's in those methods we used? Not much, really!
def get_credentials(self): # Do we have them already? If not, get'em if not self._credentials: self.creds_proxy.find_credentials() # Return what we've got, could be None return self._credentials def creds_found(self, data): # Received credentials, save them. print "creds_found", data self._credentials = data # Don't worry about get_quota yet ;-) if not self._quota_info: self.get_quota() def creds_not_found(self, data): # No credentials, remove old ones. print "creds_not_found", data self._credentials = None def creds_error(self, data): # No credentials, remove old ones. print "creds_error", data self._credentials = None
So, basically, self._credentials will hold a set of credentials, or None. Congratulations, we are now logged into Ubuntu One, so to speak.
So, let's do something useful! How about asking for how much free space there is in the account? For that, we can't use the local APIs, we have to connect to the servers, who are, after all, the ones who decide if you are over quota or not.
Access is controlled via OAuth. So, to access the API, we need to sign our requests. Here is how it's done. It's not particularly enlightening, and I did not write it, I just use it:
def sign_uri(self, uri, parameters=None): # Without credentials, return unsigned URL if not self._credentials: return uri if isinstance(uri, unicode): uri = bytes(iri2uri(uri)) print "uri:", uri method = "GET" credentials = self._credentials consumer = oauth.OAuthConsumer(credentials["consumer_key"], credentials["consumer_secret"]) token = oauth.OAuthToken(credentials["token"], credentials["token_secret"]) if not parameters: _, _, _, _, query, _ = urlparse(uri) parameters = dict(cgi.parse_qsl(query)) request = oauth.OAuthRequest.from_consumer_and_token( http_url=uri, http_method=method, parameters=parameters, oauth_consumer=consumer, token=token) sig_method = oauth.OAuthSignatureMethod_HMAC_SHA1() request.sign_request(sig_method, consumer, token) print "SIGNED:", repr(request.to_url()) return request.to_url()
And how do we ask for the quota usage? By accessing the https://one.ubuntu.com/api/quota/ entry point with the proper authorization, we would get a JSON dictionary with total and used space. So, here's a simple way to do it:
# This is on __init__ self.nam = QtNetwork.QNetworkAccessManager(self, finished=self.reply_finished) : : : def get_quota(self): """Launch quota info request.""" uri = self.sign_uri(QUOTA_API) url = QtCore.QUrl() url.setEncodedUrl(uri) self.nam.get(QtNetwork.QNetworkRequest(url))
Again, see how get_quota doesn't return the quota? What happens is that get_quota will launch a HTTP request to the Ubuntu One servers, which will, eventually, reply with the data. You don't want your app to block while you do that. So, QNetworkAccessManager will call self.reply_finished when it gets the response:
def reply_finished(self, reply): if unicode(reply.url().path()) == u'/api/quota/': # Handle quota responses self._quota_info = json.loads(unicode(reply.readAll())) print "Got quota: ", self._quota_info # Again, don't worry about update_menu yet ;-) self.update_menu()
What else would be nice to have? How about getting a call whenever the status of syncdaemon changes? For example, when sync is up to date, or when you get disconnected? Again, those are DBus signals we are connecting in our __init__:
self.status_proxy = bus.get_object( 'com.ubuntuone.SyncDaemon', '/status') self.status_iface = dbus.Interface(self.status_proxy, dbus_interface='com.ubuntuone.SyncDaemon.Status') self.status_iface.connect_to_signal( 'StatusChanged', self.status_changed) # Get the status as of right now self._last_status = self.process_status( self.status_proxy.current_status())
And what's status_changed?
def status_changed(self, status): print "New status:", status self._last_status = self.process_status(status) self.update_menu()
The process_status function is boring code to convert the info from syncdaemon's status into a human-readable thing like "Sync is up-to-date". So we store that in self._last_status and update the menu.
What menu? Well, a QSystemTrayIcon's context menu! What you have read are the main pieces you need to create something useful: a Ubuntu One tray app you can use in KDE, XFCE or openbox. Or, if you are on unity and install sni-qt, a Ubuntu One app indicator!
My Ubuntu One indicator in action.
You can find the source code for the whole example app at my u1-toys project in launchpad and here is the full source code (missing some icon resources, just get the repo)
Coming soon(ish), more example apps, and cool things to do with our APIs!