Add pagination helpers

In neutron core we want to deprecate
  neutron.db.sqlalchemyutils.paginate_query()
in favor of
  oslo_db.sqlalchemy.utils.paginate_query()

However, the oslo.db method takes different arguments. So we provide
helpers for converting the arguments to the oslo.db format.

Change-Id: I6de2945c7ed2ab48c285824a62ee0da61b030589

neutron_lib/db/utils.py

@@ -0,0 +1,53 @@
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from neutron_lib import exceptions as n_exc
+from sqlalchemy.orm import properties
+
+from neutron_lib._i18n import _
+
+
+def get_and_validate_sort_keys(sorts, model):
+    """Extract sort keys from sorts and ensure they are valid for the model.
+
+    :param sorts: list of (key, direction) tuples
+    :param model: sqlalchemy ORM model class
+    """
+
+    sort_keys = [s[0] for s in sorts]
+    for sort_key in sort_keys:
+        try:
+            sort_key_attr = getattr(model, sort_key)
+        except AttributeError:
+            # Extension attributes don't support sorting. Because it
+            # existed in attr_info, it will be caught here.
+            msg = _("'%s' is an invalid attribute for sort key") % sort_key
+            raise n_exc.BadRequest(resource=model.__tablename__, msg=msg)
+        if isinstance(sort_key_attr.property,
+                      properties.RelationshipProperty):
+            msg = _("Attribute '%(attr)s' references another resource and "
+                    "cannot be used to sort '%(resource)s' resources"
+                    ) % {'attr': sort_key, 'resource': model.__tablename__}
+            raise n_exc.BadRequest(resource=model.__tablename__, msg=msg)
+
+    return sort_keys
+
+
+def get_sort_dirs(sorts, page_reverse=False):
+    """Extract sort directions from sorts, possibly reversed.
+
+    :param sorts: list of (key, direction) tuples
+    :param page_reverse: True if sort direction is reversed
+    """
+    if page_reverse:
+        return ['desc' if s[1] else 'asc' for s in sorts]
+    return ['asc' if s[1] else 'desc' for s in sorts]

neutron_lib/tests/unit/db/test_utils.py

@@ -0,0 +1,63 @@
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+from oslo_db.sqlalchemy import models
+import sqlalchemy as sa
+from sqlalchemy.ext import declarative
+from sqlalchemy import orm
+
+from neutron_lib.db import utils
+from neutron_lib import exceptions as n_exc
+
+from neutron_lib.tests import _base as base
+
+
+class FakePort(declarative.declarative_base(cls=models.ModelBase)):
+    __tablename__ = 'fakeports'
+    port_id = sa.Column(sa.String(36), primary_key=True)
+    name = sa.Column(sa.String(64))
+    status = sa.Column(sa.String(16), nullable=False)
+
+
+class FakeRouter(declarative.declarative_base(cls=models.ModelBase)):
+    __tablename__ = 'fakerouters'
+    router_id = sa.Column(sa.String(36), primary_key=True)
+    gw_port_id = sa.Column(sa.String(36), sa.ForeignKey(FakePort.port_id))
+    gw_port = orm.relationship(FakePort, lazy='joined')
+
+
+class TestUtils(base.BaseTestCase):
+
+    def test_get_sort_dirs(self):
+        sorts = [(1, True), (2, False), (3, True)]
+        self.assertEqual(['asc', 'desc', 'asc'],
+                         utils.get_sort_dirs(sorts))
+
+    def test_get_sort_dirs_reversed(self):
+        sorts = [(1, True), (2, False), (3, True)]
+        self.assertEqual(['desc', 'asc', 'desc'],
+                         utils.get_sort_dirs(sorts, page_reverse=True))
+
+    def test_get_and_validate_sort_keys(self):
+        sorts = [('name', False), ('status', True)]
+        self.assertEqual(['name', 'status'],
+                         utils.get_and_validate_sort_keys(sorts, FakePort))
+
+    def test_get_and_validate_sort_keys_bad_key_fails(self):
+        sorts = [('master', True)]
+        self.assertRaises(n_exc.BadRequest,
+                          utils.get_and_validate_sort_keys, sorts, FakePort)
+
+    def test_get_and_validate_sort_keys_by_relationship_fails(self):
+        sorts = [('gw_port', True)]
+        self.assertRaises(n_exc.BadRequest,
+                          utils.get_and_validate_sort_keys, sorts, FakeRouter)