creyD/creyPY
1
0
Fork 0
mirror of https://github.com/creyD/creyPY.git synced 2026-04-12 19:30:30 +02:00
Code Issues Packages Projects Releases Wiki Activity

Adjusted files for isort & autopep

Browse Source
This commit is contained in:
creyD
2024-10-24 07:27:26 +00:00
committed by github-actions[bot]
parent da7ec0b28e
commit 6ce0cfbd14
1 changed files with 22 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
creyPY/fastapi/pagination.py
View File

@@ -3,7 +3,12 @@ from typing import Any, Generic, Optional, Self, Sequence, TypeVar, Union
from fastapi_pagination import Params from fastapi_pagination import Params
from fastapi_pagination.bases import AbstractPage, AbstractParams from fastapi_pagination.bases import AbstractPage, AbstractParams
from fastapi_pagination.types import GreaterEqualOne, GreaterEqualZero, AdditionalData, SyncItemsTransformer from fastapi_pagination.types import (
GreaterEqualOne,
GreaterEqualZero,
AdditionalData,
SyncItemsTransformer,
)
from fastapi_pagination.api import create_page, apply_items_transformer from fastapi_pagination.api import create_page, apply_items_transformer
from fastapi_pagination.utils import verify_params from fastapi_pagination.utils import verify_params
from fastapi_pagination.ext.sqlalchemy import create_paginate_query from fastapi_pagination.ext.sqlalchemy import create_paginate_query
@@ -77,40 +82,44 @@ def parse_page(response, page: int, size: int) -> Page:
has_prev=response.has_prev, has_prev=response.has_prev,
) )
def create_count_query(query: Select) -> Select: def create_count_query(query: Select) -> Select:
return select(func.count()).select_from(query.subquery()) return select(func.count()).select_from(query.subquery())
def unwrap_scalars( def unwrap_scalars(
items: Sequence[Sequence[T]], items: Sequence[Sequence[T]],
force_unwrap: bool = True, force_unwrap: bool = True,
) -> Union[Sequence[T], Sequence[Sequence[T]]]: ) -> Union[Sequence[T], Sequence[Sequence[T]]]:
return [item[0] if force_unwrap else item for item in items] return [item[0] if force_unwrap else item for item in items]
def paginate(connection: Session,
query: Select, def paginate(
paginationFlag: bool = True, connection: Session,
params: Optional[AbstractParams] = None, query: Select,
transformer: Optional[SyncItemsTransformer] = None, paginationFlag: bool = True,
additional_data: Optional[AdditionalData] = None): params: Optional[AbstractParams] = None,
transformer: Optional[SyncItemsTransformer] = None,
additional_data: Optional[AdditionalData] = None,
):
params, raw_params = verify_params(params, "limit-offset", "cursor") params, raw_params = verify_params(params, "limit-offset", "cursor")
count_query = create_count_query(query) count_query = create_count_query(query)
total = connection.scalar(count_query) total = connection.scalar(count_query)
if paginationFlag is False: if paginationFlag is False:
params = Params(page=1, size=total) params = Params(page=1, size=total)
query = create_paginate_query(query, params) query = create_paginate_query(query, params)
items = connection.execute(query).all() items = connection.execute(query).all()
items = unwrap_scalars(items) items = unwrap_scalars(items)
t_items = apply_items_transformer(items, transformer) t_items = apply_items_transformer(items, transformer)
return create_page( return create_page(
t_items, t_items,
params=params, params=params,
total=total, total=total,
**(additional_data or {}), **(additional_data or {}),
) )