apps/piston/models.py

   1 import urllib, time, urlparse
   2
   3 # Django imports
   4 from django.db.models.signals import post_save, post_delete
   5 from django.db import models
   6 from django.contrib.auth.models import User
   7 from django.contrib import admin
   8 from django.core.mail import send_mail, mail_admins
   9
  10 # Piston imports
  11 from managers import TokenManager, ConsumerManager, ResourceManager
  12 from signals import consumer_post_save, consumer_post_delete
  13
  14 KEY_SIZE = 18
  15 SECRET_SIZE = 32
  16 VERIFIER_SIZE = 10
  17
  18 CONSUMER_STATES = (
  19     ('pending', 'Pending'),
  20     ('accepted', 'Accepted'),
  21     ('canceled', 'Canceled'),
  22     ('rejected', 'Rejected')
  23 )
  24
  25 def generate_random(length=SECRET_SIZE):
  26     return User.objects.make_random_password(length=length)
  27
  28 class Nonce(models.Model):
  29     token_key = models.CharField(max_length=KEY_SIZE)
  30     consumer_key = models.CharField(max_length=KEY_SIZE)
  31     key = models.CharField(max_length=255)
  32
  33     def __unicode__(self):
  34         return u"Nonce %s for %s" % (self.key, self.consumer_key)
  35
  36 admin.site.register(Nonce)
  37
  38 class Consumer(models.Model):
  39     name = models.CharField(max_length=255)
  40     description = models.TextField()
  41
  42     key = models.CharField(max_length=KEY_SIZE)
  43     secret = models.CharField(max_length=SECRET_SIZE)
  44
  45     status = models.CharField(max_length=16, choices=CONSUMER_STATES, default='pending')
  46     user = models.ForeignKey(User, null=True, blank=True, related_name='consumers')
  47
  48     objects = ConsumerManager()
  49
  50     def __unicode__(self):
  51         return u"Consumer %s with key %s" % (self.name, self.key)
  52
  53     def generate_random_codes(self):
  54         """
  55         Used to generate random key/secret pairings. Use this after you've
  56         added the other data in place of save().
  57
  58         c = Consumer()
  59         c.name = "My consumer"
  60         c.description = "An app that makes ponies from the API."
  61         c.user = some_user_object
  62         c.generate_random_codes()
  63         """
  64         key = User.objects.make_random_password(length=KEY_SIZE)
  65         secret = generate_random(SECRET_SIZE)
  66
  67         while Consumer.objects.filter(key__exact=key, secret__exact=secret).count():
  68             secret = generate_random(SECRET_SIZE)
  69
  70         self.key = key
  71         self.secret = secret
  72         self.save()
  73
  74 admin.site.register(Consumer)
  75
  76 class Token(models.Model):
  77     REQUEST = 1
  78     ACCESS = 2
  79     TOKEN_TYPES = ((REQUEST, u'Request'), (ACCESS, u'Access'))
  80
  81     key = models.CharField(max_length=KEY_SIZE)
  82     secret = models.CharField(max_length=SECRET_SIZE)
  83     verifier = models.CharField(max_length=VERIFIER_SIZE)
  84     token_type = models.IntegerField(choices=TOKEN_TYPES)
  85     timestamp = models.IntegerField(default=long(time.time()))
  86     is_approved = models.BooleanField(default=False)
  87
  88     user = models.ForeignKey(User, null=True, blank=True, related_name='tokens')
  89     consumer = models.ForeignKey(Consumer)
  90
  91     callback = models.CharField(max_length=255, null=True, blank=True)
  92     callback_confirmed = models.BooleanField(default=False)
  93
  94     objects = TokenManager()
  95
  96     def __unicode__(self):
  97         return u"%s Token %s for %s" % (self.get_token_type_display(), self.key, self.consumer)
  98
  99     def to_string(self, only_key=False):
 100         token_dict = {
 101             'oauth_token': self.key,
 102             'oauth_token_secret': self.secret,
 103             'oauth_callback_confirmed': 'true',
 104         }
 105
 106         if self.verifier:
 107             token_dict.update({ 'oauth_verifier': self.verifier })
 108
 109         if only_key:
 110             del token_dict['oauth_token_secret']
 111
 112         return urllib.urlencode(token_dict)
 113
 114     def generate_random_codes(self):
 115         key = User.objects.make_random_password(length=KEY_SIZE)
 116         secret = generate_random(SECRET_SIZE)
 117
 118         while Token.objects.filter(key__exact=key, secret__exact=secret).count():
 119             secret = generate_random(SECRET_SIZE)
 120
 121         self.key = key
 122         self.secret = secret
 123         self.save()
 124
 125     # -- OAuth 1.0a stuff
 126
 127     def get_callback_url(self):
 128         if self.callback and self.verifier:
 129             # Append the oauth_verifier.
 130             parts = urlparse.urlparse(self.callback)
 131             scheme, netloc, path, params, query, fragment = parts[:6]
 132             if query:
 133                 query = '%s&oauth_verifier=%s' % (query, self.verifier)
 134             else:
 135                 query = 'oauth_verifier=%s' % self.verifier
 136             return urlparse.urlunparse((scheme, netloc, path, params,
 137                 query, fragment))
 138         return self.callback
 139
 140     def set_callback(self, callback):
 141         if callback != "oob": # out of band, says "we can't do this!"
 142             self.callback = callback
 143             self.callback_confirmed = True
 144             self.save()
 145
 146 admin.site.register(Token)
 147
 148 # Attach our signals
 149 post_save.connect(consumer_post_save, sender=Consumer)
 150 post_delete.connect(consumer_post_delete, sender=Consumer)