{i&dZddlmZddlZddlmZmZmZmZddl m Z ddl m Z ddl mZddlmZdd lmZmZdd lmZmZed e ZGd de eeZGddeeZy)z Modern pagination utility for SQLAlchemy queries with Pydantic schemas. This module provides a QueryPaginator class that handles pagination of SQLAlchemy queries with automatic serialization to Pydantic models, designed for Python 3.12+. ) annotationsN)AnyTypeVarGenericUnion) BaseModel)Query)Select)Session)selectfunc)DEFAULT_PER_PAGE MAX_PER_PAGEResponseSchemaType)boundcbeZdZUdZded<ded<ded<ded<ded<d ed <d ed <d ed <y)PaginationResultz-Pagination result container with type safety.inttotalpageper_page str | Nonenextpreviousstrfirstlastlist[ResponseSchemaType]resultN)__name__ __module__ __qualname____doc____annotations__T/var/www/html/hwPaymentPortal-be-dev/hw-payment-portal-api/src/helpers/pagination.pyrrs07 J IM  J I $$r&rceZdZdZdddedd ddZeddZeddZedd Z edd Z edd Z ddd Z edd Z eddZeddZeddZddZddZddZddZy)QueryPaginatora Modern SQLAlchemy query paginator with Pydantic schema serialization. This paginator provides efficient pagination for SQLAlchemy queries with automatic conversion to Pydantic models. It supports both ORM and raw query results with comprehensive URL generation for navigation links. Args: query: SQLAlchemy Query object or Select statement to paginate schema: Pydantic model class for response serialization db: Database session (required for Select statements) url: Base URL for generating pagination links offset: Number of records to skip (default: 0) limit: Number of records to return (default: DEFAULT_PER_PAGE) use_orm: Whether to use ORM mode for serialization (default: True) model: SQLAlchemy model class (required for Select statements) Example: >>> # For legacy Query objects >>> paginator = QueryPaginator( ... query=session.query(User), ... schema=UserSchema, ... url="/api/users", ... limit=10 ... ) >>> >>> # For modern Select statements >>> paginator = QueryPaginator( ... query=select(User), ... schema=UserSchema, ... db=session, ... model=User, ... url="/api/users", ... limit=10 ... ) >>> result = paginator.paginate() NrT)dbmodeloffsetlimituse_ormcR||_||_||_||_|j d|_t d||_t dt|t|_ ||_ t|t|_|jr|| tdd|_d|_y)N/rz=Database session and model are required for Select statements)queryschemar*r+rstripurlmaxr,minrr-r. isinstancer is_select_stmt ValueError_count_records) selfr2r3r5r*r+r,r-r.s r'__init__zQueryPaginator.__init__Ls   ::c?!Vn C|45  )7   BJ%-\] ]#' 9= r&c>|j|jrttjj |j jr|j jdn |j}|j j%|j|j j}|jj|j|_|jS|j j |_|jS)z$Get total record count with caching.r)r;r9r r count select_fromr2fromsr+ whereclausewherer*executescalar)r= count_querys r'r@zQueryPaginator.countms ;; ""%TZZ\2>>+/::+;+;DJJ$$Q' ::))5"-"3"3DJJ4J4J"KK"ggook:AAC {{#jj..0 {{r&c:|j|jzdzS)z$Get current page number (1-indexed).r1)r,r-r=s r' current_pagezQueryPaginator.current_pages tzz)Q..r&cntdtj|j|jz S)zGet total number of pages.r1)r6mathceilr@r-rIs r' total_pageszQueryPaginator.total_pagess'1dii TZZ 7899r&cN|j|jz|jkS)zCheck if there's a next page.)r,r-r@rIs r'has_nextzQueryPaginator.has_nexts {{TZZ'$**44r&c |jdkDS)z!Check if there's a previous page.r)r,rIs r' has_previouszQueryPaginator.has_previouss{{Qr&cJ|xs |j}|jd|d|S)a Build pagination URL for the given page. Args: page: Page number (1-indexed) per_page: Items per page (defaults to current limit) Returns: Formatted URL with pagination parameters z?page=z &per_page=)r-r5)r=rrs r' _build_urlzQueryPaginator._build_urls-)tzz((6$z(<Get URL for previous page, or None if no previous page exists.Nr1)rRrTrJrIs r' previous_urlzQueryPaginator.previous_urls*  t001455r&c$|jdS)zGet URL for first page.r1)rTrIs r' first_urlzQueryPaginator.first_urlsq!!r&c8|j|jS)zGet URL for last page.)rTrNrIs r'last_urlzQueryPaginator.last_urlst//00r&c |j |jS|jrv|jj|jj |j}|j j |jj}nL|jj|jj |jj} |jr:|Dcgc]}|jj|c}|_|jS|Dcgc]9}|jjt|dr|jn|;c}|_ |jScc}wcc}w#t$r+}td|jj d||d}~wwxYw)z Retrieve and serialize the paginated records. Returns: List of serialized Pydantic models for the current page Raises: ValueError: If serialization fails N_asdictz(Failed to serialize records with schema z: )r<r9r2r,r-r*rEscalarsallr.r3model_validatehasattrr^ Exceptionr:r )r=paginated_query raw_recordsrecordes r' get_recordszQueryPaginator.get_recordssn == $==    "jj// <BB4::NO''///:BBDHHJK**++DKK8>>tzzJNNPK l||R] ^!;!;F!C ^ }} #.!KK..76S\C]v~~/?cij! }}!_! lG H\H\G]]_`a_bcdjk k ls<(F8"FF.F2>F 0F F G&F>>Gc tt|j|j|j|j |j |j|j|jS)a Execute pagination and return complete result. Returns: PaginationResult containing all pagination data and serialized records Example: >>> result = paginator.paginate() >>> print(f"Page {result.page} of {math.ceil(result.total / result.per_page)}") >>> for item in result.result: ... print(item.name) rrrrrrrr) rrr@rJr-rVrXrZr\rhrIs r'paginatezQueryPaginator.paginatesX 23**""ZZ&&..##%  r&c |j}|j|j|j|j|j |j |j|jDcgc]}|jc}dScc}w)a2 Convert pagination result to dictionary format. This method provides backward compatibility with the original implementation while maintaining the same response structure. Returns: Dictionary containing pagination metadata and results rj) rkrrrrrrrr model_dump)r=paginated_resultitems r'to_dictzQueryPaginator.to_dictsz ==?%++$))(11$))(11%++$))5E5L5LMTt(M  Ns,Bc |jjd|jjd|jd|jd|j |j d Sdd S)z$String representation for debugging.z(schema=z, page=z , per_page=z, total=unknown)) __class__r r3rJr-r;r@rIs r'__repr__zQueryPaginator.__repr__s~~&&'(kk**+,%%&' |$#';;#:TZZJ   AJJ   r&)r2zUnion[Query, Select]r3ztype[ResponseSchemaType]r5rr*r r+rr,rr-rr.boolreturnNone)rwr)rwrv)N)rrrz int | Nonerwr)rwr)rwr)rwr)rwz$PaginationResult[ResponseSchemaType])rwzdict[str, Any])r r!r"r#rr>propertyr@rJrNrPrRrTrVrXrZr\rhrkrprur%r&r'r)r)%sE$X%>#>)> >  >>>>> >B(//::55 =66 66 ""11$L 0 ,  r&r))r# __future__rrLtypingrrrrpydanticrsqlalchemy.orm.queryr sqlalchemy.sqlr sqlalchemy.ormr sqlalchemyr r src.core.utils.enumsrrrrr)r%r&r'rsb# //&!"#?1C %y'*<"= %t W/0t r&