Python sqlalchemy.orm 模块，joinedload_all() 实例源码

def vip_game(self):
        """Start the process of the job as a game creation."""
        self.print_info('Hosting game %s' % self.job.match_id)

        # Copy the match data from the database
        with self.app.app_context():
            self.match = Match.query.filter_by(id=self.job.match_id).first()
            if self.match is None or self.match.status != constants.MATCH_STATUS_CREATION:
                self.dota.leave_practice_lobby()
                self.end_job_processing()
            else:
                self.players = {}
                for player in PlayerInMatch.query.\
                    options(joinedload_all('player')).\
                    filter(PlayerInMatch.match_id == self.job.match_id).\
                    all():
                    self.players[player.player_id] = player

                db.session.expunge(self.match)
                for player_id, player in self.players.items():
                    db.session.expunge(player)

                # Start the Dota lobby
                self.dota.create_practice_lobby(password=self.match.password)
                self.game_creation_call = True

def _floating_ip_get_by_address(context, address):

    # if address string is empty explicitly set it to None
    if not address:
        address = None
    try:
        result = model_query(context, models.FloatingIp).\
                    filter_by(address=address).\
                    options(joinedload_all('fixed_ip.instance')).\
                    first()

        if not result:
            raise exception.FloatingIpNotFoundForAddress(address=address)
    except db_exc.DBError:
        msg = _("Invalid floating IP %s in request") % address
        LOG.warning(msg)
        raise exception.InvalidIpAddressError(msg)

    # If the floating IP has a project ID set, check to make sure
    # the non-admin user has access.
    if result.project_id and nova.context.is_user_context(context):
        nova.context.authorize_project_context(context, result.project_id)

    return result

def _build_instance_get(context, columns_to_join=None):
    query = model_query(context, models.Instance, project_only=True).\
            options(joinedload_all('security_groups.rules')).\
            options(joinedload('info_cache'))
    if columns_to_join is None:
        columns_to_join = ['metadata', 'system_metadata']
    for column in columns_to_join:
        if column in ['info_cache', 'security_groups']:
            # Already always joined above
            continue
        if 'extra.' in column:
            query = query.options(undefer(column))
        else:
            query = query.options(joinedload(column))
    # NOTE(alaski) Stop lazy loading of columns not needed.
    for col in ['metadata', 'system_metadata']:
        if col not in columns_to_join:
            query = query.options(noload(col))
    return query

def security_group_get(context, security_group_id, columns_to_join=None):
    query = _security_group_get_query(context, project_only=True).\
                    filter_by(id=security_group_id)

    if columns_to_join is None:
        columns_to_join = []
    for column in columns_to_join:
        if column.startswith('instances'):
            query = query.options(joinedload_all(column))

    result = query.first()
    if not result:
        raise exception.SecurityGroupNotFound(
                security_group_id=security_group_id)

    return result

def security_group_get_by_name(context, project_id, group_name,
                               columns_to_join=None):
    query = _security_group_get_query(context,
                                      read_deleted="no", join_rules=False).\
            filter_by(project_id=project_id).\
            filter_by(name=group_name)

    if columns_to_join is None:
        columns_to_join = ['instances', 'rules.grantee_group']

    for column in columns_to_join:
        query = query.options(joinedload_all(column))

    result = query.first()
    if not result:
        raise exception.SecurityGroupNotFoundForProject(
                project_id=project_id, security_group_id=group_name)

    return result

def security_group_update(context, security_group_id, values,
                          columns_to_join=None):
    query = model_query(context, models.SecurityGroup).filter_by(
        id=security_group_id)
    if columns_to_join:
        for column in columns_to_join:
            query = query.options(joinedload_all(column))
    security_group_ref = query.first()

    if not security_group_ref:
        raise exception.SecurityGroupNotFound(
                security_group_id=security_group_id)
    security_group_ref.update(values)
    name = security_group_ref['name']
    project_id = security_group_ref['project_id']
    try:
        security_group_ref.save(context.session)
    except db_exc.DBDuplicateEntry:
        raise exception.SecurityGroupExists(
                project_id=project_id,
                security_group_name=name)
    return security_group_ref

def reprocess_content(self):
        """ Allows re-parsing all content as if it were imported for the first time
            but without re-hitting the source, or changing the object ids.
            Call when a code change would change the representation in the database
            """
        session = self.db
        emails = session.query(Email).filter(
                Email.source_id == self.id,
                ).options(joinedload_all(Email.parent))
        for email in emails:
            #session = self.db
            #session.add(email)
            blob = email.imported_blob
            if not isinstance(blob, native_str):
                blob = blob.decode('ascii')
            (email_object, dummy, error) = self.parse_email(blob, email)
            #session.add(email_object)
            session.commit()
            #session.remove()

        with transaction.manager:
            self.thread_mails(emails)

