def log_answer(answer):
    pin_this_thread()

    creator = answer.creator
    created = answer.created
    question = answer.question
    users = [a.creator for a in
             question.answers.select_related('creator').exclude(
                creator=creator)]
    if question.creator != creator:
        users += [question.creator]
    users = set(users)  # Remove duplicates.

    if users:
        action = Action.objects.create(
            creator=creator,
            created=created,
            url=answer.get_absolute_url(),
            content_object=answer,
            formatter_cls='questions.formatters.AnswerFormatter')
        action.users.add(*users)

    transaction.commit_unless_managed()

    # Record karma actions
    AnswerAction(answer.creator, answer.created.date()).save()
    if answer == answer.question.answers.order_by('created')[0]:
        FirstAnswerAction(answer.creator, answer.created.date()).save()

    unpin_this_thread()

def _rebuild_kb_chunk(data, **kwargs):
    """Re-render a chunk of documents."""
    log.info('Rebuilding %s documents.' % len(data))

    pin_this_thread()  # Stick to master.

    messages = []
    for pk in data:
        message = None
        try:
            document = Document.objects.get(pk=pk)
            document.html = document.current_revision.content_cleaned
            document.save()
        except Document.DoesNotExist:
            message = 'Missing document: %d' % pk
        except ValidationError as e:
            message = 'ValidationError for %d: %s' % (pk, e.messages[0])
        except SlugCollision:
            message = 'SlugCollision: %d' % pk
        except TitleCollision:
            message = 'TitleCollision: %d' % pk

        if message:
            log.debug(message)
            messages.append(message)

    if messages:
        subject = ('[%s] Exceptions raised in _rebuild_kb_chunk()' %
                   settings.PLATFORM_NAME)
        mail_admins(subject=subject, message='\n'.join(messages))
    transaction.commit_unless_managed()

    unpin_this_thread()  # Not all tasks need to do use the master.

def _rebuild_kb_chunk(data):
    """Re-render a chunk of documents.

    Note: Don't use host components when making redirects to wiki pages; those
    redirects won't be auto-pruned when they're 404s.

    """
    log.info('Rebuilding %s documents.' % len(data))

    pin_this_thread()  # Stick to master.

    messages = []
    start = time.time()
    for pk in data:
        message = None
        try:
            document = Document.objects.get(pk=pk)

            # If we know a redirect link to be broken (i.e. if it looks like a
            # link to a document but the document isn't there), log an error:
            url = document.redirect_url()
            if (url and points_to_document_view(url) and
                    not document.redirect_document()):
                log.warn('Invalid redirect document: %d' % pk)

            html = document.parse_and_calculate_links()
            if document.html != html:
                # We are calling update here to so we only update the html
                # column instead of all of them. This bypasses post_save
                # signal handlers like the one that triggers reindexing.
                # See bug 797038 and bug 797352.
                Document.objects.filter(pk=pk).update(html=html)
                statsd.incr('wiki.rebuild_chunk.change')
            else:
                statsd.incr('wiki.rebuild_chunk.nochange')
        except Document.DoesNotExist:
            message = 'Missing document: %d' % pk
        except Revision.DoesNotExist:
            message = 'Missing revision for document: %d' % pk
        except ValidationError as e:
            message = 'ValidationError for %d: %s' % (pk, e.messages[0])
        except SlugCollision:
            message = 'SlugCollision: %d' % pk
        except TitleCollision:
            message = 'TitleCollision: %d' % pk

        if message:
            log.debug(message)
            messages.append(message)
    d = time.time() - start
    statsd.timing('wiki.rebuild_chunk', int(round(d * 1000)))

    if messages:
        subject = ('[%s] Exceptions raised in _rebuild_kb_chunk()' %
                   settings.PLATFORM_NAME)
        mail_admins(subject=subject, message='\n'.join(messages))
    if not transaction.get_connection().in_atomic_block:
        transaction.commit()

    unpin_this_thread()  # Not all tasks need to do use the master.

def render_document_cascade(base):
    """Given a document, render it and all documents that may be affected."""

    # This walks along the graph of links between documents. If there is
    # a document A that includes another document B as a template, then
    # there is an edge from A to B in this graph. The goal here is to
    # process every node exactly once. This is robust to cycles and
    # diamonds in the graph, since it keeps track of what nodes have
    # been visited already.

    # In case any thing goes wrong, this guarantees we unpin the DB
    try:
        # Sends all writes to the master DB. Slaves are readonly.
        pin_this_thread()

        todo = set([base])
        done = set()

        while todo:
            d = todo.pop()
            if d in done:
                # Don't process a node twice.
                continue
            d.html = d.parse_and_calculate_links()
            d.save()
            done.add(d)
            todo.update(l.linked_from for l in d.links_to()
                        .filter(kind__in=['template', 'include']))

    finally:
        unpin_this_thread()

 def setUp(self):
     self.auth = authentication.RestSharedSecretAuthentication()
     self.profile = UserProfile.objects.get(pk=2519)
     self.profile.update(email=self.profile.email)
     self.middlewares = [APIBaseMiddleware,
                         RestSharedSecretMiddleware]
     unpin_this_thread()

