[Hejes-devel] [864] blog: more sutff for DB
hejes-devel at nytud.hu
hejes-devel at nytud.hu
Tue Jul 23 17:29:52 CEST 2013
Revision: 864
Author: mihaltz
Date: 2013-07-23 17:29:52 +0200 (Tue, 23 Jul 2013)
Log Message:
-----------
blog: more sutff for DB
Added Paths:
-----------
trunk/misc/dbblog/test.py
Added: trunk/misc/dbblog/test.py
===================================================================
--- trunk/misc/dbblog/test.py (rev 0)
+++ trunk/misc/dbblog/test.py 2013-07-23 15:29:52 UTC (rev 864)
@@ -0,0 +1,266 @@
+#!/usr/bin/env python
+# coding: utf-8
+'''
+
+Offline test for dbhelyesblog in MySQL:
+- use gluon.DAL to connect
+- create/migrate the tables
+- add test data
+
+ at author MM
+
+'''
+
+import sys
+import os
+
+sys.path.append('/opt/web2py')
+from gluon import *
+from gluon.dal import Row
+
+db = DAL("mysql://dbblogadmin:d4bl0GGr@localhost/dbhelyesblog")
+# location of .table files = ???
+
+def define_tables():
+ """
+ Define the table names and the fields + add restrictions.
+ """
+ db.define_table('images',
+ Field('file', 'upload'),
+ Field('title', unique=True)
+ )
+
+ db.define_table('tags',
+ Field('name', unique=True)
+ )
+
+ db.define_table('categories',
+ Field('name', unique=True)
+ )
+
+ db.define_table('posts',
+ Field('slug', unique=True),
+ Field('title'),
+ Field('image', 'reference images'),
+ Field('body', 'text'),
+ Field('tags', 'list:reference tags'),
+ Field('category', 'reference categories'),
+ Field('created_on', 'datetime') # TODO add in webpy: default=request.now),
+ )
+# migrate='hqa_main.table')
+
+ db.posts.slug.requires = IS_NOT_EMPTY()
+ db.posts.title.requires = IS_NOT_EMPTY()
+ db.posts.body.requires = IS_NOT_EMPTY()
+ db.posts.slug.requires = IS_NOT_EMPTY()
+ db.posts.image.requires = IS_IN_DB(db, db.images.id, '%(title)s')
+
+
+# TODO
+def create_indexes():
+ # create indexes on frequently used columns; SQLite engine only!!!
+ db.executesql('CREATE UNIQUE INDEX IF NOT EXISTS idx_hqa_main_id ON hqa_main (id);')
+ db.executesql('CREATE INDEX IF NOT EXISTS idx_hqa_main_qword ON hqa_main (qword);')
+ db.executesql('CREATE UNIQUE INDEX IF NOT EXISTS idx_hqa_prob_cat_names_id ON hqa_prob_cat_names (id);')
+ db.executesql('CREATE UNIQUE INDEX IF NOT EXISTS idx_hqa_prof_cat_names_id ON hqa_prof_cat_names (id);')
+ db.executesql('CREATE UNIQUE INDEX IF NOT EXISTS idx_hqa_remark_names_id ON hqa_remark_names (id);')
+
+# dump table and fields and types
+def dump_tables():
+ for table in db.tables:
+ print table + ':'
+ for field in db[table].fields:
+ print ' ' + field + ': ' + str(db[table][field].type)
+ print('')
+
+# Deletes all tables when new database csv file needs to be imported
+def dropdb():
+ for table_name in db.tables():
+ db[table_name].drop()
+ define_tables()
+
+# tests
+def test_add_data():
+ prof = []
+ prof.append( db.hqa_prof_cat_names.insert(name='filo') )
+ prof.append( db.hqa_prof_cat_names.insert(name='bio') )
+ db.hqa_main.insert(qword='alma', answer='alma!', prof_cat_ids=prof)
+
+def test_query0():
+ rows = db(db.hqa_main).select()
+ for row in rows:
+ print row
+
+def test_query1():
+ # 1-to-many
+ prof = []
+ prof.append( db.hqa_prof_cat_names.insert(name='filo') )
+ prof.append( db.hqa_prof_cat_names.insert(name='bio') )
+ db.hqa_main.insert(qword='alma', answer='alma!', prof_cat_ids=prof)
+ db.hqa_main.prof_cat_ids.represent = lambda ids: [db.hqa_prof_cat_names(x).name for x in ids]
+ #rows = db(db.hqa_main.prof_cat_ids.contains(prof[0])).select()
+ rows = db(db.hqa_main.prof_cat_ids.contains( 1 )).select()
+ for row in rows:
+ print row
+ print row.id, row.qword, db.hqa_main.prof_cat_ids.represent(row.prof_cat_ids)
+ #print db.hqa_prof_cat_names(1)
+
+def test_query2():
+ # inner join
+ i = db.hqa_prob_cat_names.insert(name='Hangjeloles')
+ db.hqa_main.insert(qword='alma', answer='alma!', prob_cat_id=i)
+ rows = db(db.hqa_main.prob_cat_id==db.hqa_prob_cat_names.id).select()
+ for row in rows:
+ print row
+ print row.hqa_main.id, row.hqa_main.qword, row.hqa_main.answer, row.hqa_prob_cat_names.name
+
+def test_query3():
+ # inner join
+ rows = db(db.hqa_main.prob_cat_id==db.hqa_prob_cat_names.id).select(limitby=(0, 10))
+ for row in rows:
+ print '; '.join([str(row.hqa_main.id), row.hqa_main.qword, row.hqa_main.answer, row.hqa_prob_cat_names.name])
+
+def test_query4():
+ # inner join
+ i = db.hqa_prob_cat_names.insert(name='Hangjeloles')
+ db.hqa_main.insert(qword='alma', answer='alma!', prob_cat_id=i)
+ db.hqa_main.insert(qword='barack', answer='barack!', prob_cat_id=None)
+ rows = db(db.hqa_main.prob_cat_id==db.hqa_prob_cat_names.id).select()
+ for row in rows:
+ print row
+ print row.hqa_main.id, row.hqa_main.qword, row.hqa_main.answer, row.hqa_prob_cat_names.name
+
+def test_query5():
+ # left outer join
+ i = db.hqa_prob_cat_names.insert(name='Hangjeloles')
+ db.hqa_main.insert(qword='alma', answer='alma!', prob_cat_id=i)
+ db.hqa_main.insert(qword='barack', answer='barack!', prob_cat_id=None)
+ rows = db().select(db.hqa_main.ALL, db.hqa_prob_cat_names.ALL, left=db.hqa_prob_cat_names.on(db.hqa_main.prob_cat_id==db.hqa_prob_cat_names.id))
+ for row in rows:
+ print row
+ print row.hqa_main.id, row.hqa_main.qword, row.hqa_main.answer, row.hqa_prob_cat_names.name
+
+def test_query6():
+ # left outer join
+ rows = db().select(db.hqa_main.ALL, db.hqa_prob_cat_names.ALL, left=db.hqa_prob_cat_names.on(db.hqa_main.prob_cat_id==db.hqa_prob_cat_names.id), limitby=(0, 50))
+ for row in rows:
+ print '; '.join([str(row.hqa_main.id), row.hqa_main.qword, row.hqa_main.answer, row.hqa_prob_cat_names.name])
+
+def test_update():
+ # insert + select + update
+ i = db.hqa_main.insert(qword='alma', answer='alma!')
+ row = db.hqa_main(i) # same as: row = db(db.hqa_main.id==i).select().first()
+ print row
+ row.prof_cat_ids = [1,2]
+ row.update_record()
+ print row
+ row.prof_cat_ids.append(3)
+ row.update_record()
+ print row
+ db.commit()
+
+def test_query7():
+ # select: empty result set
+ row = db.hqa_main(11111)
+ print row == None
+ print row
+
+def test_query8():
+ # select w/ left join + 1-to-many
+ db.hqa_main.prof_cat_ids.represent = lambda ids: [db.hqa_prof_cat_names(x).name for x in ids]
+ db.hqa_main.remark_ids.represent = lambda ids: [db.hqa_remark_names(x).name for x in ids]
+ rows = db().select(db.hqa_main.ALL, db.hqa_prob_cat_names.ALL, left=db.hqa_prob_cat_names.on(db.hqa_main.prob_cat_id==db.hqa_prob_cat_names.id), limitby=(0, 50))
+ for row in rows:
+ #print row
+ d = [row.hqa_main.id, row.hqa_main.qword, row.hqa_main.question, row.hqa_main.answer, row.hqa_prob_cat_names.name,
+# row.hqa_main.prof_cat_ids,
+ db.hqa_main.prof_cat_ids.represent(row.hqa_main.prof_cat_ids),
+# row.hqa_main.remark_ids,
+ db.hqa_main.remark_ids.represent(row.hqa_main.remark_ids),
+ row.hqa_main.akh]
+ print '; '.join([str(x) for x in d])
+
+def test_query9():
+ # select w/ conditions + left join + 1-to-many
+ db.hqa_main.prof_cat_ids.represent = lambda ids: [db.hqa_prof_cat_names(x).name for x in ids]
+ db.hqa_main.remark_ids.represent = lambda ids: [db.hqa_remark_names(x).name for x in ids]
+ rows = db(db.hqa_main.qword.lower().contains(u'sajt')).select(db.hqa_main.ALL, db.hqa_prob_cat_names.ALL, left=db.hqa_prob_cat_names.on(db.hqa_main.prob_cat_id==db.hqa_prob_cat_names.id), limitby=(0, 50))
+ for row in rows:
+ #print row
+ d = [row.hqa_main.id, row.hqa_main.qword, row.hqa_main.question, row.hqa_main.answer, row.hqa_prob_cat_names.name,
+ db.hqa_main.prof_cat_ids.represent(row.hqa_main.prof_cat_ids),
+ db.hqa_main.remark_ids.represent(row.hqa_main.remark_ids),
+ row.hqa_main.akh]
+ print '; '.join([str(x) for x in d])
+
+def dump_main():
+ # select + left join + 1-to-many
+ db.hqa_main.prof_cat_ids.represent = lambda ids: [db.hqa_prof_cat_names(x).name for x in ids]
+ db.hqa_main.remark_ids.represent = lambda ids: [db.hqa_remark_names(x).name for x in ids]
+ rows = db().select(db.hqa_main.ALL, db.hqa_prob_cat_names.ALL, left=db.hqa_prob_cat_names.on(db.hqa_main.prob_cat_id==db.hqa_prob_cat_names.id))
+ for row in rows:
+ #print row
+ d = [row.hqa_main.id, row.hqa_main.qword, row.hqa_main.question, row.hqa_main.answer, row.hqa_prob_cat_names.name,
+ db.hqa_main.prof_cat_ids.represent(row.hqa_main.prof_cat_ids),
+ db.hqa_main.remark_ids.represent(row.hqa_main.remark_ids),
+ row.hqa_main.akh]
+ print '\t'.join([str(x) for x in d])
+
+
+def test_query10():
+ # select using dynamic query + left join + 1-to-many
+ db.hqa_main.prof_cat_ids.represent = lambda ids: [db.hqa_prof_cat_names(x).name for x in ids]
+ db.hqa_main.remark_ids.represent = lambda ids: [db.hqa_remark_names(x).name for x in ids]
+ # method#1
+ q = db.hqa_main.id > 0 # for starters: "select * from db.hqa_main" -- is this the best solution performance-wise??...
+ q &= db.hqa_main.qword.lower().contains(u'sajt')
+ #q &= db.hqa_main.prob_cat_id == 11
+ q &= db.hqa_main.prof_cat_ids.contains(25) # 'konyha'
+ print(q)
+ # method#2
+ qs = [db.hqa_main.id > 0]
+ qs.append( db.hqa_main.qword.lower().contains(u'sajt') )
+ qs.append( db.hqa_main.prof_cat_ids.contains(25) ) # 'konyha'
+ q = reduce( lambda x,y: x & y, qs )
+ print(q)
+ rows = db(q).select(db.hqa_main.ALL, db.hqa_prob_cat_names.ALL, left=db.hqa_prob_cat_names.on(db.hqa_main.prob_cat_id==db.hqa_prob_cat_names.id), limitby=(0, 50))
+ for row in rows:
+ #print row
+ d = [row.hqa_main.id, row.hqa_main.qword, row.hqa_main.question, row.hqa_main.answer, row.hqa_prob_cat_names.name,
+ db.hqa_main.prof_cat_ids.represent(row.hqa_main.prof_cat_ids),
+ db.hqa_main.remark_ids.represent(row.hqa_main.remark_ids),
+ row.hqa_main.akh]
+ #r = Row(id=row.hqa_main.id)
+ #print r.id
+ print '\t'.join([str(x) for x in d])
+
+
+# main
+
+define_tables()
+#create_indexes_sqlite()
+#dropdb()
+
+#dump_tables()
+#test_add_data()
+#test_query0()
+#test_query1()
+#db.commit()
+#test_query2()
+#test_query4()
+#test_query5()
+#db.commit()
+#test_update()
+#test_query7()
+#test_query8()
+
+#import_csv_data()
+
+#test_query8()
+#test_query9()
+#test_query10()
+
+#dump_main()
+
+
+
Property changes on: trunk/misc/dbblog/test.py
___________________________________________________________________
Added: svn:executable
+ *
More information about the Hejes-devel
mailing list