def generate_secret(length=128):
def properties_dict(sa_object):
def column_properties(sa_variant):
def types_dict(sa_class):
def construct_bare(class_):
def object_from_dict(class_, dict_):
def copy(obj):
def get_pk(sa_variant):
def parent_references(sa_object, models):
    def get_model(table):
def parent_objects(sa_object, models, session, only_pk=False):
def query_model(session, sa_class, only_pk=False):
    def __init__(self):
    def __iter__(self):
    def listen(self, listener):
class EventRegister(object): MAX_SQL_VARIABLES = 900
INTERNAL_SESSION_ATTR = '_dbsync_internal'
def Session():
def session_closing(fn):
    def wrapped(*args, **kwargs):
def session_committing(fn):
    def wrapped(*args, **kwargs):
def committing_context():
def set_engine(engine):
def get_engine():
    def __setattr__(self, *args):
    def __init__(self, model=None, id=None):
def install(self, model):
def tracked_model(operation):
def extend(model, fieldname, fieldtype, loadfn, savefn, deletefn=None):
def _has_extensions(obj):
def _has_delete_functions(obj):
def save_extensions(obj):
def delete_extensions(old_obj, new_obj):
def toggle_listening(enabled=None):
def with_listening(enabled):
    def wrapper(proc):
        def wrapped(*args, **kwargs):
def _track_added(fn, added):
    def tracked(o, **kws):
def _track_deleted(fn, deleted, session, always=False):
    def tracked(o, **kws):
def with_transaction(include_extensions=True):
    def wrapper(proc):
        def wrapped(*args, **kwargs):
def make_content_type_id(model):
def generate_content_types(session=None):
def is_synched(obj, session=None):
def get_latest_version_id(session=None):
class ConfigurationError(Exception): pass
class tracked_record(object): class Operation(Base):
    """
    A database operation (insert, delete or update).

    The operations are grouped in versions and ordered as they are
    executed.
    """

    __tablename__ = "operations"

    row_id = Column(Integer)
    version_id = Column(
        Integer,
        ForeignKey(Version.__tablename__ + ".version_id"),
        nullable=True)
    content_type_id = Column(BigInteger)
    tracked_model = None # to be injected
    command = Column(String(1))
    command_options = ('i', 'u', 'd')
    order = Column(Integer, primary_key=True)

    version = relationship(Version, backref=backref("operations", lazy="joined"))

    @validates('command')
    def validate_command(self, key, command):
        assert command in self.command_options
        return command

    def __repr__(self):
        return u"<Operation row_id: {0}, model: {1}, command: {2}>".\
            format(self.row_id, self.tracked_model, self.command)

    def references(self, obj):
        "Whether this operation references the given object or not."
        if self.row_id != getattr(obj, get_pk(obj), None):
            return False
        model = self.tracked_model
        if model is None:
            return False # operation doesn't even refer to a tracked model
        return model is type(obj)

    def perform(operation, container, session, node_id=None):
        """
        Performs *operation*, looking for required data in
        *container*, and using *session* to perform it.

        *container* is an instance of
        dbsync.messages.base.BaseMessage.

        *node_id* is the node responsible for the operation, if known
        (else ``None``).

        If at any moment this operation fails for predictable causes,
        it will raise an *OperationError*.
        """
        model = operation.tracked_model
        if model is None:
            raise OperationError("no content type for this operation", operation)

        if operation.command == 'i':
            obj = query_model(session, model).\
                filter(getattr(model, get_pk(model)) == operation.row_id).first()
            pull_obj = container.query(model).\
                filter(attr('__pk__') == operation.row_id).first()
            if pull_obj is None:
                raise OperationError(
                    "no object backing the operation in container", operation)
            if obj is None:
                session.add(pull_obj)
            else:
                # Don't raise an exception if the incoming object is
                # exactly the same as the local one.
                if properties_dict(obj) == properties_dict(pull_obj):
                    logger.warning(u"insert attempted when an identical object "
                                   u"already existed in local database: "
                                   u"model {0} pk {1}".format(model.__name__,
                                                              operation.row_id))
                else:
                    raise OperationError(
                        u"insert attempted when the object already existed: "
                        u"model {0} pk {1}".format(model.__name__,
                                                   operation.row_id))

        elif operation.command == 'u':
            obj = query_model(session, model).\
                filter(getattr(model, get_pk(model)) == operation.row_id).first()
            if obj is None:
                # For now, the record will be created again, but is an
                # error because nothing should be deleted without
                # using dbsync
                # raise OperationError(
                #     "the referenced object doesn't exist in database", operation)
                logger.warning(
                    u"The referenced object doesn't exist in database. "
                    u"Node %s. Operation %s",
                    node_id,
                    operation)

            pull_obj = container.query(model).\
                filter(attr('__pk__') == operation.row_id).first()
            if pull_obj is None:
                raise OperationError(
                    "no object backing the operation in container", operation)
            session.merge(pull_obj)

        elif operation.command == 'd':
            obj = query_model(session, model, only_pk=True).\
                filter(getattr(model, get_pk(model)) == operation.row_id).first()
            if obj is None:
                # The object is already deleted in the server
                # The final state in node and server are the same. But
                # it's an error because nothing should be deleted
                # without using dbsync
                logger.warning(
                    "The referenced object doesn't exist in database. "
                    u"Node %s. Operation %s",
                    node_id,
                    operation)
            else:
                session.delete(obj)

        else:
            raise OperationError(
                "the operation doesn't specify a valid command ('i', 'u', 'd')",
                operation) def begin_transaction(session):
