def _unlock_many():
     tickets = []
     with self.db_engine.begin() as conn:
         cur = conn.execute(_sql('select * from slc_rad_online where nas_addr = :nas_addr'),nas_addr=nas_addr) 
         for online in cur:  
             conn.execute(bsql,**_ticket(online))
             conn.execute(_sql('delete from slc_rad_online where nas_addr = :nas_addr'),nas_addr=nas_addr)

 def set(self, key, value, expire=0):
     raw_data = self.encode_data(value)
     with self.dbengine.begin() as conn:
         _time = expire>0 and (int(time.time()) + int(expire)) or 0
         try:
             conn.execute(_sql("insert into %s values (:key, :value, :time) " % self.cache_table),
                 key=key,value=raw_data,time=_time)
         except:
             conn.execute(_sql("delete from %s where _key = :key " % self.cache_table),key=key)
             conn.execute(_sql("insert into %s values (:key, :value, :time) " % self.cache_table),
                 key=key,value=raw_data,time=_time)

 def _raw_set(self, key, value, timeout,**kwargs):
     with self.dbengine.begin() as conn:
         _time = int(time.time()) + int(timeout)
         try:
             conn.execute(_sql("insert into system_session values (:key, :value, :time) "),
                 key=key,value=value,time=_time)
         except:
             conn.execute(_sql("""update system_session 
                                 set _value=:value, _time=:time
                                 where _key=:key"""),
                                 key=key,value=value,time=_time)

 def _unlock_one():
     ticket = None
     with self.db_engine.begin() as conn:
         sql = _sql('select * from slc_rad_online where  nas_addr = :nas_addr and acct_session_id = :session_id')
         cur = conn.execute(sql,nas_addr=nas_addr,session_id=acct_session_id)
         online = cur.fetchone()
         if online:
             ticket = _ticket(online) 
             dsql = _sql('delete from slc_rad_online where nas_addr = :nas_addr and acct_session_id = :session_id')
             conn.execute(dsql,nas_addr=nas_addr,session_id=acct_session_id)
             conn.execute(bsql,**ticket)

 def _delete(self, key):
     with self.dbengine.begin() as conn:
         try:
             conn.execute(_sql("delete from system_session where _key = :key "),key=key)
         except:
             import traceback
             traceback.print_exc()

    def get(self, key):
        self.get_total += 1
        raw_data = None
        _del_func = self.delete
        with self.dbengine.begin() as conn:
            try:
                cur = conn.execute(_sql("select _value, _time from %s where _key = :key " % self.cache_table), key=key)
                _cache = cur.fetchone()
                if _cache:
                    self.hit_total += 1
                    _time = int(_cache["_time"])
                    if _time > 0 and time.time() > _time:
                        reactor.callLater(0.01, _del_func, key)
                    else:
                        raw_data = _cache["_value"]
            except:
                import traceback

                traceback.print_exc()

        try:
            if raw_data:
                return self.decode_data(raw_data)
        except:
            self.delete(key)
        return None

 def delete(self,key):
     with self.dbengine.begin() as conn:
         try:
             conn.execute(_sql("delete from %s where _key = :key " % self.cache_table),key=key)
         except:
             import traceback
             traceback.print_exc()

def create_logtable(db_engine):
    create_sql_tpl = """
    CREATE TABLE {0} (
    	id INTEGER NOT NULL PRIMARY KEY autoincrement,
    	host VARCHAR(32) NOT NULL,
    	time VARCHAR(19) NOT NULL,
    	facility VARCHAR(16) NOT NULL,
    	priority VARCHAR(16) NOT NULL,
    	username VARCHAR(16) NULL,
    	message VARCHAR(1024) NOT NULL
    );
    """
    mysql_create_sql_tpl = """
    CREATE TABLE {0} (
        id INT(11) NOT NULL PRIMARY KEY  AUTO_INCREMENT ,
        host VARCHAR(32) NOT NULL,
        time VARCHAR(19) NOT NULL,
        facility VARCHAR(16) NOT NULL,
        priority VARCHAR(16) NOT NULL,
        username VARCHAR(16) NULL,
        message VARCHAR(1024) NOT NULL
    )
    COMMENT='syslog table'
    COLLATE='utf8_general_ci'
    ENGINE=InnoDB
    AUTO_INCREMENT=1;
    """
    sql_tpl = 'mysql' in db_engine.driver and mysql_create_sql_tpl or create_sql_tpl
    table_name = "log_{0}".format(datetime.datetime.now().strftime("%Y%m%d%H"))
    sqlstr = sql_tpl.format(table_name)
    with db_engine.begin() as conn:
        try:
            conn.execute(_sql(sqlstr))
        except:
            logging.exception("create logtable error")

 def _raw_get(self, key, **kwargs):
     raw_data = None
     with self.dbengine.begin() as conn:
         try:
             cur = conn.execute(_sql("select _value, _time from system_session where _key = :key "),key=key)
             session =  cur.fetchone()
             if session:
                 _time = int(session['_time'])
                 if (time.time() - _time) > self.session_timeout:
                     conn.execute(_sql("delete from system_session where _key = :key "),key=key)
                 else:
                     raw_data = session['_value']
         except:
             import traceback
             traceback.print_exc()
     return raw_data

 def count(self):
     with self.dbengine.begin() as conn:
         try:
             cur = conn.execute(_sql("select count(_key) as count from %s " % self.cache_table))
             return int(cur.fetchone()["count"])
         except:
             self.log.error("cache count error")
             return 0

 def get_user_attr(self, username,attr_name):
     with self.db_engine.begin() as conn:
         cur = conn.execute(_sql("""select attr_value from slc_rad_account_attr
                                 where account_number = :account
                                 and attr_name = :attr_name"""),
                            account=username,attr_name=attr_name)
         b = cur.fetchone()
         return b and b['attr_value'] or None

 def check_expire(self, first_delay=0):
     if first_delay > 0:
         reactor.callLater(first_delay, self.check_expire)
     with self.dbengine.begin() as conn:
         try:
             conn.execute(_sql("delete from %s where _time > 0 and _time < :time" % self.cache_table),time=int(time.time()))
         except:
             pass
     reactor.callLater(120.0, self.check_expire)

 def getRadius(self, host):
     cache_key = 'RadiusLoader.radius.%s' % (host)
     cache = self.cache.get(cache_key)
     if cache:
         return cache
     with self.dbengine.begin() as conn:
         cur = conn.execute(_sql("""select * from tt_radius where ip_addr = :ip_addr """),ip_addr=host)
         radius = self.warp(cur.fetchone())
         self.cache.set(cache_key,radius, expire=60)
         return radius

