wsatools.DbConnect.CuSession

109 """ 110 A curation session. Any data curation activity for the archive where a 111 log is required to be recorded, should either be designed as a class that 112 inherit from this class or else an instance of this class should be invoked 113 by that curation script. This class design ensures a log is correctly 114 written, errors are correctly handled, temporary files are correctly 115 managed and that the database ArchiveCurationHistory and 116 ProgrammeCurationHistory tables are updated if required. Also a database 117 connection, temporary working directory, and programme table objects are 118 provided if required. 119 120 @warning: The destructor won't be called if the instance of a CuSession 121 object is copied to a member variable that stores it (if it used 122 only temporarily then that's OK - either the member variable uses 123 the reference to the CuSession temporarily or the reference to 124 the CuSession is passed to a temporary variable). This is because 125 a circular reference occurs on destruction. 126 127 @note: In the destructor you cannot order the destruction of the archive 128 and log objects by just calling del. Therefore, I've left their 129 order of destruction undefined - the order is not important. 130 131 @group Nested Errors and Exceptions: CuError 132 133 """

134 - class CuError(Exception):

135 """ A curation error instead of a programming error. If this is thrown 136 no exception trace is displayed and the ArchiveCurationHistory is 137 not updated. Just designed for an error that prevents a curation 138 task from even beginning. 139 """ 140 pass

141 142 #-------------------------------------------------------------------------- 143 ## Class-scope member data - all instances have the same value 144 145 _numSessions = 0 146 147 # Public class-scope member data - access as CuSession.varName 148 149 #: Exception types trapped by any CuSession in this process. 150 eTypes = set() 151 #: System constants for this CuSession. 152 sysc = SystemConstants() 153 154 #-------------------------------------------------------------------------- 155 ## Class parameters - override in subclasses 156 157 #: Curation task number. 158 cuNum = dbc.smallIntDefault() 159 #: Include day in log file name? 160 isDayStampedLog = False 161 #: Should this curation task auto-commit database transactions? 162 _autoCommit = DbSession._autoCommit 163 #: Always clean-up all temporary files. 164 _cleanUpDirs = True 165 #: Should this curation task try to re-open broken database connections? 166 _isPersistent = DbSession._isPersistent 167 #: Should only a single instance of this curation task run at a time? 168 _reqDbLock = True 169 #: In trial run mode is a cuEventID required? 170 _reqCuEventID = False 171 #: Create a temporary work directory. 172 _useWorkDir = False 173 174 #-------------------------------------------------------------------------- 175 ## Public member variable default values (access as obj.varName) 176 177 # These need to be set from command-line options 178 179 #: Descriptive comment as to why curation task is being performed. 180 comment = 'A descriptive comment' 181 #: Name of curator. 182 curator = os.getenv('USER') 183 184 # These are automatically determined but are publicly accessible 185 186 #: A connection to the archive being curated (a L{DbSession} object). 187 archive = None 188 #: Unique ID for this curation event. 189 cuEventID = None 190 #: Only curate non-survey programmes? 191 onlyNonSurveys = False 192 #: Only curate survey programmes? 193 onlySurveys = False 194 #: A L{DataFactory.ProgrammeTable} - quick access to Programme table data. 195 programme = None 196 #: ID, or set of IDs, of (all) the programme being curated this session. 197 programmeID = None 198 #: Results file for ArchiveCurationHistory. 199 resultsFilePathName = dbc.charDefault() 200 #: Unique ID tag for files on the catalogue load server share directory. 201 shareFileID = '' 202 203 #-------------------------------------------------------------------------- 204 ## Private member variable default values (access prohibited) 205 206 #: The L{DataFactory.CurationTaskTable} - quick access to CurationTask. 207 _cuTable = None 208 #: Location of the database lock file. 209 _lockPathName = None 210 #: The curation log - an instance of a L{Logger} class. 211 _log = None 212 #: Has the CU transaction rolled-back? 213 _rolledBack = None 214 #: Has the programme being curated successfully? 215 _success = None 216 217 #-------------------------------------------------------------------------- 218 ## Generic command-line interface settings for a CuSession. 219 220 CLI.progArgs.append(CLI.Argument('comment', comment, isOptional=True)) 221 CLI.progOpts.append( 222 CLI.Option('c', 'curator', 'username of curator', 'NAME', curator)) 223 224 #-------------------------------------------------------------------------- 225