def index_task(cls, id_list, **kw):
    """Index documents specified by cls and ids"""
    statsd.incr('search.tasks.index_task.%s' % cls.get_mapping_type_name())
    try:
        # Pin to master db to avoid replication lag issues and stale
        # data.
        pin_this_thread()

        qs = cls.get_model().uncached.filter(id__in=id_list).values_list(
            'id', flat=True)
        for id_ in qs:
            try:
                cls.index(cls.extract_document(id_), id_=id_)
            except UnindexMeBro:
                # If extract_document throws this, then we need to
                # remove this item from the index.
                cls.unindex(id_)

    except Exception as exc:
        retries = index_task.request.retries
        if retries >= MAX_RETRIES:
            # Some exceptions aren't pickleable and we need this to
            # throw things that are pickleable.
            raise IndexingTaskError()

        statsd.incr('search.tasks.index_task.retry', 1)
        statsd.incr('search.tasks.index_task.retry%d' % RETRY_TIMES[retries],
                    1)

        index_task.retry(exc=exc, max_retries=MAX_RETRIES,
                         countdown=RETRY_TIMES[retries])
    finally:
        unpin_this_thread()

def index_chunk_task(index, batch_id, rec_id, chunk):
    """Index a chunk of things.

    :arg index: the name of the index to index to
    :arg batch_id: the name for the batch this chunk belongs to
    :arg rec_id: the id for the record for this task
    :arg chunk: a (class, id_list) of things to index
    """
    cls, id_list = chunk

    try:
        # Pin to master db to avoid replication lag issues and stale
        # data.
        pin_this_thread()

        # Update record data.
        rec = Record.objects.get(pk=rec_id)
        rec.start_time = datetime.datetime.now()
        rec.message = u'Reindexing into %s' % index
        rec.status = Record.STATUS_IN_PROGRESS
        rec.save()

        index_chunk(cls, id_list, reraise=True)

        rec.mark_success()

    except Exception:
        rec.mark_fail(u'Errored out %s %s' % (
                sys.exc_type, sys.exc_value))
        raise

    finally:
        unpin_this_thread()

 def test_pinning_encapsulation(self):
     """Check the pinning getters and setters."""
     self.assertFalse(this_thread_is_pinned())
     pin_this_thread()
     self.assertTrue(this_thread_is_pinned())
     unpin_this_thread()
     self.assertFalse(this_thread_is_pinned())

 def test_context_manager_exception(self):
     unpin_this_thread()
     assert not this_thread_is_pinned()
     with self.assertRaises(ValueError):
         with use_master:
             assert this_thread_is_pinned()
             raise ValueError
     assert not this_thread_is_pinned()

 def process_request(self, request):
     """Set the thread's pinning flag according to the presence of the
     incoming cookie."""
     if PINNING_COOKIE in request.COOKIES or request.method not in READ_ONLY_METHODS:
         pin_this_thread()
     else:
         # In case the last request this thread served was pinned:
         unpin_this_thread()

 def test_decorator(self):
     @use_primary_db
     def check():
         assert this_thread_is_pinned()
     unpin_this_thread()
     assert not this_thread_is_pinned()
     check()
     assert not this_thread_is_pinned()

 def test_context_manager_exception(self):
     unpin_this_thread()
     self.assertFalse(this_thread_is_pinned())
     with self.assertRaises(ValueError):
         with use_master:
             self.assertTrue(this_thread_is_pinned())
             raise ValueError
     self.assertFalse(this_thread_is_pinned())

 def test_decorator(self):
     @use_master
     def check():
         assert this_thread_is_pinned()
     unpin_this_thread()
     assert not this_thread_is_pinned()
     check()
     assert not this_thread_is_pinned()

 def setUp(self):
     self.api_name = "foo"
     self.profile = UserProfile.objects.get(pk=2519)
     self.profile.update(read_dev_agreement=datetime.today())
     self.access = Access.objects.create(key="test_oauth_key", secret=generate(), user=self.profile.user)
     self.auth = authentication.RestOAuthAuthentication()
     self.middlewares = [RedirectPrefixedURIMiddleware, RestOAuthMiddleware]
     unpin_this_thread()

 def setUp(self):
     self.api_name = 'foo'
     self.profile = UserProfile.objects.get(pk=2519)
     self.profile.update(read_dev_agreement=datetime.today())
     self.access = Access.objects.create(key='test_oauth_key',
                                         secret=generate(),
                                         user=self.profile.user)
     self.auth = authentication.RestOAuthAuthentication()
     unpin_this_thread()

    def test_pinning_encapsulation(self):
        """Check the pinning getters and setters."""
        assert not this_thread_is_pinned(), "Thread started out pinned or this_thread_is_pinned() is broken."

        pin_this_thread()
        assert this_thread_is_pinned(), "pin_this_thread() didn't pin the thread."

        unpin_this_thread()
        assert not this_thread_is_pinned(), "Thread remained pinned after unpin_this_thread()."

    def test_decorator(self):

        @use_master
        def check():
            self.assertTrue(this_thread_is_pinned())

        unpin_this_thread()
        self.assertFalse(this_thread_is_pinned())
        check()
        self.assertFalse(this_thread_is_pinned())

    def test_slave_decorator_resets(self):

        @use_slave
        def check():
            self.assertFalse(this_thread_is_pinned())

        unpin_this_thread()
        self.assertFalse(this_thread_is_pinned())
        check()
        self.assertFalse(this_thread_is_pinned())

 def test_context_manager_exception(self):
     unpin_this_thread()
     try:
         assert not this_thread_is_pinned()
         with use_master:
             assert this_thread_is_pinned()
             raise ValueError
     except ValueError:
         pass
     assert not this_thread_is_pinned()

 def setUp(self):
     self.api_name = 'foo'
     self.profile = UserProfile.objects.get(pk=2519)
     self.profile.update(read_dev_agreement=datetime.today())
     self.access = Access.objects.create(key='test_oauth_key',
                                         secret='super secret',
                                         user=self.profile)
     self.auth = authentication.RestOAuthAuthentication()
     self.middlewares = [APIBaseMiddleware, RestOAuthMiddleware]
     unpin_this_thread()