olivia_finder.myrequests.request_handler

  1import queue
  2import tqdm
  3import gc
  4from typing import List, Optional
  5from .job import RequestJob
  6from .request_worker import RequestWorker
  7from ..utilities.logger import MyLogger
  8from ..utilities.config import Configuration
  9
 10class RequestHandler:
 11    '''
 12    Main class of the myrequests package. It handles the requests to the server using concurrent workers.
 13
 14    '''
 15
 16    PARALLEL_WORKERS = 8
 17
 18    def __init__(self):
 19        '''
 20        Constructor for RequestHandler class
 21        '''
 22
 23        # Get logger name from config file
 24        # self.logger = MyLogger.get_logger(Configuration().get_key('logger_myrequests', 'name'))
 25        self.logger = MyLogger.get_logger('logger_myrequests')
 26        self.logger.debug("Creating RequestHandler object")
 27
 28        # Initialize 
 29        self.jobs_queue = queue.Queue()
 30        self.num_workers = 1
 31        self.workers: List[RequestWorker] = []
 32
 33    def _clear(self):
 34        '''
 35        Reset the RequestHandler object
 36        '''
 37        
 38        # Delete the objects
 39        del self.jobs_queue
 40        del self.workers
 41        gc.collect()
 42
 43        # Initialize        
 44        self.jobs_queue = queue.Queue()        
 45        self.num_workers = 1
 46        self.workers: List[RequestWorker] = []
 47
 48    def _setup_jobs(
 49            self, 
 50            request_jobs: List[RequestJob], 
 51            num_workers: int, 
 52            progress_bar: Optional[tqdm.tqdm] = None) -> None:
 53        '''
 54        Setup the jobs
 55
 56        Parameters
 57        ----------
 58        request_jobs : List[RequestJob]
 59            List of RequestJob objects
 60        num_workers : int
 61            Number of workers
 62        progress_bar : Optional[tqdm.tqdm], optional
 63            Progress bar, by default None
 64        '''
 65        
 66        # Enqueue jobs
 67        for job in request_jobs:
 68            # job.progress_bar = progress_bar
 69            self.jobs_queue.put(job)
 70
 71        self.logger.debug(f"Jobs queue size: {self.jobs_queue.qsize()}")
 72
 73        # Workers keep working till they receive an exit string
 74        # So we need to add the number of workers times the exit string to the queue
 75        self.num_workers = num_workers
 76        for _ in range(self.num_workers):
 77            self.jobs_queue.put(RequestJob.end_job_signal())
 78
 79        # Create workers and add to the queue
 80        self.logger.debug("Creating workers")
 81
 82        for i in range(self.num_workers):
 83            worker = RequestWorker(i, self.jobs_queue, progress_bar)
 84            self.workers.append(worker)
 85
 86        self.logger.debug(f"Created workers: {len(self.workers)}")
 87
 88    def _setup_job(self, request_job: RequestJob):
 89        '''
 90        Setup a single job
 91
 92        Parameters
 93        ----------
 94        request_job : RequestJob
 95            The RequestJob object
 96        '''
 97        
 98        # Enqueue jobs
 99        self.jobs_queue.put(request_job)
100
101        # Workers keep working till they receive an exit string
102        # So we need to add the number of workers times the exit string to the queue
103        self.jobs_queue.put(RequestJob.end_job_signal())
104
105        # Create workers and add to the queue
106        self.workers.append(
107            RequestWorker(0, self.jobs_queue)
108        )
109        self.logger.debug("Job created")
110
111    def do_requests(
112            self, 
113            request_jobs: List[RequestJob], 
114            num_workers: int = PARALLEL_WORKERS, 
115            progress_bar: Optional[tqdm.tqdm] = None) -> List[RequestJob]:
116        '''
117        Do the requests
118
119        Parameters
120        ----------
121        request_jobs : List[RequestJob]
122            List of RequestJob objects
123        num_workers : int, optional
124            Number of workers, by default PARALLEL_WORKERS
125        progress_bar : Optional[tqdm.tqdm], optional
126            Progress bar, by default None
127        
128        Returns
129        -------
130        list
131            List of RequestJob objects
132
133        Examples
134        --------
135        >>> rh = RequestHandler(jobs)
136        >>> results = rh.do_requests()
137        >>> for job in results:
138        >>>     print(f'key: {job.key}, url: {job.url}, response: {job.response}')
139        '''
140
141        # Clear the RequestHandler object
142        self._clear()
143
144        self.logger.info("Starting requests")
145        self.logger.debug(f"Number of jobs: {len(request_jobs)}")
146
147        # Setup jobs
148        self._setup_jobs(request_jobs, num_workers, progress_bar)
149        
150        # Start workers
151        for worker in self.workers:
152            self.logger.debug(f"Starting worker {worker.worker_id}")
153            worker.start()
154            
155        # Join workers to wait till they finished
156        for worker in self.workers:
157
158            worker.join()
159            self.logger.debug(f"Joining worker {worker.worker_id}")
160
161        # Combine results from all workers
162        workers_finalized_jobs = []
163        for worker in self.workers:
164            self.logger.debug(f"Worker {worker.worker_id} finished")
165            workers_finalized_jobs.extend(worker.my_jobs.copy())
166
167        self.logger.info("All requests finished")
168
169        # Clear memory
170        self._clear()
171
172        return workers_finalized_jobs
173
174    def do_request(self, job: RequestJob):
175        '''
176        Do a single request
177        
178        Parameters
179        ----------
180        job : RequestJob
181            The RequestJob object
182
183        Returns
184        -------
185        RequestJob
186            The RequestJob object with the response
187
188        Examples
189        --------
190        >>> rh = RequestHandler()
191        >>> job = RequestJob("single_job", "https://www.google.com")
192        >>> result = rh.do_request(job)
193        >>> print(f'key: {result.key}, url: {result.url}, response: {result.response}')
194        '''
195
196        # Clear the RequestHandler object
197        self._clear()
198
199        # Setup job
200        self._setup_job(job)
201
202        # Start worker
203        worker = RequestWorker(0, self.jobs_queue)
204        self.logger.debug(f"Starting worker {worker.worker_id}")
205
206        self.logger.info(f"Starting request for {job.key}: {job.url}")
207        worker.start()
208
209        # Join worker to wait till it finished
210        worker.join()
211        self.logger.debug(f"Joining worker {worker.worker_id}")
212        self.logger.info(f"Request for {job.key}: {job.url} finished")
213
214        if worker.my_jobs[0].response is None:
215            self.logger.info(f"Request for {job.key}: {job.url} failed: response is None")
216
217        # Return the job
218        return worker.my_jobs[0]
219