DaemonRunnerWrapper¶

Module for creating generic, callback based wrappers.

It is little bit easier (at least for me) to use than the original daemon module.

class edeposit.amqp.daemonwrapper.DaemonRunnerWrapper(pid_filename)[source]¶

Bases: object

Generic daemon class, which allows you to daemonize your script and react to events in simple callbacks.

Parameters:	pid_filename (str) – name of daemon’s PID file, which is stored in /tmp. Class automatically adds .pid suffix.

body()[source]¶

Here should be your code loop.

Note

Loop is automatically break-ed when daemon receives one of the unix signals. After that, onExit() is called.

isRunning()[source]¶

Check PID and return true, if it looks like there is already a running instance of daemon.

PID timeout can be set thru pidfile_timeout property.

onExit()[source]¶

Called when the daemon received ?SIGTERM? and is shutting down.

Warning

You should probably put something here, by default is there only shutdown message “DaemonRunnerWrapper is shutting down.”

onIsRunning()[source]¶

Oposite of onStopFail() - this callback is called if there is already a running instance of daemon.

onStopFail()[source]¶

Called when it is not possible to stop the daemon.

This kind of event typically occurs if there is no running instance of daemon and script is called with stop parameter.

run()[source]¶

Used to handle some exceptions.

Note

No, this can’t be named differently, because it is defined in original DaemonRunner object.

Warning

DO NOT OVERRIDE THIS.

run_daemon()[source]¶

Used as daemon starter.

Warning

DO NOT OVERRIDE THIS.

PikaDaemon¶

Generic AMQP blocking communication daemon server.

Usage is simple - just inherit the class and override PikaDaemon.onMessageReceived().

You can send messages back using either PikaDaemon.sendMessage() or PikaDaemon.sendResponse(). Fist one allows you to send message everywhere, second one send message to the queue defined by constructor.

class edeposit.amqp.pikadaemon.PikaDaemon(connection_param, queue, output_exchange, output_key)[source]¶

Bases: edeposit.amqp.daemonwrapper.DaemonRunnerWrapper

Pika and Daemon wrapper for handling AMQP connections.

Parameters:	connection_param (pika.ConnectionParameters) – object setting the connection queue (str) – name of queue where the daemon should listen output_exchange (str) – name of exchange where the daemon should put responses output_key (str) – routing key for output exchange

ack(ack_delivery_tag)[source]¶

Acknowledge, that message was received.

Note

This will in some cases (depends on settings of RabbitMQ) remove the message from the message queue.

body()[source]¶

This method just handles AMQP connection details and receive loop.

Warning

Don’t override this method!

onExit()[source]¶

Called when daemon is stopped. Basically just AMQP’s .close() functions to ensure clean exit.

You can override this, but don’t forget to call it thru super(), or the AMQP communication won’t be closed properly!

onMessageReceived(method_frame, properties, body)[source]¶

Callback which is called every time when message is received.

Warning

You SHOULD override this.

Note

It is expected, that method returns True, if you want to automatically ack the received message, which can be important in some situations, because otherwise the message will be held in message queue until someone will ack it.

You don’t have to return True/False - you can ack the message yourself, by calling ack().

Note

Good design choice is to ack the message AFTER you process it, to be sure, that message is processed properly and can be removed from queue.

sendMessage(exchange, routing_key, message, properties=None, UUID=None)[source]¶

With this function, you can send message to exchange.

Parameters:

exchange (str) – name of exchange you want to message to be delivered routing_key (str) – which routing key to use in headers of message message (str) – body of message properties (dict ,optional) – properties of message - if not used, or set to None, self.content_type and delivery_mode=2 (persistent) is used UUID (str, optional) – UUID of the message. If set, it is included into properties of the message.

sendResponse(message, UUID, routing_key)[source]¶

Send message to self.output_exchange with routing key self.output_key, self.content_type in delivery_mode=2.

Parameters:	message (str) – message which will be sent UUID – unique identification of message routing_key (str) – which routing key to use to send message back

AMQPDaemon¶

This module provides generic AMQP daemon and builder of common connection informations, which are defined as constants in edeposit.amqp.settings.

Daemon is used by edeposit_amqp_alephdaemon and edeposit_amqp_calibredaemon.

class edeposit.amqp.amqpdaemon.AMQPDaemon(con_param, queue, out_exch, out_key, react_fn, glob)[source]¶

Bases: edeposit.amqp.pikadaemon.PikaDaemon

