Add paginate to winter_sqlalchemy (#194)

tests/winter_sqlalchemy/test_injector.py

@@ -15,7 +15,7 @@
 
 from winter.data import CRUDRepository
 from winter_ddd import AggregateRoot
-from winter_sqlalchemy.repository import sqla_crud
+from winter_sqlalchemy import sqla_crud
 
 
 class Fixture:

tests/winter_sqlalchemy/test_paginate.py

@@ -0,0 +1,44 @@
+import pytest
+from sqlalchemy import Column
+from sqlalchemy import Integer
+from sqlalchemy import MetaData
+from sqlalchemy import Table
+from sqlalchemy import create_engine
+from sqlalchemy import select
+
+from winter.data.pagination import PagePosition
+from winter.data.pagination import Sort
+from winter_sqlalchemy import paginate
+
+
+@pytest.mark.parametrize(
+    'limit, offset, sort, expected_ids', [
+        (None, None, None, None),
+        (None, None, Sort.by('id'), [1, 2, 3, 4, 5]),
+        (None, None, Sort.by('id').desc(), [5, 4, 3, 2, 1]),
+        (3, None, Sort.by('id'), [1, 2, 3]),
+        (6, None, Sort.by('id'), [1, 2, 3, 4, 5]),
+        (2, 2, Sort.by('id'), [3, 4]),
+        (None, 2, Sort.by('id'), [3, 4, 5]),
+    ],
+)
+def test_paginate(id_database, limit, offset, sort, expected_ids):
+    engine, table = id_database
+    statement = paginate(select([table.c.id]), PagePosition(limit, offset, sort))
+    result = engine.execute(statement)
+    ids = [row[0] for row in result]
+    if expected_ids is None:
+        assert len(ids) == 5
+    else:
+        assert ids == expected_ids
+
+
+@pytest.fixture(scope='module')
+def id_database():
+    engine = create_engine('sqlite://')
+    metadata = MetaData(engine)
+    table = Table('table', metadata, Column('id', Integer, primary_key=True))
+    metadata.create_all()
+    rows = [{'id': value} for value in range(1, 6)]
+    engine.execute(table.insert(), *rows)
+    return engine, table

tests/winter_sqlalchemy/test_sqla_crud.py

@@ -15,7 +15,7 @@
 from winter_ddd import AggregateRoot
 from winter_ddd import DomainEvent
 from winter_ddd import domain_event_handler
-from winter_sqlalchemy.repository import sqla_crud
+from winter_sqlalchemy import sqla_crud
 
 
 @dataclass
@@ -190,7 +190,7 @@ def test_find_by_id():
 def test_get_by_id():
     fixture = Fixture()
 
-    with pytest.raises(NotFoundException, match="MyEntity with ID=2 not found"):
+    with pytest.raises(NotFoundException, match='MyEntity with ID=2 not found'):
         # Act
         fixture.repository.get_by_id(2)
 

winter/__version__.py

@@ -1 +1 @@
-__version__ = '8.2.1'
+__version__ = '8.3.0'

winter_sqlalchemy/__init__.py

@@ -0,0 +1,2 @@
+from .query import paginate
+from .repository import sqla_crud

winter_sqlalchemy/query.py

@@ -0,0 +1,18 @@
+from sqlalchemy import asc
+from sqlalchemy import desc
+from sqlalchemy.sql import Select
+
+from winter.data.pagination import PagePosition
+from winter.data.pagination import SortDirection
+
+_sort_direction_map = {
+    SortDirection.ASC: asc,
+    SortDirection.DESC: desc,
+}
+
+
+def paginate(select: Select, page_position: PagePosition) -> Select:
+    if page_position.sort:
+        order_by_clauses = [_sort_direction_map[order.direction](order.field) for order in page_position.sort.orders]
+        select = select.order_by(*order_by_clauses)
+    return select.limit(page_position.limit).offset(page_position.offset)