Controlling the Response object from endpoint

[perryrobinson]

Hello, I am trying to implement pagination into an API using the fastapi-pagination library. I am having an issue with it though and wondering if I am misunderstanding how to use this library.

Basically I am wondering how to control the actual response object from the endpoint. It seems like I have to type hint the endpoint with Page[SomeModel] but I don't want to actually return my SQL alchemy model object because it contains sensitive data.

Basically the router/controller calls the service and a Customer is represented by different object types in the different layers of the app. But what I am doing below does not work unless I explicitly type hint the endpoint with Page[CustomerModel].

Any ideas?

Here is an example:

# router 
from fastapi_pagination import Page, Params as PaginationParams
from fastapi_pagination.ext.sqlalchemy import paginate
from sqlalchemy.orm import Session

@company_router.get("/{company_id}/customers")
async def get_customers_by_company(company_id: int, pagination_params: PaginationParams = Depends()) -> Page[GetCustomerResponse]:
    result: Page[Customer] = company_service.get_customers_by_company(session=db.session, company_id=company_id, pagination_params=pagination_params)
    response_objects = [c.to_get_customer_response for c in result.items]
    return Page(items=response_objects, total=result.total, page=result.page, size=result.size , pages=result.pages)
    
# service
def get_customers_by_company(self, session: Session, company_id: str, pagination_params: PaginationParams) -> Page[Customer]:
    query = session.query(CustomerModel).filter(CustomerModel.company_id == company_id)
    result: Page[CustomerModel] = paginate(query, pagination_params)
    customers = [Customer.from_sql(model=customer_model) for customer_model in result.items]
    return Page(items=customers, total=result.total, page=result.page, size=result.size , pages=result.pages)

[uriyyo]

Hi @perryrobinson,

You can use transformer to prepare your items before passing them to Page instance:

# router
from fastapi_pagination import set_page, Page, Params as PaginationParams
from fastapi_pagination.ext.sqlalchemy import paginate
from sqlalchemy.orm import Session


@company_router.get("/{company_id}/customers")
async def get_customers_by_company(company_id: int, pagination_params: PaginationParams = Depends()) -> Page[GetCustomerResponse]:
    page = company_service.get_customers_by_company(session=db.session, company_id=company_id, pagination_params=pagination_params)

    return page.copy(update={"items": [customer.to_get_customer_response for customer in page.items]})


# service
def get_customers_by_company(self, session: Session, company_id: str, pagination_params: PaginationParams) -> Page[Customer]:
    query = session.query(CustomerModel).filter(CustomerModel.company_id == company_id)

    with set_page(Page[Customer]):
        return paginate(
            query,
            pagination_params,
            transformer=lambda items: [Customer.from_sql(model=item) for item in items],
        )