def end_transaction(state, session):
def max_local(sa_class, session): class PullMessage(BaseMessage):
    """
    A pull message.

    A pull message can be queried over by version, operation or model,
    and can be filtered multiple times.

    It can be instantiated from a raw data dictionary, or can be made
    empty and filled later with specific methods (``add_version``,
    ``add_operation``, ``add_object``).
    """

    #: Datetime of creation.
    created = None

    #: List of operations to perform in the node.
    operations = None

    #: List of versions being pulled.
    versions = None

    def __init__(self, raw_data=None):
        """
        *raw_data* must be a python dictionary, normally the
        product of JSON decoding. If not given, the message will be
        empty and should be filled with the appropriate methods
        (add_*).
        """
        super(PullMessage, self).__init__(raw_data)
        if raw_data is not None:
            self._build_from_raw(raw_data)
        else:
            self.created = datetime.datetime.now()
            self.operations = []
            self.versions = []

    def _build_from_raw(self, data):
        self.created = decode(types.DateTime())(data['created'])
        self.operations = map(partial(object_from_dict, Operation),
                              imap(decode_dict(Operation), data['operations']))
        self.versions = map(partial(object_from_dict, Version),
                            imap(decode_dict(Version), data['versions']))

    def query(self, model):
        "Returns a query object for this message."
        return MessageQuery(
            model,
            dict(self.payload, **{
                    'models.Operation': self.operations,
                    'models.Version': self.versions}))

    def to_json(self):
        """
        Returns a JSON-friendly python dictionary. Structure::

            created: datetime,
            operations: list of operations,
            versions: list of versions,
            payload: dictionary with lists of objects mapped to model names
        """
        encoded = super(PullMessage, self).to_json()
        encoded['created'] = encode(types.DateTime())(self.created)
        encoded['operations'] = map(encode_dict(Operation),
                                    imap(properties_dict, self.operations))
        encoded['versions'] = map(encode_dict(Version),
                                  imap(properties_dict, self.versions))
        return encoded

    @session_closing
    def add_operation(self, op, swell=True, session=None):
        """
        Adds an operation to the message, including the required
        object if it's possible to include it.

        If *swell* is given and set to ``False``, the operation and
        object will be added bare, without parent objects. Otherwise,
        the parent objects will be added to aid in conflict
        resolution.

        A delete operation doesn't include the associated object. If
        *session* is given, the procedure won't instantiate a new
        session.

        This operation might fail, (due to database inconsitency) in
        which case the internal state of the message won't be affected
        (i.e. it won't end in an inconsistent state).

        DEPRECATED in favor of `fill_for`
        """
        model = op.tracked_model
        if model is None:
            raise ValueError("operation linked to model %s "\
                                 "which isn't being tracked" % model)
        if model not in pulled_models:
            return self
        obj = query_model(session, model).\
            filter_by(**{get_pk(model): op.row_id}).first() \
            if op.command != 'd' else None
        self.operations.append(op)
        # if the object isn't there it's because the operation is old,
        # and should be able to be compressed out when performing the
        # conflict resolution phase
        if obj is not None:
            self.add_object(obj)
            if swell:
                # add parent objects to resolve possible conflicts in merge
                for parent in parent_objects(obj, synched_models.models.keys(),
                                             session):
                    self.add_object(parent)
        return self

    @session_closing
    def add_version(self, v, swell=True, session=None):
        """
        Adds a version to the message, and all associated
        operations and objects.

        This method will either fail and leave the message instance as
        if nothing had happened, or it will succeed and return the
        modified message.

        DEPRECATED in favor of `fill_for`
        """
        for op in v.operations:
            if op.content_type_id not in synched_models.ids:
                raise ValueError("version includes operation linked "\
                                 "to model not currently being tracked", op)
        self.versions.append(v)
        for op in v.operations:
            self.add_operation(op, swell=swell, session=session)
        return self

    @session_closing
    def fill_for(self, request, swell=False, include_extensions=True,
                 session=None):
        """
        Fills this pull message (response) with versions, operations
        and objects, for the given request (PullRequestMessage).

        The *swell* parameter is deprecated and considered ``True``
        regardless of the value given. This means that parent objects
        will always be added to the message.

        *include_extensions* dictates whether the pull message will
        include model extensions or not.
        """
        assert isinstance(request, PullRequestMessage), "invalid request"
        versions = session.query(Version)
        if request.latest_version_id is not None:
            versions = versions.\
                filter(Version.version_id > request.latest_version_id)
        required_objects = {}
        required_parents = {}
        for v in versions:
            self.versions.append(v)
            for op in v.operations:
                model = op.tracked_model
                if model is None:
                    raise ValueError("operation linked to model %s "\
                                         "which isn't being tracked" % model)
                if model not in pulled_models: continue
                self.operations.append(op)
                if op.command != 'd':
                    pks = required_objects.get(model, set())
                    pks.add(op.row_id)
                    required_objects[model] = pks

        for model, pks in ((m, batch)
                           for m, pks in required_objects.iteritems()
                           for batch in grouper(pks, MAX_SQL_VARIABLES)):
            for obj in query_model(session, model).filter(
                    getattr(model, get_pk(model)).in_(list(pks))).all():
                self.add_object(obj, include_extensions=include_extensions)
                # add parent objects to resolve conflicts in merge
                for pmodel, ppk in parent_references(obj,
                                                     synched_models.models.keys()):
                    parent_pks = required_parents.get(pmodel, set())
                    parent_pks.add(ppk)
                    required_parents[pmodel] = parent_pks

        for pmodel, ppks in ((m, batch)
                             for m, pks in required_parents.iteritems()
                             for batch in grouper(pks, MAX_SQL_VARIABLES)):
            for parent in query_model(session, pmodel).filter(
                    getattr(pmodel, get_pk(pmodel)).in_(list(ppks))).all():
                self.add_object(parent, include_extensions=include_extensions)
        return self class PullRequestMessage(BaseMessage):
    """
    A pull request message.

    The message includes information for the server to decide whether
    it should send back related (parent) objects to those directly
    involved, for use in conflict resolution, or not. This is used to
    allow for thinner PullMessage(s) to be built, through the
    add_version and add_operation methods.
    """

    #: List of operation the node has performed since the last
    #  synchronization. If empty, the pull is a full 'fast-forward'
    #  thin procedure.
    operations = None

    #: The identifier used to select the operations to be included in
    #  the pull response.
    latest_version_id = None

    def __init__(self, raw_data=None):
        """
        *raw_data* must be a python dictionary. If not given, the
        message should be filled with the or
        add_unversioned_operations method.
        """
        super(PullRequestMessage, self).__init__(raw_data)
        if raw_data is not None:
            self._build_from_raw(raw_data)
        else:
            self.latest_version_id = get_latest_version_id()
            self.operations = []

    def _build_from_raw(self, data):
        self.operations = map(partial(object_from_dict, Operation),
                              imap(decode_dict(Operation), data['operations']))
        self.latest_version_id = decode(types.Integer())(
            data['latest_version_id'])

    def query(self, model):
        "Returns a query object for this message."
        return MessageQuery(
            model,
            dict(self.payload, **{'models.Operation': self.operations}))

    def to_json(self):
        "Returns a JSON-friendly python dictionary."
        encoded = super(PullRequestMessage, self).to_json()
        encoded['operations'] = map(encode_dict(Operation),
                                    imap(properties_dict, self.operations))
        encoded['latest_version_id'] = encode(types.Integer())(
            self.latest_version_id)
        return encoded

    def add_operation(self, op):
        """
        Adds an operation to the message, including the required
        object if possible.
        """
        assert op.version_id is None, "the operation {0} is already versioned".\
            format(op)
        model = op.tracked_model
        if model is None:
            raise ValueError("operation linked to model %s "\
                                 "which isn't being tracked" % model)
        if model not in pulled_models: return self
        self.operations.append(op)
        return self

    @session_closing
    def add_unversioned_operations(self, session=None):
        """
        Adds all unversioned operations to this message and
        required objects.
        """
        operations = session.query(Operation).\
            filter(Operation.version_id == None).all()
        if any(op.content_type_id not in synched_models.ids
               for op in operations):
            raise ValueError("version includes operation linked "\
                                 "to model not currently being tracked")
        for op in operations:
            self.add_operation(op)
        return self @core.session_committing
