refact(RHINENG-15309): Refactor filtering logic related to custom staleness

api/filtering/db_filters.py

@@ -16,8 +16,10 @@
 from api.filtering.db_custom_filters import build_system_profile_filter
 from api.filtering.db_custom_filters import get_host_types_from_filter
 from api.staleness_query import get_staleness_obj
+from app.auth.identity import IdentityType
+from app.config import ALL_STALENESS_STATES
 from app.config import HOST_TYPES
-from app.culling import staleness_to_conditions
+from app.culling import Conditions
 from app.exceptions import ValidationException
 from app.logging import get_logger
 from app.models import OLD_TO_NEW_REPORTER_MAP
@@ -27,9 +29,16 @@
 from app.models import db
 from app.serialization import serialize_staleness_to_dict
 from app.utils import Tag
-from lib.host_repository import ALL_STALENESS_STATES
 
-__all__ = ("canonical_fact_filter", "query_filters", "host_id_list_filter", "rbac_permissions_filter")
+__all__ = (
+    "canonical_fact_filter",
+    "query_filters",
+    "host_id_list_filter",
+    "rbac_permissions_filter",
+    "stale_timestamp_filter",
+    "staleness_to_conditions",
+    "update_query_for_owner_id",
+)
 
 logger = get_logger(__name__)
 DEFAULT_STALENESS_VALUES = ["not_culled"]
@@ -79,7 +88,7 @@ def _group_ids_filter(group_id_list: list) -> list:
     return _query_filter
 
 
-def _get_host_modified_on_time_filter(gt=None, lte=None):
+def stale_timestamp_filter(gt=None, lte=None):
     filters = []
     if gt:
         filters.append(Host.modified_on > gt)
@@ -89,18 +98,18 @@ def _get_host_modified_on_time_filter(gt=None, lte=None):
     return and_(*filters)
 
 
-def _stale_timestamp_filter(gt=None, lte=None, host_type=None):
-    return _get_host_modified_on_time_filter(gt=gt, lte=lte)
+def _host_type_filter(host_type: str | None):
+    return Host.system_profile_facts["host_type"].as_string() == host_type
 
 
-def _stale_timestamp_per_reporter_filter(gt=None, lte=None, reporter=None, host_type=None):
+def _stale_timestamp_per_reporter_filter(gt=None, lte=None, reporter=None):
     non_negative_reporter = reporter.replace("!", "")
     reporter_list = [non_negative_reporter]
     if non_negative_reporter in OLD_TO_NEW_REPORTER_MAP.keys():
         reporter_list.extend(OLD_TO_NEW_REPORTER_MAP[non_negative_reporter])
 
     if reporter.startswith("!"):
-        time_filter_ = _get_host_modified_on_time_filter(gt=gt, lte=lte)
+        time_filter_ = stale_timestamp_filter(gt=gt, lte=lte)
         return and_(
             and_(
                 not_(Host.per_reporter_staleness.has_key(rep)),
@@ -135,12 +144,13 @@ def per_reporter_staleness_filter(staleness, reporter, host_type_filter):
                 staleness,
                 host_type,
                 partial(_stale_timestamp_per_reporter_filter, reporter=reporter),
+                True,
             )
         )
         if len(host_type_filter) > 1:
             staleness_conditions.append(
                 and_(
-                    Host.system_profile_facts["host_type"].astext == host_type,
+                    _host_type_filter(host_type),
                     conditions,
                 )
             )
