3
\$\begingroup\$

PostgreSQL, Python 3.5, SQLAlchemy

I have a main database which has a journals table containing 30k+ journals. For each journal I have a RSS feed URL, using which I can parse and get new articles. There are some mandatory fields if those are missing for any of the articles then there is a third party API which I can use to retrieve additional info about the article. I need to write a python script which will run continuously and loop through the journals retrieved from main DB multiple times each day.

The attributes of the articles retrieved need to be validated before saving in a FeedResponse DB. All the valid articles will be stored in articles table and the failed articles will be saved in error table along with the error message .And eventually only the successful articles will be sent to the main DB.

Now I need some help with architecting the solution for this. What I have so far is: There will be multiple worker threads which will retrieve data from main DB and parse the rss feed to retrieve list of articles, then validate the articles and call 3rd party API wherever needed.After cleaning up and validation, it will save all the articles retrieved from that journal in FeedResponse DB.Flow Chart

My Issues:

  • Not sure if threading is the best way to achieve this or multiprocessing or anything else. I need fast processing but do not want to end up in a deadlock situation.
  • The way I have implemented threading is that efficient or I need to refactor it to break it down.
  • Should I retrieve all journals in a single call and then loop through those in memory or each worker thread should retrieve one from main DB
  • How do I ensure script continues to run and that there are no rogue/blocked threads

For brevity I have only mentioned 2 parsers but there are a couple of other parsers as well.

My Implementation

I have a Process table in FeedResponse DB which is used to identify which journal needs to be picked up by the worker thread. Schema of the DB is as follows:

class Process(Base):
   __tablename__ = 'process'
   id = Column(Integer, primary_key=True, autoincrement=True)
   worker_id = Column(Integer)
   journal_id  = Column(Integer)
   time_started = Column(DateTime(timezone=True), nullable=True)
   time_finished = Column(DateTime(timezone=True), nullable=True)
   is_finished = Column(Boolean)

manage_worker.py

def start_workers():
   count = int(sys.argv[1])
   if count > 1:
       for i in range(0, count):
           worker = Worker(i)
           worker_pool.append(worker)
   else:
       w = Worker(0, True)
       worker_pool.append(w)
   for w in worker_pool:
       w.start()

worker.py

class Worker(Thread):
   def __init__(self, worker_id, run_single=False):
       self.worker_id = worker_id
       self.current_job_journal_id = None
       self.current_job_row_id = None
       self.current_highest_id = 0
       self.is_working = False
       self.run_single = run_single
       threading.Thread.__init__(self)

   def run(self):
       self.connect_db() #Start a session with both the DBs: main DB and FeedResponseDB
       self.start_work()

   def start_work(self):
       while True:
           if not self.is_working:
               self.is_working = True
               # the max journal id from main DB is assigned to max journal id
               max_journal_id = self.get_max_journal_id()

               # tuple of new inserted row id from process table in FeedResponseDB
               new_process_row_id, new_journal_id = self.create_new_job(max_journal_id[0])
               # the current job id is assigned as the new journal id
               self.current_job_journal_id = new_journal_id
               self.current_job_row_id = new_process_row_id

               # journal data is the row from main DB returned when queried by the new journal id (which is the max journal id from main DB)
               journal_data = self.get_journal_data(new_journal_id)
               self.parse(journal_data, self.handle_is_finished)

   def parse(self, journal_data, handle_is_finished):
       time.sleep(2)       
       articles = RssParser(journal_data.rss_url, journal_data.id)
       for a in articles:
            If data_incomplete: #If article data is incomplete
                  Article = ApiParser(a)
                  # update the entry in the list of articles
       validate_articles(articles)
       handle_is_finished() 

   def create_new_job(self, max_journal_id, last_call_id=None):
       if last_call_id is None:
           # get the latest parse date from FeedResponse DB
           # get the latest id from the latest parse date from FeedResponse DB
               max_process = db.session.query(Process).filter(Process.time_started == max_date).first()
               max_process_journal_id = max_process.journal_id
           else:
               max_process_journal_id = 0
       else:
           # if last_call_id has a value, make it the latest process journal id from the  table
           max_process_journal_id = last_call_id

       # add 1 to the latest process journal id
       max_process_journal_id += 1

       # check if our max journal id is higher than the highest in main db, and if so, reset to 1
       if max_process_journal_id > max_journal_id:
           max_process_journal_id = 1

       #insert process in FeedResponse DB
       new_max_process_id = db.session.execute(
           '''
           INSERT INTO process (worker_id, journal_id, time_started, time_finished, is_finished)
           SELECT
           {} as worker_id,
           CASE
               WHEN {} <= {} THEN {}
               ELSE 1
           END as journal_id,
           clock_timestamp() as time_started,
           null as time_finished,
           'FALSE' as is_finished
           RETURNING id
           '''
           .format(self.worker_id,
                   max_process_journal_id,
                   max_journal_id,
                   max_process_journal_id).first().id
           return (new_max_process_id, max_process_journal_id)

   def handle_is_finished(self):

       # print("FINISHING PARSE PROCESS FOR ROW ID: ------->", self.current_job_row_id)
       current_process = db..session.query(Process).filter(Process.id == self.current_job_row_id).first()
       current_process.time_finished = datetime.datetime.now()
       current_process.is_finished = True
       self.current_job_journal_id = None
       self.current_job_row_id = None
       self.disconnect_db()
       self.is_working = False
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.