Parameters:

con_param (ConnectionParameters) – see getConParams() for details queue (str) – name of the queue out_exch (str) – name of the exchange for outgoing messages out_key (str) – what key will be used to send messages back react_fn (fn) – function, which can react to messages, see Note for details glob (dict) – result of globals() call - used in deserializer to automatically build classes, which are not available in this namespace of this package

Note

react_fn parameter is expected to be function, which gets two parameters - message (some form of message, it can be also namedtuple), and UUID containing unique identificator of the message.

Example of function used as react_fn parameter:

def reactToAMQPMessage(message, UUID):
    response = None
    if message == 1:
        return 2
    elif message == "Hello":
        return "Hi"
    elif type(message) == dict:
        return {1: 2}

    raise UserWarning("Unrecognized message")

As you can see, protocol is pretty easy. You get message, to which you react somehow and return response. Thats all.

get_sendback(uuid, key)[source]¶

Return function for sending progress messages back to original caller.

Parameters:	uuid (str) – UUID of the received message. key (str) – Routing key.
Returns:	Reference to function which takes only one data argument.
Return type:	fn reference

onMessageReceived(method_frame, properties, body)[source]¶

React to received message - deserialize it, add it to users reaction function stored in self.react_fn and send back result.

If Exception is thrown during process, it is sent back instead of message.

Note

In case of Exception, response message doesn’t have useful body, but in headers is stored following (string) parameters:

• exception, where the Exception’s message is stored
• exception_type, where e.__class__ is stored
• exception_name, where e.__class__.__name__ is stored
• traceback where the full traceback is stored (contains line number)

This allows you to react to unexpected cases at the other end of the AMQP communication.

parseKey(method_frame)[source]¶

process_exception(e, uuid, routing_key, body, tb=None)[source]¶

Callback called when exception was raised.

This method serializes the exception and sends it over AMQP back to caller.

Parameters:	e (obj) – Instance of the exception. uuid (str) – UUID of the message that caused the exception to raise. routing_key (str) – Which routing key was used. body (str) – Body of the exception - the longer text. tb (str, default None) – Traceback (stacktrace)v of the exception.

edeposit.amqp.amqpdaemon.getConParams(virtualhost)[source]¶

Connection object builder.

Parameters:	virtualhost (str) – selected virtualhost in rabbitmq
Returns:	object filled by constants from edeposit.amqp.settings.
Return type:	pika.ConnectionParameters

Settings¶

Module is containing all necessary global variables for package.

Module also has ability to read user-defined data from two paths: $HOME/SETTINGS_PATH and /etc/SETTINGS_PATH.

Example of the configuration file ($HOME/edeposit/amqp.json):

{
    "RABBITMQ_HOST": "localhost",
    "RABBITMQ_PORT": "2222"
}

edeposit_amqp_alephdaemon.py script¶

$ ./edeposit_amqp_alephdaemon.py -husage: edeposit_amqp_alephdaemon.py start/stop/restart [-f] FN

Aleph communicator. This daemon provides AMQP API for aleph module.

positional arguments:
  start/stop/restart  Start/stop/restart the daemon.

optional arguments:
  -h, --help          show this help message and exit
  -f, --foreground    Run at foreground, not as daemon. If not set, script is
                      will run at background as unix daemon.

./edeposit_amqp_alephdaemon.py start
started with pid 4595

$ ./edeposit_amqp_alephdaemon.py start --foreground

$ ./edeposit_amqp_alephdaemon.py stop
No handlers could be found for logger "pika.adapters.base_connection"

edeposit_amqp_calibredaemon.py script¶

$ ./edeposit_amqp_calibredaemon.py -h
usage: edeposit_amqp_calibredaemon.py start/stop/restart [-f] FN

Calibre daemon, providing AMQP interface for edeposit.amqp.calibre.

positional arguments:
  start/stop/restart  Start/stop/restart the daemon.

optional arguments:
  -h, --help          show this help message and exit
  -f, --foreground    Run at foreground, not as daemon. If not set, script is
                      will run at background as unix daemon.

$ ./edeposit_amqp_calibredaemon.py start
started with pid 10005

$ ./edeposit_amqp_calibredaemon.py start --foreground

$ ./edeposit_amqp_calibredaemon.py stop
No handlers could be found for logger "pika.adapters.base_connection"

edeposit_amqp_ftp_monitord.py script¶