@@ -156,11 +166,11 @@ def _staleness_filter(
     staleness_obj = serialize_staleness_to_dict(get_staleness_obj(identity))
     staleness_conditions = []
     for host_type in host_type_filter:
-        conditions = or_(*staleness_to_conditions(staleness_obj, staleness, host_type, _stale_timestamp_filter))
+        conditions = or_(*staleness_to_conditions(staleness_obj, staleness, host_type, stale_timestamp_filter, True))
         if len(host_type_filter) > 1:
             staleness_conditions.append(
                 and_(
-                    Host.system_profile_facts["host_type"].astext == host_type,
+                    _host_type_filter(host_type),
                     conditions,
                 )
             )
@@ -170,6 +180,21 @@ def _staleness_filter(
     return [or_(*staleness_conditions)]
 
 
+def staleness_to_conditions(
+    staleness, staleness_states, host_type, timestamp_filter_func, omit_host_type_filter: bool = False
+):
+    def _timestamp_and_host_type_filter(condition, state):
+        filter_ = timestamp_filter_func(*getattr(condition, state)())
+        if omit_host_type_filter:
+            return filter_
+        else:
+            return and_(filter_, _host_type_filter(host_type))
+
+    condition = Conditions(staleness, host_type)
+    filtered_states = (state for state in staleness_states if state != "unknown")
+    return (_timestamp_and_host_type_filter(condition, state) for state in filtered_states)
+
+
 def _registered_with_filter(registered_with: list[str], host_type_filter: set[str | None]) -> list:
     _query_filter: list = []
     if not registered_with:
@@ -257,6 +282,15 @@ def rbac_permissions_filter(rbac_filter: dict) -> list:
     return _query_filter
 
 
+def update_query_for_owner_id(identity, query):
+    # kafka based requests have dummy identity for working around the identity requirement for CRUD operations
+    logger.debug("identity auth type: %s", identity.auth_type)
+    if identity and identity.identity_type == IdentityType.SYSTEM:
+        return query.filter(and_(Host.system_profile_facts["owner_id"].as_string() == identity.system["cn"]))
+    else:
+        return query
+
+
 def query_filters(
     fqdn: str | None = None,
     display_name: str | None = None,

api/host.py

@@ -14,6 +14,7 @@
 from api.cache import CACHE
 from api.cache import delete_cached_system_keys
 from api.cache_key import make_system_cache_key
+from api.filtering.db_filters import update_query_for_owner_id
 from api.host_query import build_paginated_host_list_response
 from api.host_query import staleness_timestamps
 from api.host_query_db import get_all_hosts
@@ -58,7 +59,6 @@
 from lib.host_repository import find_existing_host
 from lib.host_repository import find_non_culled_hosts
 from lib.host_repository import get_host_list_by_id_list_from_db
-from lib.host_repository import update_query_for_owner_id
 from lib.middleware import rbac
 
 FactOperations = Enum("FactOperations", ("merge", "replace"))

api/host_query_db.py

@@ -18,9 +18,11 @@
 from api.filtering.db_filters import host_id_list_filter
 from api.filtering.db_filters import query_filters
 from api.filtering.db_filters import rbac_permissions_filter
+from api.filtering.db_filters import update_query_for_owner_id
 from api.host_query import staleness_timestamps
 from api.staleness_query import get_staleness_obj
 from app.auth import get_current_identity
+from app.config import ALL_STALENESS_STATES
 from app.exceptions import InventoryException
 from app.instrumentation import log_get_host_list_succeeded
 from app.logging import get_logger
@@ -29,8 +31,6 @@
 from app.models import HostGroupAssoc
 from app.models import db
 from app.serialization import serialize_host_for_export_svc
-from lib.host_repository import ALL_STALENESS_STATES
-from lib.host_repository import update_query_for_owner_id
 
 __all__ = (
     "get_all_hosts",

app/config.py

@@ -12,6 +12,7 @@
 PRODUCER_ACKS = {"0": 0, "1": 1, "all": "all"}
 
 HOST_TYPES = ["edge", None]
+ALL_STALENESS_STATES = ("fresh", "stale", "stale_warning")
 
 
 class Config:

app/culling.py

@@ -3,7 +3,7 @@
 from datetime import timedelta
 from datetime import timezone
 
-__all__ = ("Conditions", "staleness_to_conditions", "Timestamps", "days_to_seconds")
+__all__ = ("Conditions", "Timestamps", "days_to_seconds")
 
 
 class _Config(namedtuple("_Config", ("stale_warning_offset_delta", "culled_offset_delta"))):
@@ -93,12 +93,6 @@ def find_host_state(stale_timestamp, stale_warning_timestamp):
             return "stale warning"
 
 
-def staleness_to_conditions(staleness, staleness_states, host_type, timestamp_filter_func):
-    condition = Conditions(staleness, host_type)
-    filtered_states = (state for state in staleness_states if state != "unknown")
-    return (timestamp_filter_func(*getattr(condition, state)(), host_type=host_type) for state in filtered_states)
-
-
 def days_to_seconds(n_days: int) -> int:
     factor = 86400
     return n_days * factor

lib/host_repository.py

@@ -5,12 +5,14 @@
 from sqlalchemy import not_
 from sqlalchemy import or_
 
+from api.filtering.db_filters import stale_timestamp_filter
+from api.filtering.db_filters import staleness_to_conditions
+from api.filtering.db_filters import update_query_for_owner_id
 from api.staleness_query import get_staleness_obj
 from api.staleness_query import get_sys_default_staleness
 from app.auth import get_current_identity
-from app.auth.identity import IdentityType
+from app.config import ALL_STALENESS_STATES
 from app.config import HOST_TYPES
-from app.culling import staleness_to_conditions
 from app.exceptions import InventoryException
 from app.logging import get_logger
 from app.models import Host
@@ -27,9 +29,7 @@
     "find_host_by_multiple_canonical_facts",
     "find_hosts_by_staleness",
     "find_non_culled_hosts",
-    "stale_timestamp_filter",
     "update_existing_host",
-    "update_query_for_owner_id",
 )
 
 AddHostResult = Enum("AddHostResult", ("created", "updated"))
@@ -44,7 +44,6 @@
 IMMUTABLE_CANONICAL_FACTS = ("provider_id",)
 MUTABLE_CANONICAL_FACTS = tuple(set(ELEVATED_CANONICAL_FACT_FIELDS).difference(set(IMMUTABLE_CANONICAL_FACTS)))
 
-ALL_STALENESS_STATES = ("fresh", "stale", "stale_warning")
 NULL = None
 
 logger = get_logger(__name__)
@@ -262,15 +261,6 @@ def update_existing_host(existing_host, input_host, update_system_profile):
     return existing_host, AddHostResult.updated
 
 
-def stale_timestamp_filter(gt=None, lte=None, host_type=None):
-    filter_ = ()
-    if gt:
-        filter_ += (Host.modified_on > gt,)
-    if lte:
-        filter_ += (Host.modified_on <= lte,)
-    return and_(*filter_, (Host.system_profile_facts["host_type"].as_string() == host_type))
-
-
 def contains_no_incorrect_facts_filter(canonical_facts):
     # Does not contain any incorrect CF values
     # Incorrect value = AND( key exists, NOT( contains key:value ) )
@@ -300,15 +290,6 @@ def matches_at_least_one_canonical_fact_filter(canonical_facts):
     return or_(*filter_)
 
 
-def update_query_for_owner_id(identity, query):
-    # kafka based requests have dummy identity for working around the identity requirement for CRUD operations
-    logger.debug("identity auth type: %s", identity.auth_type)
-    if identity and identity.identity_type == IdentityType.SYSTEM:
-        return query.filter(and_(Host.system_profile_facts["owner_id"].as_string() == identity.system["cn"]))
-    else:
-        return query
-
-
 def update_system_profile(input_host, identity):
     if not input_host.system_profile_facts:
         raise InventoryException(