target.py

Go to the documentation of this file.

#!/usr/bin/env python
import os, logging
from datetime import datetime, timedelta
log = logging.getLogger(__name__)
dt_ = lambda _:datetime.strptime(_,"%Y-%m-%d %H:%M:%S")

class Target(dict):
    """
    Encapsulate DybDbi dealings here to avoid cluttering ``Scraper``

    Relevant config parameters 

    :param timefloor: None or a datetime or string such as '2010-09-18 22:57:32' used to limit the expense of validity query

    """
    def __init__(self, *args, **kwa ):
        dict.__init__(self, *args, **kwa )
        from DybPython import DBConf
        self.dbconf = DBConf( os.environ['DBCONF'] )
        self.kls = self._kls()  

    database = property(lambda self:self.dbconf.get('database'))

    # sensible defaults for easy test/mock usage
    dbi_loglevel = property(lambda self:self.get('dbi_loglevel',"INFO"))
    seed_target_tables = property(lambda self:self.get('seed_target_tables',False))
    DROP_TARGET_TABLES = property(lambda self:self.get('DROP_TARGET_TABLES',False))
    ALLOW_DROP_CREATE_TABLE = property(lambda self:self.get('ALLOW_DROP_CREATE_TABLE',False))

    def __repr__(self):
        return "Target %s:%s:%s " % ( self['target'], self.database, self['kls'] )

    def _kls(self):
        """
        :rtype DybDbi class:  returns DybDbi/genDbi class specified by ``kls`` key within this dict
        
        Also does DybDbi housekeeping:

        #. sets loglevel according to ``dbi_loglevel`` key
        #. creates target tables if both ``seed_target_tables`` key is True and ``ALLOW_DROP_CREATE_TABLE`` is True
        #. sets convenience kls attribute `kls.attnames` providing the list of attribute names : for dict consistency checking 

        .. warning:: As this is called by ctor, must set options impacting it on dict instanciation

        """
        from DybDbi import gDbi, Level          
        import DybDbi

        kls = getattr( DybDbi, self['kls'] )
        assert len(gDbi.cascader) == 1
        dbname = gDbi.cascader[0].dbname
        kls.attnames = kls.SpecKeys().aslist()

        gDbi.outputlevel = Level.From( self.dbi_loglevel )

        allow_dropcreate = self.ALLOW_DROP_CREATE_TABLE
        stt = self.seed_target_tables
        dtt = self.DROP_TARGET_TABLES
        sve = 'SUPERVISOR_ENABLED' in os.environ
        log.warn("target._kls stt %s dtt %s sve %s " % (stt,dtt,sve) )

        if sve and (stt or dtt):
            msg = "seeding/dropping target tables is not appropriate for longterm usage under supervisord : do manually with scr.py  "
            log.fatal(msg)
            raise Exception(msg)  
     
        if stt: 
            log.warn("seed_target_tables proceeding against : %r " % dict( self.dbconf, password="***") )
        elif dtt and allow_dropcreate: 
            if dbname == 'offline_db' or self.database == 'offline_db':
                log.warn("seed_target_tables proceeding against offline_db for target %r " % dict( self.dbconf, password="***") )
            kls().CreateDatabaseTables( 0, kls.__name__[1:] )     
        else:
            log.warn("seed_target_tables/DROP_TARGET_TABLES is not allowed with target %r without ALLOW_DROP_CREATE_TABLE option " % dict( self.dbconf, password="***") )
        return kls


    def instance( self, **kwa ):
        """
        Might fail with TypeError if kwa cannot be coerced, eg from aggregate queries 
        returning None when zero samples

        If the attribute names are not expected for the target kls they are skipped.
        This will be the case for the system attributes `_date_time_min` `_date_time_max` 

        """
        d = dict((k,v) for k,v in kwa.items() if k in self.kls.attnames)
        if len(kwa) != len(d):
            log.debug("target.instance skipped some attributes %r in instance preparation " % list(set(kwa).difference(set(d))) )  
        return self.kls.Create(**d)

    def writer(self, sv , localstart=None, localend=None ):
        """
        Prepare DybDbi writer for target class, with contextrange/subsite appropriate for the source instance

        Use of non-default localstart and localend type is typically only used for
        aggregate_group_by quering where the instance datetimes such as `sv[0].date_time`
        do not correspond to the contextrange of the aggregate dict. 

        :param sv:  source vector instance that contains instances of an SA mapped class
        :param localstart: default of `None` corresponds to `sv[0].date_time`
        :param localend: default of `None` corresponds to `sv[-1].date_time`
        """

        start = localstart if localstart else sv[0].date_time
        end   = localend if localend else sv[-1].date_time
        return Target._wrt( sv[0].__class__ , self.kls , start, end ) 

    def _wrt( cls , skls, tkls , localstart, localend ):
        """
        Classmethod to prepare DybDbi writer for the target class with 
        with contextrange/subsite appropriate to the source class

        :param skls: SQLAlchemy source class which determines the appropriate contextrange to use for target class writer
        :param tkls: DybDbi target class for which the writer is to be prepared 
        :param localstart: local datetime which is converted to UTC TimeStamp for timestart of CR
        :param localend: local datetime which is converted to UTC TimeStamp for timeend of CR

        Assumptions made:

        #. source tablename encodes site, subsite (IF NOT THEN MUST GET FROM INSTANCE ?) 
        #. source date_time are naive localtimes 

        """
        wrt = tkls.Wrt()
        xtn = skls.xtn   

        from DybDbi import ContextRange, TimeStamp, Site, SimFlag, DetectorId
        log.debug( "_wrt localstart %s localend %s " % ( localstart.ctime(), localend.ctime() )) 
        timestart = TimeStamp.UTCfromLocalDatetime( localstart ) 
        timeend   = TimeStamp.UTCfromLocalDatetime( localend ) 
        cr = ContextRange( xtn.sitemask,  SimFlag.kData , timestart, timeend )
        wrt.ctx( contextrange=cr, dbno=0, versiondate=TimeStamp(0,0), subsite=xtn.subsite, task=0 ) 
        return wrt
    _wrt = classmethod( _wrt )


    def require_manual(self, msg ):
        """
        Require manual operation (ie running scr.py from commandline) 
        preventing usage of rare operations/options under supervisor control 
        """
        sve = 'SUPERVISOR_ENABLED' in os.environ
        if sve:
            log.fatal(msg)
            raise Exception(msg)  
        else:
            log.info("manual running detected")            


    def seed(self, srcs, scraper , dummy=False ):
        """
        This is invoked at scraper instanciation when the conditions are met:

        #. ``seed_target_tables`` is configured True
  
        Seed entries are written to the target table. The seed validity range is configured with
        the options: `seed_timestart` `seed_timeend` and formerly the payload entry 
        was specified by the `def seed()` method implemented in the scraper class.

        Attempts to perform seeding under supervisor raises an exception, 
        to enforce this restriction.

        When testing seeding start from scratch with eg::

             mysql> drop table DcsAdTemp, DcsAdTempVld ;    
             mysql> update LOCALSEQNO set LASTUSEDSEQNO=0 where TABLENAME='DcsAdTemp' ;  

       Changes from Oct 2012, 

        #. allow use against an existing table 
        #. remove table creation functionality is removed
        #. move to payloadless seeds (removing need for dummy payload instances) 

        Motivated by the need to add new sources that contribute 
        to an existing target which has already collected data
        eg for adding ADs to the DcsAdWpHv scraper.

        """
        self.require_manual("seed is not appropriate for longterm usage under supervisord : do seeding manually ")
        if not self['seed_target_tables']:
            log.warn("improper call to seed " )
            return

        localstart, localend = self['seed_timestart'], self['seed_timeend'] 
        kls = self.kls
        for src in srcs:
            tlv = self.lastvld(src.xtn)
            if tlv: 
                log.info("tlv exists already : src %r tlv %r " % ( src,tlv.seqno))
            else:
                log.warn("writing seed for src : %r " % (src) )
                wrt = self._wrt( src, kls , localstart, localend  )
                if dummy: 
                    inst = kls.Create( **scraper.seed(src) )
                    wrt.Write( inst )
                seqno = wrt.Close()
                assert seqno, "failed to seed_update for source %r " % src 
                log.warn("wrote seed for src : %r with seqno %s  " % (src,seqno) )


    def lastvld( self, source ):
         """
         Last validity record in target database for context corresponding to `source` class.
         Query expense is restricted by the `timefloor`.  
         If `timefloor` is None a sequence of progressively 
         more expensive queries are performed to get the target last validty.

         :param source:  source context instance either an **xtn** of MockSource instance with subsite and sitemask attributes
         :param timefloor: time after which to look for validity entries in target database or None

         Note this is called only at scraper initialization, in order for the scraper to 
         find its time cursor.

         """
         timefloor = self.get('timefloor', None)
         if timefloor:
             log.debug("using timefloor %r " % timefloor )
             return self._lastvld( source, timefloor )
         else:
             log.warn("no timefloor ... inventing ")
             now = datetime.now()
             for days in (14,100,):
                 ts = now + timedelta(days=-days)  
                 tlv = self._lastvld( source , ts )
                 if tlv:
                     return tlv
             return self._lastvld(source, None ) 

    def _lastvld(self, source, timefloor ):
        """
        Obtain last target validity record for context corresponding to source class

        :param source: SA mapped class

        """ 
        from DybDbi import SimFlag
        subsite = source.subsite
        task = self.get('task', 0 )
        sqlcontext = "SiteMask & %s and SimMask & %s " % ( source.sitemask , SimFlag.kData )
        if timefloor:
            sqlcontext += " and TIMESTART > '%s' " % timefloor
        lastvld = self.kls.GetTableProxy().LastValidityRec( sqlcontext, subsite, task )
        return lastvld  




