Ignore:
Timestamp:
06/06/12 11:39:29 (8 years ago)
Author:
mnagni
Message:

Other improvement on how SA alchemy session is managed (infoDB)

File:
1 edited

Legend:

Unmodified
Added
Removed
  • mauRepo/MolesManager/trunk/src/libs/migration/InfodbEPB.py

    r8323 r8326  
    55''' 
    66from libs.epb import EPB 
    7 from libs.migration.exception.exceptions import NoDBManager 
    8 from sqlalchemy.sql.expression import text, bindparam 
    9 from sqlalchemy.types import String 
     7from sqlalchemy.sql.expression import text 
    108 
     9class InfodbEPBFactory(EPB): 
     10     
     11    def __init__(self, dbManager): 
     12        self._dbManager = dbManager 
     13 
     14    def _getSession(self): 
     15        if self._dbManager is not None: 
     16            return self._dbManager.createDbSession()                 
     17        return None 
     18         
     19    def createMigrationEPB(self): 
     20        return InfodbEPB(self._getSession()) 
    1121 
    1222class InfodbEPB(EPB): 
    1323     
    14     _infoDB = None 
    15      
    16     @classmethod 
    17     def overrrideDBManager(self, dbManager): 
    18         """ 
    19             Sets the infodbEPB libs.commons_db.DbManager 
    20         """ 
    21         InfodbEPB._infoDB = dbManager 
     24    def __init__(self, session): 
     25        self._session = session             
    2226 
    23     @classmethod 
    24     def getCedaInfoApp_dataentityByDE_ID(self, de_id, session = None): 
     27    def close(self): 
     28        return self._session.close() 
     29 
     30    def getCedaInfoApp_dataentityByDE_ID(self, de_id): 
    2531        """ 
    2632            Returns a dictionary representing the cedainfoapp_dataentity table in the cedainfo DB. 
     
    3844                    ON cedainfoapp_dataentity.responsible_officer_id=cedainfoapp_person.id \ 
    3945            WHERE cedainfoapp_dataentity.dataentity_id=:i_id") 
    40         records = InfodbEPB._infoDB.engine.execute(query_string, i_id=de_id) 
     46        records = self._session.execute(query_string, params={'i_id':de_id}) 
    4147        ret = records.fetchone() 
    4248        records.close() 
    4349        return ret 
    44  
    45  
    46     @classmethod 
    47     def getNewInfoConnection(self): 
    48         return InfodbEPB._getConnection() 
    49  
    50     @classmethod 
    51     def _getConnection(self, session = None): 
    52         if InfodbEPB._infoDB is None: 
    53             raise NoDBManager("InfodbEPB") 
    54         return EPB._getSession(InfodbEPB._infoDB, session) 
Note: See TracChangeset for help on using the changeset viewer.