226 - def __init__(self, programmeID='', curator=curator, comment=comment, 227 database=DbSession.database, isTrialRun=DbSession.isTrialRun, 228 userName=DbSession.userName, cli=None):

229 """ 230 Writes a database lock and opens a connection. Obtains a unique 231 curation event ID and prepares a Logger object to hold the verbose 232 record of this curation event, whilst reserving a placeholder in the 233 database ArchiveCurationHistory to log the record of this curation 234 event. A temporary working directory is created, if required, and all 235 member variables are initialised. 236 237 @param programmeID: Optionally supply the unique identifier of a 238 programme or survey name initials to curate 239 selected programme. Set to 'NONSURVEYS' if just 240 curating non-surveys or 'SURVEYS' for surveys only. 241 @type programmeID: str 242 @param curator: Name of curator. 243 @type curator: str 244 @param comment: Descriptive comment as to why curation task is 245 being performed. 246 @type comment: str 247 @param database: Name of the database to connect to. 248 @type database: str 249 @param isTrialRun: If True, do not perform database modifications, 250 just print the SQL statement to the terminal. 251 @type isTrialRun: bool 252 @param userName: Optionally override default database username. 253 @type userName: str 254 @param cli: Optionally initialise using the settings supplied 255 by the command-line options. 256 @type cli: L{CLI.CLI} 257 258 """ 259 if CuSession._numSessions > 0: 260 Logger.addMessage( 261 "<Info> Waiting for old CuSessions to be deleted...") 262 time.sleep(30) 263 if CuSession._numSessions > 0: 264 # @@BUG: This isn't an exception any more because when running 265 # back-to-back CuSessions, if one throws an exception, the next 266 # starts before the previous one had a chance to delete. 267 Logger.addMessage( 268 "<Warning> Cannot have more than one CuSession object " 269 "running concurrently. Delete the old object before " 270 "creating the new object.") 271 else: 272 CuSession._numSessions += 1 273 274 if cli: 275 curator = cli.getOpt("curator") 276 comment = cli.getArg("comment") 277 database = cli.getArg("database") 278 isTrialRun = cli.getOpt("test") 279 userName = cli.getOpt("user") 280 281 # Connect to database and lock 282 # @@TODO: Should DbSession have an option to manage its lock? Would 283 # make more sense. 284 serverDb = (database if '.' in database else 285 SystemConstants(database).loadServer + '.' + database) 286 287 # Strip '/' from programme acronyms 288 programmeID = programmeID.replace('/', '') 289 self._lockPathName = os.path.join(self.sysc.sysPath, 290 "dblock-%sCU%sp%s" % (serverDb, self.cuNum, programmeID or 'ALL')) 291 292 try: 293 if self._reqDbLock and not isTrialRun: 294 self._prepareDbLock() 295 296 self.archive = DbSession(database, self._autoCommit, isTrialRun, 297 userName, self._isPersistent) 298 except Exception as error: 299 # @@TODO: Consider making this exception trap more specific, 300 # including giving the time-out exception a specific type 301 # and handling specific types of DbSession exceptions. 302 raise SystemExit(error) 303 304 CuSession.sysc = self.archive.sysc 305 306 # Assign a cuEventID from ArchiveCurationHistory and prepare log file 307 self.curator = curator 308 if not isTrialRun: 309 self.cuEventID = self._getNextCuEventID() 310 Logger.addMessage("Starting cuEventID: %s" % self.cuEventID) 311 312 # Create a log file to hold the verbose record for this invocation 313 dayStamp = time.ctime().split()[0] if self.isDayStampedLog else '' 314 self._log = Logger(fileName='%s_cu%sid%s%s.log' % 315 (self.archive.database, self.cuNum, self.cuEventID, dayStamp)) 316 317 elif self._reqCuEventID: 318 self.cuEventID = 123456789 319 Logger.addMessage("Starting TRIAL cuEventID: %s" % self.cuEventID) 320 321 self.shareFileID = '%sCuID%06d' % (self.archive.database, 322 self.cuEventID if self.cuEventID else 0) 323 324 # Log this CU's description from table CurationTask 325 if self.cuNum != dbc.smallIntDefault(): 326 try: 327 self._cuTable = df.CurationTaskTable(self.archive, self.cuNum) 328 Logger.addMessage("Running Curation Use case:") 329 Logger.addMessage(self._cuTable.getDescription()) 330 except ValueError as error: 331 Logger.addMessage("<Warning> %s" % error) 332 self._cuTable = None 333 334 # Log this programme's description from table Programme 335 self.onlyNonSurveys = (programmeID.upper() == 'NONSURVEYS') 336 self.onlySurveys = (programmeID.upper() == 'SURVEYS') 337 self.programme = df.ProgrammeTable(self.archive) 338 if not programmeID or self.onlyNonSurveys or self.onlySurveys: 339 self.programmeID = set() 340 progAcronym = (programmeID or "ALL").upper() 341 if not programmeID: 342 Logger.addMessage("for all programmes") 343 elif self.onlyNonSurveys: 344 Logger.addMessage("for all non-survey programmes") 345 else: 346 Logger.addMessage("for all survey programmes") 347 else: 348 self.programmeID = self.programme.setProgID(programmeID) 349 self.onlyNonSurveys = self.programme.isNonSurvey() 350 self.onlySurveys = not self.programme.isNonSurvey() 351 progAcronym = self.programme.getAcronym(self.programmeID).upper() 352 Logger.addMessage("for programme %s: " % progAcronym 353 + self.programme.getName()) 354 355 Logger.addMessage("on curation server: " + os.getenv('HOST')) 356 357 self.comment = "[%s] " % progAcronym + (("Running CU%s" % self.cuNum) 358 if comment is CuSession.comment else comment) 359 360 # Create a working directory if required 361 if self._useWorkDir: 362 workSpace = self.sysc.tempWorkPath() 363 if os.path.exists(workSpace): 364 os.rename(workSpace, workSpace + '.old') 365 Logger.addMessage('<Warning> curationClientWorkPath %s exists,' 366 ' moved to %s.old' % (workSpace, workSpace)) 367 os.mkdir(workSpace)

