Last active
November 25, 2016 13:39
-
-
Save dz0/6c1d88c2a71b8f353bf1aa61d794a09c to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| # -*- coding: utf-8 -*- | |
| from gluon.storage import Storage | |
| # from pydal.helpers.methods import smart_query | |
| ######################################## | |
| # SEARCH FILTERS QUERY from FORM # | |
| ######################################## | |
| if "SEARCH FILTERS QUERY from FORM": | |
| # memo from sqlhtml.py class grid | |
| """ | |
| search_options = { | |
| 'string': ['=', '!=', '<', '>', '<=', '>=', 'starts with', 'ends with', 'contains', 'in', 'not in'], | |
| 'text': ['=', '!=', '<', '>', '<=', '>=', 'starts with', 'contains', 'in', 'not in'], | |
| 'date': ['=', '!=', '<', '>', '<=', '>='], | |
| 'time': ['=', '!=', '<', '>', '<=', '>='], | |
| 'datetime': ['=', '!=', '<', '>', '<=', '>='], | |
| 'integer': ['=', '!=', '<', '>', '<=', '>=', 'in', 'not in'], | |
| 'double': ['=', '!=', '<', '>', '<=', '>='], | |
| 'id': ['=', '!=', '<', '>', '<=', '>=', 'in', 'not in'], | |
| 'reference': ['=', '!='], | |
| 'boolean': ['=', '!=']} | |
| """ | |
| def queryFilter(field=None, comparison=None, name_prefix=None, # for smart way | |
| # for customisation | |
| name=None, # name of input | |
| #input=None, # INPUT(..) # TODO -- if Field is not enough.. for input.. | |
| query=None, # lambda, which expexts value from Expr to be given as filtering query | |
| target_expression=None # in case we use field of 'no_table' (this indicates the comparison target) | |
| ): | |
| """ | |
| field -- db.table.field # input field | |
| comparison -- action as in grid search fields (str) | |
| name_prefix -- based on comparison or table (if same field name in several tables) | |
| query -- should be contained in lambda, because value for comparison is not known at the time | |
| """ | |
| # comparison | |
| if not comparison: | |
| comparison = '=' | |
| if field.type in ('text', 'string', 'json'): | |
| if comparison == '=': | |
| # comparison = 'like' # a bit smarter ;) | |
| comparison = 'contains' # a bit smarter ;) | |
| # prefix | |
| prefixes = { | |
| b:a.strip().replace(' ', '_') for a, b in | |
| [ | |
| (' less or equal ','<='), | |
| (' greater or equal ','>='), | |
| (' not equal ','!='), | |
| (' equal ','='), | |
| (' less than ','<'), | |
| (' greater than ','>'), | |
| ] | |
| } | |
| if name_prefix is None: | |
| name_prefix = prefixes.get(comparison, comparison) | |
| # name | |
| name = name or str(field).replace('.', '__') | |
| if name_prefix is None: | |
| name = name_prefix +'__'+ name | |
| search_field = field.clone() # APPLY new name for search fields , but leave original field untouched | |
| search_field.name = name | |
| search_field.label += " (%s)"%name_prefix # DBG | |
| #search_field.comment = name_prefix | |
| from gluon.validators import Validator | |
| if isinstance( search_field.requires , Validator): | |
| print( name, field.requires ) | |
| search_field.requires = IS_EMPTY_OR( field.requires ) | |
| # input = TODO # if widget not available or so.. # Should get into Form elements/conmponents mangling... | |
| target_expression = target_expression or field | |
| # query | |
| query_repr = None # for dbg purposes | |
| def query4filter(field, op, value): | |
| # in general should map: | |
| # field = target_expression (Field or expression of Fields) | |
| # op = comparison | |
| if not type(field) is Field: | |
| raise TypeError('%s is not "Field type"' % field) | |
| elif not hasattr(field, '_tablename') or field._tablename=='no_table': | |
| raise TypeError('%s does not have table specified "' % field) | |
| # taken from pydal/helpers/methods.py def smart_query | |
| if op == '=': new_query = field==value; | |
| elif op == '<': new_query = field<value | |
| elif op == '>': new_query = field>value | |
| elif op == '<=': new_query = field<=value | |
| elif op == '>=': new_query = field>=value | |
| elif op == '!=': new_query = field!=value | |
| elif op == 'belongs': new_query = field.belongs(value) | |
| elif op == 'notbelongs': new_query = ~field.belongs(value) | |
| elif field.type in ('text', 'string', 'json'): | |
| if op == 'contains': new_query = field.contains(value) | |
| elif op == 'like': new_query = field.ilike(value) | |
| elif op == 'startswith': new_query = field.startswith(value) | |
| elif op == 'endswith': new_query = field.endswith(value) | |
| else: raise RuntimeError("Invalid operation") | |
| elif field._db._adapter.dbengine=='google:datastore' and \ | |
| field.type in ('list:integer', 'list:string', 'list:reference'): | |
| if op == 'contains': new_query = field.contains(value) | |
| else: raise RuntimeError("Invalid operation") | |
| else: raise RuntimeError("Invalid operation") | |
| return new_query | |
| if not query: | |
| # def query( value ): return query4filter(field, comparison, value) | |
| query = lambda value: query4filter(target_expression, comparison, value) | |
| return Storage(field=search_field, comparison=comparison, | |
| target_expression=target_expression, | |
| name=name, input=input, query=query) | |
| def searchQueryfromForm( | |
| *filters | |
| ): | |
| # FORM | |
| fields = [] | |
| formname = "" | |
| for filter in filters: | |
| if filter.field: | |
| # filter.field.writable = True | |
| # filter.field.readable = True | |
| fields.append( filter.field ) | |
| formname += filter.field.name | |
| else: | |
| raise RuntimeError("need to define input") | |
| form = SQLFORM.factory( | |
| *fields, | |
| keepvalues=True | |
| ) | |
| form.process(keepvalues=True) | |
| #DBG | |
| # db.technology.sku.name = "bla_bla" | |
| # form = SQLFORM.factory( db.technology.sku, db.technology.type ) | |
| # QUERY | |
| if type(filters[0].target_expression) is Field: | |
| first_table = filters[0].target_expression._tablename # TODO be carefull if Expression is not directly Field | |
| else: | |
| raise TypeError("Not implemented if expression is more robust") | |
| first_query = db[first_table]._id > 0 # dummy query in case no filter is selected -- could be 1==1 | |
| queries = [first_query] | |
| for filter in filters: | |
| vars = request.vars # form.vars? | |
| if filter.name in vars and vars[filter.name]: | |
| queries.append( filter.query( vars[filter.name] ) ) | |
| query = reduce(lambda a, b: (a & b), queries) | |
| # query = reduce(lambda a, b: (a & b), queries) if queries else None | |
| # query = reduce(lambda a, b: (a & b), queries) if queries else 1>0 # just True if no other stuff | |
| return Storage( form=form, query=query ) | |
| TEST = "TEST SEARCH FILTERS QUERY from FORM" | |
| def test_expr(): # ne OK... | |
| search = searchQueryfromForm( | |
| # queryFilter( db.auth_user.first_name, 'contains' ), | |
| queryFilter( Field( "first_name_with_email"), target_expression=db.auth_user.first_name + db.auth_user.email ), | |
| queryFilter( db.auth_user.email ), | |
| ) | |
| # data = SQLFORM.grid((db.auth_user.id < 10) & search.query, fields=( # can toggle | |
| data = db((db.auth_user.id < 100) & search.query).select( *( # can toggle | |
| db.auth_user.id, db.auth_user.first_name, db.auth_user.email, | |
| ) | |
| ) | |
| return dict( data = data, sql = db._lastsql, search_form=search.form, extra=response.tool ) | |
| def test_custom_field(): # OK | |
| search = searchQueryfromForm( | |
| # queryFilter( db.auth_user.first_name, 'contains' ), | |
| queryFilter( Field( "first_name__custom"), target_expression=db.auth_user.first_name ), | |
| queryFilter( Field( "first_name__custom2"), '<', target_expression=db.auth_user.first_name ), | |
| queryFilter( db.auth_user.email ), | |
| ) | |
| # data = SQLFORM.grid((db.auth_user.id < 10) & search.query, fields=( # can toggle | |
| data = db((db.auth_user.id < 100) & search.query).select( *( # can toggle | |
| db.auth_user.id, db.auth_user.first_name, db.auth_user.email, | |
| ) | |
| ) | |
| return dict( data = data, sql = db._lastsql, search_form=search.form, extra=response.tool ) | |
| def test_aggregate(): #TODO | |
| pass | |
| def test_search_query(): | |
| # db.technology.sku.name = "bla.bla" # IGNORUOJA laukus, su tašku pavadinime | |
| search = searchQueryfromForm( | |
| queryFilter( db.technology.active ), | |
| queryFilter( db.technology.sku, '==' ), | |
| queryFilter( db.technology.title, 'contains' ), | |
| queryFilter( db.technology.type ), | |
| queryFilter( db.technology.good_id ), | |
| ) | |
| return dict( | |
| searchform = search.form, | |
| data_grid = ( SQLFORM.grid(search.query, | |
| fields=[db.technology.sku, db.good.title, | |
| # tarp kitko: | |
| # db.technology.good_id, ERROR | |
| # /sqlhtml.py", line 2689, in grid | |
| # nvalue = field.represent(value, row) | |
| # TypeError: <lambda>() takes exactly 1 argument (2 given) | |
| ], | |
| left=[ db.good.on(db.technology.good_id==db.good.id)], | |
| user_signature=False) | |
| if search.query else None), | |
| data_rows = db(search.query).select() if search.query else None, | |
| extra=response.toolbar() | |
| ) | |
| #----- | |
| if "SMART JOINS BUILDER": | |
| from collections import defaultdict | |
| def find_references_and_fkeys( table ): | |
| """ | |
| returns set/dict of referenced tables (associated with grouped foreign keys) | |
| ps.: in most cases there is only one FK for referenced table, but sometimes you kave several | |
| """ | |
| result = defaultdict(list) | |
| for field in db[table]: | |
| f_type = field.type | |
| if isinstance(f_type,str) and ( | |
| f_type.startswith('reference') or | |
| f_type.startswith('list:reference')): | |
| referenced_table = f_type.split()[1].split('.')[0] | |
| result[ referenced_table ].append( field.name ) | |
| return result | |
| def db_reference_map(): | |
| # should be used as singleton -- might cache or store in session | |
| return{ x: find_references_and_fkeys(x) for x in db.tables } | |
| ## smart join smart_join feature prototype | |
| def find_or_check_connection( A, B, A_field=None, B_field=None ): | |
| """ | |
| A and B are Storages | |
| Returns pair/tuple: A_field, B_field | |
| One of fields can be given, then we look for the missing one. | |
| """ | |
| def make_sure__single_ref( refs ): | |
| ref_count = ( len(refs) ) | |
| if ref_count != 1: | |
| msgAmount = "None" if refcount==0 else ("Too many (%s) possible"%ref_count) | |
| # raise TypeError(msgAmount+" references between tables: %s -- %s " % (A+A_field , B) ) | |
| raise TypeError(msgAmount+" references between tables: %s.%s -- %s.%s " % (A, A_field, B, B_field) ) | |
| refs = db_reference_map() | |
| # if we already have full info -- both fields | |
| if A_field and B_field: | |
| # doublecheck if they are OK | |
| if B_field == db[B]._id.name and not A_field in refs[A][B] \ | |
| or A_field == db[A]._id.name and not B_field in refs[B][A] \ | |
| or A_field == db[A]._id.name and B_field == db[B]._id.name \ | |
| or A_field != db[A]._id.name and B_field != db[B]._id.name : | |
| raise ValueError("Wrong given join fields: %s.%s -- %s.%s" % ( A, A_field, B, B_field ) ) | |
| return A_field, B_field | |
| # if we have partial info -- one of fields | |
| # if it is FK | |
| if A_field: | |
| if A_field in refs[A][B]: | |
| return A_field, db[B]._id.name | |
| else: | |
| raise ValueError("Wrong given join field: %s.%s -- %s" % ( A, A_field, B) ) | |
| if B_field: | |
| if B_field in refs[B][A]: | |
| return db[A]._id.name, B_field | |
| else: | |
| raise ValueError("Wrong given join field: %s -- %s.%s" % ( A, B, B_field ) ) | |
| # if it is PK | |
| # in rare cases, it makes sense, for example: | |
| # if A has reference to B and B has reference to A at the same time. (then giving 'id' narrows possibilities) | |
| if A_field == db[A]._id.name: | |
| make_sure__single_ref( refs[B][A] ) # look for FK in B | |
| return db[A]._id.name, refs[B][A][0] | |
| if B_field == db[B]._id.name: # look for FK in A | |
| make_sure__single_ref( refs[A][B] ) | |
| return refs[A][B][0], db[B]._id.name | |
| # find out both fields | |
| if not A_field and not B_field: | |
| make_sure__single_ref( refs[A][B] + refs[B][A] ) | |
| # if there is exactly one reference -- use it | |
| if refs[A][B]: # foreign key in A | |
| return refs[A][B][0], db[B]._id.name | |
| if refs[B][A]: # fk in B | |
| return db[A]._id.name, refs[B][A][0] | |
| from gluon.dal import Expression, Table, Query | |
| # from gluon.packages.dal.pydal._globals import DEFAULT | |
| # from gluon.dal import Expression | |
| def build_joins( path ): | |
| """ | |
| first item in path is supposed to come from initial query/select | |
| Path can contain table|field names (see subfunction parse(..)) | |
| We can also use Expression db.table.on(...) in Path -- | |
| Hopefully this will alow aliases (not tested) | |
| Examples: | |
| No fields | |
| >>> ( build_joins( ['auth_user', 'auth_membership', 'auth_group']) ) | |
| >>> ( build_joins( [db.auth_user, db.auth_membership, 'auth_group']) ) | |
| >>> ( build_joins( [db.auth_user, db.auth_membership, db.auth_group, db.auth_permission]) ) | |
| Many to many with both fields | |
| >>> ( build_joins( ['auth_user', ('auth_membership', 'user_id', 'group_id'), 'auth_group']) ) | |
| >>> ( build_joins( ['auth_user', (db.auth_membership, 'user_id', 'group_id'), 'auth_group']) ) | |
| Many to many with one field | |
| >>> ( build_joins( ['auth_user', ('auth_membership', None, 'group_id'), 'auth_group']) ) | |
| >>> ( build_joins( ['auth_user', (db.auth_membership, 'user_id', None), 'auth_group']) ) | |
| First with just right field | |
| >>> ( build_joins( [ (db.auth_membership, None, 'group_id'), 'auth_group']) ) | |
| Last with just left field | |
| >>> ( build_joins( ['auth_user', (db.auth_membership, 'user_id', None)]) ) | |
| ### more experimental | |
| >>> ( build_joins( [db.auth_user, db.auth_membership.user_id ]) ) | |
| >>> ( build_joins( [db.auth_user, db.auth_membership.user_id, db.auth_group ]) ) | |
| >>> ( build_joins( ['auth_user', db.auth_membership.group_id, 'auth_group']) ) | |
| #auth_user <- auth_membership -> auth_group <- auth_permission | |
| >>> ( build_joins( ['auth_user', db.auth_membership.group_id, 'auth_group', db.auth_permission.group_id]) ) | |
| # Expresion | |
| >>> ( build_joins( ['auth_user', db.auth_membership, db.auth_group.on(db.auth_group.id == db.auth_membership.group_id), db.auth_permission.group_id]) ) | |
| """ | |
| if len(path) < 2: raise ValueError("There should be at least 2 tables mentioned in %s" % path ) | |
| prev = None | |
| nr = 0 | |
| def parse( item ): | |
| print item | |
| """ | |
| item can be either of 1, 2, or 3 parts (tuple or just table/field/string (but it must include tablename at least)) | |
| There will be lots of inference based on situation | |
| One: | |
| >>> parse( 'tablename' ) | |
| >>> parse( db.table ) | |
| >>> parse( db.table.field ) | |
| Two parts: | |
| >>> parse( (db.table.field1, 'field2' ) ) | |
| >>> parse( (db.table.field1, db.table.field2 ) ) | |
| Three parts: | |
| >>> parse( (db.table, 'field1', 'field2' ) ) | |
| Finds out the connection between previous and current table | |
| returns current table and possibly modifies prev.left_field | |
| if needed, finds out connection: prev right_field to current left_field | |
| """ | |
| left_field = right_field = undecided_field = None | |
| if isinstance(item, tuple) and len(item)==3: # 3 | |
| table, left_field, right_field = item | |
| table = str(table) | |
| if isinstance(item, tuple) and len(item)==2: # 2 | |
| left_field, right_field = item | |
| count = sum([1 for field in left_field, right_field if type(field) is Field] ) | |
| if count == 0: | |
| raise TypeError( "None of %s is <Field ...> type -- can't find out table" % item ) | |
| for field in left_field, right_field : | |
| if isinstance(field, Field): | |
| table = left_field._table | |
| # make sure values of fields are strings | |
| def just_field_name(field): | |
| return field.name if isinstance(field, Field) else field | |
| left_field = just_field_name(left_field) | |
| right_field = just_field_name(right_field) | |
| if isinstance(item, tuple) and len(item)==1: # just in case | |
| item = item[0] | |
| # if type(item) in [Field, Table, str]: # might happen MockTable or WeirdField | |
| if isinstance(item, (Field, Table, str) ): # 1 | |
| table = str(item) | |
| left_field = right_field = None | |
| print "DBG", table | |
| if '.' in table: | |
| table, undecided_field = table.split('.') # will try find_or_check_connection with undecided as left and as right | |
| if prev is None: | |
| right_field = undecided_field | |
| undecided_field = None | |
| # TypeError("Should be just table (without field), found %s" % item) # TODO could take field and later on decide if it joins to Left or Right | |
| # current = Storage( table=table, left_field=left_field, right_field=right_field, undecided_field=undecided_field ) | |
| if prev: # if previous exist - if not the starting table in chain | |
| if undecided_field: | |
| try: # try undecided as left (B_field parameter) | |
| prev.right_field, left_field = find_or_check_connection( A=prev.table, B=table, B_field=undecided_field) # | |
| # if everything OK | |
| undecided_field = None | |
| except ValueError as e: | |
| print '"Undecided" field not OK for left: "', e | |
| left_field = None | |
| # if item == path[-1]: # was error with ['auth_user', db.auth_membership.group_id, 'auth_group', db.auth_permission.group_id] as group_id`s from different tables seemd to think they are equal | |
| if nr == len(path)-1: | |
| raise ValueError('"Undecided"/unmatched field in last table %s'% item) | |
| right_field = undecided_field # try undecided as right: will trigger in next call | |
| try: | |
| prev.right_field, left_field = find_or_check_connection( prev.table, table, prev.right_field, left_field ) # if we don't know left or any fields | |
| except ValueError as e: | |
| if prev.undecided_field: # check if error was probably because of undecided field | |
| prev.right_field = None | |
| raise ValueError ('"Undecided" field not matched for %s.%s'%(prev.table, prev.undecided_field)+"\n"+str(e) ) | |
| else: | |
| raise e | |
| current = Storage( table=table, left_field=left_field, right_field=right_field, undecided_field=undecided_field ) | |
| return current | |
| # from gluon.debug import dbg | |
| # dbg.set_trace() | |
| prev = parse( path[0] ) # previous table # TODO - what if it is Expression with alias? | |
| print "\n",map(str, path), "\n" # DBG | |
| joins = [] | |
| for item in path[1:]: | |
| nr += 1 | |
| # if isinstance(item, Expression): would be right for field as well ? :/ | |
| if type(item) is Expression: # but not Field... # we can include prepaired ON's # TODO: maybe check if Expression really is "ON" ? | |
| joins.append( item ) | |
| current = Storage( table = str(item.first) ) # hopefully would work with alias? | |
| else: | |
| current = parse( item ) # current table | |
| # print "\n", current, "\n", prev # DBG | |
| joins.append( db[current.table].on( db[current.table][current.left_field] == db[prev.table][prev.right_field] ) ) | |
| print current | |
| prev = current | |
| # dbg.stop_trace() | |
| return joins | |
| def fields(table, field_names): | |
| return [ str(db[table][fname]) for fname in field_names] | |
| TEST = "TEST SMART JOINS BUILDER" | |
| #----------- | |
| def populate_fake_auth(): | |
| from gluon.contrib.populate import populate | |
| # for table in 'auth_user <- auth_membership -> auth_group <- auth_permission'.split(): | |
| for table in 'auth_user auth_group auth_permission auth_membership'.split(): | |
| populate(db[table],5) | |
| db.commit() | |
| def test(): # test with alias -- seems OK | |
| # PATOGUMAS 1! | |
| search = searchQueryfromForm( | |
| queryFilter( db.auth_user.first_name, 'contains' ), | |
| queryFilter( db.auth_user.email ), | |
| queryFilter( db.auth_membership.group_id ), | |
| queryFilter( db.auth_membership.user_id ), | |
| queryFilter( db.auth_group.role ), | |
| queryFilter( db.auth_permission.name ), | |
| ) | |
| # data = SQLFORM.grid((db.auth_user.id < 10) & search.query, fields=( # can toggle | |
| data = db((db.auth_user.id < 100) & search.query).select( *( # can toggle | |
| db.auth_user.id, db.auth_user.first_name, db.auth_user.email, | |
| db.auth_membership.id, | |
| db.auth_group.id, db.auth_group.role, | |
| db.auth_permission.id, db.auth_permission.name, db.auth_permission.table_name, | |
| # *fields( db.auth_user, 'id first_name email'.split()) | |
| ), | |
| # PATOGUMAS 2! | |
| left = build_joins( ['auth_user', db.auth_membership.group_id, db.auth_group.on( db.auth_group.id == db.auth_membership.group_id), db.auth_permission.group_id] ) | |
| ) | |
| return dict( data = data, sql = db._lastsql, search_form=search.form, extra=response.tool ) | |
| # return CAT( data , db._lastsql ) | |
| def test3(): # OK | |
| data = SQLFORM.grid(db.auth_user.id < 10, fields=( | |
| # data = db(db.auth_user.id < 10 ).select( *( | |
| db.auth_user.id, db.auth_user.first_name, db.auth_user.email, | |
| db.auth_membership.id, | |
| db.auth_group.id, db.auth_group.role, | |
| db.auth_permission.id, db.auth_permission.name, db.auth_permission.table_name, | |
| # *fields( db.auth_user, 'id first_name email'.split()) | |
| ), | |
| # *fields( 'auth_permission', 'name table_name record_id'.split() ), | |
| # left = build_joins( ['auth_user', db.auth_membership.group_id, 'auth_group', db.auth_permission.group_id] ) | |
| left = build_joins( ['auth_user', db.auth_membership, 'auth_group', db.auth_permission.group_id] ) | |
| ) | |
| return dict( data = data, sql = db._lastsql ) | |
| def test2(): # OK | |
| # tables = ['auth_user', db.auth_membership.group_id ] | |
| tables = ['auth_user', db.auth_membership ] | |
| return db().select( | |
| *[ db[table].id for table in tables], | |
| left = build_joins(['auth_user', (db.auth_membership, 'user_id', None) ]) | |
| # left = build_joins( ['auth_user', db.auth_membership.group_id ] ) | |
| ) | |
| def test_query(): | |
| # patikrinam, ką grąžina query'is | |
| # return dict(dbg=db.auth_user.id>0) | |
| return dict(dbg=Query(db, db._adapter.EQ, 1, 1) ) | |
| # return dict(dbg=Query(db, db._adapter.LT, 1, 2) ) | |
| def test_alias(): | |
| # patikrinam, ką grąžina alias'as | |
| return dict(dbg=db.auth_user.with_alias('bla')) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment