TanStack Query for Flet

[50Bytes-dev]

I want to make a similar package for Flet. I would be glad to help in realization. I think for Flet it would be a very cool addition that would allow to easily create large projects.

Source https://tanstack.com/query/latest

[50Bytes-dev]

@FeodorFitsner There's progress. I've made a framework that works fine but am stuck on the controllers. Any ideas how to make the controller react reactively to changes in the request state?

import asyncio
import flet as ft
from flet_query.hooks.use_query import use_query


class Text(ft.Text):
    def __init__(self, text):
        super().__init__(text)

    def did_mount(self):
        print("Mounted")
        return super().did_mount()

    def will_unmount(self):
        print("Unmounted")
        return super().will_unmount()

    def build(self):
        print("Building")
        return super().build()

    def before_update(self):
        print("Before updating")
        return super().before_update()

    def update(self):
        print("Updating")
        return super().update()


async def get_posts():
    await asyncio.sleep(5)
    return ["Post 1", "Post 2", "Post 3"]


async def main(page: ft.Page):
    posts = use_query(
        ("posts",),
        get_posts,
    )

    page.add(Text(posts.status))


ft.app(main)

def use_query(
    query_key: QueryKey,
    fetcher: Callable[..., Awaitable[Any]],
    enabled: bool = True,
    refetch_on_mount: Optional[RefetchOnMount] = None,
    stale_duration: Optional[int] = None,
    cache_duration: Optional[int] = None,
    refetch_interval: Optional[int] = None,
    retry_count: int = 3,
    retry_delay: int = 1500,
):
    options = UseQueryOptions(
        enabled=enabled,
        refetch_on_mount=refetch_on_mount,
        stale_duration=stale_duration,
        cache_duration=cache_duration,
        refetch_interval=refetch_interval,
        retry_count=retry_count,
        retry_delay=retry_delay,
    )

    client = QueryClient()

    observer = Observer(
        query_key=query_key,
        fetcher=fetcher,
        client=client,
        options=options,
    )

    result = UseQueryResult(
        data=observer.query.state.data,
        data_updated_at=observer.query.state.data_updated_at,
        error=observer.query.state.error,
        error_updated_at=observer.query.state.error_updated_at,
        is_error=observer.query.state.is_error,
        is_loading=observer.query.state.is_loading,
        is_fetching=observer.query.state.is_fetching,
        is_success=observer.query.state.is_success,
        status=observer.query.state.status,
        refetch=observer.fetch,
    )

    def on_state_changed():
        result.data = observer.query.state.data
        result.data_updated_at = observer.query.state.data_updated_at
        result.error = observer.query.state.error
        result.error_updated_at = observer.query.state.error_updated_at
        result.is_error = observer.query.state.is_error
        result.is_loading = observer.query.state.is_loading
        result.is_fetching = observer.query.state.is_fetching
        result.is_success = observer.query.state.is_success
        result.status = observer.query.state.status
        result.refetch = observer.fetch

        print(f"State changed {result.status}")

        ft.context.page.update()

    observer.subscribe(on_state_changed)

    ft.context.page.run_task(observer.initialize)

    return result

[50Bytes-dev]

This is the version of the code that works. What do you guys think?

import asyncio
import flet as ft
from flet_query.hooks.use_query import use_query


class StatusText(ft.Text):
    def __init__(self):
        self.posts = use_query(
            ("posts",),
            get_posts,
        )
        super().__init__(self.posts.status)

    def before_update(self):
        self.value = self.posts.status
        return super().before_update()


async def get_posts():
    await asyncio.sleep(5)
    return ["Post 1", "Post 2", "Post 3"]


async def main(page: ft.Page):
    text = StatusText()
    page.add(text)


ft.app(main)

[50Bytes-dev]

This is the version I ended up with. I don't know how to improve it yet...
I look forward to your feedback

async def get_posts(search_text: str = ""):
    print("Getting posts...")
    await asyncio.sleep(3)
    posts = ["1 Post", "2 Post", "3 Post", "4 Post", "5 Post"]
    return [post for post in posts if search_text.lower() in post.lower()]


class PostsList(ft.Column):
    def __init__(self):

        self.search_text = ft.TextField(
            value="",
            on_change=lambda e: self.posts.refetch(),
        )

        async def request_posts():
            return await get_posts(search_text=self.search_text.value or "")

        self.posts = use_query(
            ("posts", self.search_text.value),
            request_posts,
        )

        super().__init__(controls=[ft.Text("Loading posts...")])

    def before_update(self):
        if self.posts.status == QueryStatus.error:
            self.controls = [ft.Text("Error fetching posts")]

        if self.posts.status == QueryStatus.success and self.posts.data:
            self.controls = [
                self.search_text,
                *(ft.Text(post) for post in self.posts.data),
            ]


async def main(page: ft.Page):
    posts = PostsList()
    page.add(posts)


ft.app(main)

[FeodorFitsner]

So, it's sort of reactive framework but for Python?

[50Bytes-dev]

Kind of. But I don't like the way I did it. I would like to be able to implement components with query more easily

[50Bytes-dev]

Any ideas?

[ndonkoHenri]

Instead of sharing code this way, I suggest you create a repo, then share the link and explain how to test. :)

[50Bytes-dev]

@ndonkoHenri @FeodorFitsner
Published. See the tests folder. I would like to see your API implementation options that would be more convenient than what I got.
https://github.com/50Bytes-dev/flet-query

[Soif2Sang]

I'm definitely interested. I'm keeping an eye on your repo