Final steps in CrudRestController refactoring for RESTful API autogen…

…eration

tests/base.py

@@ -4,8 +4,9 @@
 
 from sqlalchemy import Column, ForeignKey, Integer, String, Text, Date
 from zope.sqlalchemy import ZopeTransactionExtension
-from sqlalchemy.orm import scoped_session, sessionmaker
+from sqlalchemy.orm import scoped_session, sessionmaker, relationship
 from sqlalchemy.ext.declarative import declarative_base
+from datetime import datetime
 
 maker = sessionmaker(autoflush=True, autocommit=False,
  extension=ZopeTransactionExtension())
@@ -14,14 +15,39 @@
 metadata = DeclarativeBase.metadata
 
 
+class Genre(DeclarativeBase):
+ __tablename__ = "genres"
+
+ genre_id = Column(Integer, primary_key=True)
+ name = Column(String(100), nullable=False)
+
+
 class Movie(DeclarativeBase):
  __tablename__ = "movies"
 
  movie_id = Column(Integer, primary_key=True)
  title = Column(String(100), nullable=False)
  description = Column(Text, nullable=True)
- release_date = Column(Date, nullable=True)
+ release_date = Column(Date, nullable=True, default=datetime.utcnow)
+
+ genre_id = Column(Integer, ForeignKey(Genre.genre_id), nullable=True)
+ genre = relationship(Genre)
+
+
+class Actor(DeclarativeBase):
+ __tablename__ = "actors"
+
+ actor_id = Column(Integer, primary_key=True)
+ name = Column(String(100), nullable=False)
+
+ movie_id = Column(Integer, ForeignKey(Movie.movie_id), nullable=True)
+ movie = relationship(Movie, backref='actors')
 
+ def __json__(self):
+ return {'name':self.name,
+ 'movie_id':self.movie_id,
+ 'actor_id':self.actor_id,
+ 'movie_title':self.movie and self.movie.title or None}
 
 class FakeModel(object):
  __file__ = 'model.py'

tests/test_crud_html.py

@@ -1,6 +1,6 @@
 from tg import TGController
 from tgext.crud import EasyCrudRestController
-from .base import CrudTest, Movie, DBSession, metadata
+from .base import CrudTest, Movie, DBSession, metadata, Actor
 
 
 class TestCrudHTML(CrudTest):
@@ -29,7 +29,7 @@ def test_post_validation(self):
  assert DBSession.query(Movie).first() is None
 
  def test_post_validation_dberror(self):
- metadata.drop_all()
+ metadata.drop_all(tables=[Movie.__table__])
 
  result = self.app.post('/movies/', params={'title':'Movie Test'})
  assert '<form action=' in result

tests/test_rest_json.py

@@ -1,9 +1,10 @@
 from tg import TGController
 from tgext.crud import EasyCrudRestController
-from .base import CrudTest, Movie, DBSession, metadata
+from .base import CrudTest, Movie, DBSession, metadata, Genre, Actor
 
+import transaction
 
-class TestRestJson(CrudTest):
+class TestRestJsonEditCreateDelete(CrudTest):
  def controller_factory(self):
  class MovieController(EasyCrudRestController):
  model = Movie
@@ -19,7 +20,7 @@ def test_post(self):
  movie = DBSession.query(Movie).first()
  assert movie is not None, result
 
- assert movie.movie_id == result.json['movie_id']
+ assert movie.movie_id == result.json['value']['movie_id']
 
  def test_post_validation(self):
  result = self.app.post('/movies.json', status=400)
@@ -29,7 +30,146 @@ def test_post_validation(self):
  assert DBSession.query(Movie).first() is None
 
  def test_post_validation_dberror(self):
- metadata.drop_all()
+ metadata.drop_all(tables=[Movie.__table__])
 
  result = self.app.post('/movies.json', params={'title':'Movie Test'}, status=400)
