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
25
from __future__ import absolute_import
28
# Use simplejson if available, much faster, and can be easily installed in
29
# older versions of python
30
import simplejson as json
32
# Is present since python 2.6
48
class LatestPublication(object):
49
"""Encapsulate how to find the latest publication for a given project."""
51
LP_API_ROOT = 'https://api.launchpad.net/1.0'
53
def __init__(self, archive, series, project):
54
self._archive = archive
55
self._project = project
56
self._setup_series_and_pocket(series)
58
def _setup_series_and_pocket(self, series):
59
"""Parse the 'series' info into a series and a pocket.
62
_setup_series_and_pocket('natty-proposed')
68
if self._series is not None and '-' in self._series:
69
self._series, self._pocket = self._series.split('-', 1)
70
self._pocket = self._pocket.title()
72
self._pocket = 'Release'
74
def _archive_URL(self):
75
"""Return the Launchpad 'Archive' URL that we will query.
76
This is everything in the URL except the query parameters.
78
return '%s/%s/+archive/primary' % (self.LP_API_ROOT, self._archive)
80
def _publication_status(self):
81
"""Handle the 'status' field.
82
It seems that Launchpad tracks all 'debian' packages as 'Pending', while
83
for 'ubuntu' we care about the 'Published' packages.
85
if self._archive == 'debian':
86
# Launchpad only tracks debian packages as "Pending", it doesn't mark
91
def _query_params(self):
92
"""Get the parameters defining our query.
93
This defines the actions we are making against the archive.
94
:return: A dict of query parameters.
96
params = {'ws.op': 'getPublishedSources',
97
'exact_match': 'true',
98
# If we need to use "" shouldn't we quote the project somehow?
99
'source_name': '"%s"' % (self._project,),
100
'status': self._publication_status(),
101
# We only need the latest one, the results seem to be properly
102
# most-recent-debian-version sorted
105
if self._series is not None:
106
params['distro_series'] = '/%s/%s' % (self._archive, self._series)
107
if self._pocket is not None:
108
params['pocket'] = self._pocket
111
def _query_URL(self):
112
"""Create the full URL that we need to query, including parameters."""
113
params = self._query_params()
114
# We sort to give deterministic results for testing
115
encoded = urllib.urlencode(sorted(params.items()))
116
return '%s?%s' % (self._archive_URL(), encoded)
118
def _get_lp_info(self):
119
"""Place an actual HTTP query against the Launchpad service."""
122
query_URL = self._query_URL()
124
req = urllib2.Request(query_URL)
125
response = urllib2.urlopen(req)
126
json_info = response.read()
127
# TODO: We haven't tested the HTTPError
128
except (urllib2.URLError, urllib2.HTTPError), e:
129
trace.mutter('failed to place query to %r' % (query_URL,))
130
trace.log_exception_quietly()
134
def _parse_json_info(self, json_info):
135
"""Parse the json response from Launchpad into objects."""
139
return json.loads(json_info)
141
trace.mutter('Failed to parse json info: %r' % (json_info,))
142
trace.log_exception_quietly()
145
def get_latest_version(self):
146
"""Get the latest published version for the given package."""
147
json_info = self._get_lp_info()
148
if json_info is None:
150
info = self._parse_json_info(json_info)
154
entries = info['entries']
155
if len(entries) == 0:
157
return entries[0]['source_package_version']
159
trace.log_exception_quietly()
163
"""Text-form for what location this represents.
166
ubuntu, natty => Ubuntu Natty
167
ubuntu, natty-proposed => Ubuntu Natty Proposed
168
:return: A string representing the location we are checking.
170
place = self._archive
171
if self._series is not None:
172
place = '%s %s' % (place, self._series)
173
if self._pocket is not None and self._pocket != 'Release':
174
place = '%s %s' % (place, self._pocket)
178
def get_latest_publication(archive, series, project):
179
"""Get the most recent publication for a given project.
181
:param archive: Either 'ubuntu' or 'debian'
182
:param series: Something like 'natty', 'sid', etc. Can be set as None. Can
183
also include a pocket such as 'natty-proposed'.
184
:param project: Something like 'bzr'
185
:return: A version string indicating the most-recent version published in
186
Launchpad. Might return None if there is an error.
188
lp = LatestPublication(archive, series, project)
189
return lp.get_latest_version()
192
def get_most_recent_tag(tag_dict, the_branch):
193
"""Get the most recent revision that has been tagged."""
194
# Note: this assumes that a given rev won't get tagged multiple times. But
195
# it should be valid for the package importer branches that we care
197
reverse_dict = dict((rev, tag) for tag, rev in tag_dict.iteritems())
198
the_branch.lock_read()
200
last_rev = the_branch.last_revision()
201
graph = the_branch.repository.get_graph()
202
stop_revisions = (None, revision.NULL_REVISION)
203
for rev_id in graph.iter_lefthand_ancestry(last_rev, stop_revisions):
204
if rev_id in reverse_dict:
205
return reverse_dict[rev_id]
210
def _get_newest_versions(the_branch, latest_pub):
211
"""Get information about how 'fresh' this packaging branch is.
213
:param the_branch: The Branch to check
214
:param latest_pub: The LatestPublication used to check most recent
216
:return: (latest_ver, branch_latest_ver)
219
latest_ver = latest_pub.get_latest_version()
220
t_latest_ver = time.time() - t
221
trace.mutter('LatestPublication.get_latest_version took: %.3fs'
223
if latest_ver is None:
226
tags = the_branch.tags.get_tag_dict()
227
t_tag_dict = time.time() - t
228
trace.mutter('LatestPublication.get_tag_dict took: %.3fs' % (t_tag_dict,))
229
if latest_ver in tags:
230
# branch might have a newer tag, but we don't really care
231
return latest_ver, latest_ver
233
best_tag = get_most_recent_tag(tags, the_branch)
234
return latest_ver, best_tag
237
def _report_freshness(latest_ver, branch_latest_ver, place, verbosity,
239
"""Report if the branch is up-to-date."""
240
if latest_ver is None:
241
if verbosity == 'all':
242
report_func('Most recent %s version: MISSING' % (place,))
243
elif verbosity == 'short':
244
report_func('%s is MISSING a version' % (place,))
246
elif latest_ver == branch_latest_ver:
247
if verbosity == 'minimal':
249
elif verbosity == 'short':
250
report_func('%s is CURRENT in %s' % (latest_ver, place))
252
report_func('Most recent %s version: %s\n'
253
'Packaging branch status: CURRENT'
254
% (place, latest_ver))
256
if verbosity in ('minimal', 'short'):
257
if branch_latest_ver is None:
258
branch_latest_ver = 'Branch'
259
report_func('%s is OUT-OF-DATE, %s has %s'
260
% (branch_latest_ver, place, latest_ver))
262
report_func('Most recent %s version: %s\n'
263
'Packaging branch version: %s\n'
264
'Packaging branch status: OUT-OF-DATE'
265
% (place, latest_ver, branch_latest_ver))
268
def report_freshness(the_branch, verbosity, latest_pub):
269
"""Report to the user how up-to-date the packaging branch is.
271
:param the_branch: A Branch object
272
:param verbosity: Can be one of:
273
off: Do not print anything, and skip all checks.
274
all: Print all information that we have in a verbose manner, this
275
includes misses, etc.
276
short: Print information, but only one-line summaries
277
minimal: Only print a one-line summary when the package branch is
279
:param latest_pub: A LatestPublication instance
281
if verbosity == 'off':
283
if verbosity is None:
285
latest_ver, branch_ver = _get_newest_versions(the_branch, latest_pub)
286
place = latest_pub.place()
287
_report_freshness(latest_ver, branch_ver, place, verbosity,
added
removed
bzrlib/plugins/launchpad/lp_api_lite.py
