| commit 28: | 8579a1562b16 |
| parent 27: | c038f3e47c5f |
| branch: | default |
Update models (stolen from piston) to be 1.0a ready, note that it will affect your Token/Consumer tables
10 months ago
django-oauth /
oauth_provider
/
models.py
| r28:8579a1562b16 | 134 loc | 4.5 KB | embed / history / annotate / raw / |
|---|
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 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 | import urllib
import urlparse
from time import time
from django.db import models
from django.contrib.auth.models import User
from managers import TokenManager, ConsumerManager, ResourceManager
from consts import KEY_SIZE, SECRET_SIZE, CONSUMER_KEY_SIZE, CONSUMER_STATES,\
PENDING, VERIFIER_SIZE
generate_random = User.objects.make_random_password
class Nonce(models.Model):
token_key = models.CharField(max_length=KEY_SIZE)
consumer_key = models.CharField(max_length=CONSUMER_KEY_SIZE)
key = models.CharField(max_length=255)
def __unicode__(self):
return u"Nonce %s for %s" % (self.key, self.consumer_key)
class Resource(models.Model):
name = models.CharField(max_length=255)
url = models.TextField(max_length=2047)
is_readonly = models.BooleanField(default=True)
objects = ResourceManager()
def __unicode__(self):
return u"Resource %s with url %s" % (self.name, self.url)
class Consumer(models.Model):
name = models.CharField(max_length=255)
description = models.TextField()
key = models.CharField(max_length=CONSUMER_KEY_SIZE)
secret = models.CharField(max_length=SECRET_SIZE, blank=True)
status = models.SmallIntegerField(choices=CONSUMER_STATES, default=PENDING)
user = models.ForeignKey(User, null=True, blank=True)
objects = ConsumerManager()
def __unicode__(self):
return u"Consumer %s with key %s" % (self.name, self.key)
def generate_random_codes(self):
"""
Used to generate random key/secret pairings.
Use this after you've added the other data in place of save().
"""
key = generate_random(length=KEY_SIZE)
secret = generate_random(length=SECRET_SIZE)
while Consumer.objects.filter(key__exact=key, secret__exact=secret).count():
secret = generate_random(length=SECRET_SIZE)
self.key = key
self.secret = secret
self.save()
class Token(models.Model):
REQUEST = 1
ACCESS = 2
TOKEN_TYPES = ((REQUEST, u'Request'), (ACCESS, u'Access'))
key = models.CharField(max_length=KEY_SIZE)
secret = models.CharField(max_length=SECRET_SIZE)
token_type = models.SmallIntegerField(choices=TOKEN_TYPES)
timestamp = models.IntegerField(default=long(time()))
is_approved = models.BooleanField(default=False)
user = models.ForeignKey(User, null=True, blank=True, related_name='tokens')
consumer = models.ForeignKey(Consumer)
resource = models.ForeignKey(Resource)
## OAuth 1.0a stuff
verifier = models.CharField(max_length=VERIFIER_SIZE)
callback = models.CharField(max_length=255, null=True, blank=True)
callback_confirmed = models.BooleanField(default=False)
objects = TokenManager()
def __unicode__(self):
return u"%s Token %s for %s" % (self.get_token_type_display(), self.key, self.consumer)
def to_string(self, only_key=False):
token_dict = {
'oauth_token': self.key,
'oauth_token_secret': self.secret,
'oauth_callback_confirmed': 'true',
}
if self.verifier:
token_dict.update({ 'oauth_verifier': self.verifier })
if only_key:
del token_dict['oauth_token_secret']
return urllib.urlencode(token_dict)
def generate_random_codes(self):
"""
Used to generate random key/secret pairings.
Use this after you've added the other data in place of save().
"""
key = generate_random(length=KEY_SIZE)
secret = generate_random(length=SECRET_SIZE)
while Token.objects.filter(key__exact=key, secret__exact=secret).count():
secret = generate_random(length=SECRET_SIZE)
self.key = key
self.secret = secret
self.save()
## OAuth 1.0a stuff
def get_callback_url(self):
if self.callback and self.verifier:
# Append the oauth_verifier.
parts = urlparse.urlparse(self.callback)
scheme, netloc, path, params, query, fragment = parts[:6]
if query:
query = '%s&oauth_verifier=%s' % (query, self.verifier)
else:
query = 'oauth_verifier=%s' % self.verifier
return urlparse.urlunparse((scheme, netloc, path, params,
query, fragment))
return self.callback
def set_callback(self, callback):
if callback != "oob": # out of band, says "we can't do this!"
self.callback = callback
self.callback_confirmed = True
self.save()
|
