(filename, no_create, update_by_name, update_static_fields)
Add or update officers from a CSV file.
Add or update officers from a CSV file.
Add or update officers from a CSV file .
def bulk_add_officers(filename, no_create, update_by_name, update_static_fields):
    """Add or update officers from a CSV file."""

    encoding = 'utf-8'

    # handles unicode errors that can occur when the file was made in Excel
    with open(filename, 'r') as f:
        if u'\ufeff' in f.readline():
            encoding = 'utf-8-sig'

    with open(filename, 'r', encoding=encoding) as f:
        ImportLog.clear_logs()
        csvfile = csv.DictReader(f)
        departments = {}

        required_fields = [
            'department_id',
            'first_name',
            'last_name',
        ]

        # Assert required fields are in CSV file
        for field in required_fields:
            if field not in csvfile.fieldnames:
                raise Exception('Missing required field {}'.format(field))
        if (not update_by_name
                and 'star_no' not in csvfile.fieldnames
                and 'unique_internal_identifier' not in csvfile.fieldnames):
            raise Exception('CSV file must include either badge numbers or unique identifiers for officers')

        for row in csvfile:
            department_id = row['department_id']
            department = departments.get(department_id)
            if row['department_id'] not in departments:
                department = Department.query.filter_by(id=department_id).one_or_none()
                if department:
                    departments[department_id] = department
                else:
                    raise Exception('Department ID {} not found'.format(department_id))

            if not update_by_name:
                # check for existing officer based on unique ID or name/badge
                if 'unique_internal_identifier' in csvfile.fieldnames and row['unique_internal_identifier']:
                    officer = Officer.query.filter_by(
                        department_id=department_id,
                        unique_internal_identifier=row['unique_internal_identifier']
                    ).one_or_none()
                elif 'star_no' in csvfile.fieldnames and row['star_no']:
                    officer = get_officer(department_id, row['star_no'],
                                          row['first_name'], row['last_name'])
                else:
                    raise Exception('Officer {} {} missing badge number and unique identifier'.format(row['first_name'],
                                                                                                      row['last_name']))
            else:
                officer = Officer.query.filter_by(
                    department_id=department_id,
                    last_name=row['last_name'],
                    first_name=row['first_name']
                ).one_or_none()

            if officer:
                update_officer_from_row(row, officer, update_static_fields)
            elif not no_create:
                create_officer_from_row(row, department_id)

        ImportLog.print_logs()
        if current_app.config['ENV'] == 'testing' or prompt_yes_no("Do you want to commit the above changes?"):
            print("Commiting changes.")
            db.session.commit()
        else:
            print("Aborting changes.")
            db.session.rollback()
            return 0, 0

        return len(ImportLog.created_officers), len(ImportLog.updated_officers)
def bulk_add_officers ( filename , no_create , update_by_name , update_static_fields ) : encoding = 'utf-8' # handles unicode errors that can occur when the file was made in Excel with open ( filename , 'r' ) as f : if u'\ufeff' in f . readline ( ) : encoding = 'utf-8-sig' with open ( filename , 'r' , encoding = encoding ) as f : ImportLog . clear_logs ( ) csvfile = csv . DictReader ( f ) departments = { } required_fields = [ 'department_id' , 'first_name' , 'last_name' , ] # Assert required fields are in CSV file for field in required_fields : if field not in csvfile . fieldnames : raise Exception ( 'Missing required field {}' . format ( field ) ) if ( not update_by_name and 'star_no' not in csvfile . fieldnames and 'unique_internal_identifier' not in csvfile . fieldnames ) : raise Exception ( 'CSV file must include either badge numbers or unique identifiers for officers' ) for row in csvfile : department_id = row [ 'department_id' ] department = departments . get ( department_id ) if row [ 'department_id' ] not in departments : department = Department . query . filter_by ( id = department_id ) . one_or_none ( ) if department : departments [ department_id ] = department else : raise Exception ( 'Department ID {} not found' . format ( department_id ) ) if not update_by_name : # check for existing officer based on unique ID or name/badge if 'unique_internal_identifier' in csvfile . fieldnames and row [ 'unique_internal_identifier' ] : officer = Officer . query . filter_by ( department_id = department_id , unique_internal_identifier = row [ 'unique_internal_identifier' ] ) . one_or_none ( ) elif 'star_no' in csvfile . fieldnames and row [ 'star_no' ] : officer = get_officer ( department_id , row [ 'star_no' ] , row [ 'first_name' ] , row [ 'last_name' ] ) else : raise Exception ( 'Officer {} {} missing badge number and unique identifier' . format ( row [ 'first_name' ] , row [ 'last_name' ] ) ) else : officer = Officer . query . filter_by ( department_id = department_id , last_name = row [ 'last_name' ] , first_name = row [ 'first_name' ] ) . one_or_none ( ) if officer : update_officer_from_row ( row , officer , update_static_fields ) elif not no_create : create_officer_from_row ( row , department_id ) ImportLog . print_logs ( ) if current_app . config [ 'ENV' ] == 'testing' or prompt_yes_no ( "Do you want to commit the above changes?" ) : print ( "Commiting changes." ) db . session . commit ( ) else : print ( "Aborting changes." ) db . session . rollback ( ) return 0 , 0 return len ( ImportLog . created_officers ) , len ( ImportLog . updated_officers )