- assert result.json['message'].startswith('(OperationalError)')
+ assert result.json['message'].startswith('(OperationalError)')
+
+ def test_put(self):
+ result = self.app.post('/movies.json', params={'title':'Movie Test'})
+ movie = result.json['value']
+
+ result = self.app.put('/movies/%s.json' % movie['movie_id'],
+ params={'title':'New Title'})
+ assert result.json['value']['title'] == 'New Title'
+
+ def test_put_validation(self):
+ result = self.app.post('/movies.json', params={'title':'Movie Test'})
+ movie = result.json['value']
+
+ result = self.app.put('/movies/%s.json' % movie['movie_id'],
+ params={'title':''}, status=400)
+ assert result.json['title'] is not None #there is an error for required title
+ assert result.json['description'] is None #there isn't any error for optional description
+
+ movie = DBSession.query(Movie).first()
+ assert movie.title == 'Movie Test'
+
+ def test_delete(self):
+ DBSession.add(Movie(title='Fifth Movie'))
+ DBSession.flush()
+ transaction.commit()
+
+ movie = DBSession.query(Movie).first()
+
+ result = self.app.delete('/movies/%s.json' % movie.movie_id)
+ assert result.json == {}
+
+ def test_delete_idempotent(self):
+ DBSession.add(Movie(title='Fifth Movie'))
+ DBSession.flush()
+ transaction.commit()
+
+ movie = DBSession.query(Movie).first()
+
+ result = self.app.delete('/movies/%s.json' % movie.movie_id)
+ assert result.json == {}
+
+ result = self.app.delete('/movies/%s.json' % movie.movie_id)
+ assert result.json == {}
+
+ def test_delete_nofilter(self):
+ DBSession.add(Movie(title='Fifth Movie'))
+ DBSession.flush()
+ transaction.commit()
+
+ movie = DBSession.query(Movie).first()
+ assert movie is not None
+
+ result = self.app.delete('/movies.json')
+ assert result.json == {}
+
+ movie = DBSession.query(Movie).first()
+ assert movie is not None
+
+class TestRestJsonRead(CrudTest):
+ def controller_factory(self):
+ class MovieController(EasyCrudRestController):
+ model = Movie
+ pagination = {'items_per_page': 3}
+
+ class ActorController(EasyCrudRestController):
+ model = Actor
+
+ class RestJsonController(TGController):
+ movies = MovieController(DBSession)
+ actors = ActorController(DBSession)
+
+ return RestJsonController()
+
+ def setUp(self):
+ super(TestRestJsonRead, self).setUp()
+ genre = Genre(name='action')
+ DBSession.add(genre)
+
+ actors = [Actor(name='James Who'), Actor(name='John Doe'), Actor(name='Man Alone')]
+ map(DBSession.add, actors)
+
+ DBSession.add(Movie(title='First Movie', genre=genre, actors=actors[:2]))
+ DBSession.add(Movie(title='Second Movie', genre=genre, actors=actors[:2]))
+ DBSession.add(Movie(title='Third Movie', genre=genre))
+ DBSession.add(Movie(title='Fourth Movie', genre=genre))
+ DBSession.add(Movie(title='Fifth Movie'))
+ DBSession.add(Movie(title='Sixth Movie'))
+ DBSession.flush()
+ transaction.commit()
+
+ def test_get_all(self):
+ result = self.app.get('/movies.json?order_by=movie_id')
+ result = result.json['value_list']
+ assert result['total'] == 6, result
+ assert result['page'] == 1, result
+ assert result['entries'][0]['title'] == 'First Movie', result
+
+ result = self.app.get('/movies.json?page=2&order_by=movie_id')
+ result = result.json['value_list']
+ assert result['total'] == 6, result
+ assert result['page'] == 2, result
+ assert result['entries'][0]['title'] == 'Fourth Movie', result
+
+ def test_get_all_filter(self):
+ actor = DBSession.query(Actor).first()
+
+ result = self.app.get('/actors.json?movie_id=%s' % actor.movie_id)
+ result = result.json['value_list']
+ assert result['total'] == 2, result
+
+ def test_get_all___json__(self):
+ actor = DBSession.query(Actor).filter(Actor.movie_id!=None).first()
+ movie_title = actor.movie.title
+
+ result = self.app.get('/actors.json?movie_id=%s' % actor.movie_id)
+ result = result.json['value_list']
+ assert result['total'] > 0, result
+
+ for entry in result['entries']:
+ assert entry['movie_title'] == movie_title
+
+ def test_get_one(self):
+ movie = DBSession.query(Movie).first()
+
+ result = self.app.get('/movies/%s.json' % movie.movie_id)
+ result = result.json
+ assert result['model'] == 'Movie', result
+ assert result['value']['title'] == movie.title
+ assert result['value']['movie_id'] == movie.movie_id
+
+ def test_get_one___json__(self):
+ actor = DBSession.query(Actor).filter(Actor.movie_id!=None).first()
+ movie_title = actor.movie.title
+
+ result = self.app.get('/actors/%s.json' % actor.actor_id)
+ result = result.json
+ assert result['model'] == 'Actor', result
+ assert result['value']['name'] == actor.name
+ assert result['value']['movie_title'] == movie_title

tgext/crud/controller.py

@@ -2,12 +2,13 @@
 """
 import tg
 from tg import expose, flash, redirect, tmpl_context, request
-from tg.decorators import without_trailing_slash, with_trailing_slash
+from tg.decorators import without_trailing_slash, with_trailing_slash, before_validate
 from tg.controllers import RestController
 
-from decorators import registered_validate, register_validators, catch_errors
-from tgext.crud.utils import SmartPaginationCollection, RequestLocalTableFiller
-from utils import create_setter, set_table_filler_getter, SortableTableBase, optional_paginate
+from tgext.crud.decorators import (registered_validate, register_validators, catch_errors,
+ optional_paginate)
+from tgext.crud.utils import (SmartPaginationCollection, RequestLocalTableFiller, create_setter,
+ set_table_filler_getter, SortableTableBase, map_args_to_pks)
 from sprox.providerselector import ProviderTypeSelector
 from sprox.fillerbase import TableFiller
 from sprox.formbase import AddRecordForm, EditableForm
@@ -224,7 +225,7 @@ def __init__(self, session, menu_items=None):
  @expose('genshi:tgext.crud.templates.get_all')
  @expose('mako:tgext.crud.templates.get_all')
  @expose('jinja:tgext.crud.templates.get_all')
- @expose('json')
+ @expose('json:')
  @optional_paginate('value_list')
  def get_all(self, *args, **kw):
  """Return all records.