def compress(session=None):
    """
    Compresses unversioned operations in the database.

    For each row in the operations table, this deletes unnecesary
    operations that would otherwise bloat the message.

    This procedure is called internally before the 'push' request
    happens, and before the local 'merge' happens.
    """
    unversioned = session.query(Operation).\
        filter(Operation.version_id == None).order_by(Operation.order.desc())
    seqs = group_by(lambda op: (op.row_id, op.content_type_id), unversioned)

    # Check errors on sequences
    for seq in seqs.itervalues():
        _assert_operation_sequence(seq, session)

    for seq in ifilter(lambda seq: len(seq) > 1, seqs.itervalues()):
        if seq[-1].command == 'i':
            if all(op.command == 'u' for op in seq[:-1]):
                # updates are superfluous
                map(session.delete, seq[:-1])
            elif seq[0].command == 'd':
                # it's as if the object never existed
                map(session.delete, seq)
        elif seq[-1].command == 'u':
            if all(op.command == 'u' for op in seq[:-1]):
                # leave a single update
                map(session.delete, seq[1:])
            elif seq[0].command == 'd':
                # leave the delete statement
                map(session.delete, seq[1:])
    session.flush()

    # repair inconsistencies
    for operation in session.query(Operation).\
            filter(Operation.version_id == None).\
            order_by(Operation.order.desc()).all():
        session.flush()
        model = operation.tracked_model
        if not model:
            logger.error(
                "operation linked to content type "
                "not tracked: %s" % operation.content_type_id)
            continue
        if operation.command in ('i', 'u'):
            if query_model(session, model, only_pk=True).\
                    filter_by(**{get_pk(model): operation.row_id}).count() == 0:
                logger.warning(
                    "deleting operation %s for model %s "
                    "for absence of backing object" % (operation, model.__name__))
                session.delete(operation)
                continue
        if operation.command == 'u':
            subsequent = session.query(Operation).\
                filter(Operation.content_type_id == operation.content_type_id,
                       Operation.version_id == None,
                       Operation.row_id == operation.row_id,
                       Operation.order > operation.order).all()
            if any(op.command == 'i' for op in subsequent) and \
                    all(op.command != 'd' for op in subsequent):
                logger.warning(
                    "deleting update operation %s for model %s "
                    "for preceding an insert operation" %\
                        (operation, model.__name__))
                session.delete(operation)
                continue
        if session.query(Operation).\
                filter(Operation.content_type_id == operation.content_type_id,
                       Operation.command == operation.command,
                       Operation.version_id == None,
                       Operation.row_id == operation.row_id,
                       Operation.order != operation.order).count() > 0:
            logger.warning(
                "deleting operation %s for model %s "
                "for being redundant after compression" %\
                    (operation, model.__name__))
            session.delete(operation)
            continue
    return session.query(Operation).\
        filter(Operation.version_id == None).\
        order_by(Operation.order.asc()).all() def compressed_operations(operations):
    """
    Compresses a set of operations so as to avoid redundant
    ones. Returns the compressed set sorted by operation order. This
    procedure doesn't perform database operations.
    """
    seqs = group_by(lambda op: (op.row_id, op.content_type_id),
                    sorted(operations, key=attr('order')))
    compressed = []
    for seq in seqs.itervalues():
        if len(seq) == 1:
            compressed.append(seq[0])
        elif seq[0].command == 'i':
            if seq[-1].command == 'd':
                pass
            else:
                compressed.append(seq[0])
        elif seq[0].command == 'u':
            if seq[-1].command == 'd':
                compressed.append(seq[-1])
            else:
                compressed.append(seq[0])
        else: # seq[0].command == 'd':
            if seq[-1].command == 'd':
                compressed.append(seq[0])
            elif seq[-1].command == 'u':
                compressed.append(seq[-1])
            else: # seq[-1].command == 'i':
                op = seq[-1]
                compressed.append(
                    Operation(order=op.order,
                              content_type_id=op.content_type_id,
                              row_id=op.row_id,
                              version_id=op.version_id,
                              command='u'))
    compressed.sort(key=attr('order'))
    return compressed def get_related_tables(sa_class):
    """
    Returns a list of related SA tables dependent on the given SA
    model by foreign key.
    """
    mapper = class_mapper(sa_class)
    models = synched_models.models.iterkeys()
    return [table for table in (class_mapper(model).mapped_table
                                for model in models)
            if mapper.mapped_table in [key.column.table
                                       for key in table.foreign_keys]] def get_fks(table_from, table_to):
    """
    Returns the names of the foreign keys that are defined in
    *table_from* SA table and that refer to *table_to* SA table. If
    the foreign keys don't exist, this procedure returns an empty
    list.
    """
    fks = filter(lambda k: k.column.table == table_to, table_from.foreign_keys)
    return [fk.parent.name for fk in fks] def find_direct_conflicts(pull_ops, unversioned_ops):
    """
    Detect conflicts where there's both unversioned and pulled
    operations, update or delete ones, referering to the same tracked
    object. This procedure relies on the uniqueness of the primary
    keys through time.
    """
    return [
        (pull_op, local_op)
        for pull_op in pull_ops
        if pull_op.command == 'u' or pull_op.command == 'd'
        for local_op in unversioned_ops
        if local_op.command == 'u' or local_op.command == 'd'
        if pull_op.row_id == local_op.row_id
        if pull_op.content_type_id == local_op.content_type_id] def find_dependency_conflicts(pull_ops, unversioned_ops, session):
    """
    Detect conflicts by relationship dependency: deletes on the pull
    message on objects that have dependent objects inserted or updated
    on the local database.
    """
    related_ids = dict(
        (pull_op, related_local_ids(pull_op, session))
        for pull_op in pull_ops
        if pull_op.command == 'd')
    return [
        (pull_op, local_op)
        for pull_op in pull_ops
        if pull_op.command == 'd'
        for local_op in unversioned_ops
        if local_op.command == 'i' or local_op.command == 'u'
        if (local_op.row_id, local_op.content_type_id) in related_ids[pull_op]] def find_reversed_dependency_conflicts(pull_ops, unversioned_ops, pull_message):
    """
    Deletes on the local database on objects that are referenced by
    inserted or updated objects in the pull message.
    """
    related_ids = dict(
        (local_op, related_remote_ids(local_op, pull_message))
        for local_op in unversioned_ops
        if local_op.command == 'd')
    return [
        (pull_op, local_op)
        for local_op in unversioned_ops
        if local_op.command == 'd'
        for pull_op in pull_ops
        if pull_op.command == 'i' or pull_op.command == 'u'
        if (pull_op.row_id, pull_op.content_type_id) in related_ids[local_op]] def find_insert_conflicts(pull_ops, unversioned_ops):
    """
    Inserts over the same object. These conflicts should be resolved
    by keeping both objects, but moving the local one out of the way
    (reinserting it to get a new primary key). It should be possible,
    however, to specify a custom handler for cases where the primary
    key is a meaningful property of the object.
    """
    return [
        (pull_op, local_op)
        for local_op in unversioned_ops
        if local_op.command == 'i'
        for pull_op in pull_ops
        if pull_op.command == 'i'
        if pull_op.row_id == local_op.row_id
        if pull_op.content_type_id == local_op.content_type_id] def find_unique_conflicts(pull_ops, unversioned_ops, pull_message, session):
    """
    Unique constraints violated in a model. Returns two lists of
    dictionaries, the first one with the solvable conflicts, and the
    second one with the proper errors. Each conflict is a dictionary
    with the following fields::

        object: the local conflicting object, bound to the session
        columns: tuple of column names in the unique constraint
        new_values: tuple of values that can be used to update the
                    conflicting object

    Each error is a dictionary with the following fields::

        model: the model (class) of the conflicting object
        pk: the value of the primary key of the conflicting object
        columns: tuple of column names in the unique constraint
    """

    def verify_constraint(model, columns, values):
        """
        Checks to see whether some local object exists with
        conflicting values.
        """
        match = query_model(session, model, only_pk=True).\
            options(*(undefer(column) for column in columns)).\
            filter_by(**dict((column, value)
                             for column, value in izip(columns, values))).first()
        pk = get_pk(model)
        return match, getattr(match, pk, None)

    def get_remote_values(model, row_id, columns):
        """
        Gets the conflicting values out of the remote object set
        (*container*).
        """
        obj = pull_message.query(model).filter(attr('__pk__') == row_id).first()
        if obj is not None:
            return tuple(getattr(obj, column) for column in columns)
        return (None,)

    # keyed to content type
    unversioned_pks = dict((ct_id, set(op.row_id for op in unversioned_ops
                                       if op.content_type_id == ct_id
                                       if op.command != 'd'))
                           for ct_id in set(operation.content_type_id
                                            for operation in unversioned_ops))
    # the lists to fill with conflicts and errors
    conflicts, errors = [], []

    for op in pull_ops:
        model = op.tracked_model

        for constraint in ifilter(lambda c: isinstance(c, UniqueConstraint),
                                  class_mapper(model).mapped_table.constraints):

            unique_columns = tuple(col.name for col in constraint.columns)
            # Unique values on the server, to check conflicts with local database
            remote_values = get_remote_values(model, op.row_id, unique_columns)

            obj_conflict, pk_conflict = verify_constraint(
                model, unique_columns, remote_values)

            is_unversioned = pk_conflict in unversioned_pks.get(
                op.content_type_id, set())

            if all(value is None for value in remote_values): continue # Null value
            if pk_conflict is None: continue # No problem
            if pk_conflict == op.row_id:
                if op.command == 'i':
                    # Two nodes created objects with the same unique
                    # value and same pk
                    errors.append(
                        {'model': type(obj_conflict),
                         'pk': pk_conflict,
                         'columns': unique_columns})
                continue

            # if pk_conflict != op.row_id:
            remote_obj = pull_message.query(model).\
                filter(attr('__pk__') == pk_conflict).first()

            if remote_obj is not None and not is_unversioned:
                old_values = tuple(getattr(obj_conflict, column)
                                   for column in unique_columns)
                # The new unique value of the conflictive object
                # in server
                new_values = tuple(getattr(remote_obj, column)
                                   for column in unique_columns)

                if old_values != new_values:
                    # Library error
                    # It's necesary to first update the unique value
                    session.refresh(obj_conflict, column_properties(obj_conflict))
                    conflicts.append(
                        {'object': obj_conflict,
                         'columns': unique_columns,
                         'new_values': new_values})
                else:
                    # The server allows two identical unique values
                    # This should be impossible
                    pass
            elif remote_obj is not None and is_unversioned:
                # Two nodes created objects with the same unique
                # values. Human error.
                errors.append(
                    {'model': type(obj_conflict),
                     'pk': pk_conflict,
                     'columns': unique_columns})
            else:
                # The conflicting object hasn't been modified on the
                # server, which must mean the local user is attempting
                # an update that collides with one from another user.
                errors.append(
                    {'model': type(obj_conflict),
                     'pk': pk_conflict,
                     'columns': unique_columns})
    return conflicts, errors def post_request(server_url, json_dict,
                 encode=None, decode=None, headers=None, timeout=None,
                 monitor=None):
    """
    Sends a POST request to *server_url* with data *json_dict* and
    returns a trio of (code, reason, body).

    *encode* is a function that transforms a python dictionary into a
    string.

    *decode* is a function that transforms a string into a python
    dictionary.

    For all dictionaries d of simple types, decode(encode(d)) == d.

    *headers* is a python dictionary with headers to send.

    *timeout* is the number of seconds to wait for a response.

    *monitor* is a routine that gets called for each chunk of the
    response received, and is given a dictionary with information. The
    key 'status' will always be in the dictionary, and other entries
    will contain additional information. Check the source code to see
    the variations.
    """
    if not server_url.startswith("http://") and \
            not server_url.startswith("https://"):
        server_url = "http://" + server_url
    enc, dec, hhs, tout = _defaults(encode, decode, headers, timeout)
    stream = inspect.isroutine(monitor)
    auth = authentication_callback(server_url) \
        if authentication_callback is not None else None
    try:
        r = requests.post(server_url, data=enc(json_dict),
                          headers=hhs or None, stream=stream,
                          timeout=tout, auth=auth)
        response = None
        if stream:
            total = r.headers.get('content-length', None)
            partial = 0
            monitor({'status': "connect", 'size': total})
            chunks = cStringIO.StringIO()
            for chunk in r:
                partial += len(chunk)
                monitor({'status': "downloading",
                         'size': total, 'received': partial})
                chunks.write(chunk)
            response = chunks.getvalue()
            chunks.close()
        else:
            response = r.content
        body = None
        try:
            body = dec(response)
        except ValueError:
            pass
        result = (r.status_code, r.reason, body)
        r.close()
        return result

    except requests.exceptions.RequestException as e:
        if stream:
            monitor({'status': "error", 'reason': "network error"})
        raise NetworkError(*e.args)

    except Exception as e:
        if stream:
            monitor({'status': "error", 'reason': "network error"})
        raise NetworkError(*e.args)
