Browse Source

Merge branch 'sqlalchemy'

Conflicts:
	tests/users_tests.py
master
Julio Biason 11 years ago
parent
commit
26b61e2765
  1. 25
      luncho/blueprints/users.py
  2. 29
      luncho/database.py
  3. 10
      luncho/helpers.py
  4. 41
      luncho/server.py
  5. 5
      manage.py
  6. 2
      requirements.txt
  7. 54
      tests/users_tests.py

25
luncho/blueprints/users.py

@ -6,13 +6,13 @@
from flask import Blueprint
from flask import request
from flask import jsonify
# from flask import current_app
from pony.orm import commit
from sqlalchemy.exc import IntegrityError
from luncho.helpers import ForceJSON
from luncho.server import User
from luncho.server import db
users = Blueprint('users', __name__)
@ -23,10 +23,19 @@ def create_user():
"""Create a new user. Request must be:
{ "username": "username", "full_name": "Full Name", "password": "hash" }"""
json = request.get_json(force=True)
new_user = User(username=json['username'],
fullname=json['full_name'],
passhash=json['password'],
validated=False)
commit()
return jsonify(status='OK')
try:
new_user = User(username=json['username'],
fullname=json['full_name'],
passhash=json['password'],
validated=False)
db.session.add(new_user)
db.session.commit()
return jsonify(status='OK')
except IntegrityError:
resp = jsonify(status='ERROR',
error='username already exists')
resp.status_code = 409
return resp

29
luncho/database.py

@ -1,29 +0,0 @@
#!/usr/bin/env python
# -*- encoding: utf-8 -*-
import datetime
from flask import current_app
from pony.orm import Database
from pony.orm import PrimaryKey
from pony.orm import Optional
from pony.orm import Required
# from pony.orm import Set
db = Database("sqlite", current_app.config['SQLITE_FILENAME'], create_db=True)
class User(db.Entity):
"""Users."""
username = PrimaryKey(unicode)
fullname = Required(unicode)
passhash = Required(unicode)
token = Optional(unicode) # 1. if the user never logged in, they will
# not have a token.
# 2. This forces the user to have a single
# login everywhere, per day.
issue_date = Optional(datetime.datetime)
validated = Required(bool, default=False)
db.generate_mapping(create_tables=True)

10
luncho/helpers.py

@ -18,8 +18,10 @@ class ForceJSON(object):
def check_json(*args, **kwargs):
json = request.get_json(force=True, silent=True)
if not json:
return jsonify(status='ERROR',
error='Request MUST be in JSON format'), 400
resp = jsonify(status='ERROR',
error='Request MUST be in JSON format')
resp.status_code = 400
return resp
# now we have the JSON, let's check if all the fields are here.
missing = []
@ -30,8 +32,10 @@ class ForceJSON(object):
if missing:
fields = ', '.join(missing)
error = 'Missing fields: {fields}'.format(fields=fields)
return jsonify(status='ERROR',
resp = jsonify(status='ERROR',
error=error)
resp.status_code = 400
return resp
return func(*args, **kwargs)
return check_json

41
luncho/server.py

@ -2,7 +2,6 @@
# -*- encoding: utf-8 -*-
import sys
import datetime
import logging
from flask import Flask
@ -12,7 +11,8 @@ from flask import Flask
# Config
# ----------------------------------------------------------------------
class Settings(object):
SQLITE_FILENAME = './luncho.db3'
SQLALCHEMY_DATABASE_URI = 'sqlite://./luncho.db3'
DEBUG = True
log = logging.getLogger('luncho.server')
@ -26,29 +26,26 @@ app.config.from_envvar('LUCNHO_CONFIG', True)
# ----------------------------------------------------------------------
# Database
# ----------------------------------------------------------------------
from pony.orm import db_session
from pony.orm import Database
from pony.orm import PrimaryKey
from pony.orm import Optional
from pony.orm import Required
from flask.ext.sqlalchemy import SQLAlchemy
db = SQLAlchemy(app)
db = Database("sqlite", app.config['SQLITE_FILENAME'], create_db=True)
class User(db.Model):
username = db.Column(db.String, primary_key=True)
fullname = db.Column(db.String, nullable=False)
passhash = db.Column(db.String, nullable=False)
token = db.Column(db.String)
issued_date = db.Column(db.Date)
validated = db.Column(db.Boolean, default=False)
class User(db.Entity):
"""Users."""
username = PrimaryKey(unicode)
fullname = Required(unicode)
passhash = Required(unicode)
token = Optional(unicode) # 1. if the user never logged in, they will
# not have a token.
# 2. This forces the user to have a single
# login everywhere, per day.
issue_date = Optional(datetime.datetime)
validated = Required(bool, default=False)
db.generate_mapping(create_tables=True)
app.wsgi_app = db_session(app.wsgi_app)
def __init__(self, username, fullname, passhash, token=None,
issued_date=None, validated=False):
self.username = username
self.fullname = fullname
self.passhash = passhash
self.token = token
self.issued_date = issued_date
self.validated = validated
# ----------------------------------------------------------------------
# Blueprints

5
manage.py

@ -9,6 +9,11 @@ from luncho.server import app
manager = Manager(app)
@manager.command
def create_db():
"""Create the database."""
if __name__ == '__main__':
logging.basicConfig(level=logging.DEBUG)
manager.run()

2
requirements.txt

@ -1,3 +1,3 @@
Flask
flask-script
pony
flask-sqlalchemy

54
tests/users_tests.py

@ -11,20 +11,68 @@ class TestUsers(unittest.TestCase):
"""Test users request."""
def setUp(self):
server.app.config['SQLITE_FILENAME'] = ':memory:'
# leave the database blank to make it in memory
server.app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite://'
server.app.config['TESTING'] = True
print server.app.config['SQLITE_FILENAME']
self.app = server.app.test_client()
server.db.create_all()
def tearDown(self):
server.db.drop_all(bind=None)
def test_create_user(self):
"""Simple user creation."""
request = {'username': 'username',
'full_name': 'full name',
'password': 'hash'}
rv = self.app.put('/user/',
data=json.dumps(request),
content_type='application/json')
self.assertEqual(rv.status_code, 200)
self.assertEqual(json.loads(rv.data), {'status': 'OK'})
def test_duplicate_user(self):
"""Check the status for trying to create a user that it is already
in the database."""
self.test_create_user() # create the first user
# now duplicate
request = {'username': 'username',
'full_name': 'full name',
'password': 'hash'}
rv = self.app.put('/user/',
data=json.dumps(request),
content_type='application/json')
print rv.data
expected = {"status": "ERROR",
"error": "username already exists"}
self.assertEqual(rv.status_code, 409)
self.assertEqual(json.loads(rv.data), expected)
def test_no_json(self):
"""Check the status when doing a request that it's not JSON."""
rv = self.app.put('/user/',
data='',
content_type='text/html')
expected = {"error": "Request MUST be in JSON format",
"status": "ERROR"}
self.assertEqual(rv.status_code, 400)
self.assertEqual(json.loads(rv.data), expected)
def test_missing_fields(self):
request = {'password': 'hash'}
rv = self.app.put('/user/',
data=json.dumps(request),
content_type='application/json')
resp = {'error': 'Missing fields: username, full_name',
'status': 'ERROR'}
self.assertEqual(rv.status_code, 400)
self.assertEqual(json.loads(rv.data), resp)
if __name__ == '__main__':
unittest.main()

Loading…
Cancel
Save