queick's Introduction

Queick

A simple inmemory job-queue manager for Python.

Feature

Written in Python only standard libraries
Job-queue manager without redis
Working for low-spec machines
Retry
Retry when network available
Scheduling

Installation

Python version >= 3.6 is required.

pip install queick

Usage

First, launch queick worker.

$ queick

Second, prepare a job file (jobfunc.py) and an application (test.py).

# jobfunc.py
import time
def function(arg):
    time.sleep(1)
    print(arg)

# test.py
from queick import JobQueue
from jobfunc import function
from time import time

q = JobQueue()
q.enqueue(function, args=("hello",))
q.enqueue_at(time() + 5, function, args=("world",)) # Run after 5 seconds

st = SchedulingTime()
st.every(minutes=1).starting_from(time.time() + 10)
q.cron(st, function, args=(1, 2,)) # Run after 10 seconds and every 1 minute

Third, run the application.

$ python test.py

For more detailed information, please refer the documentation.

Retry on network available

Jobs inside the failed queue will be dequeued when the network status changes from disconnected to connected.

Some setups are needed to use the retry mode. First, launch queick worker with --ping-host options (details below).

$ queick --ping-host asmsuechan.com # Please prepare your own ping server, do not use this.

Second, pass an option to the method.

q.enqueue(function, args=("hello",), retry_on_network_available=True)

Options

There are some options for queick worker.

name	default	description
-debug	False	if set, detailed logs will be shown
--ping-host	None	hostname for NetworkWatcher to check if the machine has the internet connection
--ping-port	80	port number for NetworkWatcher
--log-filepath	None	logfile to save all the worker log

An example usage is below:

$ queick -debug --ping-host asmsuechan.com

Testing

Unit test:

$ python -m unittest

Integration test:

$ docker build -t queick-test .
$ docker run --rm -it queick-test:latest

Development

Build queick for development.

$ python setup.py develop

Deployment

Deployed at https://pypi.org/project/queick/.

$ python setup.py sdist
$ twine upload dist/*

queick's People

Contributors

Stargazers

Watchers

queick's Issues

worker could not find the job

Hi, I followed your simplist demo on the introduction page,but I got this error.

'Queick worker could not find the job. Please check your worker's launching directory.')
queick.exceptions.NoSuchJobError: Queick worker could not find the job. Please check your worker's launching directory.

Do I need to put the job file to the directory where Queick installed?

cannot limit the workers

Hi! sorry if I'm misunderstanding this completely but I'm trying to create a way to queue jobs in my Raspberry Pi and only have them running sequentially. My test code is like this:

# test.py
from queick import JobQueue
from jobfunc import function
from time import time

q = JobQueue()
for i in range(1,50):
    q.enqueue(function, args=(f"hello{i}",), priority=i, max_workers=1)

# jobfunc.py
def function(arg):
    time.sleep(5)
    print(arg)

I was expecting this code would show me one hello{i} every 5 seconds as I'm only passing max_workers as 1 and all jobs have a different priority. Instead, after 5 seconds all jobs finish at the same time.
Any idea of what am I doing wrong? Thanks!

Recommend Projects

asmsuechan / queick Goto Github PK