'.format( ex.diag.severity, primary, detail, hint) class DBCommitError(DBError): def __init__(self, pgexc): super(DBCommitError, self).__init__(pgexc) class DBDataLineError(DBError): '''Database generated an error while the processor was running.''' def __init__(self, udl, pgexc): ''' udl An UploadDataLine instance pgexc The psycopg2 exception object ''' super(DBDataLineError, self).__init__(pgexc) self.lineno = udl.lineno self.data = udl.raw class DataLineError(UploadError): ''' Module exceptions rasied while line-by-line processing the uploaded data are derived from this class. lineno The line number e The error message descr More description of the error detail Extra HTML describing the error data The uploaded data ''' def __init__(self, lineno, e, descr='', detail='', data=''): super(DataLineError, self).__init__(e, lineno, descr, detail, data) class TooManyColsError(DataLineError): def __init__(self, lineno, e, descr='', detail='', data=''): super(TooManyColsError, self).__init__(lineno, e, descr, detail, data) # Upload processing class SQLCommand(object): ''' An SQL command that returns nothing Attributes: stmt The statement, formatted for psycopg2 substitution args Tuple of arguments used to substitute when executed. ''' def __init__(self, stmt, args, ec=None): ''' stmt The statement, formatted for psycopg2 substitution args Tuple of arguments used to substitute when executed. ec(ex) Produces the exception to raise an instance of on failure Input: ex The exception raised by psycopg2 ''' super(SQLCommand, self).__init__() self.stmt = stmt self.args = args self.ec = ec def execute(self, cur): ''' Execute the sql statement. Input: cur A psycopg2 cursor Side effects: Does something in the db. Can raise a psycopg2 error ''' try: cur.execute(self.stmt, self.args) except psycopg2.DatabaseError as ex: if self.ec == None: raise ex else: raise self.ec(ex) class LogSQLCommand(SQLCommand): '''An SQL command that logs success or failure.''' def __init__(self, stmt, args, ec=None, log_success=None, log_failure=None): ''' stmt The statement, formatted for psycopg2 substitution args Tuple of arguments used to substitute when executed. ec(ex) Produces the exception to raise an instance of on failure Input: ex The exception raised by psycopg2 ''' super(LogSQLCommand, self).__init__(stmt, args, ec) self.log_success = log_success self.log_failure = log_failure def execute(self, cur): ''' Execute the sql statement. Input: cur A psycopg2 cursor Side effects: Does something in the db. Can raise a psycopg2 error ''' try: super(LogSQLCommand, self).execute(cur) except (UploadError, psycopg2.DatabaseError) as ex: if self.log_failure: self.log_failure(ex) raise else: if self.log_success: self.log_success() class UploadLine(object): ''' Representation of a generic uploaded line Responsbile for getting data out of uploaded lines. Once out, the following attributes provide the user API: raw The "raw" line as a string tuples A sequence, the line split into columns ''' def __init__(self, line, stol, mapper): ''' line Uploaded line lineno Line number stol(st) Parse string st and convert into list mapper(st) Maps st onto desired python representation (Used for trimming whitespace) ''' super(UploadLine, self).__init__() self.raw = line self.tuples = [mapper(st) for st in stol(line)] def doublequote(st): ''' Put string in double quotes escaping according to sql's requirements for an identifier. Note that this does not work for unicode identifiers. ''' out = '"' for ch in st: if ch == '"': out += '""' else: out += ch return out + '"' class UploadHeaders(UploadLine): '''Uploaded heading line Derived from UploadLine raw The "raw" line as a string tuples As a sequence sql As a sql command fragment string ''' def __init__(self, line, stol, mapper): if mapper(line) == '': raise NoHeadersError('No column headings found on first line', 'The first line is ({0})'.format(line)) super(UploadHeaders, self).__init__(line, stol, mapper) self.sql = ', '.join(['"' + doublequote(st) + '"' for st in self.tuples]) class UploadDataLine(UploadLine): '''Uploaded data line Derived from UploadLine raw The "raw" line as a string tuples As a sequence lineno The line number ''' def __init__(self, line, lineno, stol, mapper): super(UploadDataLine, self).__init__(line, stol, mapper) self.lineno = lineno class DBData(object): '''Data that is to modify the db. It is iterable, consisting of thunks which make an UploadedLine instance. lineno Number of lines processed (including header line) ''' def __init__(self): self.lineno = 0 def __iter__(self): return self def __next__(self): ''' Iterator to return a thunk which, when called, delivers the next object to be loaded into the db.''' self.lineno += 1 return self._thunk() def _thunk(): ''' Return a thunk which, when called, delivers the next object to go into the db. ''' raise NotImpementedError() class SQLData(DBData): ''' SQL statements returning no data that execute in the db. Attributes: stmts List of SQLCommand instances ''' def __init__(self, stmts): ''' stmts List of SQLCommand instances ''' super(SQLData,self).__init__() self.stmts = stmts def gen(stmts): for stmt in stmts: yield stmt self.stmt_gen = gen(stmts) def _thunk(self): stmt = next(self.stmt_gen) return lambda: stmt class UploadData(DBData): '''Uploaded data file It is iterable, consisting of thunks which make an UploadedLine instance. headers UploadedHeaders instance cols Number of columns lineno Number of lines processed (including header line) ''' def __init__(self, fileo, file_fmt, null_data, null_rep, trim=True): ''' fileo Uploaded file object file_fmt File format: CSV or TAB null_data (boolean) Uploaded data contains nulls null_rep Uploaded string representation of null trim (boolean) Trim leading and trailing whitespace? ''' def get_headers(): try: line = next(self._fileo) except StopIteration: raise NoDataError('Uploaded file contains no data') else: self.lineno += 1 # Intuit the eol sequence if line[-2:] == '\r\n': eol = '\r\n'; line = line[:-2] elif line[-1:] == '\n': eol = '\n' line = line[:-1] elif line[-1:] == '\r': eol = '\r' line = line[:-1] else: # Give up eol = '' if file_fmt == CSV: func = lambda st: next(csv_reader((st,))) else: func = lambda st: st.split('\t') return (eol, UploadHeaders(line, func, do_trim)) super(UploadData, self).__init__() if trim: do_trim = lambda st: st.strip() else: do_trim = lambda st: st # Read lines as text with line endings intact. # newline='' required by csv.reader() self._fileo = io.TextIOWrapper(fileo, newline='') # Get column headings eol, self.headers = get_headers() neg_eollen = -len(eol) self.cols = len(self.headers.tuples) # Create parser to read raw lines into a list if file_fmt == CSV: self._parser = lambda st: self._extend(st, next(csv_reader((st,)))) else: # Tab delimited format def tab_parser(line): if neg_eollen and line[neg_eollen:] == eol: line = line[:neg_eollen] return self._extend(line, line.split('\t')) self._parser = tab_parser # Create mapper to frob data elements if null_data: null_rep = do_trim(null_rep) # psycopg2 maps None to NULL def mapper(st): st = do_trim(st) return None if st == null_rep else st self._mapper = mapper else: self._mapper = do_trim def _thunk(self): ''' Return a thunk which, when called, delivers the UploadDataLine of the next line of the uploaded file.. ''' line = next(self._fileo) return lambda: UploadDataLine(line, self.lineno, self._parser, self._mapper) def _extend(self, line, seq): '''Give the list as many elements as there are in the header. line line as string seq line as sequence If there's too many elements, raise an error. ''' if len(seq) > self.cols: raise TooManyColsError(self.lineno, 'Line has too many columns', 'More columns than column headings', data=line) return seq + [ '' for i in range(len(seq)+1, self.cols)] class DataLineProcessor(object): ''' A processor supplied uploaded lines (UploadDataLine instances) by an UploadEngine. The lines are expected to be put into a pg db. Abstract class not expected to be instantiated. Attributes: ue UploadEngine instance uh UploadHandler instance cur psycopg2 cursor Methods: eat(udl) Given an UploadDataLine instance put the line in the db. ''' def __init__(self, ue, uh): ''' ue UploadEngine instance uh UploadHandler instance ''' super(DataLineProcessor, self).__init__() self.ue = ue self.uh = uh self.cur = ue.cur def eat(udl): ''' Update the db with the content of an uploaded line udl An UploadDataLine instance ''' raise NotImplementedError class NoOpProcessor(DataLineProcessor): '''A processor that discards it's input.''' def __init__(self, ue, uh): ''' ue UploadEngine instance uh UploadHandler instance cur psycopg2 cursor ''' super(NoOpProcessor, self).__init__(ue, uh) def eat(self, udl): ''' Executes an sql command in the db. udl A DataLine instance. ''' pass class ExecuteSQL(DataLineProcessor): def __init__(self, ue, uh): ''' ue UploadEngine instance uh UploadHandler instance cur psycopg2 cursor ''' super(ExecuteSQL, self).__init__(ue, uh) def eat(self, sqlc): ''' Executes an sql command in the db. sqlc An SQLCommand instance (a command and it's args) ''' sqlc.execute(self.cur) class DBHandler(object): ''' Abstract class of a handler that modifies the db. The main logic goes here, including getting the data into and out of the db. Responsible for: Telling the form (LoadedForm) to get user data Retrieval of data file or generation of other data to go in the db (Generally involving an UploadData instance, and hence an UploadLine instance.) Validation of user supplied data; error raising: The html form data Whatever other data is uploaded or otherwise passed in Pre-processing to prepare for getting data into the db (Pre-processing is referred to as "setup" presently.) Getting the individual bits of data into and out of the db (by way of having a factory that produces a DataLineProcessor) Post-processing to wrap up after the data is in the db (Post-processing is referred to as "cleanup" presently.) Rendering the html output Attributes: request A pyramid request instance uf An UploadForm instance data Something (beyond what's on the form) that goes into db ''' def __init__(self, request): ''' request A pyramid request instance ''' super(DBHandler, self).__init__() self.request = request self.uf = self.make_form() def make_form(self): ''' Return an instantiation of the upload form needed by the upload handler. ''' raise NotImpliementedError def get_data(self): ''' Put something that will go into the db into the 'data' attribute. ''' raise NotImplimentedError def val_input(self): ''' Validate input needed beyond that required to connect to the db. Note that this occurs after read() is called. This is expected to be replaced by it's subclass. ''' return [] def read(self): '''Read the html form into the upload form instance''' self.uf.read() def write(self, result, errors): ''' Return dict needed to render an html form Input: errors List of Error instances Returns: Dict pyramid will use to render the resulting form Reserved keys: errors A list of UploadError exceptions. ''' return self.uf.write(result, errors) def cleanup(self): ''' Called after all lines are processed to do any final updates to the db. ''' pass def render(self, errors, response): ''' Return to pyramid what's needed to render the appropriate html page. Input: errors List of Error instances response Db connection result dict Returns: Dict pyramid will use to render the resulting form ''' response = self.write(response, errors) return response class SessionDBHandler(DBHandler): ''' A DBHandler that supports sessions. Attributes: uf An UploadForm instance session A pyramid session instance ''' def __init__(self, request): ''' request A pyramid request instance ''' super(SessionDBHandler, self).__init__(request) self.session = self.request.session def write(self, result, errors): ''' Add csrf token into html form to protect sessions. Input: errors List of Error instances Returns: Dict pyramid will use to render the resulting form Reserved keys: errors A list of UploadError exceptions. csrf_token Token for detecting CSRF. ''' response = super(SessionDBHandler, self).write(result, errors) response['csrf_token'] = self.session.new_csrf_token() return response class UploadHandler(SessionDBHandler): ''' Handler for uploading a file. Attributes: request A pyramid request instance uf An UploadForm instance data (optional) A DBData instance ''' def __init__(self, request): ''' request A pyramid request instance ''' super(UploadHandler, self).__init__(request) def factory(self, ue): ''' Takes an UploadEngine instance Returns a DataLineProcessor instance ''' raise NotImplementedError def val_input(self): ''' Validate input needed beyond that required to connect to the db. Returns: A list of Error instances ''' uf = self.uf errors = super(UploadHandler, self).val_input() if uf['filename'] == '': errors.append(NoFileError('No file supplied')) return errors def double_validator(self, errors): '''Utility function that can optionally be called by a val_input() function. It checks that the same file has not been uploaded twice in succession to the same db. `errors` List of errors. Appended to. ''' uf = self.uf if self.make_double_key() == uf['last_key']: errors.append(DuplicateUploadError( 'File just uploaded to this db', ('File named ({0}) just uploaded' .format(markupsafe.escape(uf['filename']))), '{0}: {1}{2}{3}
Try again to force upload.
')) def hash_sequence(self, seq): '''Produce a hash of the elements of a sequence.''' hash = hashlib.md5() for item in seq: hash.update((hash.hexdigest() + item).encode('utf-8')) return hash.hexdigest() def make_double_key(self): '''Generate unique value which identifies a double upload.''' uf = self.uf return self.hash_sequence((uf['db'], uf['filename'])) def write_double_key(self, response): '''Utility function. Optionally called from within write() to save a key which is later tested for to determine if the same file is being uploaded twice in a row. This function keys by filename and db. The key is stored in the form and submitted in POST. This way if the session times out due to inactivity the double upload is still detected. Since the data is delivered to the client and the client will likely see it anyway we use a hash not for security but just to provide a convenient single value that can be tested. Input: response Dict used by pyramid to render html form. Side effects: Modifies response. Adds 'last_key' entry used by form to store key. ''' response['last_key'] = self.make_double_key() def write(self, result, errors): ''' Add lines processed and number of errors into html form shown the user. Input: errors List of Error instances Returns: Dict pyramid will use to render the resulting form Reserved keys: errors A list of UploadError exceptions. csrf_token Token for detecting CSRF. e_cnt Number of errors. db_changed Boolean. Whether the db was changed. ''' response = super(UploadHandler, self).write(result, errors) if hasattr(self, 'data'): response['lines'] = self.data.lineno - 1 response['e_cnt'] = len(errors) response['db_changed'] = (not response['errors'] and self.uf['action'] != '') return response class TabularFileUploadHandler(UploadHandler): ''' Handler for uploading a file with rows and columns and column headings. Attributes: request A pyramid request instance uf An UploadForm instance data An UploadData instance ''' def __init__(self, request): ''' request A pyramid request instance ''' super(TabularFileUploadHandler, self).__init__(request) def cleanup(self): '''Finish after processing all lines.''' lines = self.ue.data.lineno if lines == 1: raise DataLineError( 1, 'File contains no data', ('No lines found after ' 'initial line of column headings')) class DBConnector(object): '''Abstract class supporting db connection and execution of a processor function that alters the db content. Responsible for connecting to the db, transaction management, and running the DataLineProcessor instances produced by the DBConnector's (the UploadHandler's) factory. Collects all errors raised during the processing. Telling the uploadhandler to cleanup. This is done after all upload processing and before commit or rollback (and before the connection is closed). Attributes: uh An UploadHandler instance. cur A psycopg2 cursor instance db Name of db to connect to user User to connect to db password Password to connect to db action Upload form's hidden action variable '' means not submitted, anything else means take action Methods: run() Get a DataLineProcessor instance from the upload handler's factory and feed it by iterating over data. ''' def __init__(self, uh): ''' uh An UploadHandler instance ''' super(DBConnector, self).__init__() # Configuration and response management. self.uh = uh def alter_db(self, conn): ''' Alter the contents of the db. Input: conn A database connection Returns: A list of Error instances Side Effects: Alter's db ''' raise NotImplementedError def no_connection_response(self): ''' The result dict to use when call_with_connection is not called. Matches with the result dict of call_with_connection Returns: Dict pyramid will use to render the resulting form. Reserved keys: havecreds Boolean. Supplied credentials can connect to db. ''' return {'havecreds' : False} def nodberror_factory(self): return NoDBError('No database name supplied') def nousererror_factory(self): return NoUserError('No user name supplied as login credentials') def authfailerror_factory(self): return AuthFailError('Unable to login', 'Is the database, user, and password correct?') def dryrunerror_factory(self): return DryRunError('Configured for "dry_run":' ' Transaction deliberately rolled back') def upload_data(self, data, errors): '''Put a DBData object into the db. Input: data A DBData object. errors A list of errors Side Effects: Alters db content Add errors to `errors`. ''' raise NotImplementedError def alter_db(self, conn): ''' Alter the contents of the db. Input: conn A database connection Returns: A list of Error instances Side Effects: Alter's db ''' errors = [] try: self.uh.get_data() self.data = self.uh.data self.cur = conn.cursor() # (Cannot call uh until after self is fully # initalized, including self.cur.) processor = self.uh.factory(self) except Error as ex: errors.append(ex) else: try: self.upload_data(processor, self.data, errors) # Let upload handler finish try: self.uh.cleanup() except UploadError as ex: errors.append(ex) finally: self.cur.close() return errors def call_with_connection(self, func): ''' Validate input, connect to the db, and do something with the connection. func(conn) Call this function with the connection. func(conn) must return a list of Error instances Returns: (errors, response) errors List of Error instances response Dict pyramid will use to render the resulting form. The dict returned by func(conn) plus reserved keys. Reserved keys: havecreds Boolean. Supplied credentials can connect to db. Side effects: Raises errors, calls func(conn) ''' errors = [] havecreds = False response = {} uf = self.uh.uf if self.db == '': errors.append(self.nodberror_factory()) if self.user == '': errors.append(self.nousererror_factory()) newes = self.uh.val_input() if newes: if not errors: havecreds = True errors.extend(newes) if not errors: registry = self.uh.request.registry try: conn = psycopg2.connect( database = self.db, user = self.user, password = self.password, host = registry.settings['pg_host'], port = registry.settings['pg_port']) except psycopg2.OperationalError: errors = [self.authfailerror_factory()] havecreds = False else: havecreds = True errors = self.call_alter_db(func, conn) response.update({'havecreds' : havecreds }) return (errors, response) def read_uh(self): '''Read data into the upload handler.''' self.uh.read() def read_db(self): '''Read the db from the upload handler.''' self.db = self.uh.uf['db'] def read_user(self): '''Read the username from the upload handler.''' self.user = self.uh.uf['user'] def read_password(self): '''Read the password from the upload handler.''' if 'password' in self.uh.uf: self.password = self.uh.uf['password'] else: self.password = '' def read_action(self): '''Read the action from the upload handler.''' self.action = self.uh.uf['action'] def read(self): '''Allows for custom engines to map what the handler has to what they need. ''' self.read_uh() self.read_db() self.read_user() self.read_password() self.read_action() def run(self): '''Call the alter_db function with the connection to alter the db, render form and return Response Input: Returns: (errors, response) errors List of Error instantiations response Dict containing connection result info Side effects: Maintains session (Whatever the processor() call does) Outputs html ''' self.read() if self.action == 'u': errors, response = self.call_with_connection(self.alter_db) else: errors = [] response = self.no_connection_response() return self.uh.render(errors, response) class NoTransactionEngine(DBConnector): '''An engine that does not run things in a transaction. It also does not do CSRF checking. Attributes: uh An UploadHandler instance. data An UploadData instance of the uploaded data cur A psycopg2 cursor instance db Name of db to connect to user User to connect to db password Password to connect to db action Upload form's hidden action variable '' means not submitted, anything else means take action Methods: run() Get a DataLineProcessor instance from the upload handler's factory and feed it by iterating over data. ''' def __init__(self, uh): ''' uh An UploadHandler instance ''' super(NoTransactionEngine, self).__init__(uh) def call_alter_db(self, func, conn): ''' Call a database modification function with a db connection, turning off the automatic wrapping of the activity in a transaction. func(conn) Call this function with the connection. func(conn) must return a list of Error instances Returns: errors List of Error instances Side effects: Calls func(conn) ''' conn.set_session(autocommit=True) errors = func(conn) conn.close() return errors def upload_data(self, processor, data, errors): '''Put a DBData object into the db, without a wrapping transaction. Input: processor A DataLineProcessor object. data A DBData object. errors A list of errors Side Effects: Alters db content Add errors to `errors`. ''' for thunk in data: try: udl = thunk() except DataLineError as ex: errors.append(ex) else: try: processor.eat(udl) except psycopg2.DatabaseError as ex: errors.append(DBDataLineError(udl, ex)) except DataLineError as ex: errors.append(ex) except DBError as ex: errors.append(ex) class UnsafeUploadEngine(DBConnector): '''Abstract class supporting repeatedly running a processer to put uploaded data into db. Attributes: uh An UploadHandler instance. data An UploadData instance of the uploaded data cur A psycopg2 cursor instance db Name of db to connect to user User to connect to db password Password to connect to db action Upload form's hidden action variable '' means not submitted, anything else means take action Methods: run() Get a DataLineProcessor instance from the upload handler's factory and feed it by iterating over data. eat_old_line(udl, thunk) Trap errors raised by the db while running thunk. Report any errors as due to the udl UploadDataLine instance. ''' def __init__(self, uh): ''' uh An UploadHandler instance ''' super(UnsafeUploadEngine, self).__init__(uh) def call_alter_db(self, func, conn): ''' Call a database modification function with a connection. func(conn) Call this function with the connection. func(conn) must return a list of Error instances Returns: errors List of Error instances Side effects: Calls func(conn) ''' errors = func(conn) if errors: conn.rollback() else: if ast.literal_eval(self.uh.request.registry.settings['dry_run']): conn.rollback() errors.append(self.dryrunerror_factory()) else: try: conn.commit() except psycopg2.DatabaseError as ex: errors.append(DBCommitError(ex)) conn.close() return errors def eat_old_line(self, udl, thunk): ''' Run thunk and report any errors raised by the db as due to the UploadDataLine instance udl. Returns: The result of running thunk. ''' try: result = thunk() except psycopg2.DatabaseError as ex: raise DBDataLineError(udl, ex) else: return result def upload_data(self, processor, data, errors): '''Put a DBData object into the db. Input: processor A DataLineProcessor object. data A DBData object. errors A list of errors Side Effects: Alters db content Add errors to `errors`. ''' for thunk in data: try: udl = thunk() except DataLineError as ex: errors.append(ex) else: self.cur.execute( 'SAVEPOINT line_savepoint;') try: processor.eat(udl) except psycopg2.DatabaseError as ex: self.cur.execute( 'ROLLBACK TO line_savepoint;') errors.append(DBDataLineError(udl, ex)) except DataLineError as ex: self.cur.execute( 'ROLLBACK TO line_savepoint;') errors.append(ex) except DBError as ex: self.cur.execute( 'ROLLBACK TO line_savepoint;') errors.append(ex) else: self.cur.execute( 'RELEASE line_savepoint;') class UploadEngine(UnsafeUploadEngine): '''Abstract class supporting repeatedly running a processor to alter db content. Adds cross-site request forgery checking to the UnsafeUploadEngine class. Attributes: uh An UploadHandler instance. cur A psycopg2 cursor instance db Name of db to connect to user User to connect to db password Password to connect to db action Upload form's hidden action variable '' means not submitted, anything else means take action csrf_token The client's csrf_token Methods: run() Get a DataLineProcessor instance from the upload handler's factory and feed it by iterating over data. ''' def __init__(self, uh): ''' uh An UploadHandler instance ''' super(UploadEngine, self).__init__(uh) def csrferror_factory(self): return CSRFError( 'Your request failed and you are now logged out', ('This is a security measure. ' 'Some possible causes are:'), ('