python3-postgresql 1.1.0-1build1 / usr / share / doc / python3-postgresql / html / _sources / copyman.txt

.. _pg_copyman:

***************
Copy Management
***************

The `postgresql.copyman` module provides a way to quickly move COPY data coming
from one connection to many connections. Alternatively, it can be sourced
by arbitrary iterators and target arbitrary callables.

Statement execution methods offer a way for running COPY operations
with iterators, but the cost of allocating objects for each row is too
significant for transferring gigabytes of COPY data from one connection to
another. The interfaces available on statement objects are primarily intended to
be used when transferring COPY data to and from arbitrary Python
objects.

Direct connection-to-connection COPY operations can be performed using the
high-level `postgresql.copyman.transfer` function::

	>>> from postgresql import copyman
	>>> send_stmt = source.prepare("COPY (SELECT i FROM generate_series(1, 1000000) AS g(i)) TO STDOUT")
	>>> destination.execute("CREATE TEMP TABLE loading_table (i int8)")
	>>> receive_stmt = destination.prepare("COPY loading_table FROM STDIN")
	>>> total_rows, total_bytes = copyman.transfer(send_stmt, receive_stmt)

However, if more control is needed, the `postgresql.copyman.CopyManager` class
should be used directly.


Copy Managers
=============

The `postgresql.copyman.CopyManager` class manages the Producer and the
Receivers involved in a COPY operation. Normally,
`postgresql.copyman.StatementProducer` and
`postgresql.copyman.StatementReceiver` instances. Naturally, a Producer is the
object that produces the COPY data to be given to the Manager's Receivers.

Using a Manager directly means that there is a need for more control over
the operation. The Manager is both a context manager and an iterator. The
context manager interfaces handle initialization and finalization of the COPY
state, and the iterator provides an event loop emitting information about the
amount of COPY data transferred this cycle. Normal usage takes the form::

	>>> from postgresql import copyman
	>>> send_stmt = source.prepare("COPY (SELECT i FROM generate_series(1, 1000000) AS g(i)) TO STDOUT")
	>>> destination.execute("CREATE TEMP TABLE loading_table (i int8)")
	>>> receive_stmt = destination.prepare("COPY loading_table FROM STDIN")
	>>> producer = copyman.StatementProducer(send_stmt)
	>>> receiver = copyman.StatementReceiver(receive_stmt)
	>>> 
	>>> with source.xact(), destination.xact():
	...  with copyman.CopyManager(producer, receiver) as copy:
	...   for num_messages, num_bytes in copy:
	...    update_rate(num_bytes)

As an alternative to a for-loop inside a with-statement block, the `run` method
can be called to perform the operation::

	>>> with source.xact(), destination.xact():
	...  copyman.CopyManager(producer, receiver).run()

However, there is little benefit beyond using the high-level
`postgresql.copyman.transfer` function.

Manager Interface Points
------------------------

Primarily, the `postgresql.copyman.CopyManager` provides a context manager and
an iterator for controlling the COPY operation.

 ``CopyManager.run()``
  Perform the entire COPY operation.

 ``CopyManager.__enter__()``
  Start the COPY operation. Connections taking part in the COPY should **not**
  be used until ``__exit__`` is ran.

 ``CopyManager.__exit__(typ, val, tb)``
  Finish the COPY operation. Fails in the case of an incomplete
  COPY, or an untrapped exception. Either returns `None` or raises the generalized
  exception, `postgresql.copyman.CopyFail`.

 ``CopyManager.__iter__()``
  Returns the CopyManager instance.

 ``CopyManager.__next__()``
  Transfer the next chunk of COPY data to the receivers. Yields a tuple
  consisting of the number of messages and bytes transferred,
  ``(num_messages, num_bytes)``. Raises `StopIteration` when complete.

  Raises `postgresql.copyman.ReceiverFault` when a Receiver raises an
  exception.
  Raises `postgresql.copyman.ProducerFault` when the Producer raises an
  exception. The original exception is available via the exception's
  ``__context__`` attribute.

 ``CopyManager.reconcile(faulted_receiver)``
  Reconcile a faulted receiver. When a receiver faults, it will no longer
  be in the set of Receivers. This method is used to signal to the manager that the
  problem has been corrected, and the receiver is again ready to receive.

 ``CopyManager.receivers``
  The `builtins.set` of Receivers involved in the COPY operation.

 ``CopyManager.producer``
  The Producer emitting the data to be given to the Receivers.


