From 50193471d69030dfff076b448acf03d80026e8bf Mon Sep 17 00:00:00 2001 From: Massimo Date: Thu, 14 Feb 2013 14:15:27 -0600 Subject: [PATCH] better IS_NOT_IN_DB, new scheduler API, after_connection dal callback, thanks Niphlod --- VERSION | 2 +- applications/admin/static/js/ajax_editor.js | 21 ++-- gluon/dal.py | 108 ++++++++++++++------ gluon/scheduler.py | 70 +++++++++---- gluon/tests/test_dal.py | 22 +++- gluon/validators.py | 4 +- 6 files changed, 157 insertions(+), 70 deletions(-) diff --git a/VERSION b/VERSION index 86fea6d2..91bb1985 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -Version 2.4.1-alpha.2+timestamp.2013.02.13.09.35.21 +Version 2.4.1-alpha.2+timestamp.2013.02.14.14.14.43 diff --git a/applications/admin/static/js/ajax_editor.js b/applications/admin/static/js/ajax_editor.js index d2d733ee..168236d4 100644 --- a/applications/admin/static/js/ajax_editor.js +++ b/applications/admin/static/js/ajax_editor.js @@ -11,7 +11,7 @@ function prepareMultiPartPOST(data) { var reqdata = '--' + boundary + '\r\n'; //console.log(data.length); for (var i=0;i < data.length;i++) { - reqdata += 'content-disposition: form-data; name="'; + reqdata += 'content-disposition: form-data; name="'; reqdata += data[i].Name + '"'; reqdata += "\r\n\r\n" ; reqdata += data[i].Data; @@ -44,7 +44,7 @@ function doHighlight(highlight) { window.ace_editor.gotoLine(highlight.lineno); } else if (window.mirror) { // Put the cursor at the offending line: - window.mirror.setCursor({line:highlight.lineno-1, + window.mirror.setCursor({line:highlight.lineno-1, ch:highlight.offset+1}); } else if (window.eamy) { // not implemented @@ -57,11 +57,11 @@ function doClickSave() { var data = getData(); var dataForPost = prepareMultiPartPOST(new Array( prepareDataForSave('data', data), - prepareDataForSave('file_hash', + prepareDataForSave('file_hash', jQuery("input[name='file_hash']").val()), - prepareDataForSave('saved_on', + prepareDataForSave('saved_on', jQuery("input[name='saved_on']").val()), - prepareDataForSave('saved_on', + prepareDataForSave('saved_on', jQuery("input[name='saved_on']").val()), prepareDataForSave('from_ajax','true'))); // console.info(area.textarea.value); @@ -70,7 +70,7 @@ function doClickSave() { jQuery("input[name='saved_on']").val('saving now...') jQuery.ajax({ type: "POST", - contentType: 'multipart/form-data;boundary="' + contentType: 'multipart/form-data;boundary="' + dataForPost[1] + '"', url: self.location.href, dataType: "json", @@ -106,7 +106,7 @@ function doClickSave() { doHighlight(json.highlight); } else { jQuery("input[name='saved_on']").attr('style','background-color:#99FF99'); - jQuery(".flash").delay(1000).fadeOut('slow'); + //jQuery(".flash").delay(1000).fadeOut('slow'); } // console.info(jQuery("input[name='file_hash']").val()); var output = 'exposes: '; @@ -187,13 +187,13 @@ function doToggleBreakpoint(filename, url, sel) { } else { if (json.ok==true && window.mirror) { // mark the breakpoint if ok=True - editor.setMarker(json.lineno-1, + editor.setMarker(json.lineno-1, "● %N%") } else if (json.ok==false && window.mirror) { // remove mark if ok=False editor.setMarker(json.lineno-1, "%N%") } else { - // do nothing if ok = null + // do nothing if ok = null } // alert(json.ok + json.lineno); } @@ -230,7 +230,7 @@ function doListBreakpoints(filename, url) { for (i in json.breakpoints) { lineno = json.breakpoints[i]; // mark the breakpoint if ok=True - editor.setMarker(lineno-1, + editor.setMarker(lineno-1, "● %N%"); } } @@ -250,4 +250,3 @@ function keepalive(url) { success: function(){}, error: function(x) { on_error(); } }); } - diff --git a/gluon/dal.py b/gluon/dal.py index ff411b5c..14f7c460 100644 --- a/gluon/dal.py +++ b/gluon/dal.py @@ -568,8 +568,14 @@ class ConnectionPool(object): if False and self.folder and not exists(self.folder): os.mkdir(self.folder) + def after_connection_hook(self): + """hook for the after_connection parameter""" + if callable(self._after_connection): + self._after_connection(self) + self.after_connection() + def after_connection(self): - """ this it is suppoed to be overloaded by adtapters""" + """ this it is supposed to be overloaded by adapters""" pass def reconnect(self, f=None, cursor=True): @@ -610,7 +616,7 @@ class ConnectionPool(object): self.connection = f() self.cursor = cursor and self.connection.cursor() break - self.after_connection() + self.after_connection_hook() ################################################################################### @@ -726,13 +732,14 @@ class BaseAdapter(ConnectionPool): def __init__(self, db,uri,pool_size=0, folder=None, db_codec='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={},do_connect=True): + adapter_args={},do_connect=True, after_connection=None): self.db = db self.dbengine = "None" self.uri = uri self.pool_size = pool_size self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection class Dummy(object): lastrowid = 1 def __getattr__(self, value): @@ -2144,7 +2151,7 @@ class SQLiteAdapter(BaseAdapter): def __init__(self, db, uri, pool_size=0, folder=None, db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, do_connect=True, after_connection=None): self.db = db self.dbengine = "sqlite" self.uri = uri @@ -2152,6 +2159,7 @@ class SQLiteAdapter(BaseAdapter): self.pool_size = 0 self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.find_or_make_work_folder() path_encoding = sys.getfilesystemencoding() \ or locale.getdefaultlocale()[1] or 'utf8' @@ -2210,7 +2218,7 @@ class SpatiaLiteAdapter(SQLiteAdapter): def __init__(self, db, uri, pool_size=0, folder=None, db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True, srid=4326): + adapter_args={}, do_connect=True, srid=4326, after_connection=None): self.db = db self.dbengine = "spatialite" self.uri = uri @@ -2218,6 +2226,7 @@ class SpatiaLiteAdapter(SQLiteAdapter): self.pool_size = 0 self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.find_or_make_work_folder() self.srid = srid path_encoding = sys.getfilesystemencoding() \ @@ -2315,7 +2324,7 @@ class JDBCSQLiteAdapter(SQLiteAdapter): def __init__(self, db, uri, pool_size=0, folder=None, db_codec='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, do_connect=True, after_connection=None): self.db = db self.dbengine = "sqlite" self.uri = uri @@ -2323,6 +2332,7 @@ class JDBCSQLiteAdapter(SQLiteAdapter): self.pool_size = pool_size self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.find_or_make_work_folder() path_encoding = sys.getfilesystemencoding() \ or locale.getdefaultlocale()[1] or 'utf8' @@ -2419,7 +2429,7 @@ class MySQLAdapter(BaseAdapter): def __init__(self,db,uri,pool_size=0,folder=None,db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, do_connect=True, after_connection=None): self.db = db self.dbengine = "mysql" self.uri = uri @@ -2427,6 +2437,7 @@ class MySQLAdapter(BaseAdapter): self.pool_size = pool_size self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.find_or_make_work_folder() ruri = uri.split('://',1)[1] m = self.REGEX_URI.match(ruri) @@ -2546,7 +2557,8 @@ class PostgreSQLAdapter(BaseAdapter): def __init__(self,db,uri,pool_size=0,folder=None,db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True, srid=4326): + adapter_args={}, do_connect=True, srid=4326, + after_connection=None): self.db = db self.dbengine = "postgres" self.uri = uri @@ -2554,6 +2566,7 @@ class PostgreSQLAdapter(BaseAdapter): self.pool_size = pool_size self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.srid = srid self.find_or_make_work_folder() ruri = uri.split('://',1)[1] @@ -2615,14 +2628,14 @@ class PostgreSQLAdapter(BaseAdapter): def LIKE(self,first,second): args = (self.expand(first), self.expand(second,'string')) if not first.type in ('string', 'text', 'json'): - return '(CAST(%s AS CHAR) LIKE %s)' % args + return '(CAST(%s AS CHAR(%s)) LIKE %s)' % (args[0], first.length, args[1]) else: return '(%s LIKE %s)' % args def ILIKE(self,first,second): args = (self.expand(first), self.expand(second,'string')) if not first.type in ('string', 'text', 'json'): - return '(CAST(%s AS CHAR) ILIKE %s)' % args + return '(CAST(%s AS CHAR(%s)) LIKE %s)' % (args[0], first.length, args[1]) else: return '(%s ILIKE %s)' % args @@ -2799,7 +2812,7 @@ class JDBCPostgreSQLAdapter(PostgreSQLAdapter): def __init__(self,db,uri,pool_size=0,folder=None,db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, do_connect=True, after_connection=None ): self.db = db self.dbengine = "postgres" self.uri = uri @@ -2807,6 +2820,7 @@ class JDBCPostgreSQLAdapter(PostgreSQLAdapter): self.pool_size = pool_size self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.find_or_make_work_folder() ruri = uri.split('://',1)[1] m = self.REGEX_URI.match(ruri) @@ -2926,7 +2940,7 @@ class OracleAdapter(BaseAdapter): def __init__(self,db,uri,pool_size=0,folder=None,db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, do_connect=True, after_connection=None): self.db = db self.dbengine = "oracle" self.uri = uri @@ -2934,6 +2948,7 @@ class OracleAdapter(BaseAdapter): self.pool_size = pool_size self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.find_or_make_work_folder() ruri = uri.split('://',1)[1] if not 'threaded' in driver_args: @@ -3089,7 +3104,8 @@ class MSSQLAdapter(BaseAdapter): def __init__(self,db,uri,pool_size=0,folder=None,db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True, srid=4326): + adapter_args={}, do_connect=True, srid=4326, + after_connection=None): self.db = db self.dbengine = "mssql" self.uri = uri @@ -3097,6 +3113,7 @@ class MSSQLAdapter(BaseAdapter): self.pool_size = pool_size self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.srid = srid self.find_or_make_work_folder() # ## read: http://bytes.com/groups/python/460325-cx_oracle-utf8 @@ -3312,7 +3329,8 @@ class SybaseAdapter(MSSQLAdapter): def __init__(self,db,uri,pool_size=0,folder=None,db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True, srid=4326): + adapter_args={}, do_connect=True, srid=4326, + after_connection=None): self.db = db self.dbengine = "sybase" self.uri = uri @@ -3320,6 +3338,7 @@ class SybaseAdapter(MSSQLAdapter): self.pool_size = pool_size self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.srid = srid self.find_or_make_work_folder() # ## read: http://bytes.com/groups/python/460325-cx_oracle-utf8 @@ -3448,7 +3467,7 @@ class FireBirdAdapter(BaseAdapter): def __init__(self,db,uri,pool_size=0,folder=None,db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, do_connect=True, after_connection=None): self.db = db self.dbengine = "firebird" self.uri = uri @@ -3456,6 +3475,7 @@ class FireBirdAdapter(BaseAdapter): self.pool_size = pool_size self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.find_or_make_work_folder() ruri = uri.split('://',1)[1] m = self.REGEX_URI.match(ruri) @@ -3507,7 +3527,7 @@ class FireBirdEmbeddedAdapter(FireBirdAdapter): def __init__(self,db,uri,pool_size=0,folder=None,db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, do_connect=True, after_connection=None): self.db = db self.dbengine = "firebird" self.uri = uri @@ -3515,6 +3535,7 @@ class FireBirdEmbeddedAdapter(FireBirdAdapter): self.pool_size = pool_size self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.find_or_make_work_folder() ruri = uri.split('://',1)[1] m = self.REGEX_URI.match(ruri) @@ -3615,7 +3636,7 @@ class InformixAdapter(BaseAdapter): def __init__(self,db,uri,pool_size=0,folder=None,db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, do_connect=True, after_connection=None): self.db = db self.dbengine = "informix" self.uri = uri @@ -3623,6 +3644,7 @@ class InformixAdapter(BaseAdapter): self.pool_size = pool_size self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.find_or_make_work_folder() ruri = uri.split('://',1)[1] m = self.REGEX_URI.match(ruri) @@ -3729,7 +3751,7 @@ class DB2Adapter(BaseAdapter): def __init__(self,db,uri,pool_size=0,folder=None,db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, do_connect=True, after_connection=None): self.db = db self.dbengine = "db2" self.uri = uri @@ -3737,6 +3759,7 @@ class DB2Adapter(BaseAdapter): self.pool_size = pool_size self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.find_or_make_work_folder() ruri = uri.split('://', 1)[1] def connector(cnxn=ruri,driver_args=driver_args): @@ -3793,7 +3816,7 @@ class TeradataAdapter(BaseAdapter): def __init__(self,db,uri,pool_size=0,folder=None,db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, do_connect=True, after_connection=None): self.db = db self.dbengine = "teradata" self.uri = uri @@ -3801,6 +3824,7 @@ class TeradataAdapter(BaseAdapter): self.pool_size = pool_size self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.find_or_make_work_folder() ruri = uri.split('://', 1)[1] def connector(cnxn=ruri,driver_args=driver_args): @@ -3875,7 +3899,7 @@ class IngresAdapter(BaseAdapter): def __init__(self,db,uri,pool_size=0,folder=None,db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, do_connect=True, after_connection=None): self.db = db self.dbengine = "ingres" self.uri = uri @@ -3883,6 +3907,7 @@ class IngresAdapter(BaseAdapter): self.pool_size = pool_size self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.find_or_make_work_folder() connstr = self._uri.split(':', 1)[1] # Simple URI processing @@ -4013,7 +4038,7 @@ class SAPDBAdapter(BaseAdapter): def __init__(self,db,uri,pool_size=0,folder=None,db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, do_connect=True, after_connection=None): self.db = db self.dbengine = "sapdb" self.uri = uri @@ -4021,6 +4046,7 @@ class SAPDBAdapter(BaseAdapter): self.pool_size = pool_size self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.find_or_make_work_folder() ruri = uri.split('://',1)[1] m = self.REGEX_URI.match(ruri) @@ -4056,7 +4082,7 @@ class CubridAdapter(MySQLAdapter): def __init__(self, db, uri, pool_size=0, folder=None, db_codec='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, do_connect=True, after_connection=None): self.db = db self.dbengine = "cubrid" self.uri = uri @@ -4064,6 +4090,7 @@ class CubridAdapter(MySQLAdapter): self.pool_size = pool_size self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.find_or_make_work_folder() ruri = uri.split('://',1)[1] m = self.REGEX_URI.match(ruri) @@ -4199,13 +4226,14 @@ class GoogleSQLAdapter(UseDatabaseStoredFile,MySQLAdapter): def __init__(self, db, uri='google:sql://realm:domain/database', pool_size=0, folder=None, db_codec='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, do_connect=True, after_connection=None): self.db = db self.dbengine = "mysql" self.uri = uri self.pool_size = pool_size self.db_codec = db_codec + self._after_connection = after_connection self.folder = folder or pjoin('$HOME',THREAD_LOCAL.folder.split( os.sep+'applications'+os.sep,1)[1]) ruri = uri.split("://")[1] @@ -4422,7 +4450,7 @@ class GoogleDatastoreAdapter(NoSQLAdapter): def __init__(self,db,uri,pool_size=0,folder=None,db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, do_connect=True, after_connection=None): self.types.update({ 'boolean': gae.BooleanProperty, 'string': (lambda: gae.StringProperty(multiline=True)), @@ -4451,6 +4479,7 @@ class GoogleDatastoreAdapter(NoSQLAdapter): self.folder = folder db['_lastsql'] = '' self.db_codec = 'UTF-8' + self._after_connection = after_connection self.pool_size = 0 match = self.REGEX_NAMESPACE.match(uri) if match: @@ -4887,7 +4916,7 @@ class CouchDBAdapter(NoSQLAdapter): def __init__(self,db,uri='couchdb://127.0.0.1:5984', pool_size=0,folder=None,db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, do_connect=True, after_connection=None): self.db = db self.uri = uri if do_connect: self.find_driver(adapter_args) @@ -4895,6 +4924,7 @@ class CouchDBAdapter(NoSQLAdapter): self.folder = folder db['_lastsql'] = '' self.db_codec = 'UTF-8' + self._after_connection = after_connection self.pool_size = pool_size url='http://'+uri[10:] @@ -5053,7 +5083,7 @@ class MongoDBAdapter(NoSQLAdapter): def __init__(self,db,uri='mongodb://127.0.0.1:5984/db', pool_size=0, folder=None, db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, do_connect=True, after_connection=None): self.db = db self.uri = uri @@ -5073,6 +5103,7 @@ class MongoDBAdapter(NoSQLAdapter): self.folder = folder db['_lastsql'] = '' self.db_codec = 'UTF-8' + self._after_connection = after_connection self.pool_size = pool_size #this is the minimum amount of replicates that it should wait # for on insert/update @@ -5737,7 +5768,9 @@ class IMAPAdapter(NoSQLAdapter): db_codec ='UTF-8', credential_decoder=IDENTITY, driver_args={}, - adapter_args={}, do_connect=True): + adapter_args={}, + do_connect=True, + after_connection=None): # db uri: user@example.com:password@imap.server.com:123 # TODO: max size adapter argument for preventing large mail transfers @@ -5748,6 +5781,7 @@ class IMAPAdapter(NoSQLAdapter): self.pool_size=pool_size self.folder = folder self.db_codec = db_codec + self._after_connection = after_connection self.credential_decoder = credential_decoder self.driver_args = driver_args self.adapter_args = adapter_args @@ -5852,7 +5886,7 @@ class IMAPAdapter(NoSQLAdapter): self.connection = f() self.cursor = cursor and self.connection.cursor() break - self.after_connection() + self.after_connection_hook() def get_last_message(self, tablename): last_message = None @@ -7124,7 +7158,7 @@ class DAL(object): 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): + db_uid=None, do_connect=True, after_connection=None): """ Creates a new Database Abstraction Layer instance. @@ -7133,7 +7167,9 @@ class DAL(object): :uri: string that contains information for connecting to a database. (default: 'sqlite://dummy.db') :pool_size: How many open connections to make to the database object. - :folder: + :folder: where .table files will be created. + automatically set within web2py + use an explicit path when using DAL outside web2py :db_codec: string encoding of the database (default: 'UTF-8') :check_reserved: list of adapters to check tablenames and column names against sql/nosql reserved keywords. (Default None) @@ -7150,6 +7186,11 @@ class DAL(object): :migrate_enabled (defaults to True). If set to False disables ALL migrations :fake_migrate_all (defaults to False). If sets to True fake migrates ALL tables :attempts (defaults to 5). Number of times to attempt connecting + :auto_import (defaults to False). If set, import automatically table definitions from the + databases folder + :bigint_id (defaults to False): If set, turn on bigint instead of int for id fields + :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 """ DEFAULT_URI = 'sqlite://dummy.db' @@ -7219,7 +7260,8 @@ class DAL(object): credential_decoder=credential_decoder, driver_args=driver_args or {}, adapter_args=adapter_args or {}, - do_connect=do_connect) + do_connect=do_connect, + after_connection=after_connection) self._adapter = ADAPTERS[self._dbname](**kwargs) types = ADAPTERS[self._dbname].types # copy so multiple DAL() possible @@ -7244,7 +7286,7 @@ class DAL(object): else: self._adapter = BaseAdapter(db=self,pool_size=0, uri='None',folder=folder, - db_codec=db_codec) + db_codec=db_codec, after_connection=after_connection) migrate = fake_migrate = False adapter = self._adapter self._uri_hash = hashlib_md5(adapter.uri).hexdigest() @@ -7551,7 +7593,7 @@ def index(): return Row({'status':400,'error':'invalid orderby','response':None}) if exposedfields: fields = [field for field in db[table] if str(field).split('.')[-1] in exposedfields and field.readable] - else: + else: fields = [field for field in db[table] if field.readable] count = dbset.count() try: diff --git a/gluon/scheduler.py b/gluon/scheduler.py index a866acc5..26870e85 100644 --- a/gluon/scheduler.py +++ b/gluon/scheduler.py @@ -86,7 +86,9 @@ try: except: from simplejson import loads, dumps -logger = logging.getLogger('web2py.scheduler') +IDENTIFIER = "%s#%s" % (socket.gethostname(),os.getpid()) + +logger = logging.getLogger('web2py.scheduler.%s' % IDENTIFIER) from gluon import DAL, Field, IS_NOT_EMPTY, IS_IN_SET, IS_NOT_IN_DB, IS_INT_IN_RANGE, IS_DATETIME from gluon.utils import web2py_uuid @@ -103,6 +105,7 @@ ACTIVE = 'ACTIVE' TERMINATE = 'TERMINATE' DISABLED = 'DISABLED' KILL = 'KILL' +PICK = 'PICK' EXPIRED = 'EXPIRED' SECONDS = 1 HEARTBEAT = 3 * SECONDS @@ -391,7 +394,7 @@ class MetaScheduler(threading.Thread): TASK_STATUS = (QUEUED, RUNNING, COMPLETED, FAILED, TIMEOUT, STOPPED, EXPIRED) RUN_STATUS = (RUNNING, COMPLETED, FAILED, TIMEOUT, STOPPED) -WORKER_STATUS = (ACTIVE, DISABLED, TERMINATE, KILL) +WORKER_STATUS = (ACTIVE, PICK, DISABLED, TERMINATE, KILL) class TYPE(object): @@ -431,8 +434,7 @@ class Scheduler(MetaScheduler): self.tasks = tasks self.group_names = group_names self.heartbeat = heartbeat - self.worker_name = worker_name or socket.gethostname( - ) + '#' + str(os.getpid()) + self.worker_name = worker_name or IDENTIFIER #list containing status as recorded in the table plus a boost parameter #for hibernation (i.e. when someone stop the worker acting on the worker table) self.worker_status = [RUNNING, 1] @@ -565,7 +567,7 @@ class Scheduler(MetaScheduler): break except: db.rollback() - logger.error('TICKER(%s): error assigning tasks', self.worker_name) + logger.error('TICKER: error assigning tasks') x += 1 time.sleep(0.5) @@ -580,10 +582,10 @@ class Scheduler(MetaScheduler): break except: db.rollback() - logger.error('%s: error popping tasks', self.worker_name) + logger.error(' error popping tasks') x += 1 time.sleep(0.5) - + def pop_task(self, db): now = self.now() st = self.db.scheduler_task @@ -605,7 +607,7 @@ class Scheduler(MetaScheduler): else: if self.greedy and self.is_a_ticker: #there are other tasks ready to be assigned - logger.info('TICKER (%s): greedy loop', self.worker_name) + logger.info('TICKER: greedy loop') self.wrapped_assign_tasks(db) else: logger.info('nothing to do') @@ -726,31 +728,32 @@ class Scheduler(MetaScheduler): sw, st = db.scheduler_worker, db.scheduler_task now = self.now() # record heartbeat - mybackedstatus = db( - sw.worker_name == self.worker_name).select().first() + mybackedstatus = db(sw.worker_name == self.worker_name).select().first() if not mybackedstatus: sw.insert(status=ACTIVE, worker_name=self.worker_name, first_heartbeat=now, last_heartbeat=now, group_names=self.group_names) self.worker_status = [ACTIVE, 1] # activating the process + mybackedstatus = ACTIVE else: - if mybackedstatus.status == DISABLED: + mybackedstatus = mybackedstatus.status + if mybackedstatus == DISABLED: # keep sleeping self.worker_status[0] = DISABLED if self.worker_status[1] == MAXHIBERNATION: logger.debug('........recording heartbeat') db(sw.worker_name == self.worker_name).update( last_heartbeat=now) - elif mybackedstatus.status == TERMINATE: + elif mybackedstatus == TERMINATE: self.worker_status[0] = TERMINATE logger.debug("Waiting to terminate the current task") self.give_up() return - elif mybackedstatus.status == KILL: + elif mybackedstatus == KILL: self.worker_status[0] = KILL self.die() else: - logger.debug('........recording heartbeat (%s)', self.worker_status[0]) + logger.debug('........recording heartbeat (%s) %s', self.worker_status[0], mybackedstatus) db(sw.worker_name == self.worker_name).update( last_heartbeat=now, status=ACTIVE) self.worker_status[1] = 1 # re-activating the process @@ -758,8 +761,7 @@ class Scheduler(MetaScheduler): self.worker_status[0] = ACTIVE self.do_assign_tasks = False - - if counter % 5 == 0: + if counter % 5 == 0 or mybackedstatus == PICK: try: # delete inactive workers expiration = now - datetime.timedelta(seconds=self.heartbeat * 3) @@ -769,8 +771,7 @@ class Scheduler(MetaScheduler): ' freeing workers that have not sent heartbeat') inactive_workers = db( ((sw.last_heartbeat < expiration) & (sw.status == ACTIVE)) | - ((sw.last_heartbeat < - departure) & (sw.status != ACTIVE)) + ((sw.last_heartbeat < departure) & (sw.status != ACTIVE)) ) db(st.assigned_worker_name.belongs( inactive_workers._select(sw.worker_name)))(st.status == RUNNING)\ @@ -800,7 +801,7 @@ class Scheduler(MetaScheduler): #only if this worker isn't busy, otherwise wait for a free one db(sw.worker_name == self.worker_name).update(is_ticker=True) db(sw.worker_name != self.worker_name).update(is_ticker=False) - logger.info("TICKER(%s): I'm a ticker", self.worker_name) + logger.info("TICKER: I'm a ticker") else: #giving up, only if I'm not alone if len(all_active) > 1: @@ -888,13 +889,35 @@ class Scheduler(MetaScheduler): #I'll be greedy only if tasks assigned are equal to the limit # (meaning there could be others ready to be assigned) self.greedy = x >= limit and True or False - logger.info('TICKER(%s): workers are %s', self.worker_name, len(all_workers)) - logger.info('TICKER(%s): tasks are %s', self.worker_name, x) + logger.info('TICKER: workers are %s', len(all_workers)) + logger.info('TICKER: tasks are %s', x) def sleep(self): time.sleep(self.heartbeat * self.worker_status[1]) # should only sleep until next available task + def set_worker_status(self, group_names=None, action=ACTIVE): + if not group_names: + group_names = self.group_names + elif isinstance(group_names, str): + group_names = [group_names] + for group in group_names: + self.db( + self.db.scheduler_worker.group_names.contains(group) + ).update(status=action) + + def disable(self, group_names=None): + self.set_worker_status(group_names=group_names,action=DISABLED) + + def resume(self, group_names=None): + self.set_worker_status(group_names=group_names,action=ACTIVE) + + def terminate(self, group_names=None): + self.set_worker_status(group_names=group_names,action=TERMINATE) + + def kill(self, group_names=None): + self.set_worker_status(group_names=group_names,action=KILL) + def queue_task(self, function, pargs=[], pvars={}, **kwargs): """ Queue tasks. This takes care of handling the validation of all @@ -917,6 +940,7 @@ class Scheduler(MetaScheduler): tvars = 'vars' in kwargs and kwargs.pop('vars') or dumps(pvars) tuuid = 'uuid' in kwargs and kwargs.pop('uuid') or web2py_uuid() tname = 'task_name' in kwargs and kwargs.pop('task_name') or function + immediate = 'immediate' in kwargs and kwargs.pop('immediate') or None rtn = self.db.scheduler_task.validate_and_insert( function_name=function, task_name=tname, @@ -926,6 +950,8 @@ class Scheduler(MetaScheduler): **kwargs) if not rtn.errors: rtn.uuid = tuuid + if immediate: + self.db(self.db.scheduler_worker.is_ticker == True).update(status=PICK) else: rtn.uuid = None return rtn @@ -970,7 +996,7 @@ class Scheduler(MetaScheduler): left=left, limitby=(0, 1)) ).first() - if output: + if row and output: row.result = row.scheduler_run.run_result and \ loads(row.scheduler_run.run_result, object_hook=_decode_dict) or None diff --git a/gluon/tests/test_dal.py b/gluon/tests/test_dal.py index 5dd99b85..696a5e4e 100644 --- a/gluon/tests/test_dal.py +++ b/gluon/tests/test_dal.py @@ -285,6 +285,22 @@ class TestBelongs(unittest.TestCase): db.t.drop() +class TestContains(unittest.TestCase): + def testRun(self): + db = DAL('sqlite:memory:') + db.define_table('t', Field('a', 'list:string')) + self.assertEqual(db.t.insert(a=['aaa','bbb']), 1) + self.assertEqual(db.t.insert(a=['bbb','ddd']), 2) + self.assertEqual(db.t.insert(a=['eee','aaa']), 3) + self.assertEqual(len(db(db.t.a.contains('aaa')).select()), + 2) + self.assertEqual(len(db(db.t.a.contains('bbb')).select()), + 2) + self.assertEqual(len(db(db.t.a.contains('aa')).select()), + 0) + db.t.drop() + + class TestLike(unittest.TestCase): def testRun(self): @@ -304,6 +320,11 @@ class TestLike(unittest.TestCase): 1) self.assertEqual(len(db(db.t.a.upper().like('%C')).select()), 1) db.t.drop() + db.define_table('t', Field('a', 'integer')) + self.assertEqual(db.t.insert(a=1111111111), 1) + self.assertEqual(len(db(db.t.a.like('1%')).select()), 1) + self.assertEqual(len(db(db.t.a.like('2%')).select()), 0) + db.t.drop() class TestDatetime(unittest.TestCase): @@ -690,4 +711,3 @@ class TestDALDictImportExport(unittest.TestCase): if __name__ == '__main__': unittest.main() tearDownModule() - diff --git a/gluon/validators.py b/gluon/validators.py index fc0600f1..f5942167 100644 --- a/gluon/validators.py +++ b/gluon/validators.py @@ -339,7 +339,7 @@ class IS_JSON(Validator): if value is None: return None return simplejson.dumps(value) - + class IS_IN_SET(Validator): """ @@ -626,7 +626,7 @@ class IS_NOT_IN_DB(Validator): (tablename, fieldname) = str(self.field).split('.') table = self.dbset.db[tablename] field = table[fieldname] - rows = self.dbset(field == value, ignore_common_filters=self.ignore_common_filters).select(limitby=(0, 1)) + rows = self.dbset(field == value, ignore_common_filters=self.ignore_common_filters).select(field, limitby=(0, 1)) if len(rows) > 0: if isinstance(self.record_id, dict): for f in self.record_id: