<commit_before>from gc import collect as garbage_collect

from gevent import Greenlet, joinall, sleep
from gevent.queue import Queue, Empty
from sqlalchemy.exc import DataError

from inbox.log import get_logger
logger = get_logger()
from inbox.util.concurrency import retry_with_logging, retry_and_report_killed
from inbox.util.itert import partition
from inbox.models import (Account, Folder, MAX_FOLDER_NAME_LENGTH)
from inbox.mailsync.exc import SyncException
from inbox.mailsync.reporting import report_stopped


class MailsyncError(Exception):
    pass


def verify_folder_name(account_id, old, new):
    if old is not None and old.name != new.name:
        raise SyncException(
            "Core folder on account {} changed name from '{}' to '{}'".format(
                account_id, old.name, new.name))
    return new


def save_folder_names(log, account, folder_names, db_session):
    """
    Create Folder objects & map special folder names on Account objects.

    Folders that belong to an account and no longer exist in `folder_names`
    ARE DELETED, unless they are "dangling" (do not have a 'name' set).

    We don't canonicalizing folder names to lowercase when saving
    because different backends may be case-sensitive or not. Code that
    references saved folder names should canonicalize if needed when doing
    comparisons.

    """
    assert 'inbox' in folder_names, 'Account {} has no detected inbox folder'\
        .format(account.email_address)

    all_folders = db_session.query(Folder).filter_by(
        account_id=account.id).all()
    # dangled_folders don't map to upstream account folders (may be used for
    # keeping track of e.g. special Gmail labels which are exposed as IMAP
    # flags but not folders)
    folder_for = {f.name.lower(): f for f in all_folders if f.name is not None}
    dangled_folder_for = {f.canonical_name: f for f in all_folders
                          if f.name is None}

    canonical_names = {'inbox', 'drafts', 'sent', 'spam', 'trash',
                       'starred', 'important', 'archive', 'all'}
    for canonical_name in canonical_names:
        if canonical_name in folder_names:
            backend_folder_name = folder_names[canonical_name].lower()
            if backend_folder_name not in folder_for:
                # Reconcile dangled folders which now exist on the remote
                if canonical_name in dangled_folder_for:
                    folder = dangled_folder_for[canonical_name]
                    folder.name = folder_names[canonical_name]
                    del dangled_folder_for[canonical_name]
                else:
                    folder = Folder.create(account,
                                           folder_names[canonical_name],
                                           db_session, canonical_name)
                attr_name = '{}_folder'.format(canonical_name)
                setattr(account, attr_name, verify_folder_name(
                    account.id, getattr(account, attr_name), folder))
            else:
                del folder_for[backend_folder_name]

    # Gmail labels, user-created IMAP/EAS folders, etc.
    if 'extra' in folder_names:
        for name in folder_names['extra']:
            name = name[:MAX_FOLDER_NAME_LENGTH]
            if name.lower() not in folder_for:
                folder = Folder.create(account, name, db_session)
                db_session.add(folder)
            if name.lower() in folder_for:
                del folder_for[name.lower()]

    # This may cascade to FolderItems and ImapUid (ONLY), which is what we
    # want--doing the update here short-circuits us syncing that change later.
    log.info("folders deleted from remote", folders=folder_for.keys())
    for name, folder in folder_for.iteritems():
        db_session.delete(folder)
        # TODO(emfree) delete associated tag

    # Create associated tags for any new folders.
    for folder in account.folders:
        folder.get_associated_tag(db_session)

    db_session.commit()


def gevent_check_join(log, threads, errmsg):
    """ Block until all threads have completed and throw an error if threads
        are not successful.
    """
    joinall(threads)
    errors = [thread.exception for thread in threads
              if not thread.successful()]
    if errors:
        log.error(errmsg)
        for error in errors:
            log.error(error)
        raise SyncException("Fatal error encountered")


def create_db_objects(account_id, db_session, log, folder_name, raw_messages,
                      msg_create_fn):
    new_uids = []
    # TODO: Detect which namespace to add message to. (shared folders)
    # Look up message thread,
    acc = db_session.query(Account).get(account_id)
    folder = Folder.find_or_create(db_session, acc, folder_name)
    for msg in raw_messages:
        uid = msg_create_fn(db_session, log, acc, folder, msg)
        if uid is not None:
            new_uids.append(uid)

    # imapuid, message, thread, labels
    return new_uids


def commit_uids(db_session, log, new_uids):
    new_messages = [item.message for item in new_uids]

    # Save message part blobs before committing changes to db.
    for msg in new_messages:
        threads = [Greenlet.spawn(retry_with_logging, lambda:
                                  part.save(part._data), log)
                   for part in msg.parts if hasattr(part, '_data')]
        # Fatally abort if part saves error out. Messages in this
        # chunk will be retried when the sync is restarted.
        gevent_check_join(log, threads,
                          "Could not save message parts to blob store!")
        # clear data to save memory
        for part in msg.parts:
            part._data = None

    garbage_collect()

    try:
        log.info("Committing {0} UIDs".format(len(new_uids)))
        db_session.add_all(new_uids)
        db_session.commit()
    except DataError as e:
        db_session.rollback()<commit_msg>Fix database session bug which results in creation of multiple Thread objects on Gmail.<commit_after>from gc import collect as garbage_collect

from gevent import Greenlet, joinall, sleep
from gevent.queue import Queue, Empty
from sqlalchemy.exc import DataError

from inbox.log import get_logger
logger = get_logger()
from inbox.util.concurrency import retry_with_logging, retry_and_report_killed
from inbox.util.itert import partition
from inbox.models import (Account, Folder, MAX_FOLDER_NAME_LENGTH)
from inbox.mailsync.exc import SyncException
from inbox.mailsync.reporting import report_stopped


class MailsyncError(Exception):
    pass


def verify_folder_name(account_id, old, new):
    if old is not None and old.name != new.name:
        raise SyncException(
            "Core folder on account {} changed name from '{}' to '{}'".format(
                account_id, old.name, new.name))
    return new


def save_folder_names(log, account, folder_names, db_session):
    """
    Create Folder objects & map special folder names on Account objects.

    Folders that belong to an account and no longer exist in `folder_names`
    ARE DELETED, unless they are "dangling" (do not have a 'name' set).

    We don't canonicalizing folder names to lowercase when saving
    because different backends may be case-sensitive or not. Code that
    references saved folder names should canonicalize if needed when doing
    comparisons.

    """
    assert 'inbox' in folder_names, 'Account {} has no detected inbox folder'\
        .format(account.email_address)

    all_folders = db_session.query(Folder).filter_by(
        account_id=account.id).all()
    # dangled_folders don't map to upstream account folders (may be used for
    # keeping track of e.g. special Gmail labels which are exposed as IMAP
    # flags but not folders)
    folder_for = {f.name.lower(): f for f in all_folders if f.name is not None}
    dangled_folder_for = {f.canonical_name: f for f in all_folders
                          if f.name is None}

    canonical_names = {'inbox', 'drafts', 'sent', 'spam', 'trash',
                       'starred', 'important', 'archive', 'all'}
    for canonical_name in canonical_names:
        if canonical_name in folder_names:
            backend_folder_name = folder_names[canonical_name].lower()
            if backend_folder_name not in folder_for:
                # Reconcile dangled folders which now exist on the remote
                if canonical_name in dangled_folder_for:
                    folder = dangled_folder_for[canonical_name]
                    folder.name = folder_names[canonical_name]
                    del dangled_folder_for[canonical_name]
                else:
                    folder = Folder.create(account,
                                           folder_names[canonical_name],
                                           db_session, canonical_name)
                attr_name = '{}_folder'.format(canonical_name)
                setattr(account, attr_name, verify_folder_name(
                    account.id, getattr(account, attr_name), folder))
            else:
                del folder_for[backend_folder_name]

    # Gmail labels, user-created IMAP/EAS folders, etc.
    if 'extra' in folder_names:
        for name in folder_names['extra']:
            name = name[:MAX_FOLDER_NAME_LENGTH]
            if name.lower() not in folder_for:
                folder = Folder.create(account, name, db_session)
                db_session.add(folder)
            if name.lower() in folder_for:
                del folder_for[name.lower()]

    # This may cascade to FolderItems and ImapUid (ONLY), which is what we
    # want--doing the update here short-circuits us syncing that change later.
    log.info("folders deleted from remote", folders=folder_for.keys())
    for name, folder in folder_for.iteritems():
        db_session.delete(folder)
        # TODO(emfree) delete associated tag

    # Create associated tags for any new folders.
    for folder in account.folders:
        folder.get_associated_tag(db_session)

    db_session.commit()


def gevent_check_join(log, threads, errmsg):
    """ Block until all threads have completed and throw an error if threads
        are not successful.
    """
    joinall(threads)
    errors = [thread.exception for thread in threads
              if not thread.successful()]
    if errors:
        log.error(errmsg)
        for error in errors:
            log.error(error)
        raise SyncException("Fatal error encountered")


def create_db_objects(account_id, db_session, log, folder_name, raw_messages,
                      msg_create_fn):
    new_uids = []
    # TODO: Detect which namespace to add message to. (shared folders)
    # Look up message thread,
    acc = db_session.query(Account).get(account_id)
    folder = Folder.find_or_create(db_session, acc, folder_name)
    for msg in raw_messages:
        uid = msg_create_fn(db_session, log, acc, folder, msg)
        # Must ensure message objects are flushed because they reference
        # threads, which may be new, and later messages may need to belong to
        # the same thread. If we don't flush here and disable autoflush within
        # the message creation to avoid flushing incomplete messages, we can't
        # query for the (uncommitted) new thread id.
        #
        # We should probably refactor this later to use provider-specific
        # Message constructors to avoid creating incomplete objects in the
        # first place.
        db_session.add(uid)
        db_session.flush()
        if uid is not None:
            new_uids.append(uid)

    # imapuid, message, thread, labels
    return new_uids


def commit_uids(db_session, log, new_uids):
    new_messages = [item.message for item in new_uids]

    # Save message part blobs before committing changes to db.
    for msg in new_messages:
        threads = [Greenlet.spawn(retry_with_logging, lambda:
                                  part.save(part._data), log)
                   for part in msg.parts if hasattr(part, '_data')]
        # Fatally abort if part saves error out. Messages in this
        # chunk will be retried when the sync is restarted.
        gevent_check_join(log, threads,
                          "Could not save message parts to blob store!")
        # clear data to save memory
        for part in msg.parts:
            part._data = None

    garbage_collect()

    try:
        log.info("Committing {0} UIDs".format(len(new_uids)))
        db_session.add_all(new_uids)
        db_session.commit()
    except DataError as e:
        db_session.rollback()
{'fraction_non_alphanumeric': 0.07336029819969862, 'fraction_numerical': 0.00015861686097232136, 'mean_word_length': 2.488243430152144, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '497859', 'n_tokens_mistral': 3560, 'n_tokens_neox': 3415, 'n_words': 1157}