def _get_extracts_real(request, view_def='default', ids=None, user_id=None):
    discussion = request.discussion
    user_id = user_id or Everyone
    all_extracts = discussion.db.query(Extract).filter(
        Extract.discussion_id == discussion.id
    )
    if ids:
        ids = [Extract.get_database_id(id) for id in ids]
        all_extracts = all_extracts.filter(Extract.id.in_(ids))

    all_extracts = all_extracts.options(joinedload_all(
        Extract.content))
    all_extracts = all_extracts.options(
        joinedload_all(Extract.text_fragment_identifiers).joinedload(
            TextFragmentIdentifier.extract, innerjoin=True))
    permissions = request.permissions

    return [extract.generic_json(view_def, user_id, permissions)
            for extract in all_extracts]

def process_game_dodge(self):
        """Punish players stopping game start."""
        self.print_info('Game %s cancelled because of dodge.' % self.job.match_id)

        # Say: Partie annulée - punish
        with self.app.app_context():
            match = Match.query.filter_by(id=self.job.match_id).first()
            match.status = constants.MATCH_STATUS_CANCELLED
            self.compute_player_status()
            for player in PlayerInMatch.query. \
                    options(joinedload_all('player')). \
                    filter(PlayerInMatch.match_id == self.job.match_id). \
                    all():
                if player.player.current_match == self.job.match_id:
                    player.player.current_match = None

                # Update Scoreboard
                if player.player_id in self.missing_players or player.player_id in self.wrong_team_players:
                    score = Scoreboard.query.filter_by(ladder_name=match.section, user_id=player.player_id).first()
                    if score is None:
                        score = Scoreboard(user=player.player, ladder_name=match.section)
                        db.session.add(score)
                    player.is_dodge = True
                    score.points -= 2
                    score.dodge += 1
            db.session.commit()

def select_related(self, *columns, **options):
        depth = options.pop('depth', None)
        if options:
            raise TypeError('Unexpected argument %r' % iter(options).next())
        if depth not in (None, 1):
            raise TypeError('Depth can only be 1 or None currently')
        need_all = depth is None
        columns = list(columns)
        for idx, column in enumerate(columns):
            column = column.replace('__', '.')
            if '.' in column:
                need_all = True
            columns[idx] = column
        func = (need_all and joinedload_all or joinedload)
        return self.options(func(*columns))

def select_related(self, *columns, **options):
        depth = options.pop('depth', None)
        if options:
            raise TypeError('Unexpected argument %r' % iter(options).next())
        if depth not in (None, 1):
            raise TypeError('Depth can only be 1 or None currently')
        need_all = depth is None
        columns = list(columns)
        for idx, column in enumerate(columns):
            column = column.replace('__', '.')
            if '.' in column:
                need_all = True
            columns[idx] = column
        func = (need_all and joinedload_all or joinedload)
        return self.options(func(*columns))

def floating_ip_get(context, id):
    try:
        result = model_query(context, models.FloatingIp, project_only=True).\
                     filter_by(id=id).\
                     options(joinedload_all('fixed_ip.instance')).\
                     first()

        if not result:
            raise exception.FloatingIpNotFound(id=id)
    except db_exc.DBError:
        msg = _LW("Invalid floating IP ID %s in request") % id
        LOG.warning(msg)
        raise exception.InvalidID(id=id)
    return result

def floating_ip_get_all_by_project(context, project_id):
    nova.context.authorize_project_context(context, project_id)
    # TODO(tr3buchet): why do we not want auto_assigned floating IPs here?
    return _floating_ip_get_all(context).\
                         filter_by(project_id=project_id).\
                         filter_by(auto_assigned=False).\
                         options(joinedload_all('fixed_ip.instance')).\
                         all()

def _fixed_ip_get_by_address(context, address, columns_to_join=None):
    if columns_to_join is None:
        columns_to_join = []

    try:
        result = model_query(context, models.FixedIp)
        for column in columns_to_join:
            result = result.options(joinedload_all(column))
        result = result.filter_by(address=address).first()
        if not result:
            raise exception.FixedIpNotFoundForAddress(address=address)
    except db_exc.DBError:
        msg = _("Invalid fixed IP Address %s in request") % address
        LOG.warning(msg)
        raise exception.FixedIpInvalid(msg)

    # NOTE(sirp): shouldn't we just use project_only here to restrict the
    # results?
    if (nova.context.is_user_context(context) and
            result['instance_uuid'] is not None):
        instance = _instance_get_by_uuid(
            context.elevated(read_deleted='yes'),
            result['instance_uuid'])
        nova.context.authorize_project_context(context,
                                               instance.project_id)
    return result

def security_group_rule_get_by_security_group(context, security_group_id,
                                              columns_to_join=None):
    if columns_to_join is None:
        columns_to_join = ['grantee_group.instances.system_metadata',
                           'grantee_group.instances.info_cache']
    query = (_security_group_rule_get_query(context).
             filter_by(parent_group_id=security_group_id))
    for column in columns_to_join:
        query = query.options(joinedload_all(column))
    return query.all()

