mirror of
https://github.com/Blockstream/satellite-api.git
synced 2025-03-13 11:35:20 +01:00
9d421771e7
- Preserve the SQLite database and use SQLAlchemy to wrap db interactions. - Use Alembic for database migrations. - Organize all the python modules on the new server/ directory. - Use pytest for unit tests and organize test modules at server/tests/.
40 lines
1 KiB
Python
40 lines
1 KiB
Python
import logging
|
|
import threading
|
|
import time
|
|
|
|
|
|
class Worker:
|
|
"""Worker Class
|
|
|
|
A class to execute periodic background tasks.
|
|
|
|
Args:
|
|
period : Worker period in seconds.
|
|
fcn : Function to execute every period.
|
|
args : Tuple with arguments for the given function.
|
|
name : Optional worker name.
|
|
|
|
"""
|
|
def __init__(self, period, fcn, args, name=""):
|
|
assert (isinstance(args, tuple))
|
|
self.period = period
|
|
self.fcn = fcn
|
|
self.args = args
|
|
self.name = name
|
|
self.enable = True
|
|
logging.info(f"Starting worker: {self.name}")
|
|
self.thread = threading.Thread(target=self.loop, daemon=True)
|
|
self.thread.start()
|
|
|
|
def loop(self):
|
|
next_call = time.time()
|
|
while (self.enable):
|
|
self.fcn(*self.args)
|
|
next_call += self.period
|
|
sleep = next_call - time.time()
|
|
if (sleep > 0):
|
|
time.sleep(sleep)
|
|
|
|
def stop(self):
|
|
logging.info(f"Stopping worker: {self.name}")
|
|
self.enable = False
|