$ ./edeposit_amqp_ftp_managerd.py -h
usage: edeposit_amqp_ftp_managerd.py start/stop/restart [-f] FN

ProFTPD user manager. This script allows you to add/ remove/change users in
ProFTPD server over AMQP API.

positional arguments:
  start/stop/restart  Start/stop/restart the daemon.

optional arguments:
  -h, --help          show this help message and exit
  -f, --foreground    Run at foreground, not as daemon. If not set, script is
                      will run at background as unix daemon

./edeposit_amqp_ftp_managerd.py start
started with pid 5469

$ ./edeposit_amqp_ftp_managerd.py start --foreground

$ ./edeposit_amqp_ftp_managerd.py stop
No handlers could be found for logger "pika.adapters.base_connection"

edeposit_amqp_ftp_monitord.py script¶

./edeposit_amqp_ftp_monitord.py -h
usage: edeposit_amqp_ftp_monitord.py start/stop/restart [-f] FN

ProFTPD log monitor. This script reacts to preprogrammed events from FTP
server.

positional arguments:
  start/stop/restart    Start/stop/restart the daemon.

optional arguments:
  -h, --help            show this help message and exit
  -f, --foreground      Run at foreground, not as daemon. If not set, script
                        is will run at background as unix daemon.
  -n FILENAME, --filename FILENAME
                        Path to the log file (usually
                        /var/log/proftpd/extended.log).

$ ./edeposit_amqp_ftp_monitord.py start --filename /var/log/proftpd/extended.log
Monitoring file '/var/log/proftpd/extended.log'.
started with pid 1633

$ ./edeposit_amqp_ftp_monitord.py start --filename /var/log/proftpd/extended.log --foreground
Monitoring file '/var/log/proftpd/extended.log'.

$ ./edeposit_amqp_ftp_monitord.py stop
WARNING:pika.adapters.base_connection:Unknown state on disconnect: 0
WARNING:pika.adapters.base_connection:Unknown state on disconnect: 0

edeposit_amqp_antivirusd.py script¶

$ ./edeposit_amqp_antivirusd.py -h
usage: edeposit_amqp_antivirusd.py start/stop/restart [-f] FN

AMQP daemon for ClamAV antivirus.

positional arguments:
  start/stop/restart  Start/stop/restart the daemon.

optional arguments:
  -h, --help          show this help message and exit
  -f, --foreground    Run at foreground, not as daemon. If not set, script is
                      will run at background as unix daemon.

./edeposit_amqp_antivirusd.py start
started with pid 7195

$ ./edeposit_amqp_antivirusd.py start --foreground

$ ./edeposit_amqp_antivirusd.py stop
No handlers could be found for logger "pika.adapters.base_connection"

edeposit_amqp_harvester.py script¶

./edeposit_amqp_harvester.py
usage: edeposit_amqp_harvester.py [-h] [-u] [-r]

This script is used to send data from edeposit.amqp.harvester to AMQP queue.

optional arguments:
  -h, --help      show this help message and exit
  -u, --unittest  Perform unittest.
  -r, --harvest   Harvest all data and send them to harvester queue.

$ ./edeposit_amqp_harvester.py start --harvest

$ ./edeposit_amqp_harvester.py --unittest

edeposit_amqp_ltpd.py script¶

$ ./edeposit_amqp_ltpd.py -h
usage: edeposit_amqp_ltpd.py start/stop/restart [-f/--foreground]

AMQP daemon for LTP exporter.

positional arguments:
  start/stop/restart  Start/stop/restart the daemon.

optional arguments:
  -h, --help          show this help message and exit
  -f, --foreground    Run at foreground, not as daemon. If not set, script is
                      will run at background as unix daemon.

./edeposit_amqp_ltpd.py start
started with pid 7195

$ ./edeposit_amqp_ltpd.py start --foreground

$ ./edeposit_amqp_ltpd.py stop
No handlers could be found for logger "pika.adapters.base_connection"

edeposit_amqp_pdfgend.py script¶

$ ./edeposit_amqp_pdfgend.py -h
usage: edeposit_amqp_pdfgend.py start/stop/restart [-f/--foreground]

AMQP binding for PDF generator.

positional arguments:
  start/stop/restart  Start/stop/restart the daemon.

optional arguments:
  -h, --help          show this help message and exit
  -f, --foreground    Run at foreground, not as daemon. If not set, script is
                      will run at background as unix daemon.

./edeposit_amqp_pdfgend.py start
started with pid 7195