def migration_get_in_progress_by_host_and_node(context, host, node):

    return model_query(context, models.Migration).\
            filter(or_(and_(models.Migration.source_compute == host,
                            models.Migration.source_node == node),
                       and_(models.Migration.dest_compute == host,
                            models.Migration.dest_node == node))).\
            filter(~models.Migration.status.in_(['accepted', 'confirmed',
                                                 'reverted', 'error',
                                                 'failed'])).\
            options(joinedload_all('instance.system_metadata')).\
            all()

def process_endgame_results(self):
        """After a game, process lobby results into database."""
        self.print_info('Game %s over.' % self.job.match_id)

        with self.app.app_context():
            match = Match.query.filter_by(id=self.job.match_id).first()
            match.status = constants.MATCH_STATUS_ENDED
            match.server = None
            if self.game_status.match_outcome == 2:
                match.radiant_win = True
            elif self.game_status.match_outcome == 3:
                match.radiant_win = False
            else:
                match.radiant_win = None

            self.players = {}
            for player in PlayerInMatch.query. \
                    options(joinedload_all('player')). \
                    filter(PlayerInMatch.match_id == self.job.match_id). \
                    all():
                if player.player.current_match == self.job.match_id:
                    player.player.current_match = None
                self.players[player.player_id] = player

            # Process scoreboard updates
            for player_id, player in self.players.items():
                score = Scoreboard.query.filter_by(ladder_name=match.section, user_id=player_id).first()
                if score is None:
                    score = Scoreboard(user=player.player, ladder_name=match.section)
                    db.session.add(score)
                score.matches += 1
            for player in self.game_status.members:
                if player.id == self.dota.steam_id:
                    continue
                id = player.id
                score = Scoreboard.query.filter_by(ladder_name=match.section, user_id=id).first()
                if (self.players[id].is_radiant and self.game_status.match_outcome == 2) or \
                        (not self.players[id].is_radiant and self.game_status.match_outcome == 3):
                    score.points += 1
                    score.win += 1
                elif (self.players[id].is_radiant and self.game_status.match_outcome == 3) or \
                        (not self.players[id].is_radiant and self.game_status.match_outcome == 2):
                    score.loss += 1
            for player in self.game_status.left_members:
                score = Scoreboard.query.filter_by(ladder_name=match.section, user_id=player.id).first()
                self.players[player.id].is_leaver = True
                score.points -= 3
                score.leave += 1

            db.session.commit()

def do_import_content(abstract_mbox, only_new=True):
        abstract_mbox = abstract_mbox.db.merge(abstract_mbox)
        session = abstract_mbox.db
        session.add(abstract_mbox)
        discussion_id = abstract_mbox.discussion_id

        if not os.path.isdir(abstract_mbox.filesystem_path):
            raise "There is no directory at %s" % abstract_mbox.filesystem_path
        else:
            cur_folder_path = os.path.join(abstract_mbox.filesystem_path, 'cur')
            cur_folder_present = os.path.isdir(cur_folder_path)
            new_folder_path = os.path.join(abstract_mbox.filesystem_path, 'new')
            new_folder_present = os.path.isdir(new_folder_path)
            tmp_folder_path = os.path.join(abstract_mbox.filesystem_path, 'tmp')
            tmp_folder_present = os.path.isdir(tmp_folder_path)

            if not (cur_folder_present | new_folder_present | tmp_folder_present):
                raise "Directory at %s is NOT a maildir" % abstract_mbox.filesystem_path
            else:
                #Fix the maildir in case some folders are missing
                #For instance, git cannot store empty folder
                if not cur_folder_present:
                    os.mkdir(cur_folder_path)
                if not new_folder_present:
                    os.mkdir(new_folder_path)
                if not tmp_folder_present:
                    os.mkdir(tmp_folder_path)

        mbox = mailbox.Maildir(abstract_mbox.filesystem_path, factory=None, create=False)
        mails = list(mbox.values())
        #import pdb; pdb.set_trace()
        def import_email(abstract_mbox, message_data):
            session = abstract_mbox.db
            message_string = message_data.as_string()

            (email_object, dummy, error) = abstract_mbox.parse_email(message_string)
            if error:
                raise Exception(error)
            session.add(email_object)
            transaction.commit()
            abstract_mbox = AbstractMailbox.get(abstract_mbox.id)

        if len(mails):
            [import_email(abstract_mbox, message_data) for message_data in mails]

            #We imported mails, we need to re-thread
            emails = session.query(Email).filter(
                    Email.discussion_id == discussion_id,
                    ).options(joinedload_all(Email.parent))

            AbstractMailbox.thread_mails(emails)
            transaction.commit()