Fixes to pagination

Author: rozza

flask_mongoengine/__init__.py

@@ -1,19 +1,16 @@
 # -*- coding: utf-8 -*-
 from __future__ import absolute_import
 
-import datetime
-import math
-import uuid
-
 from flask import abort
-from flask.sessions import SessionInterface, SessionMixin
-from werkzeug.datastructures import CallbackDict
 
 import mongoengine
 
 from mongoengine.queryset import MultipleObjectsReturned, DoesNotExist, QuerySet
 from mongoengine.base import ValidationError
 
+from .sessions import *
+from .pagination import *
+
 
 def _include_mongoengine(obj):
     for module in mongoengine, mongoengine.fields:
@@ -104,222 +101,3 @@ class DynamicDocument(mongoengine.DynamicDocument):
 
     meta = {'abstract': True,
             'queryset_class': BaseQuerySet}
-
-
-class Pagination(object):
-
-    def __init__(self, iterable, page, per_page):
-
-        if page < 1:
-            abort(404)
-
-        self.iterable = iterable
-        self.page = page
-        self.per_page = per_page
-        self.total = len(iterable)
-
-        start_index = (page - 1) * per_page
-        end_index = page * per_page
-
-        self.items = iterable[start_index:end_index]
-        if isinstance(self.items, QuerySet):
-            self.items = self.items.select_related()
-        if not self.items and page != 1:
-            abort(404)
-
-    @property
-    def pages(self):
-        """The total number of pages"""
-        return int(math.ceil(self.total / float(self.per_page)))
-
-    def prev(self, error_out=False):
-        """Returns a :class:`Pagination` object for the previous page."""
-        assert self.iterable is not None, ('an object is required '
-                                           'for this method to work')
-        iterable = self.iterable
-        if isinstance(iterable, QuerySet):
-            iterable._skip = None
-            iterable._limit = None
-            iterable = iterable.clone()
-        return self.__class__(iterable, self.page - 1, self.per_page)
-
-    @property
-    def prev_num(self):
-        """Number of the previous page."""
-        return self.page - 1
-
-    @property
-    def has_prev(self):
-        """True if a previous page exists"""
-        return self.page > 1
-
-    def next(self, error_out=False):
-        """Returns a :class:`Pagination` object for the next page."""
-        assert self.iterable is not None, ('an object is required '
-                                           'for this method to work')
-        iterable = self.iterable
-        if isinstance(iterable, QuerySet):
-            iterable._skip = None
-            iterable._limit = None
-            iterable = iterable.clone()
-        return self.__class__(iterable, self.page + 1, self.per_page)
-
-    @property
-    def has_next(self):
-        """True if a next page exists."""
-        return self.page < self.pages
-
-    @property
-    def next_num(self):
-        """Number of the next page"""
-        return self.page + 1
-
-    def iter_pages(self, left_edge=2, left_current=2,
-                   right_current=5, right_edge=2):
-        """Iterates over the page numbers in the pagination.  The four
-        parameters control the thresholds how many numbers should be produced
-        from the sides.  Skipped page numbers are represented as `None`.
-        This is how you could render such a pagination in the templates:
-
-        .. sourcecode:: html+jinja
-
-            {% macro render_pagination(pagination, endpoint) %}
-              <div class=pagination>
-              {%- for page in pagination.iter_pages() %}
-                {% if page %}
-                  {% if page != pagination.page %}
-                    <a href="{{ url_for(endpoint, page=page) }}">{{ page }}</a>
-                  {% else %}
-                    <strong>{{ page }}</strong>
-                  {% endif %}
-                {% else %}
-                  <span class=ellipsis>…</span>
-                {% endif %}
-              {%- endfor %}
-              </div>
-            {% endmacro %}
-        """
-        last = 0
-        for num in xrange(1, self.pages + 1):
-            if num <= left_edge or \
-               (num > self.page - left_current - 1 and
-                num < self.page + right_current) or \
-               num > self.pages - right_edge:
-                if last + 1 != num:
-                    yield None
-                yield num
-                last = num
-
-
-class ListFieldPagination(Pagination):
-
-    def __init__(self, queryset, field_name, doc_id, page, per_page,
-                 total=None):
-        """Allows an array within a document to be paginated.
-
-        Queryset must contain the document which has the array we're
-        paginating, and doc_id should be it's _id.
-        Field name is the name of the array we're paginating.
-        Page and per_page work just like in Pagination.
-        Total is an argument because it can be computed more efficiently
-        elsewhere, but we still use array.length as a fallback.
-        """
-        if page < 1:
-            abort(404)
-
-        self.page = page
-        self.per_page = per_page
-
-        self.queryset = queryset
-        self.doc_id = doc_id
-        self.field_name = field_name
-
-        start_index = (page - 1) * per_page
-
-        field_attrs = {field_name: {"$slice": [start_index, per_page]}}
-
-        self.items = getattr(queryset().fields(**field_attrs).first(), field_name)
-
-        self.total = total or len(self.items)
-
-        if not self.items and page != 1:
-            abort(404)
-
-    def prev(self, error_out=False):
-        """Returns a :class:`Pagination` object for the previous page."""
-        assert self.items is not None, ('a query object is required '
-                                        'for this method to work')
-        return self.__class__(self.queryset, self.doc_id, self.field_name,
-                              self.page - 1, self.per_page, self.total)
-
-    def next(self, error_out=False):
-        """Returns a :class:`Pagination` object for the next page."""
-        assert self.iterable is not None, ('a query object is required '
-                                           'for this method to work')
-        return self.__class__(self.queryset, self.doc_id, self.field_name,
-                              self.page + 1, self.per_page, self.total)
-
-
-class MongoEngineSession(CallbackDict, SessionMixin):
-
-    def __init__(self, initial=None, sid=None):
-        def on_update(self):
-            self.modified = True
-        CallbackDict.__init__(self, initial, on_update)
-        self.sid = sid
-        self.modified = False
-
-
-class MongoEngineSessionInterface(SessionInterface):
-    """SessionInterface for mongoengine"""
-
-    def __init__(self, db, collection='session'):
-        """
-        The MongoSessionInterface
-
-        :param db: The app's db eg: MongoEngine()
-        :param collection: The session collection name defaults to "session"
-        """
-
-        if not isinstance(collection, basestring):
-            raise ValueError('collection argument should be string or unicode')
-
-        class DBSession(db.Document):
-            sid = db.StringField(primary_key=True)
-            data = db.DictField()
-            expiration = db.DateTimeField()
-            meta = {
-                'collection': collection,
-                'indexes': [{'fields': ['expiration'],
-                             'expireAfterSeconds': 60 * 60 * 24 * 7 * 31}]
-            }
-
-        self.cls = DBSession
-
-    def get_expiration_time(self, app, session):
-        if session.permanent:
-            return app.permanent_session_lifetime
-        return datetime.timedelta(days=1)
-
-    def open_session(self, app, request):
-        sid = request.cookies.get(app.session_cookie_name)
-        if sid:
-            stored_session = self.cls.objects(sid=sid).first()
-            if stored_session and stored_session.expiration > datetime.datetime.utcnow():
-                return MongoEngineSession(initial=stored_session.data, sid=stored_session.sid)
-        return MongoEngineSession(sid=str(uuid.uuid4()))
-
-    def save_session(self, app, session, response):
-        domain = self.get_cookie_domain(app)
-        if not session:
-            if session.modified:
-                response.delete_cookie(app.session_cookie_name, domain=domain)
-            return
-
-        expiration = datetime.datetime.now() + self.get_expiration_time(app, session)
-
-        if session.modified:
-            self.cls(sid=session.sid, data=session, expiration=expiration).save()
-
-        response.set_cookie(app.session_cookie_name, session.sid,
-                            expires=expiration, httponly=True, domain=domain)