Public Member Functions | |
| def | __init__ |
| def | __repr__ |
| def | log |
| def | errlog |
| def | drop |
| def | create |
| def | spawn |
| def | tables |
| def | process |
| def | populate |
| def | fixture |
| def | mysqldump |
| def | dump |
| def | fixture_ |
Public Attributes | |
| url_ | |
| origname | |
| user | |
| pswd | |
| fix | |
| verbosity | |
| name | |
Static Public Attributes | |
| string | prefix = 'tmp_' |
| tuple | url = property( lambda self:os.path.expanduser(os.path.expandvars(self.url_.GetUrl())), lambda self,n:self.url_.SetUrl(n) ) |
| tuple | name = property( lambda self:self.url_.GetFile(), lambda self,n:self.url_.SetFile(n) ) |
| tuple | host = property( lambda self:self.url_.GetHost(), lambda self,n:self.url_.SetHost(n) ) |
| tuple | ascii = property( lambda self:self.url_.GetAnchor(), lambda self,n:self.url_.SetAnchor(n) ) |
| tuple | opts = property( lambda self:self.url_.GetOptions(), lambda self,n:self.url_.SetOptions(n) ) |
| tuple | creds = property( lambda self:dict( name=self.name, host=self.host, user=self.user, pswd=self.pswd )) |
| tuple | server = property( _get_server, doc=_get_server.__doc__ ) |
| tuple | isconnected = property(lambda self:self.server.IsConnected()) |
Private Member Functions | |
| def | _get_server |
| def | _spawn |
Private Attributes | |
| _server | |
| _attempt | |
Static Private Attributes | |
| string | _mysqldump = "mysqldump --no-defaults --host=%(host)s --user=%(user)s --password=%(pswd)s %(opt)s %(name)s " |
Dictionary holding parameters to connect to a DB and provides functionality to drop/create databases and run updates/queries against them.
Definition at line 23 of file dbcas.py.
| def DybPython::dbcas::DBCon::__init__ | ( | self, | ||
| url, | ||||
| user, | ||||
| pswd, | ||||
| kwa | ||||
| ) |
Definition at line 29 of file dbcas.py.
00033 : 00034 self.url_ = TUrl(url) 00035 assert self.url_.IsValid() 00036 self.origname = self.name 00037 self.user = user 00038 self.pswd = pswd 00039 self.fix = kwa.get('fix', None) 00040 self.verbosity = kwa.get('verbosity', 0) 00041 name = kwa.get('name', None) 00042 if name: self.name = name ## using the setter property
| def DybPython::dbcas::DBCon::__repr__ | ( | self | ) |
Definition at line 50 of file dbcas.py.
00050 :self.url_.GetAnchor(), lambda self,n:self.url_.SetAnchor(n) ) 00051 opts = property( lambda self:self.url_.GetOptions(), lambda self,n:self.url_.SetOptions(n) ) 00052 creds = property( lambda self:dict( name=self.name, host=self.host, user=self.user, pswd=self.pswd ))
| def DybPython::dbcas::DBCon::log | ( | self | ) |
| def DybPython::dbcas::DBCon::errlog | ( | self, | ||
| dbn | ||||
| ) |
| def DybPython::dbcas::DBCon::_get_server | ( | self | ) | [private] |
If the connection attempt fails, try again without specifying the DB name, see :root:`TMySQLServer`
.. todo::
Find way to avoid/capture the error after failure to connect
Definition at line 62 of file dbcas.py.
00062 : 00063 self[dbn] = [] 00064 self[dbn].append( self.server.GetErrorMsg() ) 00065 00066 def _get_server(self): 00067 """ 00068 If the connection attempt fails, try again without specifying the DB name, see :root:`TMySQLServer` 00069 00070 .. todo:: 00071 00072 Find way to avoid/capture the error after failure to connect 00073 00074 """ 00075 if not(self._server): 00076 self._server = TSQLServer.Connect( self.url , self.user , self.pswd ) 00077 self._attempt += 1 00078 if self._server == None: 00079 if self.name != "" and self._attempt < 3: 00080 print "attempt %s failed to connect to %s, try again without specifying the DB " % ( self._attempt, self.url ) 00081 self.name = "" 00082 self._get_server() 00083 else: print "attempt %s failed to connect to %s " % ( self._attempt , self.url )
| def DybPython::dbcas::DBCon::drop | ( | self, | ||
| dbn | ||||
| ) |
Definition at line 86 of file dbcas.py.
00090 : 00091 if dbn == self.origname: 00092 raise DBExc("cannot drop DB of the originating name \"%s\" for safety" % self.origname ) 00093 if self.server.DropDataBase( dbn ) == 0: 00094 if self.verbosity>1: 00095 print "succeeded to drop db \"%s\" " % dbn else:
| def DybPython::dbcas::DBCon::create | ( | self, | ||
| dbn, | ||||
autoclobber = True | ||||
| ) |
Definition at line 96 of file dbcas.py.
00096 : 00097 print "failed to drop db \"%s\" " % dbn 00098 self.errlog(dbn) 00099 00100 def create(self, dbn , autoclobber=True ): 00101 if dbn == self.origname: 00102 raise DBExc("cannot create DB of the originating name \"%s\" for safety" % self.origname ) 00103 self.server.EnableErrorOutput(False) 00104 if self.server.CreateDataBase( dbn ) == 0: 00105 if self.verbosity>1: 00106 print "succeeded to create db \"%s\" " % dbn 00107 else: 00108 err, msg = self.server.GetErrorCode(), self.server.GetErrorMsg() 00109 if err == 1007 and autoclobber == True and len(self.get(dbn,())) < 5: 00110 if self.verbosity>2: 00111 print "failed to create db \"%s\" as it exists already ... " % dbn 00112 self.errlog(dbn) 00113 self.drop(dbn) 00114 self.create(dbn) 00115 else: print "failed to create db \"%s\" due to %s \"%s\" " % (dbn, err, msg )
| def DybPython::dbcas::DBCon::_spawn | ( | self, | ||
| kwa | ||||
| ) | [private] |
| def DybPython::dbcas::DBCon::spawn | ( | self, | ||
fixpass = False | ||||
| ) |
Create new DB with prefixed name and spawn a DBCon to talk to it with When *fixpass* is True the DB is neither created or dropped, but it is assumed to exist. This is used when doing DBI double dipping, used for example in :dybgaudi:`Database/DBWriter/tests`
Definition at line 124 of file dbcas.py.
00128 : 00129 """ 00130 Create new DB with prefixed name and spawn a DBCon to talk to it with 00131 00132 When *fixpass* is True the DB is neither created or dropped, but it is assumed 00133 to exist. This is used when doing DBI double dipping, used for example in 00134 :dybgaudi:`Database/DBWriter/tests` 00135 00136 """ 00137 name = "%s%s" % (DBCon.prefix, self.origname) if not fixpass:
| def DybPython::dbcas::DBCon::tables | ( | self | ) |
| def DybPython::dbcas::DBCon::process | ( | self, | ||
| sql | ||||
| ) |
Attempts to create prepared statement from sql then processes it
Definition at line 143 of file dbcas.py.
00145 :self.server.IsConnected()) 00146 00147 def process(self, sql): 00148 """ 00149 Attempts to create prepared statement from sql then processes it 00150 """ 00151 if sql.strip() == "": 00152 if self.verbosity>2: 00153 print "skipping blank line" 00154 return True 00155 00156 if sql.endswith("insert into DbiDemoData3 values( data supplied by DbiDemodata3 - see DbiValidate::PrepareDatabases() )"): 00157 if self.verbosity>0: 00158 print "skipping DbiValidate atrocity " 00159 return True 00160 00161 #print "process [%s] " % sql 00162 if not(self.isconnected): 00163 print "not connected, cannot process \"%s\" " % sql 00164 return 00165 st = self.server.Statement( sql ) 00166 if not(st): 00167 print "failed to create statement ... probably an error in the sql " 00168 return 00169 ok = st.Process()
| def DybPython::dbcas::DBCon::populate | ( | self | ) |
| def DybPython::dbcas::DBCon::fixture | ( | self | ) |
| def DybPython::dbcas::DBCon::mysqldump | ( | self, | ||
| opt | ||||
| ) |
| def DybPython::dbcas::DBCon::dump | ( | self, | ||
path = None, |
||||
opt = "" | ||||
| ) |
| def DybPython::dbcas::DBCon::fixture_ | ( | self, | ||
| path_ | ||||
| ) |
Definition at line 187 of file dbcas.py.
00191 : 00192 if not(self.name.startswith(DBCon.prefix)): 00193 raise DBExc("as safety measure cannot apply fixture to DB %s, name must start with prefix \"%s\" " % ( self.name, DBCon.prefix) ) 00194 path = os.path.expandvars(os.path.expanduser(path_)) 00195 print "%s fixture reading path %s " % (self, path_ ) for sql in open(path).read().split(";"):
string DybPython::dbcas::DBCon::prefix = 'tmp_' [static] |
tuple DybPython::dbcas::DBCon::url = property( lambda self:os.path.expanduser(os.path.expandvars(self.url_.GetUrl())), lambda self,n:self.url_.SetUrl(n) ) [static] |
tuple DybPython::dbcas::DBCon::name = property( lambda self:self.url_.GetFile(), lambda self,n:self.url_.SetFile(n) ) [static] |
tuple DybPython::dbcas::DBCon::host = property( lambda self:self.url_.GetHost(), lambda self,n:self.url_.SetHost(n) ) [static] |
tuple DybPython::dbcas::DBCon::ascii = property( lambda self:self.url_.GetAnchor(), lambda self,n:self.url_.SetAnchor(n) ) [static] |
tuple DybPython::dbcas::DBCon::opts = property( lambda self:self.url_.GetOptions(), lambda self,n:self.url_.SetOptions(n) ) [static] |
tuple DybPython::dbcas::DBCon::server = property( _get_server, doc=_get_server.__doc__ ) [static] |
tuple DybPython::dbcas::DBCon::isconnected = property(lambda self:self.server.IsConnected()) [static] |
DybPython::dbcas::DBCon::_server [private] |
DybPython::dbcas::DBCon::_attempt [private] |
1.4.7