pystalk documentation

pystalk is a simple Python module for interacting with the beanstalk task queueing daemon. It doesn’t provide much magic, but is suitable for building all sorts of functionality on top of.

This work is available under the terms of the ISC License.

See also

https://github.com/easypost/beancmd

Contents

• pystalk ChangeLog

Members

class pystalk.BeanstalkClient(host: str, port: int = 11300, socket_timeout: float | None = None, auto_decode: bool = False)

Simple wrapper around the Beanstalk API.

Parameters:

• host (str) – Hostname or IP address to connect to

• port (int) – Port to connect to

• socket_timeout (float) – Timeout to set on the socket.

• auto_decode (bool) – Attempt to decode job bodies as UTF-8 when reading them

Doesn’t provide any fanciness for writing consumers or producers. Just lets you invoke methods to call beanstalk functions.

Warning

Setting socket timeout to a value lower than the value you pass to blocking functions like reserve_job() will cause errors!

bury_job(job_id, pri=65536)

Mark the given job_id as buried. The job must have been previously reserved by this connection

Parameters:

• job_id – Job to bury

• pri (int) – Priority for the newly-buried job. If not passed, will keep its current priority

close()

Close any open connection to the Beanstalk server.

This object is still safe to use after calling close() ; it will automatically reconnect and re-establish any open watches / uses.

It is a logic error to close the connection while you have a reserved job

delete_job(job_id)

Delete the given job id. The job must have been previously reserved by this connection

classmethod from_uri(uri, socket_timeout=None, auto_decode=False)

Construct a synchronous Beanstalk Client from a URI.

The URI may be of the form beanstalk://host:port or beanstalkd://host:port

IPv6 literals must be wrapped in brackets as per RFC 2732.

ignore(tube)

Remove the given tube from the watchlist.

Parameters:

tube – Name of tube to remove from the watchlist

If all tubes are ignore() d, beanstalk will auto-add “default” to the watchlist to prevent the list from being empty. See watch() for more unformation.

kick_job(job_id)

Kick the given job id. The job must either be in the DELAYED or BURIED state and will be immediately moved to the READY state.

kick_jobs(num_jobs)

Kick some number of jobs from the buried queue onto the ready queue.

Parameters:

num_jobs (int) – Number of jobs to kick

If not that many jobs are in the buried queue, it will kick as many as it can.

list_tubes()

Return a list of tubes that this beanstalk instance knows about

Return type:

list of tubes

pause_tube(tube, delay=3600)

Pause a tube for some number of seconds, preventing it from issuing jobs.

Parameters:

delay (int) – Time to pause for, in seconds

There is no way to permanently pause a tube; passing 0 for delay actually un-pauses the tube.

See also

unpause_tube()

peek_buried()

Peek at the top job on the buried queue

peek_buried_iter()

Peek at buried jobs in sequence

peek_delayed()

Peek at the job job on the delayed queue

peek_delayed_iter()

Peek at delayed jobs in sequence

peek_ready()

Peek at the job job on the ready queue.

Return type:

Job

peek_ready_iter()

Peek at ready jobs in sequence

ping() → bool

Check whether a connection to the beanstalk server can be opened.

put_job(data: str | bytes, pri: int = 65536, delay: int = 0, ttr: int = 120)

Insert a new job into whatever queue is currently USEd

Parameters:

• data (Text (either str which will be encoded as utf-8, or bytes which are already utf-8) – Job body

• pri (int) – Priority for the job

• delay (int) – Delay in seconds before the job should be placed on the ready queue

• ttr (int) – Time to reserve (how long a worker may work on this job before we assume the worker is blocked and give the job to another worker

See also

put_job_into()

Put a job into a specific tube

using()

Insert a job using an external guard

put_job_into(tube_name: str, data: str | bytes, pri: int = 65536, delay: int = 0, ttr: int = 120)

Insert a new job into a specific queue. Wrapper around put_job().

Parameters:

• tube_name (str) – Tube name

• data (Text (either str which will be encoded as utf-8, or bytes which are already utf-8) – Job body

• pri (int) – Priority for the job

• delay (int) – Delay in seconds before the job should be placed on the ready queue

• ttr (int) – Time to reserve (how long a worker may work on this job before we assume the worker is blocked and give the job to another worker

See also

put_job()

Put a job into whatever the current tube is

using()

Insert a job using an external guard

release_job(job_id, pri=65536, delay=0)

Put a job back on the queue to be processed (indicating that you’ve aborted it)

You can only release a job which you have reserved using reserve_job() or reserve_iter().

Parameters:

• job_id – Job ID to return

• pri (int) – New priority (if not passed, will use old priority)

• delay (int) – New delay for job (if not passed, will use 0)

reserve_iter()

Reserve jobs as an iterator. Ends iteration when there are no more jobs immediately available

reserve_job(timeout=5)

Reserve a job for this connection. Blocks for TIMEOUT secionds and raises TIMED_OUT if no job was available

Parameters:

timeout (int) – Time to wait for a job, in seconds.

stats()

Return a dictionary with a bunch of instance-wide statistics

Return type:

dict

stats_job(job_id)

Fetch statistics about a single job

Return type:

dict

stats_tube(tube_name)

Fetch statistics about a single tube

Parameters:

tube_name – Tube to fetch stats about

Return type:

dict

unpause_tube(tube)

Unpause a tube which was previously paused with pause_tube().

See also

pause_tube()

use(tube)

Start producing jobs into the given tube.

Parameters:

tube – Name of the tube to USE

Subsequent calls to put_job() insert jobs into this tube.

using(tube)

Context-manager to insert jobs into a specific tube

Parameters:

tube – Tube to insert to

Yields out an instance of BeanstalkInsertingProxy to insert items into that tube

See also

use()

Change the default tube

put_job()

Put a job into whatever the current tube is

put_job_into()

Put a job into a specific tube

watch(tube)

Add the given tube to the watchlist.

Parameters:

tube – Name of the tube to add to the watchlist

Note: Initially, all connections are watching a tube named “default”. If you manually call watch(), we will un-watch the “default” tube. To keep it in your list, first call watch() with the other tubes, then call watch() with “default”.

property watchlist

class pystalk.BeanstalkError(message: str | bytes)

class pystalk.BeanstalkConnectionError(host, port, err)

Raised when the underlying socket connection to beanstalkd fails.

Distinguishes connection-level failures (e.g. connection refused, timeout, DNS errors) from beanstalk protocol errors. Carries the host and port that were being connected to, and preserves the original socket error both on err and via implicit exception chaining (__context__).

This is a subclass of BeanstalkError so that existing except BeanstalkError callers continue to catch connection failures.

class pystalk.client.Job(job_id, job_data)

Structure holding a job returned from Beanstalk

Variables:

• job_id – Opaque identifier for the job (to be passed to BeanstalkClient.release_job() or BeanstalkClient.stats_job()).

• job_data – Blob of the data. str if BeanstalkClient.auto_decode is True; otherwise bytes

class pystalk.client.BeanstalkInsertingProxy(beanstalk_client: BeanstalkClient, tube: str)

Proxy object yielded by BeanstalkClient.using()

put_job(data: str | bytes, pri: int = 65536, delay: int = 0, ttr: int = 120)

Method to insert a job into the tube selected with BeanstalkClient.using().

Parameters:

• data (Text (either str which will be encoded as utf-8, or bytes which are already utf-8) – Job body

• pri (int) – Priority for the job

• delay (int) – Delay in seconds before the job should be placed on the ready queue

• ttr (int) – Time to reserve (how long a worker may work on this job before we assume the worker is blocked and give the job to another worker

Indices and tables

• Index

• Module Index

• Search Page