<commit_before>#!/usr/bin/env python3
from src.core.errlog import errdata,mklog
import psycopg2 as psql
import time


# Main push-to-psql entry point
def export(data,project,config):
    db  = config['general']['database']
    tbl = config['general']['table']
    fields = data[0]._fields
    # handle custom-inserion instance if needed.
    if 'conversions' in config:
        ins = custom_insertion(fields,config['conversions'])
    # default is just standard psycopg2 formatting...
    else: ins = ','.join(['%s'] * len(fields))
    cmd = 'INSERT INTO {} VALUES ({})'.format(tbl,ins)
    errors,errtxt,duplicates = handle_push(data,cmd,db)
    if duplicates:
        print('duplicate rows ignored: ',len(duplicates))
    # save any rows which raised unexpexted errors
    # to a csv with prefix `psqlerr`.
    if errors: errdata(project,errors,txt='psqlerr')
    # if `save-duplicates` is flagged `true`, save
    # all duplicate rows to the errors directory.
    if duplicates and 'save-duplicates' in config['general']:
        if config['general']['save-duplicates']:
            errdata(project,duplicates,txt='psqldups')
    # append any unexpected errors to
    # the project's main error log.
    for err in errtxt: mklog(project,err)

# Controller for the actual push attempt.
# Forces all rows to be attempted at least once.
# This is a workaround for the fact that psql,
# until recently, will invalidate entire transaction
# blocks if too many duplacates are encountered.
def handle_push(data,cmd,db):
    rowtotal = len(data)
    # make a copy of data to avoid
    # side-effects from pops/dels.
    rows = [r for r in data]
    uploaded,ignored = [],[]
    errors,errtxt = [],[]
    while len(rows) > 0:
        rows,upld,igns,err,txt = push_rows(rows,cmd,db)
        uploaded += upld
        ignored += igns
        if err: errors.append(err)
        if txt: errtxt.append(txt)
    assert rowtotal == sum(len(errors),len(ignored),len(uploaded))
    return errors,errtxt,ignored

# Attempts to push rows to db.  Removes rows
# from data as they are pushed.  Halts & returns
# remaining rows upon first exception.
def push_rows(data,cmd,db):
    ignored,uploaded = [],[]
    error,text = None,None
    with psql.connect(database=db) as con:
        con.set_session(autocommit=True)
        for i in range(len(data)):
            row = data.pop(0)
            try:
                with con.cursor() as cur:
                    cur.execute(cmd,row)
                uploaded.append(row)
            except Exception as err:
                if 'duplicate key' in str(err):
                    ignored.append(row)<commit_msg>fixed minor bug in a unit test & added pre-attempt printout to<commit_after>#!/usr/bin/env python3
from src.core.errlog import errdata,mklog
import psycopg2 as psql
import time


# Main push-to-psql entry point
def export(data,project,config):
    db  = config['general']['database']
    tbl = config['general']['table']
    fields = data[0]._fields
    # handle custom-inserion instance if needed.
    if 'conversions' in config:
        ins = custom_insertion(fields,config['conversions'])
    # default is just standard psycopg2 formatting...
    else: ins = ','.join(['%s'] * len(fields))
    cmd = 'INSERT INTO {} VALUES ({})'.format(tbl,ins)
    print('pushing {} rows to psql...'.format(len(data)))
    errors,errtxt,duplicates = handle_push(data,cmd,db)
    if duplicates:
        print('duplicate rows ignored: ',len(duplicates))
    # save any rows which raised unexpexted errors
    # to a csv with prefix `psqlerr`.
    if errors: errdata(project,errors,txt='psqlerr')
    # if `save-duplicates` is flagged `true`, save
    # all duplicate rows to the errors directory.
    if duplicates and 'save-duplicates' in config['general']:
        if config['general']['save-duplicates']:
            errdata(project,duplicates,txt='psqldups')
    # append any unexpected errors to
    # the project's main error log.
    for err in errtxt: mklog(project,err)

# Controller for the actual push attempt.
# Forces all rows to be attempted at least once.
# This is a workaround for the fact that psql,
# until recently, will invalidate entire transaction
# blocks if too many duplacates are encountered.
def handle_push(data,cmd,db):
    rowtotal = len(data)
    # make a copy of data to avoid
    # side-effects from pops/dels.
    rows = [r for r in data]
    uploaded,ignored = [],[]
    errors,errtxt = [],[]
    while len(rows) > 0:
        rows,upld,igns,err,txt = push_rows(rows,cmd,db)
        uploaded += upld
        ignored += igns
        if err: errors.append(err)
        if txt: errtxt.append(txt)
    assert rowtotal == sum((len(errors),len(ignored),len(uploaded)))
    return errors,errtxt,ignored

# Attempts to push rows to db.  Removes rows
# from data as they are pushed.  Halts & returns
# remaining rows upon first exception.
def push_rows(data,cmd,db):
    ignored,uploaded = [],[]
    error,text = None,None
    with psql.connect(database=db) as con:
        con.set_session(autocommit=True)
        for i in range(len(data)):
            row = data.pop(0)
            try:
                with con.cursor() as cur:
                    cur.execute(cmd,row)
                uploaded.append(row)
            except Exception as err:
                if 'duplicate key' in str(err):
                    ignored.append(row)
{'fraction_non_alphanumeric': 0.11451942740286299, 'fraction_numerical': 0.002230897936419409, 'mean_word_length': 2.9212827988338192, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6770710', 'n_tokens_mistral': 1667, 'n_tokens_neox': 1626, 'n_words': 501}