# coding: utf-8
"""
Order data model/view/handler
"""
from svv.controllers import PageHandler
from svv.html import tags
from svv import database as db
import datetime
import logging
import collections
log = logging.getLogger('svv.orders')
class FormError (Exception) :
"""
A user-level error in a form field
"""
def __init__ (self, field, value, error) :
"""
field - name of field with error
value - the errenous value in the form that we recieved it
may be None if it was the *lack* of a value that caused the issue
error - descriptive text for user
"""
self.field = field
self.value = value
super(FormError, self).__init__(error)
class OrderForm (object) :
"""
A single instance of a <form>, where we can process submitted data from the client, storing the associated
Order-related data, and then render a form for any of that related data.
"""
# any POST data we have processed, updated from process()
data = None
def __init__ (self, app) :
"""
app - bind this form to the app state (db etc)
"""
self.app = app
# accumulated errors
self.errors = collections.defaultdict(list)
def defaults (self) :
"""
Update our attributes with default values
"""
self.customer_id = None
self.customer_name = None
self.contact_id = None
self.contact_name = None
self.contact_phone = None
self.contact_email = None
self.contact_customer = None
self.event_name = None
self.event_description = None
tomorrow = datetime.date.today() + datetime.timedelta(days=1)
# default to tomorrow afternoon
self.event_start = datetime.datetime.combine(tomorrow, datetime.time(16, 00))
# automatically determined once start is set
self.event_end = None
def process_raw_field (self, name, default=None, required=None) :
"""
Process a generic incoming data field.
default - value to return if no value was present
required - raise a FormError if no value present
Returns the value as a str, or default
"""
if name in self.data :
return self.data[name]
elif required :
raise FormError(name, None, "Required field")
else :
return default
def process_string_field (self, name, default=None, required=None, strip=True) :
"""
Process a generic incoming string field.
Trims extra whitespace from around the value, unless strip=False is given.
Returns the value as unicode, or default.
"""
value = self.process_raw_field(name, required=required)
if value is None :
return default
try :
# XXX: decode somehow, or can werkzeug handle that?
value = unicode(value)
except UnicodeDecodeError :
raise FormError(name, value, "Failed to decode Unicode characters")
if strip :
value = value.strip()
return value
def process_integer_field (self, name, default=None, required=None) :
"""
Process a generic incoming int field.
Returns the value as int, or default.
"""
value = self.process_raw_field(name, required=required)
if value is None :
return default
try :
return int(value)
except ValueError :
raise FormError(name, value, "Must be a number")
DATETIME_FORMAT = "%d.%m.%Y %H:%M"
def process_datetime_field (self, name, default=None, required=None, format=DATETIME_FORMAT) :
"""
Process an incoming datetime field.
Returns the value as datetime, or default.
"""
value = self.process_raw_field(name, required=required)
if value is None :
return default
try :
return datetime.datetime.strptime(value, format)
except ValueError, ex :
raise FormError(name, value, "Invalid date/time value: " + str(ex))
def process_multifield (self, table, id, fields) :
"""
Process a set of user-given field values for an object with an unique id, and some set of additional fields.
If the id is given, look up the corresponding field values, and return those.
If any of the fields are given, either look up a matching row, or create a new one, and return its id.
Returns an (id_name, field_name, ...) N-tuple.
"""
id_name, id_col, id_value = id
if id_value :
# look up object from db
columns = [col for name, col, value in fields]
sql = db.select(columns, (id_col == id_value))
for row in self.app.query(sql) :
# XXX: sanity-check row values vs our values
# new values
fields = tuple(
(name, col, row[col]) for name, col, value in fields
)
# ok, just use the first one
break
else :
# not found!
raise FormError(id_name, id_value, "Item selected does not seem to exist")
log.info("Lookup %s=%d -> %s", id_name, id_value, dict((name, value) for name, col, value in fields))
elif any(value for name, col, value in fields) :
# look up identical object from db?
sql = db.select([id_col], db.and_(*[(col == value) for name, col, value in fields]))
for row in self.app.query(sql) :
if id_value :
log.warn("Duplicate %s=%d for %s", id_name, id_value, dict((name, value) for name, col, value in fields))
# found object's id
id_value = row[id_col]
log.info("Found %s -> %d", dict((name, value) for name, col, value in fields), id_value)
# create new object?
if not id_value :
sql = db.insert(table).values(dict((col, value) for name, col, value in fields))
id_value, = self.app.insert(sql)
log.info("Create %s -> %d", dict((name, value) for name, col, value in fields), id_value)
else :
# not known
log.debug("No %s known for order...", id_name)
# return full set of values
return (id_value, ) + tuple(value for name, col, value in fields)
def process_customer (self) :
"""
Process the incoming customer_* fields, returning (customer_id, customer_name).
"""
try :
customer_id = self.process_integer_field('customer_id')
customer_name = self.process_string_field('customer_name')
if not customer_id and not customer_name :
raise FormError('customer_name', None, "Must enter a customer")
return self.process_multifield(db.customers,
('customer_id', db.customers.c.id, customer_id),
(
('customer_name', db.customers.c.name, customer_name),
),
)
except FormError, e :
# list it
self.fail_field(e, 'customer_name')
return None, None
def process_contact (self, customer_id) :
"""
Process the incoming contact_* fields, returning
(contact_id, contact_name, contact_phone, contact_email, contact_customer)
"""
try :
contact_id = self.process_integer_field('contact_id')
contact_name = self.process_string_field('contact_name')
contact_phone = self.process_string_field('contact_phone')
contact_email = self.process_string_field('contact_email')
contact_customer = customer_id
if not contact_id and not (contact_name or contact_phone or contact_email) :
raise FormError('contact_name', None, "Must enter a contact")
return self.process_multifield(db.contacts,
('contact_id', db.contacts.c.id, contact_id),
(
('contact_name', db.contacts.c.name, contact_name),
('contact_phone', db.contacts.c.phone, contact_phone),
('contact_email', db.contacts.c.email, contact_email),
('contact_customer', db.contacts.c.customer, contact_customer),
),
)
except FormError, e :
# list it
self.fail_field(e, 'contact_name' if e.field == 'contact_id' else None)
return None, None, None, None, None
def process_event (self) :
"""
Process the incoming event_* fields, returning
(event_name, event_description, event_start, event_end)
"""
try :
event_name = self.process_string_field('event_name')
event_description = self.process_string_field('event_description', strip=False)
event_start = self.process_datetime_field('event_start')
event_end = self.process_datetime_field('event_end')
if event_end < event_start :
raise FormError('event_start', event_end, "Event must end after start")
return (event_name, event_description, event_start, event_end)
except FormError, e :
# list it
self.fail_field(e)
return None, None, None, None
def process (self, data) :
"""
Bind ourselves to the given incoming POST data, and update our order field attributes.
data - the submitted POST data as a MultiDict
Returns True if all fields were processed without errors, False otherwise.
"""
# bind the raw post data
self.data = data
# customer
self.customer_id, self.customer_name = self.process_customer()
# contact
self.contact_id, self.contact_name, self.contact_phone, self.contact_email, self.contact_customer = self.process_contact(self.customer_id)
if self.contact_customer and not self.customer_id :
# TODO: be really smart?
pass
# event
self.event_name, self.event_description, self.event_start, self.event_end = self.process_event()
return not self.errors
def fail_field (self, form_error, field=None) :
"""
Mark the field mentioned inside the given FormError as failed.
form_error - the FormError to store
field - the name of the field to store the error under, if not the same as in form_error
"""
field = field or form_error.field
log.warn("Marking field %s as failed: %s", field, form_error)
self.errors[field].append(form_error)
def build_customer_list (self) :
"""
Query a (id, name) list of customers.
"""
sql = db.select([db.customers.c.id, db.customers.c.name])
return self.app.query(sql)
def build_contact_list (self, customer_id=None) :
"""
Query a (id, name, phone, email) list of contacts, optionally for given customer if given.
"""
sql = db.select([db.contacts.c.id, db.contacts.c.name, db.contacts.c.phone, db.contacts.c.email])
if customer_id :
sql = sql.where((db.contacts.c.customer == customer_id))
return self.app.query(sql)
def render_text_input (self, name, value=None, multiline=False) :
"""
Render HTML for a generic text field input.
name - field name, as used for POST
value - default field value
multiline - use a multi-line <textarea> instead
"""
if multiline :
# XXX: textarea can't be self-closing for some reason?
return tags.textarea(name=name, id=name, _selfclosing=False, _whitespace_sensitive=True)(value)
else :
return tags.input(type='text', name=name, id=name, value=value)
def render_select_input (self, name, options, value=None) :
"""
Render HTML for a generic select control.
name - field name, as used for POST
options - sequence of (value, title) options. `value` may be None to omit.
value - the selected value
"""
return tags.select(name=name, id=name)(
(
tags.option(value=opt_value, selected=('selected' if opt_value == value else None))(opt_title)
) for opt_value, opt_title in options
)
def render_datetime_input (self, name, value=None) :
"""
Render HTML for a generic datetime control (using jQuery).
name - field name, as used for POST
value - selected date
"""
return (
self.render_text_input(name, (value.strftime(self.DATETIME_FORMAT) if value else None)),
tags.script("$(document).ready(function () { $('#" + name + "').datetimepicker(); });"),
)
def render_customer_input (self) :
"""
Render HTML for customer_id/name field inputs.
"""
# all known customers
customers = list(self.build_customer_list())
return (
self.render_select_input('customer_id', [(0, u"Luo uusi")] + customers, self.customer_id),
self.render_text_input('customer_name', self.customer_name),
tags.script(r"$(document).ready(function () { $('#customer_id').formSelectPreset({textTarget: $('#customer_name')}); });"),
)
def render_contact_input (self) :
"""
Render HTML for contact name field <input>s
"""
# recommended contacts for selected customer, if known
contacts = self.build_contact_list(self.customer_id)
return (
self.render_select_input('contact_id', [(0, u"Luo uusi")] + [(id, name) for id, name, phone, email in contacts], self.contact_id),
self.render_text_input('contact_name', self.contact_name),
tags.script(r"$(document).ready(function () { $('#contact_id').formSelectPreset({textTarget: $('#contact_name')}); });"),
)
def render_event_input (self) :
"""
Render HTML for event start/end field <input>s
"""
return (
self.render_datetime_input('event_start', self.event_start),
" - ",
self.render_datetime_input('event_end', self.event_end),
tags.script(r"""
$(document).ready(function () {
var event_start = $('#event_start');
var event_end = $('#event_end');
/* Buggy shit doesn't work
{
beforeShow: function (input, inst) {
// copy default value from event_start
event_end.datetimepicker("option", "defaultDate", event_start.datetimepicker("getDate"));
}
}
event_start.change(function () {
// copy value as default
var start_date = event_start.datetimepicker("getDate");
event_end.datetimepicker("option", "defaultDate", start_date);
});
// init default as well
event_start.change();
*/
});""" ),
)
def render_form_field (self, name, title, description, inputs) :
"""
Render the label, input control, error note and description for a single field, along with their containing <li>.
"""
# any errors for this field?
errors = self.errors[name]
return tags.li(class_='field' + (' failed' if errors else ''))(
tags.label((
title,
tags.strong(u"(Virheellinen)") if errors else None,
), for_=name),
inputs,
tags.p(description),
# possible errors
tags.ul(class_='errors')(tags.li(error.message) for error in errors) if errors else None,
)
def render (self, action, submit=u"Tallenna") :
"""
Render the entire <form>, using any loaded/processed values.
action - the target URL for the form to POST to
submit - label for the submit button
"""
return tags.form(action=action, method='POST')(
(
tags.h3(u"Lomakkeessa oli virheitä"),
tags.p(u"Korjaa lomake ja lähetä uudelleen"),
tags.ul(class_='errors')(
tags.li(tags.a(href='#' + error.field)(
tags.em(error.field),
error.message,
)) for field_errors in self.errors.itervalues() for error in field_errors
),
) if self.errors else None,
tags.fieldset(
tags.legend(u"Tilaaja"),
tags.ol(
self.render_form_field('customer_name', u"Tilaaja", u"Tilaavan yhdistyksen/henkilön nimi", (
self.render_customer_input()
)),
self.render_form_field('contact_name', u"Yhteyshenkilö", u"Yhteyshenkilön nimi, jos eri kun tilaaja", (
self.render_contact_input()
)),
self.render_form_field('contact_phone', u"Puhelin", u"Yhteyshenkilön puhelinnumero", (
self.render_text_input('contact_phone', self.contact_phone)
)),
self.render_form_field('contact_email', u"Sähköposti", u"Yhteyshenkilön sähköpostiosoite", (
self.render_text_input('contact_email', self.contact_email)
)),
),
),
tags.fieldset(
tags.legend(u"Tapahtuma"),
tags.ol(
self.render_form_field('event_name', u"Tapahtuma", u"Tapahtuman lyhyt nimi", (
self.render_text_input('event_name', self.event_name)
)),
self.render_form_field('event_description', u"Lisätiedot", u"Tapahtuman tarkemmat tiedot", (
self.render_text_input('event_description', self.event_description, multiline=True)
)),
self.render_form_field('event_start', u"Ajankohta", u"Tapahtuman ajankohta (kamat noudetaan - palautetaan)", (
self.render_event_input()
)),
),
),
tags.input(type='submit', value=submit),
)
class OrdersView (PageHandler) :
def render_content (self) :
return tags.h1("Orders list")
class OrderView (PageHandler) :
def render_content (self, id) :
return tags.h1("Order info for #%d" % (id, ))
class NewOrderView (PageHandler) :
"""
Render form for input, let the user correct their errors, create the order, and redirect out.
"""
def create (self, form) :
"""
Create the new order from the given form data, returning the new order's ID
"""
# if we've gotten this far, then we can create it!
sql = db.insert(db.orders).values(
customer = form.customer_id,
contact = form.contact_id,
event_name = form.event_name,
event_description = form.event_description,
event_start = form.event_start,
event_end = form.event_end,
)
# go!
order_id, = self.app.insert(sql)
# great
return order_id
def process (self) :
"""
Set up up our form.
"""
self.form = OrderForm(self.app)
# use either POST data or defaults
if self.POST :
# try and process the input, checking for any failures...
if self.form.process(self.POST) :
# should be good, create it!
order_id = self.create(self.form)
# redirect there now that our business is done and the order exists
return self.redirect_for(OrderView, id=order_id)
else :
# errors in form input
pass
else :
# init from defaults
self.form.defaults()
def render_content (self) :
"""
Render our form
"""
return (
tags.h1(u"Uusi tilaus"),
self.form.render(action=self.url_for(NewOrderView))
)