ddmee / polling2 Goto Github PK

Hi,

Can you tell me how I can abort, exit, kill, cancel or any other method the poll ?

Because when I use Ctrl+c or stop my script, I wai a long time before I can regain control of the console.

Thank you for this good project and support.

I want to poll the return value of a function:

def myfunction():
   df=df_function()
   global value
   value=df[df['date']==date]
   return True

polling2.poll(
   lambda: myfunction == True,
   step=30,
   timeout=3600)
   print(value)

But does not work: the polling script runs indefinitely.

When the timeout argument is set to 0, polling will run forever due to this:

timeout = time.time() + timeout if timeout else None

evaluating the timeout argument to False, which in turn sets timeout to None, which will result in the loop never throwing a TimeoutException:

if timeout is not None and time.time() >= timeout

Is this behavior intentional, especially considering that this works without explicitly setting poll_forever to True? My assumption was that timeout=0 would run the target function once and then immediately timeout if check_success doesn't pass.

$ bandit polling2.py
[main]	INFO	profile include tests: None
[main]	INFO	profile exclude tests: None
[main]	INFO	cli include tests: None
[main]	INFO	cli exclude tests: None
[main]	INFO	running on Python 3.8.10
[node_visitor]	WARNING	Unable to find qualified name for module: polling2.py
Run started:2022-10-14 22:09:04.481341

Test results:
>> Issue: [B101:assert_used] Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.
   Severity: Low   Confidence: High
   CWE: CWE-703 (https://cwe.mitre.org/data/definitions/703.html)
   Location: polling2.py:166:4
   More Info: https://bandit.readthedocs.io/en/1.7.4/plugins/b101_assert_used.html
165	
166	    assert (timeout is not None or max_tries is not None) or poll_forever, \
167	        ('You did not specify a maximum number of tries or a timeout. Without either of these set, the polling '
168	         'function will poll forever. If this is the behavior you want, pass "poll_forever=True"')
169	
170	    assert not ((timeout is not None or max_tries is not None) and poll_forever), \

--------------------------------------------------
>> Issue: [B101:assert_used] Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.
   Severity: Low   Confidence: High
   CWE: CWE-703 (https://cwe.mitre.org/data/definitions/703.html)
   Location: polling2.py:170:4
   More Info: https://bandit.readthedocs.io/en/1.7.4/plugins/b101_assert_used.html
169	
170	    assert not ((timeout is not None or max_tries is not None) and poll_forever), \
171	        'You cannot specify both the option to poll_forever and max_tries/timeout.'
172	
173	    kwargs = kwargs or dict()

--------------------------------------------------

Code scanned:
	Total lines of code: 161
	Total lines skipped (#nosec): 0
	Total potential issues skipped due to specifically being disabled (e.g., #nosec BXXX): 0

Run metrics:
	Total issues (by severity):
		Undefined: 0
		Low: 2
		Medium: 0
		High: 0
	Total issues (by confidence):
		Undefined: 0
		Low: 0
		Medium: 0
		High: 2
Files skipped (0):

Consider raising ValueError instead.

I would like to use polling2 to send requests and take an action when status requirements are valid and also would like to monitor and store the response codes from the target endpoint. It could be either written as a log file to troubleshoot. Log part I added recently but it does not work like this to see the what is response code or what this function actually doing.
Could u help me understand how can I implement the response code from the target URL? Thanks in advance

my current usage of a piece of code like this

continueToPoll = True
while continueToPoll:
     try:
         r = polling2.poll(lambda: requests.request("GET", Url, auth=(userName, password)),step=60,poll_forever=True,verify=False,log_error=logging.DEBUG)

     except:
         print("Connection refused by the server..")
         weekNumber = datetime.datetime.today().weekday()
         start = datetime.time(5, 0, 0)
         end = datetime.time(18, 0, 0)
         current = datetime.datetime.now().time()

         if current>= start and current <= end:
           if weekNumber < 5:
               time.sleep(300)
           else:
                time.sleep(3600)
         else:
              time.sleep(3600)
         continue

     x = r.json()
     r.close()
     currentChangeStatus = x['fields']['status']['name']
     if currentStatus == expectedStatus: continueToPoll = False

I am using poetry and nix to do package management. I run the following command to setup: poetry lock -vv --no-update && nix build

It fails with the following error:

error: builder for '/nix/store/jykc399kb5ylimj9sric45mg1f35d0da-python3.7-polling2-0.5.0.drv' failed with exit code 2;
       last 10 log lines:
       >   File "<frozen importlib._bootstrap>", line 1006, in _gcd_import
       >   File "<frozen importlib._bootstrap>", line 983, in _find_and_load
       >   File "<frozen importlib._bootstrap>", line 953, in _find_and_load_unlocked
       >   File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
       >   File "<frozen importlib._bootstrap>", line 1006, in _gcd_import
       >   File "<frozen importlib._bootstrap>", line 983, in _find_and_load
       >   File "<frozen importlib._bootstrap>", line 965, in _find_and_load_unlocked
       > ModuleNotFoundError: No module named 'setuptools'
       >
       >

Port from request in the original repo, before I noticed this repo exists.
justiniso/polling#24

Would be great if check_success would have the ability to not just receive a callback, but also a list of callbacks.

check_success: Union[Callback, List[Callback]]

The (very real) use case I have here is that when making a REST call, there are different verifications that need to be made before continuing the code. These verifications are mix-and-match, depending on the scenario and I would love to do something like:

result = polling.poll(
    client.get_data,
    check_success=[
        lambda response: error_check(response, 200),
        validate_data,
        lambda response: check_something_else(response, some_other_args),
    ]
)

So check_success would only pass if all callables return True (if list, otherwise call the one callable passed in check_success, as-is now).

For example, if a user writes a call:

polling2.poll(target=wait_for_something, step=1, timeout=60)

Generally a user will want to log how long the polling is going to take. B/c wait_for_something might hang.

So the user should typically write

logger.debug('Waiting 60secs for wait_for_something()')
polling2.poll(target=wait_for_something, step=1, timeout=60)

We can just make poll() log the automatically for the user.

Instead of calling the poll function directly, it would be nice to be able to decorate a function with an @poll decorator instead. A web search yields https://github.com/benjamin-hodgson/poll which was last committed to 6 years ago, but which implemented a polling decorator and could serve as a reference.