$ ./edeposit_amqp_pdfgend.py start --foreground

$ ./edeposit_amqp_pdfgend.py stop
No handlers could be found for logger "pika.adapters.base_connection"

edeposit_amqp_downloaderd.py script¶

$ ./edeposit_amqp_downloaderd.py -h
usage: edeposit_amqp_downloaderd.py start/stop/restart [-f/--foreground]

AMQP binding for downloader.

positional arguments:
  start/stop/restart  Start/stop/restart the daemon.

optional arguments:
  -h, --help          show this help message and exit
  -f, --foreground    Run at foreground, not as daemon. If not set, script is
                      will run at background as unix daemon.

./edeposit_amqp_downloaderd.py start
started with pid 7195

$ ./edeposit_amqp_downloaderd.py start --foreground

$ ./edeposit_amqp_downloaderd.py stop
No handlers could be found for logger "pika.adapters.base_connection"

edeposit_amqp_storaged.py script¶

..storaged

members:
undoc-members:
show-inheritance:

$ ./edeposit_amqp_storaged.py start/stop/restart [-f/--foreground]

AMQP binding for storage.

positional arguments:
  start/stop/restart  Start/stop/restart the daemon.

optional arguments:
  -h, --help          show this help message and exit
  -f, --foreground    Run at foreground, not as daemon. If not set, script is
                      will run at background as unix daemon.

./edeposit_amqp_storaged.py start
started with pid 7195

$ ./edeposit_amqp_storaged.py start --foreground

$ ./edeposit_amqp_storaged.py stop
No handlers could be found for logger "pika.adapters.base_connection"

edeposit_amqp_marcxml2modsd.py script¶

..marcxml2modsd

members:
undoc-members:
show-inheritance:

$ ./edeposit_amqp_marcxml2modsd.py marcxml2modsd.py start/stop/restart [-f/--foreground]

AMQP binding for marcxml2mods.

positional arguments:
  start/stop/restart  Start/stop/restart the daemon.

optional arguments:
  -h, --help          show this help message and exit
  -f, --foreground    Run at foreground, not as daemon. If not set, script is
                      will run at background as unix daemon.

./edeposit_amqp_marcxml2modsd.py start
started with pid 7195

$ ./edeposit_amqp_marcxml2modsd.py start --foreground

$ ./edeposit_amqp_marcxml2modsd.py stop
No handlers could be found for logger "pika.adapters.base_connection"

edeposit_amqp_aleph_link_exportd.py script¶

..aleph_link_exportd

members:
undoc-members:
show-inheritance:

$ ./edeposit_amqp_aleph_link_exportd.py start/stop/restart [-f/--foreground]

AMQP binding for aleph_link_export.

positional arguments:
  start/stop/restart  Start/stop/restart the daemon.

optional arguments:
  -h, --help          show this help message and exit
  -f, --foreground    Run at foreground, not as daemon. If not set, script is
                      will run at background as unix daemon.

./edeposit_amqp_aleph_link_exportd.py start
started with pid 7195

$ ./edeposit_amqp_aleph_link_exportd.py start --foreground

$ ./edeposit_amqp_aleph_link_exportd.py stop
No handlers could be found for logger "pika.adapters.base_connection"

edeposit_amqp_tool.py script¶

AMQP tool used for debugging and automatic RabbitMQ schema making.

edeposit_amqp_tool.create_blocking_connection(*args, **kwargs)[source]¶

Return properly created blocking connection.

Parameters:	host (str) – Host as it is defined in get_amqp_settings().

Uses edeposit.amqp.amqpdaemon.getConParams().

edeposit_amqp_tool.create_schema(host)[source]¶

Create exchanges, queues and route them.

Parameters:	host (str) – One of the possible hosts.

edeposit_amqp_tool.get_list_of_hosts()[source]¶

Returns:	List of strings with names of possible hosts.
Return type:	list

edeposit_amqp_tool.main()[source]¶

edeposit_amqp_tool.receive(host, timeout)[source]¶

Print all messages in queue.

Parameters:	host (str) – Specified –host. timeout (int) – How log should script wait for message.

edeposit_amqp_tool.send_message(host, data, timeout=None, properties=None)[source]¶

Send message to given host.

Parameters:	host (str) – Specified host: aleph/ftp/whatever available host. data (str) – JSON data. timeout (int, default None) – How much time wait for connection.

edeposit_amqp_tool.test_virtualhost(fn)[source]¶