Coverage for /opt/conda/envs/apienv/lib/python3.10/site-packages/daiquiri/core/metadata/ispyalchemy/utils.py: 85%

1import logging 

2from daiquiri.core.utils import get_start_end 

3 

4logger = logging.getLogger(__name__) 

5 

6 

7def order(query, sort_map, default=None, **kwargs): 

8 """Sort a result set by a field 

9 

10 Args: 

11 query (sqlalchemy.query): The current query 

12 sort_map (dict): A mapping of field(str) -> sqlalchemy.column 

13 

14 Kwargs: 

15 order_by (str): Field to sort by 

16 order (str): Asc or desc 

17 

18 Returns 

19 query (sqlalchemy.query): The ordered query 

20 """ 

21 sort_field = kwargs.get("order_by") 

22 sort_order = kwargs.get("order") 

23 

24 if not (sort_field and sort_order) and not default: 

25 return query 

26 

27 if default: 

28 if not sort_field: 

29 sort_field = default[0] 

30 if not sort_order: 

31 sort_order = default[1] 

32 

33 if sort_field not in sort_map: 

34 logger.warning(f"Unknown sort_field {sort_field}") 

35 return query 

36 

37 return query.order_by(getattr(sort_map[sort_field], sort_order)()) 

38 

39 

40def page(query, **kwargs): 

41 """Paginate a result set 

42 

43 Kwargs: 

44 per_page (str): Number of rows per page 

45 page (str): Page number to display 

46 last (bool): Reverse order, last page first 

47 

48 Returns 

49 query (sqlalchemy.query): The paginated query 

50 """ 

51 lims = get_start_end(kwargs) 

52 return query.limit(lims["per_page"]).offset(lims["st"])