olivia_finder.myrequests.job

 1from __future__ import annotations
 2from typing import Dict, Optional
 3import requests
 4
 5class RequestJob:
 6    '''
 7    A class to represent a job to be done by a worker.
 8            
 9    '''
10
11    FINALIZE_KEY = "FINALIZE"
12
13    def __init__(self, key: object, url: Optional[str], params: Optional[Dict[str, str]] = None):
14        '''
15        Constructor
16
17        Parameters
18        ----------
19        key : object
20            Key of the job
21        url : Optional[str]
22            Url of the request
23        params : Dict[str, str]
24            Parameters of the request
25        '''
26
27        self.key: object = key
28        self.url: Optional[str] = url
29        self.response: Optional[requests.Response] = None
30        self.params: Optional[Dict[str, str]] = params
31
32    def set_response(self, job_response: requests.Response):
33        '''
34        Set the response of the job
35
36        Parameters
37        ----------
38        job_response : requests.Response
39            Response of the request
40        '''
41
42        self.response = job_response
43
44    @staticmethod
45    def end_job_signal() -> RequestJob:
46        '''
47        Returns a job that signals the worker to end
48
49        Returns
50        -------
51        RequestJob
52            A job that signals the worker to end
53            
54        '''
55        
56        return RequestJob(
57            key=RequestJob.FINALIZE_KEY,
58            url=None
59        )