[Hejes-devel] [1100] updated trunk/misc/dbblog/test.py to current data model
hejes-devel at nytud.hu
hejes-devel at nytud.hu
Fri Feb 14 12:31:32 CET 2014
Revision: 1100
Author: mihaltz
Date: 2014-02-14 12:31:32 +0100 (Fri, 14 Feb 2014)
Log Message:
-----------
updated trunk/misc/dbblog/test.py to current data model
Modified Paths:
--------------
trunk/misc/dbblog/test.py
Modified: trunk/misc/dbblog/test.py
===================================================================
--- trunk/misc/dbblog/test.py 2014-02-14 10:49:40 UTC (rev 1099)
+++ trunk/misc/dbblog/test.py 2014-02-14 11:31:32 UTC (rev 1100)
@@ -30,47 +30,49 @@
"""
Define the table names and the fields + add restrictions.
"""
- dbblog.define_table('images',
- Field('file', 'upload'),
- Field('title', unique=True),
- migrate='blog_images.table'
- )
-
dbblog.define_table('tags',
Field('name', unique=True),
Field('posts', 'list:reference posts'),
migrate='blog_tags.table'
- )
+ )
dbblog.define_table('categories',
+ Field('showing', 'boolean'),
+ Field('ordering', 'integer', unique=True),
Field('name', unique=True),
format = '%(name)s',
migrate='blog_categories.table'
- )
-
+ )
+
dbblog.define_table('posts',
+ Field('showing', 'boolean'),
Field('slug', unique=True),
Field('title'),
- Field('image_url'),
+ Field('image_filename'),
+ Field('lead'),
Field('body', 'text'),
Field('tags', 'list:string'),
Field('category', 'reference categories'),
Field('created_on', 'datetime'), #, default=request.now),
+ Field('author'),
migrate='blog_posts.table'
- )
+ )
- dbblog.posts.slug.requires = IS_NOT_EMPTY()
+ dbblog.posts.slug.requires = [IS_SLUG(), IS_NOT_IN_DB(dbblog, dbblog.posts.slug)]
dbblog.posts.title.requires = IS_NOT_EMPTY()
+# dbblog.posts.image_filename.requires = IS_IN_SET(list_image_files('applications/'+URL('static', 'blog')))
+ dbblog.posts.lead.requires = IS_NOT_EMPTY()
dbblog.posts.body.requires = IS_NOT_EMPTY()
- dbblog.posts.slug.requires = IS_NOT_EMPTY()
dbblog.posts.category.requires = IS_IN_DB(dbblog, dbblog.categories.id, '%(name)s')
dbblog.posts.tags.requires = IS_LIST_OF(IS_LOWER())
+ dbblog.posts.author.requires = IS_NOT_EMPTY()
+ # Set callbacks on table posts to automatically maintain table tags
dbblog.posts._after_update.append( _on_update_posts)
dbblog.posts._after_insert.append( _on_insert_posts)
dbblog.posts._after_delete.append( _on_delete_posts)
- dbblog.posts._before_update.append( _bef_upd)
-
+
+ # make sure everything gets thru (?)
dbblog.commit()
@@ -87,8 +89,7 @@
def create_indexes():
- """Create indexes (if they didn't exist) on frequently used columns. Note: tables must already exist (otherwise raises exception)."""
- create_index_if_not_exists(dbblog, 'images', 'idx_images_title', 'title', True)
+ # Create indexes (indices?) (if they didn't exist)
create_index_if_not_exists(dbblog, 'tags', 'idx_tags_name', 'name', True)
create_index_if_not_exists(dbblog, 'categories', 'idx_categories_name', 'name', True)
create_index_if_not_exists(dbblog, 'posts', 'idx_posts_created_on', 'created_on')
@@ -208,6 +209,7 @@
print('after update: {0}, {1}'.format(s,f))
print(s.query)
+
def _on_update_posts(s, f):
"""Callback functions on update operations on table posts in order to maintain table tags.
s: Set object, records updated, f: dictionary, field names and values passed to update
@@ -215,8 +217,9 @@
#print('after update: {0}, {1}'.format(s, f))
dbblog.executesql('LOCK TABLES tags WRITE, posts READ;') # prevent other sessions from accessing while we're doing maintenance
id = long(s.query.second) # s.select(dbblog.posts.id).first().id # id of post just updated
- # 1. check whether each new tag is referred to from table tags
- for tag in f['tags']: # tags just assigned to post
+ mytags = f['tags'] if f['showing'] else [] # set of tags in just updated post; if post is not showing, emulate empty set of tags
+ # 1. check whether each new tag is referred to from table tags
+ for tag in mytags: # tags just assigned to post
r = dbblog(dbblog.tags.name == tag).select() # get matching record from table tags
if len(r) == 1: # matching record exists:
if id not in r[0].posts: # check if contains reference to post, if not, update it
@@ -227,9 +230,9 @@
# 2. for each record in table tags that refers to post: check if post still has tag, if not, delete reference/record
to_delete = [] # ids of records that have to be deleted from table tags
for row in dbblog(dbblog.tags.posts.contains(id)).select():
- if row.name not in f['tags']:
+ if row.name not in mytags:
if len(row.posts) > 1:
- dbblog(dbblog.tags.id == rid).update(posts = [x for x in row.posts if x != id]) # remove reference to post in question
+ dbblog(dbblog.tags.id == row.id).update(posts = [x for x in row.posts if x != id]) # remove reference to post in question
else:
to_delete.append(row.id) # schedule for deleting
for rid in to_delete: # delete
@@ -242,10 +245,11 @@
"""Callback functions on insert operations on table posts in order to maintain table tags.
f: a dict of fields passed to insert, id: the id of the newly inserted record
"""
- print('after insert: {0}, {1}'.format(f, id))
- #dbblog.executesql('LOCK TABLES tags WRITE, posts READ;') # prevent other sessions from accessing
+ #print('after insert: {0}, {1}'.format(f, id))
+ dbblog.executesql('LOCK TABLES tags WRITE, posts READ;') # prevent other sessions from accessing
+ mytags = f['tags'] if f['showing'] else [] # set of tags in just updated post; if post is not showing, emulate empty set of tags
# Add each tag to table tags
- for tag in f['tags']:
+ for tag in mytags:
r = dbblog(dbblog.tags.name == tag).select() # get matching record from table tags
if len(r) == 1: # matching record exists: update it
dbblog(dbblog.tags.name == tag).update(posts = r.first().posts + [id])
@@ -259,15 +263,14 @@
"""Callback functions on insert operations on table posts in order to maintain table tags.
s: Set object, records deleted
"""
- print('after delete: {0}'.format(s))
- #dbblog.executesql('LOCK TABLES tags WRITE, posts READ;') # prevent other sessions from accessing while we're doing maintenance
+ #print('after delete: {0}'.format(s))
+ dbblog.executesql('LOCK TABLES tags WRITE, posts READ;') # prevent other sessions from accessing while we're doing maintenance
id = long(s.query.second) # id of post just deleted (gluon.DAL uses <Set (table.id = id)> for delete/update)
- print('id of post just deleted: {0}'.format(id))
# Delete references from matching records in table tags or delete matching records that only referenced this post
to_delete = [] # ids of records that have to be deleted from table tags
for row in dbblog(dbblog.tags.posts.contains(id)).select():
if len(row.posts) > 1:
- print('update tags set posts={0} where id={1}'.format([x for x in row.posts if x != id], row.id))
+ #print('update tags set posts={0} where id={1}'.format([x for x in row.posts if x != id], row.id))
dbblog(dbblog.tags.id == row.id).update(posts = [x for x in row.posts if x != id]) # remove reference to post in question
dbblog.commit()
else:
@@ -277,6 +280,7 @@
dbblog.commit()
dbblog.executesql('UNLOCK TABLES;') # remove lock
+
def test_callbacks():
#dbblog.posts._before_update.append(lambda s,f: print('before update: {0}, {1}'.format(s,f)))
#dbblog.posts._before_update.append( _bef_upd)
@@ -286,7 +290,15 @@
dbblog(dbblog.posts.id==1).update(tags=['új verzió'])
#dbblog.commit()
+def test_notfound():
+# cats = dbblog(dbblog.categories.showing == True).select(orderby=dbblog.categories.ordering)
+# tags = dbblog().select(dbblog.tags.ALL, orderby=dbblog.tags.name)
+ slug = 'afasdfs' # 'on-is-hozzaszolhat'
+ rows = dbblog(dbblog.posts.slug==slug).select()
+ print(len(rows))
+ print(rows)
+
#########
# main
#########
@@ -298,7 +310,7 @@
#regenerate_tags_table()
#test_set()
#test_callbacks()
-test_validator()
+test_notfound()
#query_categories()
#query_posts()
More information about the Hejes-devel
mailing list