Is it possible to copy query instance?

[uriyyo]

Hi,

I am working on a pagination library, and for me it will be great to be able to copy instance of query.

Is there an existing API that allows to do it? Unfortunately, I didn't find anything(

Ideally, it can look like this:

q = User.select()

count_q = q.copy()
select_q = q.copy()

[dantownsend]

Yeah, it's a good point. We don't have an API for this yet, but I would like to add it.

Using this might be sufficient:

import copy

copy.deepcopy(my_query)

But I haven't tested it much.

[uriyyo]

Unfortunately, it's not working when query has a relation filter.
Here is an example:

import os
from asyncio import run
from copy import deepcopy
from itertools import count
from pathlib import Path
from random import choice

from faker import Faker
from piccolo.columns import ForeignKey, Integer, Text
from piccolo.conf.apps import AppConfig, AppRegistry
from piccolo.engine import SQLiteEngine, engine_finder
from piccolo.table import Table

os.environ["PICCOLO_CONF"] = __name__

_counter = count().__next__

faker = Faker()


class _Group(Table):
    id = Integer(default=_counter, primary_key=True)

    name = Text()


class _User(Table):
    id = Integer(default=_counter, primary_key=True)
    name = Text(required=False, null=True)
    email = Text(required=False, null=True)
    group = ForeignKey(_Group, null=True)


DB = SQLiteEngine()
APP_REGISTRY = AppRegistry()

APP_CONFIG = AppConfig(
    app_name="example",
    migrations_folder_path=None,
    table_classes=[_User],
)


async def setup() -> None:
    engine: SQLiteEngine = engine_finder()

    p = Path(engine.path)
    p.unlink(missing_ok=True)

    await engine.prep_database()
    await _Group.create_table().run()
    await _User.create_table().run()

    groups = [_Group(name=name) for name in ("admins", "users", "guests")]
    await _Group.insert(*groups)

    users = [
        _User(
            name=faker.name(),
            email=faker.email(),
            group=choice(groups),
        )
        for _ in range(100)
    ]
    await _User.insert(*users)


async def main() -> None:
    await setup()

    q = _User.select().where(_User.group.name == "admins")

    q_copy = deepcopy(q)

    await q_copy.run()


if __name__ == "__main__":
    run(main())

Traceback (most recent call last):
  File "/home/uriyyo/.cache/pypoetry/virtualenvs/fastapi-pagination-qCLVCiBY-py3.11/lib/python3.11/site-packages/piccolo/query/base.py", line 266, in querystrings
    return self.sqlite_querystrings
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/uriyyo/.cache/pypoetry/virtualenvs/fastapi-pagination-qCLVCiBY-py3.11/lib/python3.11/site-packages/piccolo/query/base.py", line 236, in sqlite_querystrings
    raise NotImplementedError
NotImplementedError

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/mnt/d/projects/fastapi-pagination/foo.py", line 79, in <module>
    run(main())
  File "/usr/lib/python3.11/asyncio/runners.py", line 190, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/asyncio/base_events.py", line 653, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/mnt/d/projects/fastapi-pagination/foo.py", line 75, in main
    await q_copy.run()
  File "/home/uriyyo/.cache/pypoetry/virtualenvs/fastapi-pagination-qCLVCiBY-py3.11/lib/python3.11/site-packages/piccolo/query/methods/select.py", line 836, in run
    results = await super().run(node=node, in_pool=in_pool)
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/uriyyo/.cache/pypoetry/virtualenvs/fastapi-pagination-qCLVCiBY-py3.11/lib/python3.11/site-packages/piccolo/query/base.py", line 197, in run
    return await self._run(node=node, in_pool=in_pool)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/uriyyo/.cache/pypoetry/virtualenvs/fastapi-pagination-qCLVCiBY-py3.11/lib/python3.11/site-packages/piccolo/query/base.py", line 176, in _run
    querystrings = self.querystrings
                   ^^^^^^^^^^^^^^^^^
  File "/home/uriyyo/.cache/pypoetry/virtualenvs/fastapi-pagination-qCLVCiBY-py3.11/lib/python3.11/site-packages/piccolo/query/base.py", line 268, in querystrings
    return self.default_querystrings
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/uriyyo/.cache/pypoetry/virtualenvs/fastapi-pagination-qCLVCiBY-py3.11/lib/python3.11/site-packages/piccolo/query/methods/select.py", line 797, in default_querystrings
    args.append(self.where_delegate._where.querystring)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/uriyyo/.cache/pypoetry/virtualenvs/fastapi-pagination-qCLVCiBY-py3.11/lib/python3.11/site-packages/piccolo/columns/combination.py", line 208, in querystring
    args.append(self.values_querystring)
                ^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/uriyyo/.cache/pypoetry/virtualenvs/fastapi-pagination-qCLVCiBY-py3.11/lib/python3.11/site-packages/piccolo/columns/combination.py", line 196, in values_querystring
    raise ValueError("values is undefined")
ValueError: values is undefined

[sinisaos]

@uriyyo I assume you need a copy of query instance in your library for this. I'm really sorry if I'm wrong and your solution works as it should, but couldn't it be done much simpler, something like this

__all__ = ["paginate"]

from typing import Any, Optional, Type, TypeVar, Union

from piccolo.query import Select
from piccolo.table import Table

from ..api import apply_items_transformer, create_page
from ..bases import AbstractParams
from ..types import AdditionalData, SyncItemsTransformer
from ..utils import verify_params
from .utils import generic_query_apply_params

T = TypeVar("T", bound=Table, covariant=True)


async def paginate(
    query: Union[Select[T], Type[T]],
    params: Optional[AbstractParams] = None,
    *,
    transformer: Optional[SyncItemsTransformer] = None,
    additional_data: Optional[AdditionalData] = None,
) -> Any:
    params, raw_params = verify_params(params, "limit-offset")

    if not isinstance(query, Select):
        query = query.select()

    total = len(await query)

    items = await generic_query_apply_params(query, raw_params)
    t_items = await apply_items_transformer(items, transformer, async_=True)

    return create_page(
        t_items,
        total=total,
        params=params,
        **(additional_data or {}),
    )

I tried it like this

@app.get("/users/default", response_model=Page[UserOut])
@app.get("/users/limit-offset", response_model=LimitOffsetPage[UserOut])
async def get_users() -> Any:
    # also work without FK relation (return await paginate(_User))
    return await paginate(_User.select().where(_User.group.name == "admins"))

and everything works. Again, I'm deeply sorry if I'm wrong and in that case please ignore my comment.

[uriyyo]

@sinisaos Your solution will work as expected but has one little performance issue.

    total = len(await query)

This code will fetch all rows and then call python len on it. Let's imagine a case when you have 100,000 rows, you will fetch all those rows in memory to calculate len which can lead to performance issues.

[sinisaos]

@uriyyo I agree with you that SQL COUNT is much more efficient than calculating len() in Python, but that was just an idea. In the end, any Page or LimitOffset pagination will lead to performance problems with large data sets. Sorry again and forget about my comment.

[uriyyo]

@sinisaos No worries, thanks for your suggestion

[KianAnbarestani]

Piccolo ORM doesn't support .copy() on query objects, but you can use Python's copy module:

import copy

q = User.select()
count_q = copy.copy(q)
select_q = copy.copy(q)

Alternatively, use a function to return a fresh query each time for better control.