ddmee / polling2 Goto Github PK
View Code? Open in Web Editor NEWPowerful polling utility in Python
Home Page: https://polling2.readthedocs.io/
License: MIT License
Powerful polling utility in Python
Home Page: https://polling2.readthedocs.io/
License: MIT License
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.
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.
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
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.
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.
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'
>
>
Lines 166 to 171 in b9244a1
$ 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.
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.
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.