[20a692]: migrate-3rdparty / allura_import.py History

Child: [218b8a] (diff)

Download this file

allura_import.py    65 lines (52 with data), 2.3 kB 

 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
import sys

import urllib

import urllib2

import urlparse

import hmac

import hashlib

from optparse import OptionParser

from pprint import pprint

from datetime import datetime





def parse_options():

    optparser = OptionParser(usage=''' %prog --api-key= --secret-key= <JSON dump>



Import project data dump in JSON format into Allura project.''')

    optparser.add_option('-a', '--api-key', dest='api_key', help='API key')

    optparser.add_option('-s', '--secret-key', dest='secret_key', help='Secret key')

    optparser.add_option('-u', '--base-url', dest='base_url', default='https://sourceforge.net', help='Base Allura URL (%default)')

    optparser.add_option('--validate', dest='validate', action='store_true', help='Validate import data')

    options, args = optparser.parse_args()

    if len(args) != 1:

        optparser.error("Wrong number of arguments.")

    if not options.api_key or not options.secret_key:

        optparser.error("Keys are required.")

    return options, args





class AlluraRestClient(object):



    def __init__(self, base_url, api_key, secret_key):

        self.base_url = base_url

        self.api_key = api_key

        self.secret_key = secret_key

        

    def sign(self, path, params):

        params.append(('api_key', self.api_key))

        params.append(('api_timestamp', datetime.utcnow().isoformat()))

        message = path + '?' + urllib.urlencode(sorted(params))

        digest = hmac.new(self.secret_key, message, hashlib.sha256).hexdigest()

        params.append(('api_signature', digest))

        return params



    def call(self, url, **params):

        url = urlparse.urljoin(options.base_url, url)

        params = self.sign(urlparse.urlparse(url).path, params.items())



        try:

            result = urllib2.urlopen(url, urllib.urlencode(params))

            return result.read()

        except urllib2.HTTPError, e:

            error_content = e.read()

            e.msg += '. Error response:\n' + error_content

            raise e



    

if __name__ == '__main__':

    options, args = parse_options()

    if options.validate:

        url = '/rest/p/test/bugs/validate_import'

    else:

        url = '/rest/p/test/bugs/perform_import'



    cli = AlluraRestClient(options.base_url, options.api_key, options.secret_key)

    print cli.call(url, doc=open(args[0]).read())