dinislage/web2py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge ssh://github.com/web2py/web2py

Browse Source
This commit is contained in:
Michele Comitini
2013-08-22 11:09:03 +02:00
parent 9228204722 1d14949e17
commit 21b1b0fa48
8 changed files with 399 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
VERSION
+1 -1
View File
@@ -1 +1 @@
Version 2.6.0-development+timestamp.2013.08.21.04.28.27
Version 2.6.0-development+timestamp.2013.08.22.03.46.07
gluon/contrib/memdb.py
-2
View File
@@ -302,8 +302,6 @@ class Table(DALStorage):
def __call__(self, id):
return self.get(id)
def __getitem__(self,id):
return self.get(id)
class Expression(object):
gluon/contrib/pdfinvoice.pdf
+161
View File
@@ -0,0 +1,161 @@
"""
BSD license - created by Massimo Di Pierro
"""
from reportlab.pdfgen.canvas import Canvas
from reportlab.platypus import Table
from reportlab.lib.pagesizes import A4
from reportlab.lib.units import cm
from decimal import Decimal
import cStringIO
import datetime
def listify(item):
if isinstance(item,basestring):
item = item.split('\n')
return item
class PDF(object):
def __init__(self, page_size=A4, font_face='Helvetica'):
self.page_size = page_size
self.font_face = font_face
self.logo = None
def format_currency(self,value):
a = list(str(int(value)))
for k in range(len(a)-3,0,-3):
a.insert(k,',')
a = ''.join(a)
b = ("%.2f" % (value-int(value)))[2:]
return "%s.%s" % (a,b)
def draw(self, invoice, items_page=10):
""" Draws the invoice """
buffer = cStringIO.StringIO()
invoice_items = invoice['items']
pages = max((len(invoice_items)-2)/items_page+1,1)
canvas = Canvas(buffer, pagesize=self.page_size)
for page in range(pages):
canvas.translate(0, 29.7 * cm)
canvas.setFont(self.font_face, 10)
canvas.saveState()
canvas.setStrokeColorRGB(0.9, 0.5, 0.2)
canvas.setFillColorRGB(0.2, 0.2, 0.2)
canvas.setFont(self.font_face, 16)
canvas.drawString(1 * cm, -1 * cm, invoice.get('title',''))
if self.logo:
canvas.drawInlineImage(self.logo, 1 * cm, -1 * cm, 250, 16)
canvas.setLineWidth(4)
canvas.line(0, -1.25 * cm, 21.7 * cm, -1.25 * cm)
canvas.restoreState()
canvas.saveState()
notes = listify(invoice.get('notes',''))
textobject = canvas.beginText(1 * cm, -25 * cm)
for line in notes:
textobject.textLine(line)
canvas.drawText(textobject)
textobject = canvas.beginText(18 * cm, -28 * cm)
textobject.textLine('Pag.%s/%s' % (page+1,pages))
canvas.drawText(textobject)
canvas.restoreState()
canvas.saveState()
business_details = listify(invoice.get('from','FROM:'))
canvas.setFont(self.font_face, 9)
textobject = canvas.beginText(13 * cm, -2.5 * cm)
for line in business_details:
textobject.textLine(line)
canvas.drawText(textobject)
canvas.restoreState()
canvas.saveState()
client_info = listify(invoice.get('to','TO:'))
textobject = canvas.beginText(1.5 * cm, -2.5 * cm)
for line in client_info:
textobject.textLine(line)
canvas.drawText(textobject)
canvas.restoreState()
textobject = canvas.beginText(1.5 * cm, -6.75 * cm)
textobject.textLine(u'Invoice ID: %s' % invoice.get('id','<invoice id>'))
textobject.textLine(u'Invoice Date: %s' % invoice.get('date',datetime.date.today()))
textobject.textLine(u'Client: %s' % invoice.get('client_name','<invoice client>'))
canvas.drawText(textobject)
items = invoice_items[1:][page*items_page:(page+1)*items_page]
if items:
data = [invoice_items[0]]
for item in items:
data.append([
self.format_currency(x)
if isinstance(x,float) else x
for x in item])
righta = [k for k,v in enumerate(items[0])
if isinstance(v,(int,float,Decimal))]
if page == pages-1:
total = self.format_currency(invoice['total'])
else:
total = ''
data.append(['']*(len(items[0])-1)+[total])
colWidths = [2.5*cm]*len(items[0])
colWidths[1] = (21.5-2.5*len(items[0]))*cm
table = Table(data, colWidths=colWidths)
table.setStyle([
('FONT', (0, 0), (-1, -1), self.font_face),
('FONTSIZE', (0, 0), (-1, -1), 8),
('TEXTCOLOR', (0, 0), (-1, -1), (0.2, 0.2, 0.2)),
('GRID', (0, 0), (-1, -2), 1, (0.7, 0.7, 0.7)),
('GRID', (-1, -1), (-1, -1), 1, (0.7, 0.7, 0.7)),
('BACKGROUND', (0, 0), (-1, 0), (0.8, 0.8, 0.8)),
]+[('ALIGN',(k,0),(k,-1),'RIGHT') for k in righta])
tw, th, = table.wrapOn(canvas, 15 * cm, 19 * cm)
table.drawOn(canvas, 1 * cm, -8 * cm - th)
if page == pages-1:
items = invoice['totals'][1:]
if items:
data = [invoice['totals'][0]]
for item in items:
data.append([
self.format_currency(x)
if isinstance(x,float) else x
for x in item])
righta = [k for k,v in enumerate(items[0])
if isinstance(v,(int,float,Decimal))]
total = self.format_currency(invoice['total'])
data.append(['']*(len(items[0])-1)+[total])
colWidths = [2.5*cm]*len(items[0])
colWidths[1] = (21.5-2.5*len(items[0]))*cm
table = Table(data, colWidths=colWidths)
table.setStyle([
('FONT', (0, 0), (-1, -1), self.font_face),
('FONTSIZE', (0, 0), (-1, -1), 8),
('TEXTCOLOR', (0, 0), (-1, -1), (0.2, 0.2, 0.2)),
('GRID', (0, 0), (-1, -2), 1, (0.7, 0.7, 0.7)),
('GRID', (-1, -1), (-1, -1), 1, (0.7, 0.7, 0.7)),
('BACKGROUND', (0, 0), (-1, 0), (0.8, 0.8, 0.8)),
]+[('ALIGN',(k,0),(k,-1),'RIGHT') for k in righta])
tw, th, = table.wrapOn(canvas, 15 * cm, 19 * cm)
table.drawOn(canvas, 1 * cm, -18 * cm - th)
canvas.showPage()
canvas.save()
return buffer.getvalue()
if __name__=='__main__':
invoice = {
'title': 'Invoice - web2py.com',
'id': '00001',
'date': '10/10/2013',
'client_name': 'Nobody',
'from': 'FROM:\nweb2py.com\nWabash ave\nChicago',
'to': 'TO:\nNobody\nHis address',
'notes': 'no comment!',
'total': 650.00,
'items': [
['Codice','Desc','Quantity','Unit price','Total']]+[
['000001','Chair',2,10.0,20.0] for k in range(30)],
'totals': [
['Codice','Desc','Total']]+[
['000001','Chairs',600.0],
['','Tax',50.0]],
}
print PDF().draw(invoice,items_page=20)
gluon/dal.py
+192 -90
View File
@@ -6932,7 +6932,7 @@ def sqlhtml_validators(field):
refs = reduce(lambda a,b:a&b, [count(ids[i:i+30]) for i in rx])
else:
refs = db(id.belongs(ids)).select(id)
return (refs and ', '.join(str(f(r,x.id)) for x in refs) or '')
return (refs and ', '.join(f(r,x.id) for x in refs) or '')
field.represent = field.represent or list_ref_repr
if hasattr(referenced, '_format') and referenced._format:
requires = validators.IS_IN_DB(db,referenced._id,
@@ -7252,7 +7252,7 @@ class DAL(object):
or
db = DAL(**{"uri": ..., "tables": [...]...}) # experimental
db = DAL({"uri": ..., "items": ...}) # experimental
db.define_table('tablename', Field('fieldname1'),
Field('fieldname2'))
@@ -7367,9 +7367,8 @@ class DAL(object):
migrate_enabled=True, fake_migrate_all=False,
decode_credentials=False, driver_args=None,
adapter_args=None, attempts=5, auto_import=False,
bigint_id=False, debug=False, lazy_tables=False,
db_uid=None, do_connect=True,
after_connection=None, tables=None):
bigint_id=False,debug=False,lazy_tables=False,
db_uid=None, do_connect=True, after_connection=None):
"""
Creates a new Database Abstraction Layer instance.
@@ -7381,7 +7380,7 @@ class DAL(object):
experimental: you can specify a dictionary as uri
parameter i.e. with
db = DAL({"uri": "sqlite://storage.sqlite",
"tables": {...}, ...})
"items": {...}, ...})
for an example of dict input you can check the output
of the scaffolding db model with
@@ -7425,6 +7424,18 @@ class DAL(object):
:lazy_tables (defaults to False): delay table definition until table access
:after_connection (defaults to None): a callable that will be execute after the connection
"""
items = None
if isinstance(uri, dict):
if "items" in uri:
items = uri.pop("items")
try:
newuri = uri.pop("uri")
except KeyError:
newuri = DEFAULT_URI
locals().update(uri)
uri = newuri
if uri == '<zombie>' and db_uid is not None: return
if not decode_credentials:
credential_decoder = lambda cred: cred
@@ -7517,20 +7528,31 @@ class DAL(object):
self._fake_migrate = fake_migrate
self._migrate_enabled = migrate_enabled
self._fake_migrate_all = fake_migrate_all
if auto_import or tables:
if auto_import or items:
self.import_table_definitions(adapter.folder,
tables=tables)
items=items)
@property
def tables(self):
return self._tables
def import_table_definitions(self, path, migrate=False,
fake_migrate=False, tables=None):
fake_migrate=False, items=None):
pattern = pjoin(path,self._uri_hash+'_*.table')
if tables:
for table in tables:
self.define_table(**table)
if items:
for tablename, table in items.iteritems():
# TODO: read all field/table options
fields = []
# remove unsupported/illegal Table arguments
[table.pop(name) for name in ("name", "fields") if
name in table]
if "items" in table:
for fieldname, field in table.pop("items").iteritems():
# remove unsupported/illegal Field arguments
[field.pop(key) for key in ("requires", "name",
"compute", "colname") if key in field]
fields.append(Field(str(fieldname), **field))
self.define_table(str(tablename), *fields, **table)
else:
for filename in glob.glob(pattern):
tfile = self._adapter.file_open(filename, 'r')
@@ -7828,14 +7850,8 @@ def index():
):
if not fields and 'fields' in args:
fields = args.get('fields',())
if not isinstance(tablename, str):
if isinstance(tablename, unicode):
try:
tablename = str(tablename)
except UnicodeEncodeError:
raise SyntaxError("invalid unicode table name")
else:
raise SyntaxError("missing table name")
if not isinstance(tablename,str):
raise SyntaxError("missing table name")
elif hasattr(self,tablename) or tablename in self.tables:
if not args.get('redefine',False):
raise SyntaxError('table already defined: %s' % tablename)
@@ -7899,40 +7915,48 @@ def index():
if on_define: on_define(table)
return table
def as_dict(self, flat=False, sanitize=True):
db_uid = uri = None
def as_dict(self, flat=False, sanitize=True, field_options=True):
dbname = db_uid = uri = None
if not sanitize:
uri, db_uid = (self._uri, self._db_uid)
db_as_dict = dict(tables=[], uri=uri, db_uid=db_uid,
**dict([(k, getattr(self, "_" + k, None))
for k in 'pool_size','folder','db_codec',
uri, dbname, db_uid = (self._uri, self._dbname, self._db_uid)
db_as_dict = dict(items={}, tables=[], uri=uri, dbname=dbname,
db_uid=db_uid,
**dict([(k, getattr(self, "_" + k)) for
k in 'pool_size','folder','db_codec',
'check_reserved','migrate','fake_migrate',
'migrate_enabled','fake_migrate_all',
'decode_credentials','driver_args',
'adapter_args', 'attempts',
'bigint_id','debug','lazy_tables',
'do_connect']))
for table in self:
db_as_dict["tables"].append(table.as_dict(flat=flat,
sanitize=sanitize))
tablename = str(table)
db_as_dict["tables"].append(tablename)
db_as_dict["items"][tablename] = table.as_dict(flat=flat,
sanitize=sanitize,
field_options=field_options)
return db_as_dict
def as_xml(self, sanitize=True):
def as_xml(self, sanitize=True, field_options=True):
if not have_serializers:
raise ImportError("No xml serializers available")
d = self.as_dict(flat=True, sanitize=sanitize)
d = self.as_dict(flat=True, sanitize=sanitize,
field_options=field_options)
return serializers.xml(d)
def as_json(self, sanitize=True):
def as_json(self, sanitize=True, field_options=True):
if not have_serializers:
raise ImportError("No json serializers available")
d = self.as_dict(flat=True, sanitize=sanitize)
d = self.as_dict(flat=True, sanitize=sanitize,
field_options=field_options)
return serializers.json(d)
def as_yaml(self, sanitize=True):
def as_yaml(self, sanitize=True, field_options=True):
if not have_serializers:
raise ImportError("No YAML serializers available")
d = self.as_dict(flat=True, sanitize=sanitize)
d = self.as_dict(flat=True, sanitize=sanitize,
field_options=field_options)
return serializers.yaml(d)
def __contains__(self, tablename):
@@ -8300,9 +8324,7 @@ class Table(object):
if len(self._primarykey)==1:
self._id = [f for f in fields if isinstance(f,Field) \
and f.name==self._primarykey[0]][0]
elif not [f for f in fields if (isinstance(f,Field) and
f.type=='id') or (isinstance(f, dict) and
f.get("type", None)=="id")]:
elif not [f for f in fields if isinstance(f,Field) and f.type=='id']:
field = Field('id', 'id')
newfields.append(field)
fieldnames.add('id')
@@ -8320,7 +8342,8 @@ class Table(object):
if field.db is not None:
field = copy.copy(field)
include_new(field)
elif isinstance(field, dict) and not field['fieldname'] in fieldnames:
elif isinstance(field, dict) and 'fieldname' and \
not field['fieldname'] in fieldnames:
include_new(Field(**field))
elif isinstance(field, Table):
table = field
@@ -8709,6 +8732,32 @@ class Table(object):
response.id = None
return response
def validate_and_update(self, _key=DEFAULT, **fields):
response = Row()
response.errors = Row()
new_fields = copy.copy(fields)
for key,value in fields.iteritems():
value,error = self[key].validate(value)
if error:
response.errors[key] = "%s" % error
else:
new_fields[key] = value
if _key is DEFAULT:
record = self(**values)
elif isinstance(_key,dict):
record = self(**_key)
else:
record = self(_key)
if not response.errors and record:
row = self._db(self._id==_key)
response.id = row.update(**fields)
else:
response.id = None
return response
def update_or_insert(self, _key=DEFAULT, **values):
if _key is DEFAULT:
record = self(**values)
@@ -8875,8 +8924,9 @@ class Table(object):
if id_map and cid is not None:
id_map_self[long(line[cid])] = new_id
def as_dict(self, flat=False, sanitize=True):
table_as_dict = dict(tablename=str(self), fields=[],
def as_dict(self, flat=False, sanitize=True, field_options=True):
tablename = str(self)
table_as_dict = dict(name=tablename, items={}, fields=[],
sequence_name=self._sequence_name,
trigger_name=self._trigger_name,
common_filter=self._common_filter, format=self._format,
@@ -8884,26 +8934,31 @@ class Table(object):
for field in self:
if (field.readable or field.writable) or (not sanitize):
table_as_dict["fields"].append(field.as_dict(
flat=flat, sanitize=sanitize))
table_as_dict["fields"].append(field.name)
table_as_dict["items"][field.name] = \
field.as_dict(flat=flat, sanitize=sanitize,
options=field_options)
return table_as_dict
def as_xml(self, sanitize=True):
def as_xml(self, sanitize=True, field_options=True):
if not have_serializers:
raise ImportError("No xml serializers available")
d = self.as_dict(flat=True, sanitize=sanitize)
d = self.as_dict(flat=True, sanitize=sanitize,
field_options=field_options)
return serializers.xml(d)
def as_json(self, sanitize=True):
def as_json(self, sanitize=True, field_options=True):
if not have_serializers:
raise ImportError("No json serializers available")
d = self.as_dict(flat=True, sanitize=sanitize)
d = self.as_dict(flat=True, sanitize=sanitize,
field_options=field_options)
return serializers.json(d)
def as_yaml(self, sanitize=True):
def as_yaml(self, sanitize=True, field_options=True):
if not have_serializers:
raise ImportError("No YAML serializers available")
d = self.as_dict(flat=True, sanitize=sanitize)
d = self.as_dict(flat=True, sanitize=sanitize,
field_options=field_options)
return serializers.yaml(d)
def with_alias(self, alias):
@@ -9384,13 +9439,8 @@ class Field(Expression):
self.op = None
self.first = None
self.second = None
if isinstance(fieldname, unicode):
try:
fieldname = str(fieldname)
except UnicodeEncodeError:
raise SyntaxError('Field: invalid unicode field name')
self.name = fieldname = cleanup(fieldname)
if not isinstance(fieldname, str) or hasattr(Table, fieldname) or \
if not isinstance(fieldname,str) or hasattr(Table,fieldname) or \
fieldname[0] == '_' or REGEX_PYTHON_KEYWORDS.match(fieldname):
raise SyntaxError('Field: invalid field name: %s' % fieldname)
self.type = type if not isinstance(type, (Table,Field)) else 'reference %s' % type
@@ -9590,61 +9640,113 @@ class Field(Expression):
def count(self, distinct=None):
return Expression(self.db, self.db._adapter.COUNT, self, distinct, 'integer')
def as_dict(self, flat=False, sanitize=True):
attrs = ("name", 'authorize', 'represent', 'ondelete',
'custom_store', 'autodelete', 'custom_retrieve',
'filter_out', 'uploadseparate', 'widget', 'uploadfs',
'update', 'custom_delete', 'uploadfield', 'uploadfolder',
'custom_qualifier', 'unique', 'writable', 'compute',
'map_none', 'default', 'type', 'required', 'readable',
'requires', 'comment', 'label', 'length', 'notnull',
'custom_retrieve_file_properties', 'filter_in')
serializable = (int, long, basestring, float, tuple,
bool, type(None))
def as_dict(self, flat=False, sanitize=True, options=True):
attrs = ('type', 'length', 'default', 'required',
'ondelete', 'notnull', 'unique', 'uploadfield',
'widget', 'label', 'comment', 'writable', 'readable',
'update', 'authorize', 'autodelete', 'represent',
'uploadfolder', 'uploadseparate', 'uploadfs',
'compute', 'custom_store', 'custom_retrieve',
'custom_retrieve_file_properties', 'custom_delete',
'filter_in', 'filter_out', 'custom_qualifier',
'map_none', 'name')
SERIALIZABLE_TYPES = (int, long, basestring, dict, list,
float, tuple, bool, type(None))
def flatten(obj):
if isinstance(obj, dict):
return dict((flatten(k), flatten(v)) for k, v in
obj.items())
elif isinstance(obj, (tuple, list, set)):
return [flatten(v) for v in obj]
elif isinstance(obj, serializable):
return obj
elif isinstance(obj, (datetime.datetime,
datetime.date, datetime.time)):
return str(obj)
else:
if flat:
if isinstance(obj, flatten.__class__):
return str(type(obj))
elif isinstance(obj, type):
try:
return str(obj).split("'")[1]
except IndexError:
return str(obj)
elif not isinstance(obj, SERIALIZABLE_TYPES):
return str(obj)
elif isinstance(obj, dict):
newobj = dict()
for k, v in obj.items():
newobj[k] = flatten(v)
return newobj
elif isinstance(obj, (list, tuple, set)):
return [flatten(v) for v in obj]
else:
return obj
elif isinstance(obj, (dict, set)):
return obj.copy()
else: return obj
def filter_requires(t, r, options=True):
if sanitize and any([keyword in str(t).upper() for
keyword in ("CRYPT", "IS_STRONG")]):
return None
d = dict()
if not (sanitize and not (self.readable or self.writable)):
for attr in attrs:
if flat:
d.update({attr: flatten(getattr(self, attr))})
else:
d.update({attr: getattr(self, attr)})
d["fieldname"] = d.pop("name")
if not isinstance(r, dict):
if options and hasattr(r, "options"):
if callable(r.options):
r.options()
newr = r.__dict__.copy()
else:
newr = r.copy()
# remove options if not required
if not options and newr.has_key("labels"):
[newr.update({key:None}) for key in
("labels", "theset") if (key in newr)]
for k, v in newr.items():
if k == "other":
if isinstance(v, dict):
otype, other = v.popitem()
else:
otype = flatten(type(v))
other = v
newr[k] = {otype: filter_requires(otype, other,
options=options)}
else:
newr[k] = flatten(v)
return newr
if isinstance(self.requires, (tuple, list, set)):
requires = dict([(flatten(type(r)),
filter_requires(type(r), r,
options=options)) for
r in self.requires])
else:
requires = {flatten(type(self.requires)):
filter_requires(type(self.requires),
self.requires, options=options)}
d = dict(colname="%s.%s" % (self.tablename, self.name),
requires=requires)
d.update([(attr, flatten(getattr(self, attr))) for attr in attrs])
return d
def as_xml(self, sanitize=True):
def as_xml(self, sanitize=True, options=True):
if have_serializers:
xml = serializers.xml
else:
raise ImportError("No xml serializers available")
d = self.as_dict(flat=True, sanitize=sanitize)
d = self.as_dict(flat=True, sanitize=sanitize,
options=options)
return xml(d)
def as_json(self, sanitize=True):
def as_json(self, sanitize=True, options=True):
if have_serializers:
json = serializers.json
else:
raise ImportError("No json serializers available")
d = self.as_dict(flat=True, sanitize=sanitize)
d = self.as_dict(flat=True, sanitize=sanitize,
options=options)
return json(d)
def as_yaml(self, sanitize=True):
def as_yaml(self, sanitize=True, options=True):
if have_serializers:
d = self.as_dict(flat=True, sanitize=sanitize)
d = self.as_dict(flat=True, sanitize=sanitize,
options=options)
return serializers.yaml(d)
else:
raise ImportError("No YAML serializers available")
gluon/globals.py
+10 -2
View File
@@ -196,8 +196,11 @@ class Request(Storage):
if (body and
env.request_method in ('POST', 'PUT', 'DELETE', 'BOTH') and
not is_json):
query_string = env.pop('QUERY_STRING') if 'QUERY_STRING' in env else None
dpost = cgi.FieldStorage(fp=body, environ=env, keep_blank_values=1)
post_vars.update(dpost)
if query_string is not None:
env['QUERY_STRING'] = query_string
# The same detection used by FieldStorage to detect multipart POSTs
is_multipart = dpost.type[:10] == 'multipart/'
body.seek(0)
@@ -880,7 +883,9 @@ class Session(Storage):
if response.session_new:
return
# Get session data out of the database
(record_id, unique_key) = response.session_id.split(':')
if response.session_id is None:
return
(record_id, sep, unique_key) = response.session_id.partition(':')
if record_id.isdigit() and long(record_id)>1:
new_unique_key = web2py_uuid()
@@ -912,9 +917,12 @@ class Session(Storage):
if response.session_id:
rcookies[response.session_id_name] = response.session_id
rcookies[response.session_id_name]['path'] = '/'
if response.session_cookie_expires:
if isinstance(response.session_cookie_expires,datetime.datetime):
rcookies[response.session_id_name]['expires'] = \
response.session_cookie_expires.strftime(FMT)
elif isinstance(response.session_cookie_expires,str):
rcookies[response.session_id_name]['expires'] = \
response.session_cookie_expires
def clear(self):
previous_session_hash = self.pop('_session_hash', None)
gluon/sqlhtml.py
+1 -4
View File
@@ -3095,10 +3095,7 @@ class ExporterHTML(ExportClass):
ExportClass.__init__(self, rows)
def export(self):
if self.rows:
return self.rows.xml()
else:
return '<html>\n<body>\n<table>\n</table>\n</body>\n</html>'
return '<html>\n<head>\n<meta http-equiv="content-type" content="text/html; charset=UTF-8" />\n</head>\n<body>\n%s\n</body>\n</html>' % (self.rows.xml() or '')
class ExporterXML(ExportClass):
label = 'XML'
gluon/tests/test_dal.py
+33 -29
View File
@@ -701,12 +701,13 @@ class TestDALDictImportExport(unittest.TestCase):
assert isinstance(dbdict, dict)
uri = dbdict["uri"]
assert isinstance(uri, basestring) and uri
assert len(dbdict["tables"]) == 2
assert len(dbdict["tables"][0]["fields"]) == 3
assert dbdict["tables"][0]["fields"][1]["type"] == db.person.name.type
assert dbdict["tables"][0]["fields"][1]["default"] == db.person.name.default
assert len(dbdict["items"]) == 2
assert len(dbdict["items"]["person"]["items"]) == 3
assert dbdict["items"]["person"]["items"]["name"]["type"] == db.person.name.type
assert dbdict["items"]["person"]["items"]["name"]["default"] == db.person.name.default
assert dbdict
db2 = DAL(**dbdict)
db2 = DAL(dbdict, check_reserved=['all'])
assert len(db.tables) == len(db2.tables)
assert hasattr(db2, "pet") and isinstance(db2.pet, Table)
assert hasattr(db2.pet, "friend") and isinstance(db2.pet.friend, Field)
@@ -724,7 +725,7 @@ class TestDALDictImportExport(unittest.TestCase):
unicode_keys = True
if sys.version < "2.6.5":
unicode_keys = False
db3 = DAL(**serializers.loads_json(dbjson,
db3 = DAL(serializers.loads_json(dbjson,
unicode_keys=unicode_keys))
assert hasattr(db3, "person") and hasattr(db3.person, "uuid") and\
db3.person.uuid.type == db.person.uuid.type
@@ -735,19 +736,18 @@ class TestDALDictImportExport(unittest.TestCase):
mpfc = "Monty Python's Flying Circus"
dbdict4 = {"uri": DEFAULT_URI,
"tables":[{"tablename": "staff",
"fields": [{"fieldname": "name",
"default":"Michael"},
{"fieldname": "food",
"default":"Spam"},
{"fieldname": "tvshow",
"type": "reference tvshow"}]},
{"tablename": "tvshow",
"fields": [{"fieldname": "name",
"default":mpfc},
{"fieldname": "rating",
"type":"double"}]}]}
db4 = DAL(**dbdict4)
"items":{"staff":{"items": {"name":
{"default":"Michael"},
"food":
{"default":"Spam"},
"tvshow":
{"type": "reference tvshow"}
}},
"tvshow":{"items": {"name":
{"default":mpfc},
"rating":
{"type":"double"}}}}}
db4 = DAL(dbdict4, check_reserved=['all'])
assert "staff" in db4.tables
assert "name" in db4.staff
assert db4.tvshow.rating.type == "double"
@@ -761,19 +761,20 @@ class TestDALDictImportExport(unittest.TestCase):
db4.commit()
dbdict5 = {"uri": DEFAULT_URI}
db5 = DAL(**dbdict5)
db5 = DAL(dbdict5, check_reserved=['all'])
assert db5.tables in ([], None)
assert not (str(db5) in ("", None))
dbdict6 = {"uri": DEFAULT_URI,
"tables":[{"tablename": "staff"},
{"tablename": "tvshow",
"fields": [{"fieldname": "name"},
{"fieldname": "rating", "type":"double"}
]
}]
}
db6 = DAL(**dbdict6)
"items":{"staff":{},
"tvshow":{"items": {"name": {},
"rating":
{"type":"double"}
}
}
}
}
db6 = DAL(dbdict6, check_reserved=['all'])
assert len(db6["staff"].fields) == 1
assert "name" in db6["tvshow"].fields
@@ -781,12 +782,15 @@ class TestDALDictImportExport(unittest.TestCase):
assert db6.staff.insert() is not None
assert db6(db6.staff).select().first().id == 1
db6.staff.drop()
db6.tvshow.drop()
db6.commit()
if __name__ == '__main__':
unittest.main()
tearDownModule()
gluon/tools.py
+1 -2
View File
@@ -1225,8 +1225,7 @@ class Auth(object):
response = current.response
if auth and auth.remember:
# when user wants to be logged in for longer
response.cookies[response.session_id_name]["expires"] = \
auth.expiration
response.session_cookie_expires = auth.expiration
if signature:
self.define_signature()
else: