sfa/senslab/slabpostgres.py

   1 import sys
   2
   3 from sqlalchemy import create_engine, and_
   4 from sqlalchemy.orm import sessionmaker
   5
   6 from sfa.util.config import Config
   7 from sfa.util.sfalogging import logger
   8
   9 from sqlalchemy import Column, Integer, String, DateTime
  10 from sqlalchemy import Table, Column, MetaData, join, ForeignKey
  11 import sfa.storage.model as model
  12
  13 from sqlalchemy.ext.declarative import declarative_base
  14 from sqlalchemy.orm import relationship, backref
  15
  16
  17 from sqlalchemy import MetaData, Table
  18 from sqlalchemy.exc import NoSuchTableError
  19
  20 #Dict holding the columns names of the table as keys
  21 #and their type, used for creation of the table
  22 slice_table = {'record_id_user':'integer PRIMARY KEY references X ON DELETE CASCADE ON UPDATE CASCADE','oar_job_id':'integer DEFAULT -1',  'record_id_slice':'integer', 'slice_hrn':'text NOT NULL'}
  23
  24 #Dict with all the specific senslab tables
  25 tablenames_dict = {'slice_senslab': slice_table}
  26
  27 ##############################
  28
  29
  30
  31 SlabBase = declarative_base()
  32
  33
  34
  35
  36 class SliceSenslab (SlabBase):
  37     __tablename__ = 'slice_senslab'
  38     record_id_user = Column(Integer, primary_key=True)
  39     oar_job_id = Column( Integer,default = -1)
  40     record_id_slice = Column(Integer)
  41     slice_hrn = Column(String,nullable = False)
  42
  43     def __init__ (self, slice_hrn =None, oar_job_id=None, record_id_slice=None, record_id_user= None):
  44         if record_id_slice:
  45             self.record_id_slice = record_id_slice
  46         if slice_hrn:
  47             self.slice_hrn = slice_hrn
  48         if oar_job_id:
  49             self.oar_job_id = oar_job_id
  50         if slice_hrn:
  51             self.slice_hrn = slice_hrn
  52         if record_id_user:
  53             self.record_id_user= record_id_user
  54
  55     def __repr__(self):
  56         result="<Record id user =%s, slice hrn=%s, oar_job id=%s,Record id slice =%s" % \
  57                 (self.record_id_user, self.slice_hrn, self.oar_job_id, self.record_id_slice)
  58         result += ">"
  59         return result
  60
  61     def dumpquerytodict(self):
  62         dict = {'slice_hrn':self.slice_hrn,
  63         'record_id':self.record_id_slice,
  64         'record_id_user':self.record_id_user,
  65         'oar_job_id':self.oar_job_id,
  66          'record_id_slice':self.record_id_slice,
  67          'slice_hrn':self.slice_hrn}
  68         return dict
  69 #class PeerSenslab(SlabBase):
  70     #__tablename__ = 'peer_senslab'
  71     #peername = Column(String, nullable = False)
  72     #peerid = Column( Integer,primary_key=True)
  73
  74     #def __init__ (self,peername = None ):
  75         #if peername:
  76             #self.peername = peername
  77
  78
  79       #def __repr__(self):
  80         #result="<Peer id  =%s, Peer name =%s" % (self.peerid, self.peername)
  81         #result += ">"
  82         #return result
  83
  84 class SlabDB:
  85     def __init__(self,config):
  86         self.sl_base = SlabBase
  87
  88         dbname="slab_sfa"
  89         # will be created lazily on-demand
  90         self.slab_session = None
  91         # the former PostgreSQL.py used the psycopg2 directly and was doing
  92         #self.connection.set_client_encoding("UNICODE")
  93         # it's unclear how to achieve this in sqlalchemy, nor if it's needed at all
  94         # http://www.sqlalchemy.org/docs/dialects/postgresql.html#unicode
  95         # we indeed have /var/lib/pgsql/data/postgresql.conf where
  96         # this setting is unset, it might be an angle to tweak that if need be
  97         # try a unix socket first - omitting the hostname does the trick
  98         unix_url = "postgresql+psycopg2://%s:%s@:%s/%s"%\
  99             (config.SFA_DB_USER,config.SFA_DB_PASSWORD,config.SFA_DB_PORT,dbname)
 100         print >>sys.stderr, " \r\n \r\n SLAPOSTGRES INIT unix_url %s" %(unix_url)
 101         # the TCP fallback method
 102         tcp_url = "postgresql+psycopg2://%s:%s@%s:%s/%s"%\
 103             (config.SFA_DB_USER,config.SFA_DB_PASSWORD,config.SFA_DB_HOST,config.SFA_DB_PORT,dbname)
 104         for url in [ unix_url, tcp_url ] :
 105             try:
 106                 self.slab_engine = create_engine (url,echo_pool=True,echo=True)
 107                 self.check()
 108                 self.url=url
 109                 return
 110             except:
 111                 pass
 112         self.slab_engine=None
 113         raise Exception,"Could not connect to database"
 114
 115     def check (self):
 116         self.slab_engine.execute ("select 1").scalar()
 117
 118
 119     def session (self):
 120         if self.slab_session is None:
 121             Session=sessionmaker ()
 122             self.slab_session=Session(bind=self.slab_engine)
 123         return self.slab_session
 124
 125
 126
 127
 128     #Close connection to database
 129     def close(self):
 130         if self.connection is not None:
 131             self.connection.close()
 132             self.connection = None
 133
 134
 135
 136
 137     def exists(self, tablename):
 138         """
 139         Checks if the table specified as tablename exists.
 140
 141         """
 142
 143         try:
 144             metadata = MetaData (bind=self.slab_engine)
 145             table=Table (tablename, metadata, autoload=True)
 146
 147             return True
 148         except NoSuchTableError:
 149             print>>sys.stderr, " \r\n \r\n \t SLABPOSTGRES EXISTS NOPE! tablename %s " %(tablename)
 150             return False
 151
 152
 153     def createtable(self, tablename ):
 154         """
 155         Creates the specifed table. Uses the global dictionnary holding the tablenames and
 156         the table schema.
 157
 158         """
 159
 160         print>>sys.stderr, " \r\n \r\n \t SLABPOSTGRES createtable SlabBase.metadata.sorted_tables %s \r\n engine %s" %(SlabBase.metadata.sorted_tables , slab_engine)
 161         SlabBase.metadata.create_all(slab_engine)
 162         return
 163
 164
 165     def update_job(self, job_id, hrn):
 166         slice_rec = slab_dbsession.query(SliceSenslab).filter_by(slice_hrn = hrn).first()
 167         print>>sys.stderr, " \r\n \r\n \t SLABPOSTGRES  update_job slice_rec %s"%(slice_rec)
 168         slice_rec.oar_job_id = job_id
 169         slab_dbsession.commit()
 170
 171     def find (self, name = None, filter_dict = None):
 172         print>>sys.stderr, " \r\n \r\n \t SLABPOSTGRES find  filter_dict %s"%(filter_dict)
 173
 174         #Filter_by can not handle more than one argument, hence these functions
 175         def filter_id_user(query, user_id):
 176             print>>sys.stderr, " \r\n \r\n \t SLABPOSTGRES find  filter_id_user"
 177             return query.filter_by(record_id_user = user_id)
 178
 179         def filter_job(query, job):
 180             print>>sys.stderr, " \r\n \r\n \t SLABPOSTGRES find filter_job "
 181             return query.filter_by(oar_job_id = job)
 182
 183         def filer_id_slice (query, id_slice):
 184             print>>sys.stderr, " \r\n \r\n \t SLABPOSTGRES find  filer_id_slice"
 185             return query.filter_by(record_id_slice = id_slice)
 186
 187         def filter_slice_hrn(query, hrn):
 188             print>>sys.stderr, " \r\n \r\n \t SLABPOSTGRES find  filter_slice_hrn"
 189             return query.filter_by(slice_hrn = hrn)
 190
 191
 192         extended_filter = {'record_id_user': filter_id_user,
 193          'oar_job_id':filter_job,
 194          'record_id_slice': filer_id_slice,
 195          'slice_hrn': filter_slice_hrn}
 196
 197         Q = slab_dbsession.query(SliceSenslab)
 198
 199         if filter_dict is not None:
 200             for k in filter_dict:
 201                 try:
 202                   newQ= extended_filter[k](Q, filter_dict[k])
 203                   Q = newQ
 204                 except KeyError:
 205                     print>>sys.stderr, "\r\n \t\t FFFFFFFFFFFFFFFFUUUUUUUUFUFUFU!!!!!!!!"
 206         print>>sys.stderr, " HEEEEEEEEEEEEY %s " %(Q.first())
 207         rec = Q.first()
 208         print>>sys.stderr, " \r\n \r\n \t SLABPOSTGRES find  rec %s" %(rec)
 209         return dict(zip(['record_id_user','oar_job_id', 'record_id_slice','slice_hrn'],[rec.record_id_user,rec.oar_job_id,rec.record_id_slice, rec.slice_hrn]))
 210         #reclist = []
 211         ##for rec in Q.all():
 212             #reclist.append(dict(zip(['record_id_user','oar_job_id', 'record_id_slice','slice_hrn'],[rec.record_id_user,rec.oar_job_id,rec.record_id_slice, rec.slice_hrn])))
 213         #print>>sys.stderr, " \r\n \r\n \t SLABPOSTGRES find  reclist %s" %(reclist)
 214         #return reclist
 215
 216
 217
 218
 219 from sfa.util.config import Config
 220
 221 slab_alchemy= SlabDB(Config())
 222 slab_engine=slab_alchemy.slab_engine
 223 slab_dbsession=slab_alchemy.session()