368 369 #-------------------------------------------------------------------------- 370

371 - def __del__(self):

372 """ 373 Finalises curation session - updates the ArchiveCurationHistory and 374 ProgrammeCurationHistory tables, if required. Cleans-up any temporary 375 files and files in the catalogue server share directory. The database 376 lock is removed, whilst the connection is automatically dropped, as 377 well as the log being automatically written a file. 378 379 @note: Order of deletion of self.archive and self._log is random, so 380 the disconnect from database message isn't also saved to the log 381 file, but it will always be displayed. 382 """ 383 try: 384 if self._log and self._success is not None \ 385 and self.archive and not self.archive.isTrialRun: 386 self._updateArchiveCurationHistory() 387 self._updateProgrammeCurationHistory() 388 elif self.cuEventID: 389 self._scrubACHPlaceHolder() 390 finally: 391 # Reset DbSession log requirement marker, since this is only used 392 # by CuSession objects and only one CuSession object is allowed in 393 # memory at once. 394 DbSession._isLogReq.update((k, False) for k in DbSession._isLogReq) 395 CuSession._numSessions -= 1 396 397 if self._success is None: 398 if self.archive and self.archive.isTrialRun: 399 Logger.addMessage("Database unmodified.") 400 else: 401 Logger.addMessage("<Warning> No curation task performed.") 402 elif self._success: 403 Logger.addMessage("All done! Successful curation.") 404 else: 405 Logger.addMessage("**** THERE WERE ERRORS (see above) ****") 406 407 # Remove working directory 408 if self._cleanUpDirs and self._useWorkDir \ 409 and os.path.exists(self.sysc.tempWorkPath()): 410 try: 411 shutil.rmtree(self.sysc.tempWorkPath()) 412 except OSError as error: 413 Logger.addMessage( 414 "<Warning> Unable to remove temporary working directory " 415 "%s: %s" % (self.sysc.tempWorkPath(), error)) 416 417 # Clean load-server NFS file share space of remaining CU temp files 418 try: 419 shareFiles = ([] if not self.shareFileID else 420 [self.archive.sharePath(fileName) 421 for fileName in os.listdir(self.archive.sharePath()) 422 if self.shareFileID in fileName]) 423 except OSError: 424 shareFiles = [] 425 426 if not self._success and self._cleanUpDirs and shareFiles: 427 msg = \ 428 "Remove outgest/ingest files from the share path (yes/no)? >" 429 if os.getenv("USER") != "scos": 430 answer = raw_input(msg).lower() 431 self._cleanUpDirs = not answer or 'y' in answer 432 else: 433 # No user-input if scos in case it's a background run, 434 # safest to keep the files on failure in this case - the 435 # files in question will be logged. 436 self._cleanUpDirs = False 437 438 if self._cleanUpDirs and shareFiles: 439 Logger.addMessage("Removing outgest/ingest files in " 440 + self.archive.sharePath()) 441 try: 442 for fileName in shareFiles: 443 os.remove(fileName) 444 except OSError as error: 445 Logger.addMessage("<Warning> Attempt to clean file share " 446 "failed: %s" % error) 447 elif shareFiles: 448 Logger.addMessage("<Info> Files left in file share:\n" + 449 '\n'.join(shareFiles)) 450 451 # Remove database lock 452 if self._lockPathName and os.path.exists(self._lockPathName): 453 thisTask = '%s.%s' % (os.getenv('HOST'), os.getpid()) 454 lockTask = file(self._lockPathName).read().split()[-1] 455 if thisTask == lockTask: 456 os.remove(self._lockPathName) 457 print("unlocked " + os.path.basename(self._lockPathName)) 458 459 elif self._reqDbLock \ 460 and self.archive and not self.archive.isTrialRun: 461 462 Logger.addMessage("<WARNING> Tried to remove DbLock, " 463 "but the file doesn't exist.")

