1
# Copyright (C) 2011 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
"""Tools for dealing with the Launchpad API without using launchpadlib.
19
The api itself is a RESTful interface, so we can make HTTP queries directly.
20
loading launchpadlib itself has a fairly high overhead (just calling
21
Launchpad.login_anonymously() takes a 500ms once the WADL is cached, and 5+s to
26
# Use simplejson if available, much faster, and can be easily installed in
27
# older versions of python
28
import simplejson as json
30
# Is present since python 2.6
45
class LatestPublication(object):
46
"""Encapsulate how to find the latest publication for a given project."""
48
LP_API_ROOT = 'https://api.launchpad.net/1.0'
50
def __init__(self, archive, series, project):
51
self._archive = archive
52
self._project = project
53
self._setup_series_and_pocket(series)
55
def _setup_series_and_pocket(self, series):
56
"""Parse the 'series' info into a series and a pocket.
59
_setup_series_and_pocket('natty-proposed')
65
if self._series is not None and '-' in self._series:
66
self._series, self._pocket = self._series.split('-', 1)
67
self._pocket = self._pocket.title()
69
self._pocket = 'Release'
71
def _archive_URL(self):
72
"""Return the Launchpad 'Archive' URL that we will query.
73
This is everything in the URL except the query parameters.
75
return '%s/%s/+archive/primary' % (self.LP_API_ROOT, self._archive)
77
def _publication_status(self):
78
"""Handle the 'status' field.
79
It seems that Launchpad tracks all 'debian' packages as 'Pending', while
80
for 'ubuntu' we care about the 'Published' packages.
82
if self._archive == 'debian':
83
# Launchpad only tracks debian packages as "Pending", it doesn't mark
88
def _query_params(self):
89
"""Get the parameters defining our query.
90
This defines the actions we are making against the archive.
91
:return: A dict of query parameters.
93
params = {'ws.op': 'getPublishedSources',
94
'exact_match': 'true',
95
# If we need to use "" shouldn't we quote the project somehow?
96
'source_name': '"%s"' % (self._project,),
97
'status': self._publication_status(),
98
# We only need the latest one, the results seem to be properly
99
# most-recent-debian-version sorted
102
if self._series is not None:
103
params['distro_series'] = '/%s/%s' % (self._archive, self._series)
104
if self._pocket is not None:
105
params['pocket'] = self._pocket
108
def _query_URL(self):
109
"""Create the full URL that we need to query, including parameters."""
110
params = self._query_params()
111
# We sort to give deterministic results for testing
112
encoded = urllib.urlencode(sorted(params.items()))
113
return '%s?%s' % (self._archive_URL(), encoded)
115
def _get_lp_info(self):
116
"""Place an actual HTTP query against the Launchpad service."""
119
query_URL = self._query_URL()
121
req = urllib2.Request(query_URL)
122
response = urllib2.urlopen(req)
123
json_info = response.read()
124
# TODO: We haven't tested the HTTPError
125
except (urllib2.URLError, urllib2.HTTPError), e:
126
trace.mutter('failed to place query to %r' % (query_URL,))
127
trace.log_exception_quietly()
131
def _parse_json_info(self, json_info):
132
"""Parse the json response from Launchpad into objects."""
136
return json.loads(json_info)
138
trace.mutter('Failed to parse json info: %r' % (json_info,))
139
trace.log_exception_quietly()
142
def get_latest_version(self):
143
"""Get the latest published version for the given package."""
144
json_info = self._get_lp_info()
145
if json_info is None:
147
info = self._parse_json_info(json_info)
151
entries = info['entries']
152
if len(entries) == 0:
154
return entries[0]['source_package_version']
156
trace.log_exception_quietly()
160
def get_latest_publication(archive, series, project):
161
"""Get the most recent publication for a given project.
163
:param archive: Either 'ubuntu' or 'debian'
164
:param series: Something like 'natty', 'sid', etc. Can be set as None. Can
165
also include a pocket such as 'natty-proposed'.
166
:param project: Something like 'bzr'
167
:return: A version string indicating the most-recent version published in
168
Launchpad. Might return None if there is an error.
170
lp = LatestPublication(archive, series, project)
171
return lp.get_latest_version()
174
def get_most_recent_tag(tag_dict, the_branch):
175
"""Get the most recent revision that has been tagged."""
176
# Note: this assumes that a given rev won't get tagged multiple times. But
177
# it should be valid for the package importer branches that we care
179
reverse_dict = dict((rev, tag) for tag, rev in tag_dict.iteritems())
180
the_branch.lock_read()
182
last_rev = the_branch.last_revision()
183
graph = the_branch.repository.get_graph()
184
stop_revisions = (None, revision.NULL_REVISION)
185
for rev_id in graph.iter_lefthand_ancestry(last_rev, stop_revisions):
186
if rev_id in reverse_dict:
187
return reverse_dict[rev_id]