Faults
======

The CopyManager generalizes any exceptions that occur during transfer. While
inside the context manager, `postgresql.copyman.Fault` may be raised if a
Receiver or a Producer raises an exception. A `postgresql.copyman.ProducerFault`
in the case of the Producer, and `postgresql.copyman.ReceiverFault` in the case
of the Receivers.

.. note::
 Faults are only raised by `postgresql.copyman.CopyManager.__next__`. The
 ``run()`` method will only raise `postgresql.copyman.CopyFail`.

Receiver Faults
---------------

The Manager assumes the Fault is fatal to a Receiver, and immediately removes
it from the set of target receivers. Additionally, if the Fault exception goes
untrapped, the copy will ultimately fail.

The Fault exception references the Manager that raised the exception, and the
actual exceptions that occurred associated with the Receiver that caused them.

In order to identify the exception that caused a Fault, the ``faults`` attribute
on the `postgresql.copyman.ReceiverFault` must be referenced::

	>>> from postgresql import copyman
	>>> send_stmt = source.prepare("COPY (SELECT i FROM generate_series(1, 1000000) AS g(i)) TO STDOUT")
	>>> destination.execute("CREATE TEMP TABLE loading_table (i int8)")
	>>> receive_stmt = destination.prepare("COPY loading_table FROM STDIN")
	>>> producer = copyman.StatementProducer(send_stmt)
	>>> receiver = copyman.StatementReceiver(receive_stmt)
	>>> 
	>>> with source.xact(), destination.xact():
	...  with copyman.CopyManager(producer, receiver) as copy:
	...   while copy.receivers:
	...    try:
	...     for num_messages, num_bytes in copy:
	...      update_rate(num_bytes)
	...     break
	...    except copyman.ReceiverFault as cf:
	...     # Access the original exception using the receiver as the key.
	...     original_exception = cf.faults[receiver]
	...     if unknown_failure(original_exception):
	...      ...
	...      raise


ReceiverFault Properties
~~~~~~~~~~~~~~~~~~~~~~~~

The following attributes exist on `postgresql.copyman.ReceiverFault` instances:

 ``ReceiverFault.manager``
  The subject `postgresql.copyman.CopyManager` instance.

 ``ReceiverFault.faults``
  A dictionary mapping the Receiver to the exception raised by that Receiver.


Reconciliation
~~~~~~~~~~~~~~

When a `postgresql.copyman.ReceiverFault` is raised, the Manager immediately
removes the Receiver so that the COPY operation can continue. Continuation of
the COPY can occur by trapping the exception and continuing the iteration of the
Manager. However, if the fault is recoverable, the
`postgresql.copyman.CopyManager.reconcile` method must be used to reintroduce the
Receiver into the Manager's set. Faults must be trapped from within the
Manager's context::

	>>> import socket
	>>> from postgresql import copyman
	>>> send_stmt = source.prepare("COPY (SELECT i FROM generate_series(1, 1000000) AS g(i)) TO STDOUT")
	>>> destination.execute("CREATE TEMP TABLE loading_table (i int8)")
	>>> receive_stmt = destination.prepare("COPY loading_table FROM STDIN")
	>>> producer = copyman.StatementProducer(send_stmt)
	>>> receiver = copyman.StatementReceiver(receive_stmt)
	>>> 
	>>> with source.xact(), destination.xact():
	...  with copyman.CopyManager(producer, receiver) as copy:
	...   while copy.receivers:
	...    try:
	...     for num_messages, num_bytes in copy:
	...      update_rate(num_bytes)
	...    except copyman.ReceiverFault as cf:
	...     if isinstance(cf.faults[receiver], socket.timeout):
	...      copy.reconcile(receiver)
	...     else:
	...      raise