464 465 #-------------------------------------------------------------------------- 466

467 - def run(self):

468 """ Executes the _onRun() method of the curation session within a 469 try block, so that all the standard exceptions may be dealt with 470 correctly. 471 472 @todo: Replace init and del with a context manager for database 473 connections and locks, implemented as a with-block in this 474 function. 475 """ 476 try: 477 self._onRun() 478 self.archive.commitTransaction() 479 self._rolledBack = False 480 self._success = True 481 except BaseException as error: 482 # Block future keyboard interrupts until CU has finished cleanly 483 with utils.noInterrupt(): 484 stdErrMsg = CuSession.logException(error) 485 if error.args and self._log: 486 stdErrMsg += ": see log " + self._log.pathName 487 488 self._onException() 489 if isinstance(error, CuSession.CuError) and \ 490 not any(self.archive._isLogReq.values()): 491 self._success = None 492 493 # @@NOTE: If raise is in with-block then interrupts won't restore. 494 raise SystemExit(stdErrMsg)

495 496 #-------------------------------------------------------------------------- 497 498 @staticmethod

499 - def logException(error):

500 """ 501 Logs the given trapped exception. 502 503 @param error: Exception caught. 504 @type error: Exception 505 506 @return: Derived error type (for full list see L{CuSession.eTypes}). 507 @rtype: str 508 509 """ 510 wantTrace = False 511 if isinstance(error, SystemExit): 512 eType = SystemExit.__name__ 513 elif isinstance(error, odbc.Error): 514 eType = "SQL or Database Error" 515 elif isinstance(error, ExtError): 516 eType = "External-process Error" 517 elif isinstance(error, CuSession.CuError): 518 eType = "Curation Failure" 519 elif isinstance(error, KeyboardInterrupt): 520 eType = KeyboardInterrupt.__name__ 521 # No user-input for scos runs in case it's a background job 522 wantTrace = (os.getenv("USER") == "scos") 523 if not wantTrace: 524 prompt = "Would you like to see trace (yes/no)? >" 525 answer = None 526 while answer is None: 527 try: # Don't accept another Ctrl-C as an answer! 528 answer = raw_input(prompt) 529 except KeyboardInterrupt: 530 print('') # Just output a carriage return 531 except EOFError: 532 # No stdin available, original traceback will be lost 533 answer = "no" 534 break 535 wantTrace = not answer or 'y' in answer.lower() 536 else: 537 sysErrors = (OSError, IOError, DbSession.CatalogueServerError) 538 eType = ("System Failure" if isinstance(error, sysErrors) else 539 "Programming Error") 540 wantTrace = True 541 542 if not str(error).startswith(tuple(CuSession.eTypes)): 543 # Don't log the message if it's a stderr message from an exception 544 # previously thrown by a CuSession. 545 Logger.addExceptionMessage(error, eType, wantTrace) 546 547 # Keep track of eTypes thrown (see above) as messages for stderr 548 CuSession.eTypes.add(eType) 549 550 return eType

