| 9 | from ..models import ContactEntryModel |
| 10 | |
| 11 | class ContactState(SessionState): |
| 12 | form_data: dict = {} |
| 13 | entries: List['ContactEntryModel'] = [] |
| 14 | did_submit: bool = False |
| 15 | |
| 16 | @rx.var |
| 17 | def thank_you(self): |
| 18 | first_name = self.form_data.get("first_name") or "" |
| 19 | return f"Thank you {first_name}".strip() + "!" |
| 20 | |
| 21 | async def handle_submit(self, form_data: dict): |
| 22 | """Handle the form submit.""" |
| 23 | # print(form_data) |
| 24 | self.form_data = form_data |
| 25 | data = {} |
| 26 | for k,v in form_data.items(): |
| 27 | if v == "" or v is None: |
| 28 | continue |
| 29 | data[k] = v |
| 30 | if self.my_user_id is not None: |
| 31 | data['user_id'] = self.my_user_id |
| 32 | if self.my_userinfo_id is not None: |
| 33 | data['userinfo_id'] = self.my_userinfo_id |
| 34 | with rx.session() as session: |
| 35 | db_entry = ContactEntryModel( |
| 36 | **data |
| 37 | ) |
| 38 | session.add(db_entry) |
| 39 | session.commit() |
| 40 | self.did_submit = True |
| 41 | yield |
| 42 | await asyncio.sleep(2) |
| 43 | self.did_submit = False |
| 44 | yield |
| 45 | |
| 46 | def list_entries(self): |
| 47 | with rx.session() as session: |
| 48 | entries = session.exec( |
| 49 | select(ContactEntryModel) |
| 50 | ).all() |
| 51 | self.entries = entries |
nothing calls this directly
no outgoing calls
no test coverage detected