import collections
from sqlalchemy.orm import make_transient
from dbsync.lang import *
from dbsync.utils import class_mapper, get_pk, query_model
from dbsync import core
from dbsync.models import Operation
from dbsync import dialects
from dbsync.messages.pull import PullMessage, PullRequestMessage
from dbsync.client.compression import compress, compressed_operations
from dbsync.client.conflicts import (
    get_related_tables,
    get_fks,
    find_direct_conflicts,
    find_dependency_conflicts,
    find_reversed_dependency_conflicts,
    find_insert_conflicts,
    find_unique_conflicts)
from dbsync.client.net import post_request
"""
Pull, merge and related operations.
"""





# Utilities specific to the merge

def max_local(model, session):
    """
    Returns the maximum value for the primary key of the given model
    in the local database.
    """
    if model is None:
        raise ValueError("null model given to max_local query")
    return dialects.max_local(model, session)


def max_remote(model, container):
    """
    Returns the maximum value for the primary key of the given model
    in the container.
    """
    return max(getattr(obj, get_pk(obj)) for obj in container.query(model))


def update_local_id(old_id, new_id, model, session):
    """
    Updates the tuple matching *old_id* with *new_id*, and updates all
    dependent tuples in other tables as well.
    """
    # Updating either the tuple or the dependent tuples first would
    # cause integrity violations if the transaction is flushed in
    # between. The order doesn't matter.
    if model is None:
        raise ValueError("null model given to update_local_id subtransaction")
    # must load fully, don't know yet why
    obj = query_model(session, model).\
        filter_by(**{get_pk(model): old_id}).first()
    setattr(obj, get_pk(model), new_id)

    # Then the dependent ones

    related_tables = get_related_tables(model)