def execute_sqls(config, sqlstr):
    sh.info("exec sql >> %s" % sqlstr)
    results = []
    with get_engine(config).begin() as conn:
        try:
            results = conn.execute(_sql(sqlstr))
        except Exception as err:
            return sh.err("exec sql error: %s" % str(err))
    sh.info("exec sql done")
    print_result(results)

 def getMasterRadius(self):
     cache_key = 'RadiusLoader.master_radius'
     cache = self.cache.get(cache_key)
     if cache:
         return cache
     with self.dbengine.begin() as conn:
         cur = conn.execute(_sql("select * from trw_radius where serv_type = 1"))
         radius = self.warp(cur.fetchone())
         self.cache.set(cache_key,radius, expire=600)
         return radius

 def add_ticket(self,ticket):
     _ticket = ticket.copy()
     for _key in _ticket:
         if _key not in ticket_fds:
             del ticket[_key]
     with self.db_engine.begin() as conn:
         keys = ','.join(ticket.keys())
         vals = ",".join(["'%s'"% c for c in ticket.values()])
         sql = _sql('insert into slc_rad_ticket (%s) values(%s)'%(keys,vals))
         conn.execute(sql)

 def _raw_replace(self, key, value, timeout,**kwargs):
     with self.dbengine.begin() as conn:
         _time = int(time.time()) + int(timeout)
         try:
             conn.execute(_sql("""update system_session 
                                 set _value=:value, _time=:time
                                 where _key=:key"""),
                                 key=key,value=value,time=_time)
         except:
             import traceback
             traceback.print_exc()

    def process(self):
        table_name = "log_{0}".format((datetime.datetime.now() + datetime.timedelta(hours=1)).strftime("%Y%m%d%H"))
        sqlstr = self.sql_tpl.format(table_name)
        with self.dbengine.begin() as conn:
            try:
                conn.execute(_sql(sqlstr))
                log.msg("create table {0} success;".format(table_name))
            except Exception as err:
                log.msg('create table error {0}'.format(err.message))

        reactor.callLater(60 * 20, self.process, )

 def update(self, key, value, expire=0):
     raw_data = self.encode_data(value)
     with self.dbengine.begin() as conn:
         _time = expire>0 and (int(time.time()) + int(expire)) or 0
         try:
             conn.execute(_sql("""update %s 
                                 set _value=:value, _time=:time
                                 where _key=:key""" % self.cache_table),
                                 key=key,value=raw_data,time=_time)
         except:
             import traceback
             traceback.print_exc()

 def update_billing(self,billing,time_length=0,flow_length=0):
     with self.db_engine.begin() as conn:
         # update account
         balan_sql = _sql("""update slc_rad_account set 
             balance = :balance,
             time_length=:time_length,
             flow_length=:flow_length
             where account_number = :account
         """)
         conn.execute(balan_sql,
             balance=billing.balance,
             time_length=time_length,
             flow_length=flow_length,
             account=billing.account_number
         )
         
         # update online
         online_sql = _sql("""update slc_rad_online set 
             billing_times = :billing_times,
             input_total = :input_total,
             output_total = :output_total
             where nas_addr = :nas_addr and acct_session_id = :session_id
         """)
         conn.execute(online_sql,
             billing_times=billing.acct_session_time,
             input_total=billing.input_total,
             output_total=billing.output_total,
             nas_addr=billing.nas_addr,
             session_id=billing.acct_session_id
         )
         
         # update billing
         keys = ','.join(billing.keys())
         vals = ",".join(["'%s'"% c for c in billing.values()])
         billing_sql = _sql('insert into slc_rad_billing (%s) values(%s)'%(keys,vals))
         conn.execute(billing_sql)
         
         
     self.update_user_cache(billing.account_number)