from functools import cached_property
from datetime import date
from onegov.activity import InvoiceItem
from typing import Literal, TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Collection, Iterator
from sqlalchemy.orm import Session
from uuid import UUID
[docs]
class InvoiceAction:
def __init__(
self,
session: 'Session',
id: 'UUID',
action: Literal['mark-paid', 'mark-unpaid', 'remove-manual'],
extend_to: Literal['invoice', 'family'] | None = None,
text: str | None = None
) -> None:
[docs]
self.extend_to = extend_to
@cached_property
[docs]
def item(self) -> InvoiceItem | None:
return self.session.query(InvoiceItem).filter_by(id=self.id).first()
@property
[docs]
def valid(self) -> bool:
if self.action not in ('mark-paid', 'mark-unpaid', 'remove-manual'):
return False
if self.extend_to not in (None, 'invoice', 'family'):
return False
if not self.item:
return False
if self.extend_to == 'family' and self.action != 'remove-manual':
return False
return True
@property
[docs]
def targets(self) -> 'Iterator[InvoiceItem]':
item = self.item
if item:
yield item
if self.extend_to == 'invoice':
q = self.session.query(InvoiceItem)
q = q.filter(InvoiceItem.id != item.id)
q = q.filter(InvoiceItem.invoice_id == item.invoice_id)
yield from q
if self.extend_to == 'family':
q = self.session.query(InvoiceItem)
q = q.filter(InvoiceItem.id != item.id)
q = q.filter(InvoiceItem.family == item.family)
yield from q
[docs]
def execute(self) -> None:
if self.action == 'mark-paid':
self.execute_mark_paid(tuple(self.targets))
elif self.action == 'mark-unpaid':
self.execute_mark_unpaid(tuple(self.targets))
elif self.action == 'remove-manual':
assert self.extend_to in (None, 'family')
self.execute_remove_manual(tuple(self.targets))
else:
raise NotImplementedError()
[docs]
def assert_safe_to_change(
self,
targets: 'Collection[InvoiceItem]'
) -> None:
for target in targets:
if target.invoice.disable_changes_for_items((target, )):
raise RuntimeError('Item was paid online')
[docs]
def execute_mark_paid(self, targets: 'Collection[InvoiceItem]') -> None:
self.assert_safe_to_change(targets)
for target in targets:
target.payment_date = date.today()
target.paid = True
[docs]
def execute_mark_unpaid(self, targets: 'Collection[InvoiceItem]') -> None:
self.assert_safe_to_change(targets)
for target in targets:
target.payment_date = None
target.paid = False
target.tid = None
target.source = None
[docs]
def execute_remove_manual(
self,
targets: 'Collection[InvoiceItem]'
) -> None:
self.assert_safe_to_change(targets)
for target in targets:
assert target.family and target.family.startswith('manual-')
self.session.delete(target)