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
40
from urllib.parse import urlencode
41
except ImportError: # python < 3
42
from urllib import urlencode
44
import urllib.request as urllib2
45
except ImportError: # python < 3
54
class LatestPublication(object):
55
"""Encapsulate how to find the latest publication for a given project."""
57
LP_API_ROOT = 'https://api.launchpad.net/1.0'
59
def __init__(self, archive, series, project):
60
self._archive = archive
61
self._project = project
62
self._setup_series_and_pocket(series)
64
def _setup_series_and_pocket(self, series):
65
"""Parse the 'series' info into a series and a pocket.
68
_setup_series_and_pocket('natty-proposed')
74
if self._series is not None and '-' in self._series:
75
self._series, self._pocket = self._series.split('-', 1)
76
self._pocket = self._pocket.title()
78
self._pocket = 'Release'
80
def _archive_URL(self):
81
"""Return the Launchpad 'Archive' URL that we will query.
82
This is everything in the URL except the query parameters.
84
return '%s/%s/+archive/primary' % (self.LP_API_ROOT, self._archive)
86
def _publication_status(self):
87
"""Handle the 'status' field.
88
It seems that Launchpad tracks all 'debian' packages as 'Pending', while
89
for 'ubuntu' we care about the 'Published' packages.
91
if self._archive == 'debian':
92
# Launchpad only tracks debian packages as "Pending", it doesn't mark
97
def _query_params(self):
98
"""Get the parameters defining our query.
99
This defines the actions we are making against the archive.
100
:return: A dict of query parameters.
102
params = {'ws.op': 'getPublishedSources',
103
'exact_match': 'true',
104
# If we need to use "" shouldn't we quote the project somehow?
105
'source_name': '"%s"' % (self._project,),
106
'status': self._publication_status(),
107
# We only need the latest one, the results seem to be properly
108
# most-recent-debian-version sorted
111
if self._series is not None:
112
params['distro_series'] = '/%s/%s' % (self._archive, self._series)
113
if self._pocket is not None:
114
params['pocket'] = self._pocket
117
def _query_URL(self):
118
"""Create the full URL that we need to query, including parameters."""
119
params = self._query_params()
120
# We sort to give deterministic results for testing
121
encoded = urlencode(sorted(params.items()))
122
return '%s?%s' % (self._archive_URL(), encoded)
124
def _get_lp_info(self):
125
"""Place an actual HTTP query against the Launchpad service."""
128
query_URL = self._query_URL()
130
req = urllib2.Request(query_URL)
131
response = urllib2.urlopen(req)
132
json_info = response.read()
133
# TODO: We haven't tested the HTTPError
134
except (urllib2.URLError, urllib2.HTTPError) as e:
135
trace.mutter('failed to place query to %r' % (query_URL,))
136
trace.log_exception_quietly()
140
def _parse_json_info(self, json_info):
141
"""Parse the json response from Launchpad into objects."""
145
return json.loads(json_info)
147
trace.mutter('Failed to parse json info: %r' % (json_info,))
148
trace.log_exception_quietly()
151
def get_latest_version(self):
152
"""Get the latest published version for the given package."""
153
json_info = self._get_lp_info()
154
if json_info is None:
156
info = self._parse_json_info(json_info)
160
entries = info['entries']
161
if len(entries) == 0:
163
return entries[0]['source_package_version']
165
trace.log_exception_quietly()
169
"""Text-form for what location this represents.
172
ubuntu, natty => Ubuntu Natty
173
ubuntu, natty-proposed => Ubuntu Natty Proposed
174
:return: A string representing the location we are checking.
176
place = self._archive
177
if self._series is not None:
178
place = '%s %s' % (place, self._series)
179
if self._pocket is not None and self._pocket != 'Release':
180
place = '%s %s' % (place, self._pocket)
184
def get_latest_publication(archive, series, project):
185
"""Get the most recent publication for a given project.
187
:param archive: Either 'ubuntu' or 'debian'
188
:param series: Something like 'natty', 'sid', etc. Can be set as None. Can
189
also include a pocket such as 'natty-proposed'.
190
:param project: Something like 'bzr'
191
:return: A version string indicating the most-recent version published in
192
Launchpad. Might return None if there is an error.
194
lp = LatestPublication(archive, series, project)
195
return lp.get_latest_version()
198
def get_most_recent_tag(tag_dict, the_branch):
199
"""Get the most recent revision that has been tagged."""
200
# Note: this assumes that a given rev won't get tagged multiple times. But
201
# it should be valid for the package importer branches that we care
203
reverse_dict = dict((rev, tag) for tag, rev in tag_dict.items())
204
with the_branch.lock_read():
205
last_rev = the_branch.last_revision()
206
graph = the_branch.repository.get_graph()
207
stop_revisions = (None, revision.NULL_REVISION)
208
for rev_id in graph.iter_lefthand_ancestry(last_rev, stop_revisions):
209
if rev_id in reverse_dict:
210
return reverse_dict[rev_id]
213
def _get_newest_versions(the_branch, latest_pub):
214
"""Get information about how 'fresh' this packaging branch is.
216
:param the_branch: The Branch to check
217
:param latest_pub: The LatestPublication used to check most recent
219
:return: (latest_ver, branch_latest_ver)
222
latest_ver = latest_pub.get_latest_version()
223
t_latest_ver = time.time() - t
224
trace.mutter('LatestPublication.get_latest_version took: %.3fs'
226
if latest_ver is None:
229
tags = the_branch.tags.get_tag_dict()
230
t_tag_dict = time.time() - t
231
trace.mutter('LatestPublication.get_tag_dict took: %.3fs' % (t_tag_dict,))
232
if latest_ver in tags:
233
# branch might have a newer tag, but we don't really care
234
return latest_ver, latest_ver
236
best_tag = get_most_recent_tag(tags, the_branch)
237
return latest_ver, best_tag
240
def _report_freshness(latest_ver, branch_latest_ver, place, verbosity,
242
"""Report if the branch is up-to-date."""
243
if latest_ver is None:
244
if verbosity == 'all':
245
report_func('Most recent %s version: MISSING' % (place,))
246
elif verbosity == 'short':
247
report_func('%s is MISSING a version' % (place,))
249
elif latest_ver == branch_latest_ver:
250
if verbosity == 'minimal':
252
elif verbosity == 'short':
253
report_func('%s is CURRENT in %s' % (latest_ver, place))
255
report_func('Most recent %s version: %s\n'
256
'Packaging branch status: CURRENT'
257
% (place, latest_ver))
259
if verbosity in ('minimal', 'short'):
260
if branch_latest_ver is None:
261
branch_latest_ver = 'Branch'
262
report_func('%s is OUT-OF-DATE, %s has %s'
263
% (branch_latest_ver, place, latest_ver))
265
report_func('Most recent %s version: %s\n'
266
'Packaging branch version: %s\n'
267
'Packaging branch status: OUT-OF-DATE'
268
% (place, latest_ver, branch_latest_ver))
271
def report_freshness(the_branch, verbosity, latest_pub):
272
"""Report to the user how up-to-date the packaging branch is.
274
:param the_branch: A Branch object
275
:param verbosity: Can be one of:
276
off: Do not print anything, and skip all checks.
277
all: Print all information that we have in a verbose manner, this
278
includes misses, etc.
279
short: Print information, but only one-line summaries
280
minimal: Only print a one-line summary when the package branch is
282
:param latest_pub: A LatestPublication instance
284
if verbosity == 'off':
286
if verbosity is None:
288
latest_ver, branch_ver = _get_newest_versions(the_branch, latest_pub)
289
place = latest_pub.place()
290
_report_freshness(latest_ver, branch_ver, place, verbosity,
added
removed
breezy/plugins/launchpad/lp_api_lite.py