Recovering from Faults does add significant complexity to a COPY operation,
so, often, it's best to avoid conditions in which reconciliable Faults may
occur.


Producer Faults
---------------

Producer faults are normally fatal to the COPY operation and should rarely be
trapped. However, the Manager makes no state changes when a Producer faults,
so, unlike Receiver Faults, no reconciliation process is necessary; rather,
if it's safe to continue, the Manager's iterator should continue to be
processed.

ProducerFault Properties
~~~~~~~~~~~~~~~~~~~~~~~~

The following attributes exist on `postgresql.copyman.ProducerFault` instances:

 ``ReceiverFault.manager``
  The subject `postgresql.copyman.CopyManager`.

 ``ReceiverFault.__context__``
  The original exception raised by the Producer.


Failures
========

When a COPY operation is aborted, either by an exception or by the iterator
being broken, a `postgresql.copyman.CopyFail` exception will be raised by the
Manager's ``__exit__()`` method. The `postgresql.copyman.CopyFail` exception
offers to record any exceptions that occur during the exit of the context
managers of the Producer and the Receivers.


CopyFail Properties
-------------------

The following properties exist on `postgresql.copyman.CopyFail` exceptions:

 ``CopyFail.manager``
  The Manager whose COPY operation failed.

 ``CopyFail.receiver_faults``
  A dictionary mapping a `postgresql.copyman.Receiver` to the exception raised
  by that Receiver's ``__exit__``. `None` if no exceptions were raised by the
  Receivers.

 ``CopyFail.producer_fault``
  The exception Raised by the `postgresql.copyman.Producer`. `None` if none.


Producers
=========

The following Producers are available:

 ``postgresql.copyman.StatementProducer(postgresql.api.Statement)``
  Given a Statement producing COPY data, construct a Producer.

 ``postgresql.copyman.IteratorProducer(collections.Iterator)``
  Given an Iterator producing *chunks* of COPY lines, construct a Producer to
  manage the data coming from the iterator.


Receivers
=========

 ``postgresql.copyman.StatementReceiver(postgresql.api.Statement)``
  Given a Statement producing COPY data, construct a Producer.

 ``postgresql.copyman.CallReceiver(callable)``
  Given a callable, construct a Receiver that will transmit COPY data in chunks
  of lines. That is, the callable will be given a list of COPY lines for each
  transfer cycle.


Terminology
===========

The following terms are regularly used to describe the implementation and
processes of the `postgresql.copyman` module:

 Manager
  The object used to manage data coming from a Producer and being given to the
  Receivers. It also manages the necessary initialization and finalization steps
  required by those factors.

 Producer
  The object used to produce the COPY data to be given to the Receivers. The
  source.

 Receiver
  An object that consumes COPY data. A target.

 Fault
  Specifically, `postgresql.copyman.Fault` exceptions. A Fault is raised
  when a Receiver or a Producer raises an exception during the COPY operation.

 Reconciliation
  Generally, the steps performed by the "reconcile" method on
  `postgresql.copyman.CopyManager` instances. More precisely, the
  necessary steps for a Receiver's reintroduction into the COPY operation after
  a Fault.

 Failed Copy
  A failed copy is an aborted COPY operation. This occurs in situations of
  untrapped exceptions or an incomplete COPY. Specifically, the COPY will be
  noted as failed in cases where the Manager's iterator is *not* ran until
  exhaustion.

 Realignment
  The process of providing compensating data to the Receivers so that the
  connection will be on a message boundary. Occurs when the COPY operation
  is aborted.