Source code for galaxy.model.index_filter_util

"""Utility functions used to adapt galaxy.util.search to Galaxy model index queries."""
from typing import (
    List,
    Union,
)

from sqlalchemy import (
    and_,
    or_,
)
from sqlalchemy.orm import (
    aliased,
    InstrumentedAttribute,
)
from sqlalchemy.sql.expression import BinaryExpression

from galaxy import model
from galaxy.util.search import (
    FilteredTerm,
    RawTextTerm,
)


[docs]def text_column_filter(column, term: FilteredTerm):
    if term.quoted:
        filter = column == term.text
    else:
        filter = column.ilike(f"%{term.text}%")
    return filter


RawTextSearchableT = Union[BinaryExpression, InstrumentedAttribute]


[docs]def raw_text_column_filter(columns: List[RawTextSearchableT], term: RawTextTerm):
    like_text = f"%{term.text}%"
    return or_(*[c.ilike(like_text) if isinstance(c, InstrumentedAttribute) else c for c in columns])


[docs]def tag_filter(assocation_model_class, term_text, quoted: bool = False):
    if ":" in term_text:
        key, value = term_text.rsplit(":", 1)
        if not quoted:
            return and_(
                assocation_model_class.user_tname.ilike(key),
                assocation_model_class.user_value.ilike(f"%{value}%"),
            )
        else:
            return and_(
                assocation_model_class.user_tname == key,
                assocation_model_class.user_value == value,
            )
    else:
        if not quoted:
            return assocation_model_class.user_tname.ilike(f"%{term_text}%")
        else:
            return assocation_model_class.user_tname == term_text


[docs]def append_user_filter(query, model_class, term: FilteredTerm):
    alias = aliased(model.User)
    query = query.outerjoin(model_class.user.of_type(alias))
    query = query.filter(text_column_filter(alias.username, term))
    return query