if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)

    target = Target( kls='GDcsAdWpHv', target="tmp_offline_db", timefloor  = dt_("2010-01-01 00:00:00"), seed_target_tables=True )
    log.debug( "kls %r with attnames %s " % (target.kls, target.kls.attnames) )

    class MockSource(object):
        def __init__(self, sitemask_subsite):
             sitemask, subsite = sitemask_subsite.split()
             self.sitemask = int(sitemask)
             self.subsite  = int(subsite)
        __repr__ = lambda self:"MockSource (%s,%s) " % (self.sitemask, self.subsite)

    ## adhoc query to obtain all sources feeding the table already in low level way 
    from DybPython import DB
    db = DB(target.dbconf.sect)
    sql = "select distinct(concat(sitemask,\" \",subsite)) as ss from %sVld" % target.kls.__name__[1:]
    sss = map(lambda _:_['ss'],db(sql))

    srcs = map(MockSource,sss + ["4 4","2 2"])   ## extend sources for new ADs
    for s in srcs:   
        tlv = target.lastvld( s )
        timeend = tlv.contextrange.timeend if tlv else "-"           ## UTC TimeStamp
        log.info( "%r timeend %s  " % (s, timeend) )

        #limeend = timeend.UTCtoNaiveLocalDatetime     
        #log.info( "timeend %s  (localtime)" % limeend )

    target['seed_timestart'] = dt_("2010-01-01 00:00:00")
    target['seed_timeend']   = dt_("2011-01-01 00:00:00")
    target.seed( srcs, None , dummy=False )