/usr/lib/python2.7/dist-packages/pgxnclient/api.py is in pgxnclient 1.2.1-3.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | """
pgxnclient -- client API stub
"""
# Copyright (C) 2011-2012 Daniele Varrazzo
# This file is part of the PGXN client
from __future__ import with_statement
from urllib import urlencode
from pgxnclient import network
from pgxnclient.utils import load_json
from pgxnclient.errors import NetworkError, NotFound, ResourceNotFound
from pgxnclient.utils.uri import expand_template
class Api(object):
def __init__(self, mirror):
self.mirror = mirror
def dist(self, dist, version=''):
try:
with self.call(version and 'meta' or 'dist',
{'dist': dist, 'version': version}) as f:
return load_json(f)
except ResourceNotFound:
raise NotFound("distribution '%s' not found" % dist)
def ext(self, ext):
try:
with self.call('extension', {'extension': ext}) as f:
return load_json(f)
except ResourceNotFound:
raise NotFound("extension '%s' not found" % ext)
def meta(self, dist, version, as_json=True):
with self.call('meta', {'dist': dist, 'version': version}) as f:
if as_json:
return load_json(f)
else:
return f.read().decode('utf-8')
def readme(self, dist, version):
with self.call('readme', {'dist': dist, 'version': version}) as f:
return f.read()
def download(self, dist, version):
dist = dist.lower()
version = version.lower()
return self.call('download', {'dist': dist, 'version': version})
def mirrors(self):
with self.call('mirrors') as f:
return load_json(f)
def search(self, where, query):
"""Search into PGXN.
:param where: where to search. The server currently supports "docs",
"dists", "extensions"
:param query: list of strings to search
"""
# convert the query list into a string
q = ' '.join([' ' in s and ('"%s"' % s) or s for s in query])
with self.call('search', {'in': where}, query={'q': q}) as f:
return load_json(f)
def stats(self, arg):
with self.call('stats', {'stats': arg}) as f:
return load_json(f)
def user(self, username):
with self.call('user', {'user': username}) as f:
return load_json(f)
def call(self, meth, args=None, query=None):
url = self.get_url(meth, args, query)
try:
return network.get_file(url)
except ResourceNotFound:
# check if it is one of the broken URLs as reported in
# https://groups.google.com/group/pgxn-users/browse_thread/thread/e41fbc202680c92c
version = args and args.get('version')
if not (version and version.trail):
raise
args = args.copy()
args['version'] = str(version).replace('-', '', 1)
url = self.get_url(meth, args, query)
return network.get_file(url)
def get_url(self, meth, args=None, query=None):
tmpl = self.get_template(meth)
url = expand_template(tmpl, args or {})
url = self.mirror.rstrip('/') + url
if query is not None:
url = url + '?' + urlencode(query)
return url
def get_template(self, meth):
return self.get_index()[meth]
_api_index = None
def get_index(self):
if self._api_index is None:
url = self.mirror.rstrip('/') + '/index.json'
try:
with network.get_file(url) as f:
self._api_index = load_json(f)
except ResourceNotFound:
raise NetworkError("API index not found at '%s'" % url)
return self._api_index
|