One of the nice things about work­ing at Canon­i­cal is that we pro­duce open source soft­ware. I, specif­i­cal­ly, work in the team that does the desk­top clients for Ubun­tu One which is a re­al­ly cool job, and a re­al­ly cool piece of soft­ware. How­ev­er, one thing not enough peo­ple know, is that we of­fer damn nice APIs for de­vel­op­er­s. We have to, since all our client code is open source, so we need those APIs for our­selves.

So, here is a small tu­to­ri­al about us­ing some of those APIs. I did it us­ing Python and PyQt for sev­er­al rea­son­s:

• Both are great tools for pro­­to­­typ­ing

• Both have good sup­­port for the re­quired stuff (D­Bus, HTTP, OAu­th)

• It's what I know and en­joy. Since I did this code on a sun­­day, I am not go­ing to use oth­­er things.

Hav­ing said that, there is noth­ing python-spe­cif­ic or Qt-spe­cif­ic in the code. Where I do a HTTP re­quest us­ing Qt­Net­work, you are free to use lib­soup, or what­ev­er.

So, on to the nuts and bolt­s. The main pieces of Ubun­tu One, from a in­fra­struc­ture per­spec­tive, are Ubun­tu SSO Clien­t, that han­dles us­er reg­is­tra­tion and login, and Sync­Dae­mon, which han­dles file syn­chro­niza­tion.

To in­ter­act with them, on Lin­ux, they of­fer DBus in­ter­faces. So, for ex­am­ple, this is a frag­ment of code show­ing a way to get the Ubun­tu One cre­den­tials (this would nor­mal­ly be part of an ob­jec­t'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 no­ticed that get_­cre­den­tials does­n't ac­tu­al­ly re­turn the cre­den­tial­s. What it does is, it tells Sync­Dae­mon to fetch the cre­den­tial­s, and then, when/if they are there, one of the sig­nals will be emit­ted, and one of the con­nect­ed meth­ods will be called. This is nice, be­cause it means you don't have to wor­ry about your app block­ing while Sync­Dae­mon is do­ing all this.

But what's in those meth­ods we used? Not much, re­al­ly!

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, ba­si­cal­ly, self­._­cre­den­tials will hold a set of cre­den­tial­s, or None. Con­grat­u­la­tion­s, we are now logged in­to Ubun­tu One, so to speak.

So, let's do some­thing use­ful! How about ask­ing for how much free space there is in the ac­coun­t? For that, we can't use the lo­cal APIs, we have to con­nect to the server­s, who are, af­ter al­l, the ones who de­cide if you are over quo­ta or not.

Ac­cess is con­trolled via OAuth. So, to ac­cess the API, we need to sign our re­quest­s. Here is how it's done. It's not par­tic­u­lar­ly en­light­en­ing, 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 quo­ta us­age? By ac­cess­ing the http­s://one.ubun­tu.­com/api/quo­ta/ en­try point with the prop­er au­tho­riza­tion, we would get a JSON dic­tio­nary with to­tal and used space. So, here's a sim­ple 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_quo­ta does­n't re­turn the quo­ta? What hap­pens is that get_quo­ta will launch a HTTP re­quest to the Ubun­tu One server­s, which will, even­tu­al­ly, re­ply with the da­ta. You don't want your app to block while you do that. So, QNet­workAc­cess­Man­ag­er will call self­.re­ply_fin­ished when it gets the re­spon­se:

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 get­ting a call when­ev­er the sta­tus of sync­dae­mon changes? For ex­am­ple, when sync is up to date, or when you get dis­con­nect­ed? Again, those are DBus sig­nals we are con­nect­ing 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 sta­tus_changed?

def status_changed(self, status):
    print "New status:", status
    self._last_status = self.process_status(status)
    self.update_menu()

The pro­cess_s­ta­tus func­tion is bor­ing code to con­vert the in­fo from sync­dae­mon's sta­tus in­to a hu­man-read­able thing like "Sync is up­-­to-­date". So we store that in self­._last_s­ta­tus and up­date the menu.

What menu? Well, a QSys­tem­Tray­I­con's con­text menu! What you have read are the main pieces you need to cre­ate some­thing use­ful: a Ubun­tu One tray app you can use in KDE, XFCE or open­box. Or, if you are on uni­ty and in­stall sni-qt, a Ubun­tu One app in­di­ca­tor!

You can find the source code for the whole ex­am­ple app at my u1-­toys project in launch­pad and here is the full source code (miss­ing some icon re­sources, just get the re­po)

Com­ing soon(ish), more ex­am­ple app­s, and cool things to do with our APIs!