rattail/wuttapos
1
0
Fork 0
Code Issues Pull requests Wiki Activity

Add support for suspend/resume txn

Browse source
This commit is contained in:
Lance Edgar 2023-10-11 23:14:43 -05:00
parent 52c35ed85c
commit e7ecd88e64
3 changed files with 336 additions and 84 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

150
wuttapos/controls/lookup.py
View file

@ -40,10 +40,12 @@ class WuttaLookup(WuttaControl):
long_scroll_delta = 500
def __init__(self, *args, **kwargs):
self.show_search = kwargs.pop('show_search', True)
self.initial_search = kwargs.pop('initial_search', None)
self.allow_empty_query = kwargs.pop('allow_empty_query', False)
self.on_select = kwargs.pop('on_select', None)
self.on_cancel = kwargs.pop('on_cancel', None)
super().__init__(*args, **kwargs)
# track current selection
@ -52,11 +54,6 @@ class WuttaLookup(WuttaControl):
def build(self):
self.searchbox = ft.TextField("", text_size=self.font_size * 0.8,
on_submit=self.lookup,
autofocus=True,
expand=True)
self.search_results = ft.DataTable(
columns=[ft.DataColumn(self.make_cell_text(text))
for text in self.get_results_columns()],
@ -99,67 +96,75 @@ class WuttaLookup(WuttaControl):
scroll=ft.ScrollMode.AUTO,
)
return ft.Column(
[
controls = []
if self.show_search:
self.searchbox = ft.TextField("", text_size=self.font_size * 0.8,
on_submit=self.lookup,
autofocus=True,
expand=True)
controls.extend([
ft.Row(
[
ft.Text("SEARCH FOR:"),
self.searchbox,
ft.Container(
content=ft.Text("Lookup", size=self.font_size * 0.8),
alignment=ft.alignment.center,
height=self.default_button_height_dlg * 0.8,
width=self.default_button_height_dlg * 1.3,
border=ft.border.all(1, 'black'),
border_radius=ft.border_radius.all(5),
on_click=self.lookup,
bgcolor='blue',
),
ft.Container(
content=ft.Text("Reset", size=self.font_size * 0.8),
alignment=ft.alignment.center,
height=self.default_button_height_dlg * 0.8,
width=self.default_button_height_dlg * 1.3,
border=ft.border.all(1, 'black'),
border_radius=ft.border_radius.all(5),
on_click=self.reset,
bgcolor='yellow',
),
ft.Container(
content=ft.Text("Cancel", size=self.font_size * 0.8),
alignment=ft.alignment.center,
height=self.default_button_height_dlg * 0.8,
width=self.default_button_height_dlg * 1.3,
border=ft.border.all(1, 'black'),
border_radius=ft.border_radius.all(5),
on_click=self.cancel,
),
self.make_button("Lookup",
font_size=self.font_size * 0.8,
height=self.default_button_height_dlg * 0.8,
width=self.default_button_height_dlg * 1.3,
on_click=self.lookup,
bgcolor='blue'),
self.make_button("Reset",
font_size=self.font_size * 0.8,
height=self.default_button_height_dlg * 0.8,
width=self.default_button_height_dlg * 1.3,
on_click=self.reset,
bgcolor='yellow'),
],
),
ft.Divider(),
WuttaKeyboard(self.config, on_keypress=self.keypress,
on_long_backspace=self.long_backspace),
ft.Divider(),
ft.Row(
[
self.search_results_wrapper,
ft.VerticalDivider(),
ft.Column(
[
self.select_button,
ft.Row(),
ft.Row(),
ft.Row(),
ft.Row(),
ft.Row(),
self.up_button,
self.down_button,
],
),
],
vertical_alignment=ft.CrossAxisAlignment.START,
),
],
])
controls.extend([
ft.Divider(),
ft.Row(
[
self.search_results_wrapper,
ft.VerticalDivider(),
ft.Column(
[
self.select_button,
ft.Row(),
ft.Row(),
ft.Row(),
ft.Row(),
ft.Row(),
self.up_button,
self.down_button,
ft.Row(),
ft.Row(),
ft.Row(),
ft.Row(),
ft.Row(),
self.make_button("Cancel",
font_size=self.font_size * 0.8,
height=self.default_button_height_dlg * 0.8,
width=self.default_button_height_dlg * 1.3,
on_click=self.cancel),
],
),
],
vertical_alignment=ft.CrossAxisAlignment.START,
),
])
return ft.Container(
content=ft.Column(controls=controls),
height=None if self.show_search else 600,
)
def get_results_columns(self):
@ -167,7 +172,8 @@ class WuttaLookup(WuttaControl):
def did_mount(self):
if self.initial_search is not None:
self.searchbox.value = self.initial_search
if self.show_search:
self.searchbox.value = self.initial_search
self.initial_search = None # only do it once
self.update()
self.lookup()
@ -205,11 +211,15 @@ class WuttaLookup(WuttaControl):
return obj
def lookup(self, e=None):
entry = self.searchbox.value
if not entry and not self.allow_empty_query:
self.searchbox.focus()
self.update()
return
if self.show_search:
entry = self.searchbox.value
if not entry and not self.allow_empty_query:
self.searchbox.focus()
self.update()
return
else:
entry = None
session = self.app.make_session()
results = self.get_results(session, entry)
@ -231,21 +241,27 @@ class WuttaLookup(WuttaControl):
self.no_results.visible = False
else:
self.no_results.value = f"NO RESULTS FOR: {entry}"
if self.show_search:
self.no_results.value = f"NO RESULTS FOR: {entry}"
else:
self.no_results.value = "NO RESULTS FOUND"
self.no_results.visible = True
self.searchbox.focus()
if self.show_search:
self.searchbox.focus()
self.update()
def reset(self, e):
self.searchbox.value = ""
if self.show_search:
self.searchbox.value = ""
self.search_results.rows.clear()
self.no_results.visible = False
self.selected_uuid = None
self.selected_datarow = None
self.select_button.disabled = True
self.select_button.bgcolor = self.disabled_bgcolor
self.searchbox.focus()
if self.show_search:
self.searchbox.focus()
self.update()
def set_selection(self, row):

91
wuttapos/controls/txnlookup.py Normal file
View file

@ -0,0 +1,91 @@
# -*- coding: utf-8; -*-
################################################################################
#
# WuttaPOS -- Pythonic Point of Sale System
# Copyright © 2023 Lance Edgar
#
# This file is part of WuttaPOS.
#
# WuttaPOS is free software: you can redistribute it and/or modify it under the
# terms of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later
# version.
#
# WuttaPOS is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# WuttaPOS. If not, see <http://www.gnu.org/licenses/>.
#
################################################################################
"""
WuttaPOS - transaction lookup control
"""
from .lookup import WuttaLookup
class WuttaTransactionLookup(WuttaLookup):
def __init__(self, *args, **kwargs):
# nb. this forces first query
kwargs.setdefault('initial_search', True)
# TODO: how to deal with 'modes'
self.mode = kwargs.pop('mode', None)
if not self.mode:
raise ValueError("must specify mode")
if self.mode != 'resume':
raise ValueError("only 'resume' mode is supported")
kwargs.setdefault('show_search', False)
super().__init__(*args, **kwargs)
def get_results_columns(self):
return [
"Date/Time",
"Terminal",
"Txn ID",
"Cashier",
"Customer",
"Balance",
]
def get_results(self, session, entry):
model = self.app.model
# TODO: how to deal with 'modes'
assert self.mode == 'resume'
query = session.query(model.POSBatch)\
.filter(model.POSBatch.status_code == model.POSBatch.STATUS_SUSPENDED)\
.filter(model.POSBatch.executed == None)\
.order_by(model.POSBatch.created.desc())
transactions = []
for batch in query:
# TODO: should use 'suspended' timestamp instead here?
dt = self.app.localtime(batch.created, from_utc=True)
transactions.append({
'uuid': batch.uuid,
'datetime': self.app.render_datetime(dt),
'terminal': batch.terminal_id,
'txnid': batch.id_str,
'cashier': str(batch.cashier or ''),
'customer': str(batch.customer or ''),
'balance': self.app.render_currency(batch.get_balance()),
})
return transactions
def make_result_row(self, txn):
return [
txn['datetime'],
txn['terminal'],
txn['txnid'],
txn['cashier'],
txn['customer'],
txn['balance'],
]

179
wuttapos/views/pos.py
View file

@ -35,6 +35,7 @@ from wuttapos.controls.loginform import WuttaLoginForm
from wuttapos.controls.custlookup import WuttaCustomerLookup
from wuttapos.controls.itemlookup import WuttaProductLookup
from wuttapos.controls.deptlookup import WuttaDepartmentLookup
from wuttapos.controls.txnlookup import WuttaTransactionLookup
from wuttapos.controls.txnitem import WuttaTxnItem
from wuttapos.controls.tenkey import WuttaTenkeyMenu
@ -708,6 +709,21 @@ class POSView(WuttaView):
),
expand=0)
self.suspend_menu = ft.Container(
content=ft.Column(
[
ft.Row(
[
self.make_suspend_button(),
self.make_resume_button(),
],
spacing=0,
),
],
spacing=0,
),
expand=0)
self.set_quantity = ft.Text(value=None, data=None, weight=ft.FontWeight.BOLD, size=40)
return [
@ -748,7 +764,12 @@ class POSView(WuttaView):
self.meta_menu,
],
),
self.context_menu,
ft.Row(
[
self.context_menu,
self.suspend_menu,
],
),
],
),
],
@ -796,6 +817,124 @@ class POSView(WuttaView):
'tender_name': "Check"}
return self.make_tender_button(check, **kwargs)
def make_suspend_button(self, **kwargs):
return self.make_button("SUSPEND", bgcolor='purple',
font_size=self.default_font_size,
height=self.default_button_size,
width=self.default_button_size * 2,
on_click=self.suspend_click)
def suspend_click(self, e):
session = self.app.make_session()
batch = self.get_current_batch(session, create=False)
session.close()
# nothing to suspend if no batch
if not batch:
self.show_snackbar("NO TRANSACTION", bgcolor='yellow')
self.reset()
return
def confirm(e):
dlg.open = False
self.page.update()
# nb. do this just in case we must show login dialog
# cf. https://github.com/flet-dev/flet/issues/1670
time.sleep(0.1)
self.authorized_action('pos.suspend', self.suspend_transaction,
message="Suspend Transaction")
def cancel(e):
dlg.open = False
self.reset()
# prompt to suspend
dlg = ft.AlertDialog(
title=ft.Text("Confirm SUSPEND"),
content=ft.Text("Really SUSPEND transaction?"),
actions=[
self.make_button(f"Yes, SUSPEND", font_size=self.default_font_size,
height=self.default_button_size,
width=self.default_button_size * 3,
bgcolor='yellow',
on_click=confirm),
self.make_button("Cancel", font_size=self.default_font_size,
height=self.default_button_size,
width=self.default_button_size * 2.5,
on_click=cancel),
])
self.page.dialog = dlg
dlg.open = True
self.page.update()
def suspend_transaction(self, user):
session = self.app.make_session()
batch = self.get_current_batch(session)
user = session.get(user.__class__, user.uuid)
handler = self.get_batch_handler()
handler.suspend_transaction(batch, user)
session.commit()
session.close()
self.clear_all()
self.reset()
def make_resume_button(self, **kwargs):
return self.make_button("RESUME", bgcolor='purple',
font_size=self.default_font_size,
height=self.default_button_size,
width=self.default_button_size * 2,
on_click=self.resume_click)
def resume_click(self, e):
session = self.app.make_session()
batch = self.get_current_batch(session, create=False)
session.close()
# can't resume if txn in progress
if batch:
self.show_snackbar("TRANSACTION IN PROGRESS", bgcolor='yellow')
self.reset()
return
def select(uuid):
session = self.app.make_session()
user = self.get_current_user(session)
handler = self.get_batch_handler()
# TODO: this would need to work differently if suspended
# txns are kept in a central server DB
batch = session.get(self.app.model.POSBatch, uuid)
batch = handler.resume_transaction(batch, user)
session.commit()
session.refresh(batch)
self.load_batch(batch)
session.close()
dlg.open = False
self.reset()
def cancel(e):
dlg.open = False
self.reset()
# prompt to choose txn
dlg = ft.AlertDialog(
title=ft.Text("Resume Transaction"),
content=WuttaTransactionLookup(self.config, mode='resume',
on_select=select, on_cancel=cancel),
)
self.page.dialog = dlg
dlg.open = True
self.page.update()
def get_current_user(self, session):
uuid = self.page.session.get('user_uuid')
if uuid:
@ -819,30 +958,36 @@ class POSView(WuttaView):
session = self.app.make_session()
batch = self.get_current_batch(session, create=False)
if batch:
handler = self.get_batch_handler()
self.page.session.set('txn_display', handler.get_screen_txn_display(batch))
self.page.session.set('cust_uuid', batch.customer_uuid)
self.page.session.set('cust_display', handler.get_screen_cust_display(batch=batch))
self.items.controls.clear()
for row in batch.active_rows():
self.add_row_item(row)
self.items.scroll_to(offset=-1, duration=100)
self.refresh_totals(batch)
self.load_batch(batch)
else:
self.page.session.set('txn_display', None)
self.page.session.set('cust_uuid', None)
self.page.session.set('cust_display', None)
self.informed_refresh()
self.informed_refresh()
session.commit()
# TODO: i think commit() was for when it auto-created the
# batch, so that can go away now..right?
#session.commit()
session.close()
self.page.update()
def load_batch(self, batch):
"""
Load the given batch as the current transaction.
"""
handler = self.get_batch_handler()
self.page.session.set('txn_display', handler.get_screen_txn_display(batch))
self.page.session.set('cust_uuid', batch.customer_uuid)
self.page.session.set('cust_display', handler.get_screen_cust_display(batch=batch))
self.items.controls.clear()
for row in batch.active_rows():
self.add_row_item(row)
self.items.scroll_to(offset=-1, duration=100)
self.refresh_totals(batch)
self.informed_refresh()
def not_supported(self, e=None, feature=None):
# test error handler