@@ -240,7 +241,8 @@ def get_all(self, *args, **kw):
  paginator.paginate_page = 0
 
  if tg.request.response_type == 'application/json':
- return dict(value_list=self.table_filler.get_value(**kw))
+ count, entries = self.table_filler._do_get_provider_count_and_objs(**kw)
+ return dict(value_list=entries)
 
  if not getattr(self.table.__class__, '__retrieves_own_value__', False):
  kw.pop('limit', None)
@@ -276,28 +278,29 @@ def get_all(self, *args, **kw):
  @expose('genshi:tgext.crud.templates.get_one')
  @expose('mako:tgext.crud.templates.get_one')
  @expose('jinja:tgext.crud.templates.get_one')
- @expose('json')
+ @expose('json:')
  def get_one(self, *args, **kw):
  """get one record, returns HTML or json"""
  #this would probably only be realized as a json stream
+ kw = map_args_to_pks(args, {})
+
+ if tg.request.response_type == 'application/json':
+ return dict(model=self.model.__name__,
+ value=self.provider.get_obj(self.model, kw))
+
  tmpl_context.widget = self.edit_form
- pks = self.provider.get_primary_fields(self.model)
- kw = {}
- for i, pk in enumerate(pks):
- kw[pk] = args[i]
  value = self.edit_filler.get_value(kw)
- return dict(value=value,model=self.model.__name__)
+ return dict(value=value, model=self.model.__name__)
 
  @expose('genshi:tgext.crud.templates.edit')
  @expose('mako:tgext.crud.templates.edit')
  @expose('jinja:tgext.crud.templates.edit')
  def edit(self, *args, **kw):
  """Display a page to edit the record."""
- tmpl_context.widget = self.edit_form
  pks = self.provider.get_primary_fields(self.model)
- kw = {}
- for i, pk in enumerate(pks):
-  kw[pk] = args[i]
+ kw = map_args_to_pks(args, {})
+
+ tmpl_context.widget = self.edit_form
  value = self.edit_filler.get_value(kw)
  value['_method'] = 'PUT'
  return dict(value=value, model=self.model.__name__, pk_count=len(pks))
@@ -319,20 +322,17 @@ def post(self, *args, **kw):
  obj = self.provider.create(self.model, params=kw)
 
  if tg.request.response_type == 'application/json':
- return dict(**self.provider.dictify(obj))
+ return dict(model=self.model.__name__, value=obj)
 
- raise redirect('./', params=self._kept_params())
+ return redirect('./', params=self._kept_params())
 
- @expose()
+ @expose(content_type='text/html')
+ @expose('json:', content_type='application/json')
+ @before_validate(map_args_to_pks)
  @registered_validate(error_handler=edit)
  @catch_errors(errors, error_handler=edit)
  def put(self, *args, **kw):
  """update"""
- pks = self.provider.get_primary_fields(self.model)
- for i, pk in enumerate(pks):
- if pk not in kw and i < len(args):
- kw[pk] = args[i]
-
  omit_fields = []
  if getattr(self, 'edit_form', None):
  omit_fields.extend(self.edit_form.__omit_fields__)
@@ -342,18 +342,31 @@ def put(self, *args, **kw):
  if value is None or value == '':
  omit_fields.append(remembered_value)
 
- self.provider.update(self.model, params=kw, omit_fields=omit_fields)
- redirect('../' * len(pks), params=self._kept_params())
+ obj = self.provider.update(self.model, params=kw, omit_fields=omit_fields)
+ if tg.request.response_type == 'application/json':
+ return dict(model=self.model.__name__, value=obj)
+
+ pks = self.provider.get_primary_fields(self.model)
+ return redirect('../' * len(pks), params=self._kept_params())
 
- @expose()
+ @expose(content_type='text/html')
+ @expose('json:', content_type='application/json')
  def post_delete(self, *args, **kw):
  """This is the code that actually deletes the record"""
+ kw = map_args_to_pks(args, {})
+
+ obj = None
+ if kw:
+ obj = self.provider.get_obj(self.model, kw)
+
+ if obj is not None:
+ self.provider.delete(self.model, kw)
+
+ if tg.request.response_type == 'application/json':
+ return dict()
+
  pks = self.provider.get_primary_fields(self.model)
- d = {}
- for i, arg in enumerate(args):
- d[pks[i]] = arg
- self.provider.delete(self.model, d)
- redirect('./' + '../' * (len(pks) - 1), params=self._kept_params())
+ return redirect('./' + '../' * (len(pks) - 1), params=self._kept_params())
 
  @expose('genshi:tgext.crud.templates.get_delete')
  @expose('jinja:tgext.crud.templates.get_delete')