551 552 #-------------------------------------------------------------------------- 553

554 - def logBrokenFile(self, fileDetails, logFilePath=None):

555 """ 556 Appends the details of the broken file to end of the provided log file. 557 Ensures that file has correct permisions if it does not already exist. 558 If in test mode the broken file is not logged. 559 560 @param fileDetails: Details of broken file, at least the file name. 561 @type fileDetails: str 562 @param logFilePath: Full-path to the log file. 563 @type logFilePath: str 564 565 """ 566 logFilePath = logFilePath or self.sysc.filesBrokenPath() 567 try: 568 if not self.archive.isTrialRun: 569 line = "%s %s\n" % (fileDetails, self._log.pathName) 570 utils.ensureDirExist(os.path.dirname(logFilePath)) 571 if os.path.exists(logFilePath): 572 file(logFilePath, 'a').write(line) 573 else: 574 file(logFilePath, 'w').write(line) 575 os.chmod(logFilePath, 0666) 576 except: 577 Logger.addMessage("<Warning> Cannot update file %s" % logFilePath)

578 579 #-------------------------------------------------------------------------- 580

581 - def _onRun(self):

582 """ Curation sessions that inherit from this class should override this 583 method, executing within all of the tasks that required of this 584 curation session. 585 """ 586 pass

587 588 #-------------------------------------------------------------------------- 589

590 - def _onException(self):

591 """ Attempts to rollback transaction and sets the database state flags, 592 so that the destructor realises the curation session failed. 593 Can override in subclasses to tidy up run tasks specific to that CU 594 following an exception, but always call this method. 595 """ 596 self._success = False 597 self._rolledBack = not any(self.archive._isLogReq.values()) 598 # Not strictly required as a commit won't occur, but why not. 599 self.archive.rollbackTransaction()

600 601 #-------------------------------------------------------------------------- 602

603 - def _getNextCuEventID(self):

