ocrd_network.rabbitmq_utils package

ocrd_network.rabbitmq_utils.check_if_queue_exists(logger: Logger, rmq_data: Dict, processor_name: str) → bool[source]
ocrd_network.rabbitmq_utils.connect_rabbitmq_consumer(logger: Logger, rmq_data: Dict) → RMQConsumer[source]
ocrd_network.rabbitmq_utils.connect_rabbitmq_publisher(logger: Logger, rmq_data: Dict, enable_acks: bool = True) → RMQPublisher[source]
ocrd_network.rabbitmq_utils.create_message_queues(logger: Logger, rmq_publisher: RMQPublisher, queue_names: List[str]) → None[source]
ocrd_network.rabbitmq_utils.verify_and_parse_mq_uri(rabbitmq_address: str)[source]

Check the full list of available parameters in the docs here: https://pika.readthedocs.io/en/stable/_modules/pika/connection.html#URLParameters

ocrd_network.rabbitmq_utils.verify_rabbitmq_available(logger: Logger, rabbitmq_address: str) → None[source]
class ocrd_network.rabbitmq_utils.RMQConsumer(host: str = 'localhost', port: int = 5672, vhost: str = '/')[source]

Bases: RMQConnector

authenticate_and_connect(username: str, password: str) → None[source]
setup_defaults() → None[source]
get_one_message(queue_name: str, auto_ack: bool = False) → Any | None[source]
configure_consuming(queue_name: str, callback_method: Any) → None[source]
start_consuming() → None[source]
get_waiting_message_count() → int | None[source]
ack_message(delivery_tag: int) → None[source]
class ocrd_network.rabbitmq_utils.RMQConnector(host: str = 'localhost', port: int = 5672, vhost: str = '/')[source]

Bases: object

close_connection(reply_code: int = 200, reply_text: str = 'Normal shutdown')[source]
static declare_and_bind_defaults(connection: BlockingConnection, channel: BlockingChannel) → None[source]
static open_blocking_connection(credentials: PlainCredentials, host: str = 'localhost', port: int = 5672, vhost: str = '/') → BlockingConnection[source]
static open_blocking_channel(connection: BlockingConnection) → BlockingChannel | None[source]
static exchange_bind(channel: BlockingChannel, destination_exchange: str, source_exchange: str, routing_key: str, arguments: Any | None = None) → None[source]
static exchange_declare(channel: BlockingChannel, exchange_name: str, exchange_type: str, passive: bool = False, durable: bool = False, auto_delete: bool = False, internal: bool = False, arguments: Any | None = None) → None[source]
static exchange_delete(channel: BlockingChannel, exchange_name: str, if_unused: bool = False) → None[source]
static exchange_unbind(channel: BlockingChannel, destination_exchange: str, source_exchange: str, routing_key: str, arguments: Any | None = None) → None[source]
static queue_bind(channel: BlockingChannel, queue_name: str, exchange_name: str, routing_key: str, arguments: Any | None = None) → None[source]
static queue_declare(channel: BlockingChannel, queue_name: str, passive: bool = False, durable: bool = False, exclusive: bool = False, auto_delete: bool = False, arguments: Any | None = None) → None[source]
static queue_delete(channel: BlockingChannel, queue_name: str, if_unused: bool = False, if_empty: bool = False) → None[source]
static queue_purge(channel: BlockingChannel, queue_name: str) → None[source]
static queue_unbind(channel: BlockingChannel, queue_name: str, exchange_name: str, routing_key: str, arguments: Any | None = None) → None[source]
create_queue(queue_name: str, exchange_name: str | None = 'ocrd-network-default', exchange_type: str | None = 'direct', passive: bool = False) → None[source]
static set_qos(channel: BlockingChannel, prefetch_size: int = 0, prefetch_count: int = 1, global_qos: bool = False) → None[source]
static confirm_delivery(channel: BlockingChannel) → None[source]
static basic_publish(channel: BlockingChannel, exchange_name: str, routing_key: str, message_body: bytes, properties: BasicProperties) → None[source]
class ocrd_network.rabbitmq_utils.RMQPublisher(host: str = 'localhost', port: int = 5672, vhost: str = '/')[source]

Bases: RMQConnector

authenticate_and_connect(username: str, password: str) → None[source]
setup_defaults() → None[source]
publish_to_queue(queue_name: str, message: bytes, exchange_name: str | None = 'ocrd-network-default', properties: BasicProperties | None = None) → None[source]
enable_delivery_confirmations() → None[source]
class ocrd_network.rabbitmq_utils.OcrdProcessingMessage(job_id: str, processor_name: str, created_time: int, input_file_grps: List[str], output_file_grps: List[str] | None, path_to_mets: str | None, workspace_id: str | None, page_id: str | None, result_queue_name: str | None, callback_url: str | None, internal_callback_url: str | None, parameters: Dict[str, Any] | None = None)[source]

Bases: object

static encode_yml(ocrd_processing_message: OcrdProcessingMessage, encode_type: str = 'utf-8') → bytes[source]
static decode_yml(ocrd_processing_message: bytes, decode_type: str = 'utf-8') → OcrdProcessingMessage[source]
class ocrd_network.rabbitmq_utils.OcrdResultMessage(job_id: str, state: str, path_to_mets: str | None, workspace_id: str | None = '')[source]

Bases: object

static encode_yml(ocrd_result_message: OcrdResultMessage, encode_type: str = 'utf-8') → bytes[source]
static decode_yml(ocrd_result_message: bytes, decode_type: str = 'utf-8') → OcrdResultMessage[source]

Submodules