queue_pmh_rt_record.py

importargparse
fromtimeimportsleep
fromtimeimporttime

fromsqlalchemyimporttext

fromappimportdb
fromappimportlogger
frompmh_recordimportPmhRecord
fromrecordthresher.recordimportRecordthresherParentRecord
fromrecordthresher.record_makerimportPmhRecordMaker
fromutilimportelapsed
fromutilimportsafe_commit

importendpoint# magic


classQueuePmhRTRecord:
defworker_run(self, **kwargs):
single_id=kwargs.get("pmh_id", None)
chunk_size=kwargs.get("chunk", 100)
limit=kwargs.get("limit", None)

iflimitisNone:
limit=float("inf")

ifsingle_id:
pmh=PmhRecord.query.filter(PmhRecord.id==single_id).scalar()

ifrecord:=PmhRecordMaker.make_record(pmh):
db.session.merge(record)
secondary_records=PmhRecordMaker.make_secondary_repository_responses(record)
forsecondary_recordinsecondary_records:
db.session.merge(secondary_record)
db.session.merge(
RecordthresherParentRecord(
record_id=secondary_record.id,
parent_record_id=record.id
 )
 )

safe_commit(db) orlogger.info("COMMIT fail")
else:
num_updated=0

whilenum_updated<limit:
start_time=time()

pmh_ids=self.fetch_queue_chunk(chunk_size)

ifnotpmh_ids:
logger.info('no queued pmh records ready to update. waiting...')
sleep(5)
continue

secondary_records= {}
parent_relationships= {}

forpmh_idinpmh_ids:
ifpmh:=PmhRecord.query.filter(PmhRecord.id==pmh_id).scalar():
ifrecord:=PmhRecordMaker.make_record(pmh):
db.session.merge(record)
record_secondary_records=PmhRecordMaker.make_secondary_repository_responses(record)
forrecord_secondary_recordinrecord_secondary_records:
secondary_records[record_secondary_record.id] =record_secondary_record
parent_relationships[record_secondary_record.id] =RecordthresherParentRecord(
record_id=record_secondary_record.id,
parent_record_id=record.id
 )

forsecondary_recordinsecondary_records.values():
print(secondary_record)
db.session.merge(secondary_record)

forparent_relationshipinparent_relationships.values():
db.session.merge(parent_relationship)

db.session.execute(
text('''
 delete from recordthresher.pmh_record_queue q
 where q.pmh_id = any(:pmh_ids)
 ''').bindparams(pmh_ids=pmh_ids)
 )

commit_start_time=time()
safe_commit(db) orlogger.info("commit fail")
logger.info(f'commit took {elapsed(commit_start_time, 2)} seconds')

num_updated+=chunk_size
logger.info(f'processed {len(pmh_ids)} PMH records in {elapsed(start_time, 2)} seconds')

deffetch_queue_chunk(self, chunk_size):
logger.info("looking for new jobs")

queue_query=text("""
 with queue_chunk as (
 select pmh_id
 from recordthresher.pmh_record_queue
 where started is null
 order by rand
 limit :chunk
 for update skip locked
 )
 update recordthresher.pmh_record_queue q
 set started = now()
 from queue_chunk
 where q.pmh_id = queue_chunk.pmh_id
 returning q.pmh_id;
 """).bindparams(chunk=chunk_size)

job_time=time()
pmh_id_list= [row[0] forrowindb.engine.execute(queue_query.execution_options(autocommit=True)).all()]
logger.info(f'got {len(pmh_id_list)} ids, took {elapsed(job_time)} seconds')

returnpmh_id_list


if__name__=="__main__":
parser=argparse.ArgumentParser()
parser.add_argument('--pmh_id', nargs="?", type=str, help="pmh_id you want to update the RT record for")
parser.add_argument('--limit', "-l", nargs="?", type=int, help="how many records to update")
parser.add_argument('--chunk', "-ch", nargs="?", default=100, type=int, help="how many records to update at once")

parsed_args=parser.parse_args()

my_queue=QueuePmhRTRecord()
my_queue.worker_run(**vars(parsed_args))