604 """ 605 Obtain the next available cuEventID. Whilst putting a placeholder row 606 into the ArchiveCurationHistory table to reserve this cuEventID, 607 preventing it from being hijacked by a parallel process. 608 609 """ 610 cuEventID = self._getNextID(dbc.curationEventUIDAttributeName(), 611 dbc.archiveCurationHistoryTableName()) 612 613 if self.sysc.isVSA(): 614 dbID = 2 if self.archive.database == "VSAVVV" else 1 615 cuEventID = 10 * (cuEventID // 10 + 1) + dbID 616 617 isLogReq = self.archive._isLogReq[self.archive.database] 618 619 # Write it back in immediately to reserve this UID. 620 try: 621 self.archive.insertData(dbc.archiveCurationHistoryTableName(), 622 rowData=[cuEventID, self.cuNum, dbc.charDefault(), 623 dbc.charDefault(), utils.makeMssqlTimeStamp(), 624 self.curator, "Placeholder for CU currently " 625 "in progress", dbc.yes()], 626 enforcePKC=True) 627 628 self.archive.commitTransaction() 629 630 except odbc.InternalError as error: 631 if "database is read-only" not in str(error): 632 raise 633 634 raise SystemExit("<ERROR> Cannot curate databases that are mounted" 635 " read-only. Even if the CU only performs outgests" 636 " it still needs write permissions to construct" 637 " temporary views to enable BCP to work.") 638 639 except odbc.ProgrammingError as error: 640 if "permission was denied" not in str(error): 641 raise 642 643 raise SystemExit("<ERROR> You do not have write permission " 644 "on database " + self.archive.database) 645 finally: 646 self.archive._isLogReq[self.archive.database] = isLogReq 647 648 return cuEventID

649 650 #-------------------------------------------------------------------------- 651

652 - def _getNextID(self, attrName, tableName, where=''):

653 """ 654 Obtain next available ID number in the db for a given ID attribute. 655 656 @param attrName: Name of ID attribute. 657 @type attrName: str 658 @param tableName: Name of the table where attribute resides. 659 @type tableName: str 660 @param where: Optional SQL WHERE clause. 661 @type where: str 662 663 @return: Next available ID number. 664 @rtype: int 665 666 """ 667 maxID = self.archive.queryAttrMax(attrName, tableName, where) 668 669 if maxID: 670 return maxID + 1 671 else: 672 return 1

673 674 #-------------------------------------------------------------------------- 675

676 - def _scrubACHPlaceHolder(self):

677 """ 678 Scrubs the placeholder default row in the ArchiveCurationHistory table, 679 generated by _getNextCuEventID(). 680 681 """ 682 self.archive.delete(dbc.archiveCurationHistoryTableName(), 683 whereStr="cuEventID=%s" % self.cuEventID)

684 685 #-------------------------------------------------------------------------- 686

687 - def _updateArchiveCurationHistory(self):

688 """ 689 Adds an entry for this curation session to the ArchiveCurationHistory 690 database table. 691 692 """ 693 row = self.cuEventID, self.cuNum, self._log.pathName, \ 694 self.resultsFilePathName, utils.makeMssqlTimeStamp(), \ 695 self.curator, self.comment, (dbc.yes() if self._rolledBack 696 else dbc.no()) 697 698 Logger.addMessage("Updating archive curation history with this row:") 699 Logger.addMessage(repr(row)) 700 701 self._scrubACHPlaceHolder() 702 self.archive.insertData(dbc.archiveCurationHistoryTableName(), row) 703 704 if self._cuTable: 705 Logger.addMessage("completed " + self._cuTable.getDescription())

706 707 #-------------------------------------------------------------------------- 708

709 - def _updateProgrammeCurationHistory(self):

710 """ 711 Adds an entry for this curation session to the ProgrammeCurationHistory 712 database table. 713 714 """ 715 if not self.programmeID: 716 return 717 elif type(self.programmeID) is not set: 718 self.programmeID = [self.programmeID] 719 720 if len(self.programmeID) > 1: 721 Logger.addMessage( 722 "for programmes: " + utils.numberRange(self.programmeID)) 723 else: 724 for progID in self.programmeID: 725 Logger.addMessage("for programme %s: %s" % 726 (progID, self.programme.getName(programmeID=progID))) 727 728 tableName = dbc.programmeCurationHistoryTableName() 729 row = self.cuEventID, utils.makeMssqlTimeStamp(), \ 730 (dbc.yes() if self._success else dbc.no()) 731 732 Logger.addMessage("Updating programme curation history with this row:") 733 Logger.addMessage(repr(row)) 734 735 for progID in self.programmeID: 736 self.archive.insertData(tableName, (progID,) + row)

737 738 #-------------------------------------------------------------------------- 739

740 - def _prepareDbLock(self):

741 """ 742 Writes database lock. If lock already present then pause and wait for 743 database to unlock. 744 745 """ 746 numTries = 0 747 maxTries = 100 748 delay = 2 749 while os.path.exists(self._lockPathName): 750 time.sleep(delay) 751 numTries += 1 752 if numTries == 1: 753 print("A lock file exists: " + self._lockPathName) 754 print("End the task that owns the lock and/or manually " 755 "delete the lock file and this curation session will " 756 "automatically start.") 757 elif numTries > maxTries: 758 raise Exception( 759 "Timed out in waiting for database lock to be freed") 760 761 file(self._lockPathName, 'w').write("DB locked by CU%s. PID = %s.%s" % 762 (self.cuNum, os.getenv('HOST'), os.getpid()))

Source Code for Module wsatools.DbConnect.CuSession