[Python] How to work with client_storage without the "NotImplementedError" in flet, Python...

Discussão em 'Python' iniciado por Stack, Março 9, 2025.

  1. Stack

    Stack Membro Participativo

    Well, I want to write an async app on Flet with Telethon. It is my first time so do not judge strictly pls.

    That's my code:

    main.py

    import flet as ft
    import main_page


    async def main(page: ft.Page):
    page.title = "Telegram resender 3000"
    page.horizontal_alignment = ft.CrossAxisAlignment.START
    await page.update_async()

    await page.add_async(main_page.MainPage(page))


    if __name__ == "__main__":
    ft.app(target=main)


    main_page.py

    import flet as ft
    import telethon
    from .load_env import *
    import asyncio
    import os

    import support_classes as sc


    class MainPage(ft.UserControl):

    def __init__(self, page):
    super().__init__()
    self.page = page

    def build(self):
    self.idx = 0
    # create text
    self.add_account_text = ft.Text(value="Add your tg account here")

    # create input for writing phone number and ADD btn
    self.add_phone_number = ft.TextField(
    hint_text="Enter your phone number")
    self.add_phone_btn = ft.FloatingActionButton(
    icon=ft.icons.ADD, on_click=self.register_new_acc)

    self.accs_list = ft.Row(controls=[]) # create field for tg accs
    self.upload_tg_accs_btn = ft.IconButton(
    icon=ft.icons.DOWNLOAD,
    icon_color="blue400",
    icon_size=30,
    tooltip="Upload existing telegram accounts",
    on_click=self.get_tg_accs,
    )

    self.add_account_part = ft.Column(
    width=300,
    controls=[
    ft.Row(
    controls=[
    self.add_account_text,
    self.add_phone_number,
    self.add_phone_btn
    ]
    ),
    ft.Row(controls=[
    self.accs_list,
    self.upload_tg_accs_btn,
    ]),
    ])

    return self.add_account_part

    async def register_new_acc(self, e):

    # add field for code
    self.code_recieve_input = ft.TextField(
    hint_text="Enter the code and wait...")

    self.add_account_part.controls.append(self.code_recieve_input)
    await self.add_account_part.update_async()

    # create a session and get request for a code
    client = telethon.TelegramClient(
    str(self.idx), api_id=api_id, api_hash=api_hash)

    # save the phone
    phone = self.add_phone_number.value

    # make input empty
    self.add_phone_number.value = ""

    await client.start(phone=phone, code_callback=lambda: self.code_receive(phone))

    async def code_receive(self, phone):
    # wait for user's answer
    await asyncio.sleep(30)

    # remove code input
    self.add_account_part.controls.remove(self.code_recieve_input)
    await self.add_account_part.update_async()

    # add a new acc to a list
    acc = sc.TGAcc(
    num=self.code_recieve_input.value,
    idx=self.idx,
    del_func=self.delete_acc
    )

    self.accs_list.controls.append(acc)
    await self.accs_list.update_async()

    # add number to a storage
    tg_accs = {}
    # if we already have nums
    if self.page.client_storage.contains_key_async("tg_accs"):
    tg_accs = self.page.client_storage.get("tg_accs")

    self.page.client_storage.set("tg_accs", tg_accs + {self.idx: phone})

    self.idx += 1 # increase the idx (of accounts)
    return self.code_recieve_input.value

    async def delete_acc(self, acc):
    # delete from list and update
    self.accs_list.controls.remove(acc)
    await self.accs_list.update_async()

    # remove session file
    os.remove(f'{acc.idx}.session')

    async def get_tg_accs(self, e):

    # create list of all connected accs
    if self.page.client_storage.contains_key('tg_accs'):
    accs_list = [sc.TGAcc(num=v, idx=k, del_func=self.delete_acc)
    for k, v in self.page.client_storage.get("tg_accs").items()]

    # set our data
    self.accs_list.controls = accs_list

    await self.accs_list.update_async()


    tg_acc.py

    import flet as ft
    import os

    class TGAcc(ft.UserControl):

    def __init__(self, num, idx, del_func):
    super().__init__()
    self.num = num
    self.idx = idx
    self.del_func = del_func

    def build(self):
    # build the container
    self.container = ft.Column(
    controls=[
    ft.Row(
    controls=[
    ft.Text(f"{self.idx} = {self.num}"),
    ft.IconButton(
    icon=ft.icons.DELETE_FOREVER_ROUNDED,
    icon_color="pink600",
    icon_size=30,
    tooltip="Delete account",
    on_click=self.delete_acc
    )
    ]
    )
    ]
    )

    return self.container

    async def delete_acc(self, e):
    await self.del_func(self)


    Also I have load_env.py file there I load all secret data.

    And I "sorted" these file into folders so that's the screen. I also added screen of UI. [​IMG]

    [​IMG]

    **So, the problem: ** I need to load data from client_storage in startup of a program and after write it into self.accs_list.controls. idk how do it on startup of all, so I created a download btn, by clicking on it data should load but all that loads is an exception raise NotImplementedError().

    I tried to delete self.page = page in init of MainPage and just use self.page everywhere in code below but no result has appeared(

    Here an error, that's always the same

    Task exception was never retrieved
    future: <Task finished name='Task-15' coro=<__connect_internal_async.<locals>.on_event() done, defined at /home/felix/anaconda3/envs/telegram_bot/lib/python3.10/site-packages/flet_runtime/app.py:425> exception=NotImplementedError()>
    Traceback (most recent call last):
    File "/home/felix/anaconda3/envs/telegram_bot/lib/python3.10/site-packages/flet_runtime/app.py", line 427, in on_event
    await conn.sessions[e.sessionID].on_event_async(
    File "/home/felix/anaconda3/envs/telegram_bot/lib/python3.10/site-packages/flet_core/page.py", line 482, in on_event_async
    await handler(ce)
    File "/home/felix/Documents/code/flet-telegram-repost/main_page/main_page.py", line 115, in get_tg_accs
    if self.page.client_storage.contains_key('tg_accs'):
    File "/home/felix/anaconda3/envs/telegram_bot/lib/python3.10/site-packages/flet_core/client_storage.py", line 46, in contains_key
    self.__page.invoke_method(
    File "/home/felix/anaconda3/envs/telegram_bot/lib/python3.10/site-packages/flet_core/page.py", line 859, in invoke_method
    result = self._send_command(
    File "/home/felix/anaconda3/envs/telegram_bot/lib/python3.10/site-packages/flet_core/page.py", line 700, in _send_command
    return self.__conn.send_command(
    File "/home/felix/anaconda3/envs/telegram_bot/lib/python3.10/site-packages/flet_core/connection.py", line 21, in send_command
    raise NotImplementedError()
    NotImplementedError


    idk how to fix my problem, in docs any examples with the client_storage + ft.UserControl I didnt find. That's my first time so I can do veeeery stupid mistakes.

    If anyone knows how to load data from client_storage on startup of Flet app write pls, I listen carefully)

    Continue reading...

Compartilhe esta Página