Merge branch 'release_25.0' into dev

        return self._populated_optimized

    def expire_populated_state(self):

        required_object_session(self).expire(self, ("populated_state",))

    @property

    def allow_implicit_mapping(self):

        return self.collection_type != "record"

    def __init__(self) -> None:

        self.objects: dict[type, dict] = defaultdict(dict)

    def execute(self, query: Any, *args, **kwargs) -> Any:

        pass

    def delete(self, obj: model.RepresentById) -> None:

        self.objects[obj.__class__].pop(obj.id, None)

    def scalars(self, query: Any, *args, **kwargs) -> Any:

        pass

    def commit(self) -> None:

        pass

from typing import (

    Optional,

    TYPE_CHECKING,

    Union,

)

from sqlalchemy.exc import IntegrityError

if TYPE_CHECKING:

    from galaxy.model import User

    from galaxy.model.store import SessionlessContext

log = logging.getLogger(__name__)

    Manages CRUD operations related to tagging objects.

    """

    def __init__(self, sa_session: scoped_session, galaxy_session: Optional[GalaxySession] = None) -> None:

    def __init__(

        self, sa_session: Union[scoped_session, "SessionlessContext"], galaxy_session: Optional[GalaxySession] = None

    ) -> None:

        self.sa_session = sa_session

        # Minimum tag length.

        self.min_tag_len = 1

    def create_tag_handler_session(self, galaxy_session: Optional[GalaxySession]):

        # Creates a transient tag handler that avoids repeated flushes

        if isinstance(self.sa_session, scoped_session):

            return GalaxyTagHandlerSession(self.sa_session, galaxy_session=galaxy_session)

        else:

            return self

    def add_tags_from_list(self, user, item, new_tags_list, flush=True):

        new_tags_set = set(new_tags_list)

    def _create_tag_instance(self, tag_name):

        # For good performance caller should first check if there's already an appropriate tag

        tag = Tag(type=0, name=tag_name)

        if not self.sa_session:

        if not isinstance(self.sa_session, scoped_session):

            return tag

        Session = sessionmaker(self.sa_session.bind)

        with Session() as separate_session:

                    # If we are not waiting for elements, there was some

                    # problem creating the collection. Collection will never

                    # be populated.

                    # We want to be certain of this however, so refresh attribute ...

                    replacement.collection.expire_populated_state()

                    # ... and repeat check to avoid race condition

                    if not replacement.collection.populated:

                        raise modules.FailWorkflowEvaluation(

                            why=InvocationFailureCollectionFailed(

                                reason=FailureReason.collection_failed,

        "metadata_bam",

        "job_properties",

        "from_work_dir_glob",

        "gx_group_tag",

    ]

)