narrow: Use NarrowParameter instead of dictionary.

Earlier we were using the type `OptionalNarrowListT` for all functions that required "narrow" as a parameter. This commit changes all the functions accepting a "narrow" to use a list of the new `NarrowParameter` instead of `OptionalNarrowListT` which is a list of dicts.
2025-11-04 14:03:30 +00:00 · 2024-06-19 19:17:34 +05:30
parent 4e05381897
commit 15accfc983
4 changed files with 231 additions and 215 deletions
--- a/zerver/lib/narrow.py
+++ b/zerver/lib/narrow.py
@@ -150,7 +150,7 @@ class NarrowParameter(BaseModel):
        return self


-def is_spectator_compatible(narrow: Iterable[Dict[str, Any]]) -> bool:
+def is_spectator_compatible(narrow: Iterable[NarrowParameter]) -> bool:
    # This implementation should agree with is_spectator_compatible in hash_parser.ts.
    supported_operators = [
        *channel_operators,
@@ -163,15 +163,13 @@ def is_spectator_compatible(narrow: Iterable[Dict[str, Any]]) -> bool:
        "id",
    ]
    for element in narrow:
-        operator = element["operator"]
-        if "operand" not in element:
-            return False
+        operator = element.operator
        if operator not in supported_operators:
            return False
    return True


-def is_web_public_narrow(narrow: Optional[Iterable[Dict[str, Any]]]) -> bool:
+def is_web_public_narrow(narrow: Optional[Iterable[NarrowParameter]]) -> bool:
    if narrow is None:
        return False

@@ -179,9 +177,9 @@ def is_web_public_narrow(narrow: Optional[Iterable[Dict[str, Any]]]) -> bool:
        # Web-public queries are only allowed for limited types of narrows.
        # term == {'operator': 'channels', 'operand': 'web-public', 'negated': False}
        # or term == {'operator': 'streams', 'operand': 'web-public', 'negated': False}
-        term["operator"] in channels_operators
-        and term["operand"] == "web-public"
-        and term["negated"] is False
+        term.operator in channels_operators
+        and term.operand == "web-public"
+        and term.negated is False
        for term in narrow
    )

@@ -204,8 +202,6 @@ class BadNarrowOperatorError(JsonableError):

 ConditionTransform: TypeAlias = Callable[[ClauseElement], ClauseElement]

-OptionalNarrowListT: TypeAlias = Optional[List[Dict[str, Any]]]
-
 # These delimiters will not appear in rendered messages or HTML-escaped topics.
 TS_START = "<ts-match>"
 TS_STOP = "</ts-match>"
@@ -307,7 +303,7 @@ class NarrowBuilder:
                "No message can be both a channel message and direct message"
            )

-    def add_term(self, query: Select, term: Dict[str, Any]) -> Select:
+    def add_term(self, query: Select, term: NarrowParameter) -> Select:
        """
        Extend the given query to one narrowed by the given term, and return the result.

@@ -320,10 +316,10 @@ class NarrowBuilder:
        # methods to the same criterion.  See the class's block comment
        # for details.

-        operator = term["operator"]
-        operand = term["operand"]
+        operator = term.operator
+        operand = term.operand

-        negated = term.get("negated", False)
+        negated = term.negated

        if operator in self.by_method_map:
            method = self.by_method_map[operator]
@@ -804,7 +800,9 @@ class NarrowBuilder:


 def ok_to_include_history(
-    narrow: OptionalNarrowListT, user_profile: Optional[UserProfile], is_web_public_query: bool
+    narrow: Optional[List[NarrowParameter]],
+    user_profile: Optional[UserProfile],
+    is_web_public_query: bool,
 ) -> bool:
    # There are occasions where we need to find Message rows that
    # have no corresponding UserMessage row, because the user is
@@ -830,16 +828,16 @@ def ok_to_include_history(
    include_history = False
    if narrow is not None:
        for term in narrow:
-            if term["operator"] in channel_operators and not term.get("negated", False):
-                operand: Union[str, int] = term["operand"]
+            if term.operator in channel_operators and not term.negated:
+                operand: Union[str, int] = term.operand
                if isinstance(operand, str):
                    include_history = can_access_stream_history_by_name(user_profile, operand)
                else:
                    include_history = can_access_stream_history_by_id(user_profile, operand)
            elif (
-                term["operator"] in channels_operators
-                and term["operand"] == "public"
-                and not term.get("negated", False)
+                term.operator in channels_operators
+                and term.operand == "public"
+                and not term.negated
                and user_profile.can_access_public_streams()
            ):
                include_history = True
@@ -847,24 +845,24 @@ def ok_to_include_history(
        # that's a property on the UserMessage table.  There cannot be
        # historical messages in these cases anyway.
        for term in narrow:
-            if term["operator"] == "is" and term["operand"] not in {"resolved", "followed"}:
+            if term.operator == "is" and term.operand not in {"resolved", "followed"}:
                include_history = False

    return include_history


 def get_channel_from_narrow_access_unchecked(
-    narrow: OptionalNarrowListT, realm: Realm
+    narrow: Optional[List[NarrowParameter]], realm: Realm
 ) -> Optional[Stream]:
    if narrow is not None:
        for term in narrow:
-            if term["operator"] in channel_operators:
-                return get_stream_by_narrow_operand_access_unchecked(term["operand"], realm)
+            if term.operator in channel_operators:
+                return get_stream_by_narrow_operand_access_unchecked(term.operand, realm)
    return None


 def exclude_muting_conditions(
-    user_profile: UserProfile, narrow: OptionalNarrowListT
+    user_profile: UserProfile, narrow: Optional[List[NarrowParameter]]
 ) -> List[ClauseElement]:
    conditions: List[ClauseElement] = []
    channel_id = None
@@ -957,7 +955,7 @@ def add_narrow_conditions(
    user_profile: Optional[UserProfile],
    inner_msg_id_col: ColumnElement[Integer],
    query: Select,
-    narrow: OptionalNarrowListT,
+    narrow: Optional[List[NarrowParameter]],
    is_web_public_query: bool,
    realm: Realm,
 ) -> Tuple[Select, bool]:
@@ -974,15 +972,15 @@ def add_narrow_conditions(
    # our query, but we need to collect the search operands and handle
    # them after the loop.
    for term in narrow:
-        if term["operator"] == "search":
-            search_operands.append(term["operand"])
+        if term.operator == "search":
+            search_operands.append(term.operand)
        else:
            query = builder.add_term(query, term)

    if search_operands:
        is_search = True
        query = query.add_columns(topic_column_sa(), column("rendered_content", Text))
-        search_term = dict(
+        search_term = NarrowParameter(
            operator="search",
            operand=" ".join(search_operands),
        )
@@ -992,7 +990,9 @@ def add_narrow_conditions(


 def find_first_unread_anchor(
-    sa_conn: Connection, user_profile: Optional[UserProfile], narrow: OptionalNarrowListT
+    sa_conn: Connection,
+    user_profile: Optional[UserProfile],
+    narrow: Optional[List[NarrowParameter]],
 ) -> int:
    # For anonymous web users, all messages are treated as read, and so
    # always return LARGER_THAN_MAX_MESSAGE_ID.
@@ -1253,7 +1253,7 @@ class FetchedMessages(LimitedMessages[Row]):

 def fetch_messages(
    *,
-    narrow: OptionalNarrowListT,
+    narrow: Optional[List[NarrowParameter]],
    user_profile: Optional[UserProfile],
    realm: Realm,
    is_web_public_query: bool,
--- a/zerver/tests/test_message_fetch.py
+++ b/zerver/tests/test_message_fetch.py
@@ -34,6 +34,7 @@ from zerver.lib.narrow import (
    LARGER_THAN_MAX_MESSAGE_ID,
    BadNarrowOperatorError,
    NarrowBuilder,
+    NarrowParameter,
    exclude_muting_conditions,
    find_first_unread_anchor,
    is_spectator_compatible,
@@ -122,31 +123,31 @@ class NarrowBuilderTest(ZulipTestCase):
        self.othello_email = self.example_user("othello").email

    def test_add_term_using_not_defined_operator(self) -> None:
-        term = dict(operator="not-defined", operand="any")
+        term = NarrowParameter(operator="not-defined", operand="any")
        self.assertRaises(BadNarrowOperatorError, self._build_query, term)

    def test_add_term_using_channel_operator(self) -> None:
-        term = dict(operator="channel", operand="Scotland")
+        term = NarrowParameter(operator="channel", operand="Scotland")
        self._do_add_term_test(term, "WHERE recipient_id = %(recipient_id_1)s")

    def test_add_term_using_channel_operator_and_negated(self) -> None:  # NEGATED
-        term = dict(operator="channel", operand="Scotland", negated=True)
+        term = NarrowParameter(operator="channel", operand="Scotland", negated=True)
        self._do_add_term_test(term, "WHERE recipient_id != %(recipient_id_1)s")

    def test_add_term_using_channel_operator_and_non_existing_operand_should_raise_error(
        self,
    ) -> None:  # NEGATED
-        term = dict(operator="channel", operand="non-existing-channel")
+        term = NarrowParameter(operator="channel", operand="non-existing-channel")
        self.assertRaises(BadNarrowOperatorError, self._build_query, term)

    def test_add_term_using_channels_operator_and_invalid_operand_should_raise_error(
        self,
    ) -> None:  # NEGATED
-        term = dict(operator="channels", operand="invalid_operands")
+        term = NarrowParameter(operator="channels", operand="invalid_operands")
        self.assertRaises(BadNarrowOperatorError, self._build_query, term)

    def test_add_term_using_channels_operator_and_public_operand(self) -> None:
-        term = dict(operator="channels", operand="public")
+        term = NarrowParameter(operator="channels", operand="public")
        self._do_add_term_test(
            term,
            "WHERE recipient_id IN (__[POSTCOMPILE_recipient_id_1])",
@@ -182,7 +183,7 @@ class NarrowBuilderTest(ZulipTestCase):
        )

    def test_add_term_using_channels_operator_and_public_operand_negated(self) -> None:
-        term = dict(operator="channels", operand="public", negated=True)
+        term = NarrowParameter(operator="channels", operand="public", negated=True)
        self._do_add_term_test(
            term,
            "WHERE (recipient_id NOT IN (__[POSTCOMPILE_recipient_id_1]))",
@@ -218,28 +219,28 @@ class NarrowBuilderTest(ZulipTestCase):
        )

    def test_add_term_using_is_operator_and_dm_operand(self) -> None:
-        term = dict(operator="is", operand="dm")
+        term = NarrowParameter(operator="is", operand="dm")
        self._do_add_term_test(term, "WHERE (flags & %(flags_1)s) != %(param_1)s")

    def test_add_term_using_is_operator_dm_operand_and_negated(self) -> None:  # NEGATED
-        term = dict(operator="is", operand="dm", negated=True)
+        term = NarrowParameter(operator="is", operand="dm", negated=True)
        self._do_add_term_test(term, "WHERE (flags & %(flags_1)s) = %(param_1)s")

    def test_add_term_using_is_operator_and_non_dm_operand(self) -> None:
        for operand in ["starred", "mentioned", "alerted"]:
-            term = dict(operator="is", operand=operand)
+            term = NarrowParameter(operator="is", operand=operand)
            self._do_add_term_test(term, "WHERE (flags & %(flags_1)s) != %(param_1)s")

    def test_add_term_using_is_operator_and_unread_operand(self) -> None:
-        term = dict(operator="is", operand="unread")
+        term = NarrowParameter(operator="is", operand="unread")
        self._do_add_term_test(term, "WHERE (flags & %(flags_1)s) = %(param_1)s")

    def test_add_term_using_is_operator_and_unread_operand_and_negated(self) -> None:  # NEGATED
-        term = dict(operator="is", operand="unread", negated=True)
+        term = NarrowParameter(operator="is", operand="unread", negated=True)
        self._do_add_term_test(term, "WHERE (flags & %(flags_1)s) != %(param_1)s")

    def test_add_term_using_is_operator_non_dm_operand_and_negated(self) -> None:  # NEGATED
-        term = dict(operator="is", operand="starred", negated=True)
+        term = NarrowParameter(operator="is", operand="starred", negated=True)
        where_clause = "WHERE (flags & %(flags_1)s) = %(param_1)s"
        params = dict(
            flags_1=UserMessage.flags.starred.mask,
@@ -247,7 +248,7 @@ class NarrowBuilderTest(ZulipTestCase):
        )
        self._do_add_term_test(term, where_clause, params)

-        term = dict(operator="is", operand="alerted", negated=True)
+        term = NarrowParameter(operator="is", operand="alerted", negated=True)
        where_clause = "WHERE (flags & %(flags_1)s) = %(param_1)s"
        params = dict(
            flags_1=UserMessage.flags.has_alert_word.mask,
@@ -255,7 +256,7 @@ class NarrowBuilderTest(ZulipTestCase):
        )
        self._do_add_term_test(term, where_clause, params)

-        term = dict(operator="is", operand="mentioned", negated=True)
+        term = NarrowParameter(operator="is", operand="mentioned", negated=True)
        where_clause = "WHERE (flags & %(flags_1)s) = %(param_1)s"
        mention_flags_mask = (
            UserMessage.flags.mentioned.mask
@@ -270,63 +271,63 @@ class NarrowBuilderTest(ZulipTestCase):
        self._do_add_term_test(term, where_clause, params)

    def test_add_term_using_is_operator_for_resolved_topics(self) -> None:
-        term = dict(operator="is", operand="resolved")
+        term = NarrowParameter(operator="is", operand="resolved")
        self._do_add_term_test(term, "WHERE (subject LIKE %(subject_1)s || '%%'")

    def test_add_term_using_is_operator_for_negated_resolved_topics(self) -> None:
-        term = dict(operator="is", operand="resolved", negated=True)
+        term = NarrowParameter(operator="is", operand="resolved", negated=True)
        self._do_add_term_test(term, "WHERE (subject NOT LIKE %(subject_1)s || '%%'")

    def test_add_term_using_is_operator_for_followed_topics(self) -> None:
-        term = dict(operator="is", operand="followed", negated=False)
+        term = NarrowParameter(operator="is", operand="followed", negated=False)
        self._do_add_term_test(
            term,
            "EXISTS (SELECT 1 \nFROM zerver_usertopic \nWHERE zerver_usertopic.user_profile_id = %(param_1)s AND zerver_usertopic.visibility_policy = %(param_2)s AND upper(zerver_usertopic.topic_name) = upper(zerver_message.subject) AND zerver_usertopic.recipient_id = zerver_message.recipient_id)",
        )

    def test_add_term_using_is_operator_for_negated_followed_topics(self) -> None:
-        term = dict(operator="is", operand="followed", negated=True)
+        term = NarrowParameter(operator="is", operand="followed", negated=True)
        self._do_add_term_test(
            term,
            "NOT (EXISTS (SELECT 1 \nFROM zerver_usertopic \nWHERE zerver_usertopic.user_profile_id = %(param_1)s AND zerver_usertopic.visibility_policy = %(param_2)s AND upper(zerver_usertopic.topic_name) = upper(zerver_message.subject) AND zerver_usertopic.recipient_id = zerver_message.recipient_id))",
        )

    def test_add_term_using_non_supported_operator_should_raise_error(self) -> None:
-        term = dict(operator="is", operand="non_supported")
+        term = NarrowParameter(operator="is", operand="non_supported")
        self.assertRaises(BadNarrowOperatorError, self._build_query, term)

    def test_add_term_using_topic_operator_and_lunch_operand(self) -> None:
-        term = dict(operator="topic", operand="lunch")
+        term = NarrowParameter(operator="topic", operand="lunch")
        self._do_add_term_test(term, "WHERE upper(subject) = upper(%(param_1)s)")

    def test_add_term_using_topic_operator_lunch_operand_and_negated(self) -> None:  # NEGATED
-        term = dict(operator="topic", operand="lunch", negated=True)
+        term = NarrowParameter(operator="topic", operand="lunch", negated=True)
        self._do_add_term_test(term, "WHERE upper(subject) != upper(%(param_1)s)")

    def test_add_term_using_topic_operator_and_personal_operand(self) -> None:
-        term = dict(operator="topic", operand="personal")
+        term = NarrowParameter(operator="topic", operand="personal")
        self._do_add_term_test(term, "WHERE upper(subject) = upper(%(param_1)s)")

    def test_add_term_using_topic_operator_personal_operand_and_negated(self) -> None:  # NEGATED
-        term = dict(operator="topic", operand="personal", negated=True)
+        term = NarrowParameter(operator="topic", operand="personal", negated=True)
        self._do_add_term_test(term, "WHERE upper(subject) != upper(%(param_1)s)")

    def test_add_term_using_sender_operator(self) -> None:
-        term = dict(operator="sender", operand=self.othello_email)
+        term = NarrowParameter(operator="sender", operand=self.othello_email)
        self._do_add_term_test(term, "WHERE sender_id = %(param_1)s")

    def test_add_term_using_sender_operator_and_negated(self) -> None:  # NEGATED
-        term = dict(operator="sender", operand=self.othello_email, negated=True)
+        term = NarrowParameter(operator="sender", operand=self.othello_email, negated=True)
        self._do_add_term_test(term, "WHERE sender_id != %(param_1)s")

    def test_add_term_using_sender_operator_with_non_existing_user_as_operand(
        self,
    ) -> None:  # NEGATED
-        term = dict(operator="sender", operand="non-existing@zulip.com")
+        term = NarrowParameter(operator="sender", operand="non-existing@zulip.com")
        self.assertRaises(BadNarrowOperatorError, self._build_query, term)

    def test_add_term_using_dm_operator_and_not_the_same_user_as_operand(self) -> None:
-        term = dict(operator="dm", operand=self.othello_email)
+        term = NarrowParameter(operator="dm", operand=self.othello_email)
        self._do_add_term_test(
            term,
            "WHERE (flags & %(flags_1)s) != %(param_1)s AND realm_id = %(realm_id_1)s AND (sender_id = %(sender_id_1)s AND recipient_id = %(recipient_id_1)s OR sender_id = %(sender_id_2)s AND recipient_id = %(recipient_id_2)s)",
@@ -336,15 +337,15 @@ class NarrowBuilderTest(ZulipTestCase):
        expected_error_message = (
            "Invalid narrow operator: No message can be both a channel message and direct message"
        )
-        term1 = dict(operator="dm", operand=self.othello_email)
+        term1 = NarrowParameter(operator="dm", operand=self.othello_email)
        self._build_query(term1)

-        topic_term = dict(operator="topic", operand="bogus")
+        topic_term = NarrowParameter(operator="topic", operand="bogus")
        with self.assertRaises(BadNarrowOperatorError) as error:
            self._build_query(topic_term)
        self.assertEqual(expected_error_message, str(error.exception))

-        channels_term = dict(operator="channels", operand="public")
+        channels_term = NarrowParameter(operator="channels", operand="public")
        with self.assertRaises(BadNarrowOperatorError) as error:
            self._build_query(channels_term)
        self.assertEqual(expected_error_message, str(error.exception))
@@ -352,14 +353,14 @@ class NarrowBuilderTest(ZulipTestCase):
    def test_add_term_using_dm_operator_not_the_same_user_as_operand_and_negated(
        self,
    ) -> None:  # NEGATED
-        term = dict(operator="dm", operand=self.othello_email, negated=True)
+        term = NarrowParameter(operator="dm", operand=self.othello_email, negated=True)
        self._do_add_term_test(
            term,
            "WHERE NOT ((flags & %(flags_1)s) != %(param_1)s AND realm_id = %(realm_id_1)s AND (sender_id = %(sender_id_1)s AND recipient_id = %(recipient_id_1)s OR sender_id = %(sender_id_2)s AND recipient_id = %(recipient_id_2)s))",
        )

    def test_add_term_using_dm_operator_the_same_user_as_operand(self) -> None:
-        term = dict(operator="dm", operand=self.hamlet_email)
+        term = NarrowParameter(operator="dm", operand=self.hamlet_email)
        self._do_add_term_test(
            term,
            "WHERE (flags & %(flags_1)s) != %(param_1)s AND realm_id = %(realm_id_1)s AND sender_id = %(sender_id_1)s AND recipient_id = %(recipient_id_1)s",
@@ -368,7 +369,7 @@ class NarrowBuilderTest(ZulipTestCase):
    def test_add_term_using_dm_operator_the_same_user_as_operand_and_negated(
        self,
    ) -> None:  # NEGATED
-        term = dict(operator="dm", operand=self.hamlet_email, negated=True)
+        term = NarrowParameter(operator="dm", operand=self.hamlet_email, negated=True)
        self._do_add_term_test(
            term,
            "WHERE NOT ((flags & %(flags_1)s) != %(param_1)s AND realm_id = %(realm_id_1)s AND sender_id = %(sender_id_1)s AND recipient_id = %(recipient_id_1)s)",
@@ -378,7 +379,7 @@ class NarrowBuilderTest(ZulipTestCase):
        myself_and_other = (
            f"{self.example_user('hamlet').email},{self.example_user('othello').email}"
        )
-        term = dict(operator="dm", operand=myself_and_other)
+        term = NarrowParameter(operator="dm", operand=myself_and_other)
        self._do_add_term_test(
            term,
            "WHERE (flags & %(flags_1)s) != %(param_1)s AND realm_id = %(realm_id_1)s AND (sender_id = %(sender_id_1)s AND recipient_id = %(recipient_id_1)s OR sender_id = %(sender_id_2)s AND recipient_id = %(recipient_id_2)s)",
@@ -387,7 +388,7 @@ class NarrowBuilderTest(ZulipTestCase):
    def test_add_term_using_dm_operator_more_than_one_user_as_operand_no_huddle(self) -> None:
        # If the group doesn't exist, it's a flat false
        two_others = f"{self.example_user('cordelia').email},{self.example_user('othello').email}"
-        term = dict(operator="dm", operand=two_others)
+        term = NarrowParameter(operator="dm", operand=two_others)
        self._do_add_term_test(term, "WHERE false")

    def test_add_term_using_dm_operator_more_than_one_user_as_operand(self) -> None:
@@ -400,7 +401,7 @@ class NarrowBuilderTest(ZulipTestCase):
            ]
        )
        two_others = f"{self.example_user('cordelia').email},{self.example_user('othello').email}"
-        term = dict(operator="dm", operand=two_others)
+        term = NarrowParameter(operator="dm", operand=two_others)
        self._do_add_term_test(term, "WHERE recipient_id = %(recipient_id_1)s")

    def test_add_term_using_dm_operator_self_and_user_as_operand_and_negated(
@@ -409,7 +410,7 @@ class NarrowBuilderTest(ZulipTestCase):
        myself_and_other = (
            f"{self.example_user('hamlet').email},{self.example_user('othello').email}"
        )
-        term = dict(operator="dm", operand=myself_and_other, negated=True)
+        term = NarrowParameter(operator="dm", operand=myself_and_other, negated=True)
        self._do_add_term_test(
            term,
            "WHERE NOT ((flags & %(flags_1)s) != %(param_1)s AND realm_id = %(realm_id_1)s AND (sender_id = %(sender_id_1)s AND recipient_id = %(recipient_id_1)s OR sender_id = %(sender_id_2)s AND recipient_id = %(recipient_id_2)s))",
@@ -420,7 +421,7 @@ class NarrowBuilderTest(ZulipTestCase):
    ) -> None:  # NEGATED
        # If the group doesn't exist, it's a flat true
        two_others = f"{self.example_user('cordelia').email},{self.example_user('othello').email}"
-        term = dict(operator="dm", operand=two_others, negated=True)
+        term = NarrowParameter(operator="dm", operand=two_others, negated=True)
        self._do_add_term_test(term, "WHERE true")

    def test_add_term_using_dm_operator_more_than_one_user_as_operand_and_negated(
@@ -435,26 +436,28 @@ class NarrowBuilderTest(ZulipTestCase):
            ]
        )
        two_others = f"{self.example_user('cordelia').email},{self.example_user('othello').email}"
-        term = dict(operator="dm", operand=two_others, negated=True)
+        term = NarrowParameter(operator="dm", operand=two_others, negated=True)
        self._do_add_term_test(term, "WHERE recipient_id != %(recipient_id_1)s")

    def test_add_term_using_dm_operator_with_comma_noise(self) -> None:
-        term = dict(operator="dm", operand=" ,,, ,,, ,")
+        term = NarrowParameter(operator="dm", operand=" ,,, ,,, ,")
        self.assertRaises(BadNarrowOperatorError, self._build_query, term)

    def test_add_term_using_dm_operator_with_existing_and_non_existing_user_as_operand(
        self,
    ) -> None:
-        term = dict(operator="dm", operand=self.othello_email + ",non-existing@zulip.com")
+        term = NarrowParameter(
+            operator="dm", operand=self.othello_email + ",non-existing@zulip.com"
+        )
        self.assertRaises(BadNarrowOperatorError, self._build_query, term)

    def test_add_term_using_dm_including_operator_with_logged_in_user_email(self) -> None:
-        term = dict(operator="dm-including", operand=self.hamlet_email)
+        term = NarrowParameter(operator="dm-including", operand=self.hamlet_email)
        self._do_add_term_test(term, "WHERE (flags & %(flags_1)s) != %(param_1)s")

    def test_add_term_using_dm_including_operator_with_different_user_email(self) -> None:
        # Test without any such group direct messages existing
-        term = dict(operator="dm-including", operand=self.othello_email)
+        term = NarrowParameter(operator="dm-including", operand=self.othello_email)
        self._do_add_term_test(
            term,
            "WHERE (flags & %(flags_1)s) != %(param_1)s AND realm_id = %(realm_id_1)s AND (sender_id = %(sender_id_1)s AND recipient_id = %(recipient_id_1)s OR sender_id = %(sender_id_2)s AND recipient_id = %(recipient_id_2)s OR recipient_id IN (__[POSTCOMPILE_recipient_id_3]))",
@@ -465,7 +468,7 @@ class NarrowBuilderTest(ZulipTestCase):
            self.user_profile, [self.example_user("othello"), self.example_user("cordelia")]
        )

-        term = dict(operator="dm-including", operand=self.othello_email)
+        term = NarrowParameter(operator="dm-including", operand=self.othello_email)
        self._do_add_term_test(
            term,
            "WHERE (flags & %(flags_1)s) != %(param_1)s AND realm_id = %(realm_id_1)s AND (sender_id = %(sender_id_1)s AND recipient_id = %(recipient_id_1)s OR sender_id = %(sender_id_2)s AND recipient_id = %(recipient_id_2)s OR recipient_id IN (__[POSTCOMPILE_recipient_id_3]))",
@@ -474,37 +477,37 @@ class NarrowBuilderTest(ZulipTestCase):
    def test_add_term_using_dm_including_operator_with_different_user_email_and_negated(
        self,
    ) -> None:  # NEGATED
-        term = dict(operator="dm-including", operand=self.othello_email, negated=True)
+        term = NarrowParameter(operator="dm-including", operand=self.othello_email, negated=True)
        self._do_add_term_test(
            term,
            "WHERE NOT ((flags & %(flags_1)s) != %(param_1)s AND realm_id = %(realm_id_1)s AND (sender_id = %(sender_id_1)s AND recipient_id = %(recipient_id_1)s OR sender_id = %(sender_id_2)s AND recipient_id = %(recipient_id_2)s OR recipient_id IN (__[POSTCOMPILE_recipient_id_3])))",
        )

    def test_add_term_using_id_operator_integer(self) -> None:
-        term = dict(operator="id", operand=555)
+        term = NarrowParameter(operator="id", operand=555)
        self._do_add_term_test(term, "WHERE id = %(param_1)s")

    def test_add_term_using_id_operator_string(self) -> None:
-        term = dict(operator="id", operand="555")
+        term = NarrowParameter(operator="id", operand="555")
        self._do_add_term_test(term, "WHERE id = %(param_1)s")

    def test_add_term_using_id_operator_invalid(self) -> None:
-        term = dict(operator="id", operand="")
+        term = NarrowParameter(operator="id", operand="")
        self.assertRaises(BadNarrowOperatorError, self._build_query, term)

-        term = dict(operator="id", operand="notanint")
+        term = NarrowParameter(operator="id", operand="notanint")
        self.assertRaises(BadNarrowOperatorError, self._build_query, term)

-        term = dict(operator="id", operand=str(Message.MAX_POSSIBLE_MESSAGE_ID + 1))
+        term = NarrowParameter(operator="id", operand=str(Message.MAX_POSSIBLE_MESSAGE_ID + 1))
        self.assertRaises(BadNarrowOperatorError, self._build_query, term)

    def test_add_term_using_id_operator_and_negated(self) -> None:  # NEGATED
-        term = dict(operator="id", operand=555, negated=True)
+        term = NarrowParameter(operator="id", operand=555, negated=True)
        self._do_add_term_test(term, "WHERE id != %(param_1)s")

    @override_settings(USING_PGROONGA=False)
    def test_add_term_using_search_operator(self) -> None:
-        term = dict(operator="search", operand='"french fries"')
+        term = NarrowParameter(operator="search", operand='"french fries"')
        self._do_add_term_test(
            term,
            "WHERE (content ILIKE %(content_1)s OR subject ILIKE %(subject_1)s) AND (search_tsvector @@ plainto_tsquery(%(param_4)s, %(param_5)s))",
@@ -512,7 +515,7 @@ class NarrowBuilderTest(ZulipTestCase):

    @override_settings(USING_PGROONGA=False)
    def test_add_term_using_search_operator_and_negated(self) -> None:  # NEGATED
-        term = dict(operator="search", operand='"french fries"', negated=True)
+        term = NarrowParameter(operator="search", operand='"french fries"', negated=True)
        self._do_add_term_test(
            term,
            "WHERE NOT (content ILIKE %(content_1)s OR subject ILIKE %(subject_1)s) AND NOT (search_tsvector @@ plainto_tsquery(%(param_4)s, %(param_5)s))",
@@ -520,113 +523,113 @@ class NarrowBuilderTest(ZulipTestCase):

    @override_settings(USING_PGROONGA=True)
    def test_add_term_using_search_operator_pgroonga(self) -> None:
-        term = dict(operator="search", operand='"french fries"')
+        term = NarrowParameter(operator="search", operand='"french fries"')
        self._do_add_term_test(term, "WHERE search_pgroonga &@~ escape_html(%(escape_html_1)s)")

    @override_settings(USING_PGROONGA=True)
    def test_add_term_using_search_operator_and_negated_pgroonga(self) -> None:  # NEGATED
-        term = dict(operator="search", operand='"french fries"', negated=True)
+        term = NarrowParameter(operator="search", operand='"french fries"', negated=True)
        self._do_add_term_test(
            term, "WHERE NOT (search_pgroonga &@~ escape_html(%(escape_html_1)s))"
        )

    def test_add_term_using_has_operator_and_attachment_operand(self) -> None:
-        term = dict(operator="has", operand="attachment")
+        term = NarrowParameter(operator="has", operand="attachment")
        self._do_add_term_test(term, "WHERE has_attachment")

    def test_add_term_using_has_operator_attachment_operand_and_negated(self) -> None:  # NEGATED
-        term = dict(operator="has", operand="attachment", negated=True)
+        term = NarrowParameter(operator="has", operand="attachment", negated=True)
        self._do_add_term_test(term, "WHERE NOT has_attachment")

    def test_add_term_using_has_operator_and_image_operand(self) -> None:
-        term = dict(operator="has", operand="image")
+        term = NarrowParameter(operator="has", operand="image")
        self._do_add_term_test(term, "WHERE has_image")

    def test_add_term_using_has_operator_image_operand_and_negated(self) -> None:  # NEGATED
-        term = dict(operator="has", operand="image", negated=True)
+        term = NarrowParameter(operator="has", operand="image", negated=True)
        self._do_add_term_test(term, "WHERE NOT has_image")

    def test_add_term_using_has_operator_and_link_operand(self) -> None:
-        term = dict(operator="has", operand="link")
+        term = NarrowParameter(operator="has", operand="link")
        self._do_add_term_test(term, "WHERE has_link")

    def test_add_term_using_has_operator_link_operand_and_negated(self) -> None:  # NEGATED
-        term = dict(operator="has", operand="link", negated=True)
+        term = NarrowParameter(operator="has", operand="link", negated=True)
        self._do_add_term_test(term, "WHERE NOT has_link")

    def test_add_term_using_has_operator_and_reaction_operand(self) -> None:
-        term = dict(operator="has", operand="reaction")
+        term = NarrowParameter(operator="has", operand="reaction")
        self._do_add_term_test(
            term,
            "EXISTS (SELECT 1 \nFROM zerver_reaction \nWHERE zerver_message.id = zerver_reaction.message_id)",
        )

    def test_add_term_using_has_operator_and_reaction_operand_and_negated(self) -> None:
-        term = dict(operator="has", operand="reaction", negated=True)
+        term = NarrowParameter(operator="has", operand="reaction", negated=True)
        self._do_add_term_test(
            term,
            "NOT (EXISTS (SELECT 1 \nFROM zerver_reaction \nWHERE zerver_message.id = zerver_reaction.message_id))",
        )

    def test_add_term_using_has_operator_non_supported_operand_should_raise_error(self) -> None:
-        term = dict(operator="has", operand="non_supported")
+        term = NarrowParameter(operator="has", operand="non_supported")
        self.assertRaises(BadNarrowOperatorError, self._build_query, term)

    def test_add_term_using_in_operator(self) -> None:
        mute_channel(self.realm, self.user_profile, "Verona")
-        term = dict(operator="in", operand="home")
+        term = NarrowParameter(operator="in", operand="home")
        self._do_add_term_test(term, "WHERE (recipient_id NOT IN (__[POSTCOMPILE_recipient_id_1]))")

    def test_add_term_using_in_operator_and_negated(self) -> None:
        # negated = True should not change anything
        mute_channel(self.realm, self.user_profile, "Verona")
-        term = dict(operator="in", operand="home", negated=True)
+        term = NarrowParameter(operator="in", operand="home", negated=True)
        self._do_add_term_test(term, "WHERE (recipient_id NOT IN (__[POSTCOMPILE_recipient_id_1]))")

    def test_add_term_using_in_operator_and_all_operand(self) -> None:
        mute_channel(self.realm, self.user_profile, "Verona")
-        term = dict(operator="in", operand="all")
+        term = NarrowParameter(operator="in", operand="all")
        query = self._build_query(term)
        self.assertEqual(get_sqlalchemy_sql(query), "SELECT id \nFROM zerver_message")

    def test_add_term_using_in_operator_all_operand_and_negated(self) -> None:
        # negated = True should not change anything
        mute_channel(self.realm, self.user_profile, "Verona")
-        term = dict(operator="in", operand="all", negated=True)
+        term = NarrowParameter(operator="in", operand="all", negated=True)
        query = self._build_query(term)
        self.assertEqual(get_sqlalchemy_sql(query), "SELECT id \nFROM zerver_message")

    def test_add_term_using_in_operator_and_not_defined_operand(self) -> None:
-        term = dict(operator="in", operand="not_defined")
+        term = NarrowParameter(operator="in", operand="not_defined")
        self.assertRaises(BadNarrowOperatorError, self._build_query, term)

    def test_add_term_using_near_operator(self) -> None:
-        term = dict(operator="near", operand="operand")
+        term = NarrowParameter(operator="near", operand="operand")
        query = self._build_query(term)
        self.assertEqual(get_sqlalchemy_sql(query), "SELECT id \nFROM zerver_message")

    def test_add_term_non_web_public_channel_in_web_public_query(self) -> None:
        self.make_stream("non-web-public-channel", realm=self.realm)
-        term = dict(operator="channel", operand="non-web-public-channel")
+        term = NarrowParameter(operator="channel", operand="non-web-public-channel")
        builder = NarrowBuilder(self.user_profile, column("id", Integer), self.realm, True)

-        def _build_query(term: Dict[str, Any]) -> Select:
+        def _build_query(term: NarrowParameter) -> Select:
            return builder.add_term(self.raw_query, term)

        self.assertRaises(BadNarrowOperatorError, _build_query, term)

    # Test "is:private" (legacy alias for "is:dm")
    def test_add_term_using_is_operator_and_private_operand(self) -> None:
-        term = dict(operator="is", operand="private")
+        term = NarrowParameter(operator="is", operand="private")
        self._do_add_term_test(term, "WHERE (flags & %(flags_1)s) != %(param_1)s")

    def test_add_term_using_is_operator_private_operand_and_negated(self) -> None:  # NEGATED
-        term = dict(operator="is", operand="private", negated=True)
+        term = NarrowParameter(operator="is", operand="private", negated=True)
        self._do_add_term_test(term, "WHERE (flags & %(flags_1)s) = %(param_1)s")

    # Test that "pm-with" (legacy alias for "dm") works.
    def test_add_term_using_pm_with_operator(self) -> None:
-        term = dict(operator="pm-with", operand=self.hamlet_email)
+        term = NarrowParameter(operator="pm-with", operand=self.hamlet_email)
        self._do_add_term_test(
            term,
            "WHERE (flags & %(flags_1)s) != %(param_1)s AND realm_id = %(realm_id_1)s AND sender_id = %(sender_id_1)s AND recipient_id = %(recipient_id_1)s",
@@ -634,7 +637,7 @@ class NarrowBuilderTest(ZulipTestCase):

    # Test that the underscore version of "pm-with" works.
    def test_add_term_using_underscore_version_of_pm_with_operator(self) -> None:
-        term = dict(operator="pm_with", operand=self.hamlet_email)
+        term = NarrowParameter(operator="pm_with", operand=self.hamlet_email)
        self._do_add_term_test(
            term,
            "WHERE (flags & %(flags_1)s) != %(param_1)s AND realm_id = %(realm_id_1)s AND sender_id = %(sender_id_1)s AND recipient_id = %(recipient_id_1)s",
@@ -642,11 +645,11 @@ class NarrowBuilderTest(ZulipTestCase):

    # Test that deprecated "group-pm-with" (replaced by "dm-including" ) works.
    def test_add_term_using_dm_including_operator_with_non_existing_user(self) -> None:
-        term = dict(operator="dm-including", operand="non-existing@zulip.com")
+        term = NarrowParameter(operator="dm-including", operand="non-existing@zulip.com")
        self.assertRaises(BadNarrowOperatorError, self._build_query, term)

    def test_add_term_using_group_pm_operator_and_not_the_same_user_as_operand(self) -> None:
-        term = dict(operator="group-pm-with", operand=self.othello_email)
+        term = NarrowParameter(operator="group-pm-with", operand=self.othello_email)
        self._do_add_term_test(
            term,
            "WHERE (flags & %(flags_1)s) != %(param_1)s AND realm_id = %(realm_id_1)s AND recipient_id IN (__[POSTCOMPILE_recipient_id_1])",
@@ -655,19 +658,19 @@ class NarrowBuilderTest(ZulipTestCase):
    def test_add_term_using_group_pm_operator_not_the_same_user_as_operand_and_negated(
        self,
    ) -> None:  # NEGATED
-        term = dict(operator="group-pm-with", operand=self.othello_email, negated=True)
+        term = NarrowParameter(operator="group-pm-with", operand=self.othello_email, negated=True)
        self._do_add_term_test(
            term,
            "WHERE NOT ((flags & %(flags_1)s) != %(param_1)s AND realm_id = %(realm_id_1)s AND recipient_id IN (__[POSTCOMPILE_recipient_id_1]))",
        )

    def test_add_term_using_group_pm_operator_with_non_existing_user_as_operand(self) -> None:
-        term = dict(operator="group-pm-with", operand="non-existing@zulip.com")
+        term = NarrowParameter(operator="group-pm-with", operand="non-existing@zulip.com")
        self.assertRaises(BadNarrowOperatorError, self._build_query, term)

    # Test that the underscore version of "group-pm-with" works.
    def test_add_term_using_underscore_version_of_group_pm_with_operator(self) -> None:
-        term = dict(operator="group_pm_with", operand=self.othello_email)
+        term = NarrowParameter(operator="group_pm_with", operand=self.othello_email)
        self._do_add_term_test(
            term,
            "WHERE (flags & %(flags_1)s) != %(param_1)s AND realm_id = %(realm_id_1)s AND recipient_id IN (__[POSTCOMPILE_recipient_id_1])",
@@ -675,42 +678,42 @@ class NarrowBuilderTest(ZulipTestCase):

    # Test that "stream" (legacy alias for "channel" operator) works.
    def test_add_term_using_stream_operator(self) -> None:
-        term = dict(operator="stream", operand="Scotland")
+        term = NarrowParameter(operator="stream", operand="Scotland")
        self._do_add_term_test(term, "WHERE recipient_id = %(recipient_id_1)s")

    def test_add_term_using_stream_operator_and_negated(self) -> None:  # NEGATED
-        term = dict(operator="stream", operand="Scotland", negated=True)
+        term = NarrowParameter(operator="stream", operand="Scotland", negated=True)
        self._do_add_term_test(term, "WHERE recipient_id != %(recipient_id_1)s")

    def test_add_term_using_stream_operator_and_non_existing_operand_should_raise_error(
        self,
    ) -> None:  # NEGATED
-        term = dict(operator="stream", operand="non-existing-channel")
+        term = NarrowParameter(operator="stream", operand="non-existing-channel")
        self.assertRaises(BadNarrowOperatorError, self._build_query, term)

    # Test that "streams" (legacy alias for "channels" operator) works.
    def test_add_term_using_streams_operator_and_invalid_operand_should_raise_error(
        self,
    ) -> None:  # NEGATED
-        term = dict(operator="streams", operand="invalid_operands")
+        term = NarrowParameter(operator="streams", operand="invalid_operands")
        self.assertRaises(BadNarrowOperatorError, self._build_query, term)

    def test_add_term_using_streams_operator_and_public_operand(self) -> None:
-        term = dict(operator="streams", operand="public")
+        term = NarrowParameter(operator="streams", operand="public")
        self._do_add_term_test(
            term,
            "WHERE recipient_id IN (__[POSTCOMPILE_recipient_id_1])",
        )

    def test_add_term_using_streams_operator_and_public_operand_negated(self) -> None:
-        term = dict(operator="streams", operand="public", negated=True)
+        term = NarrowParameter(operator="streams", operand="public", negated=True)
        self._do_add_term_test(
            term,
            "WHERE (recipient_id NOT IN (__[POSTCOMPILE_recipient_id_1]))",
        )

    def _do_add_term_test(
-        self, term: Dict[str, Any], where_clause: str, params: Optional[Dict[str, Any]] = None
+        self, term: NarrowParameter, where_clause: str, params: Optional[Dict[str, Any]] = None
    ) -> None:
        query = self._build_query(term)
        if params is not None:
@@ -718,7 +721,7 @@ class NarrowBuilderTest(ZulipTestCase):
            self.assertEqual(actual_params, params)
        self.assertIn(where_clause, get_sqlalchemy_sql(query))

-    def _build_query(self, term: Dict[str, Any]) -> Select:
+    def _build_query(self, term: NarrowParameter) -> Select:
        return self.builder.add_term(self.raw_query, term)


@@ -977,62 +980,86 @@ class NarrowLibraryTest(ZulipTestCase):

    def test_is_spectator_compatible(self) -> None:
        self.assertTrue(is_spectator_compatible([]))
-        self.assertTrue(is_spectator_compatible([{"operator": "has", "operand": "attachment"}]))
-        self.assertTrue(is_spectator_compatible([{"operator": "has", "operand": "image"}]))
-        self.assertTrue(is_spectator_compatible([{"operator": "search", "operand": "magic"}]))
-        self.assertTrue(is_spectator_compatible([{"operator": "near", "operand": "15"}]))
        self.assertTrue(
-            is_spectator_compatible(
-                [{"operator": "id", "operand": "15"}, {"operator": "has", "operand": "attachment"}]
-            )
+            is_spectator_compatible([NarrowParameter(operator="has", operand="attachment")])
        )
+        self.assertTrue(is_spectator_compatible([NarrowParameter(operator="has", operand="image")]))
        self.assertTrue(
-            is_spectator_compatible([{"operator": "sender", "operand": "hamlet@zulip.com"}])
+            is_spectator_compatible([NarrowParameter(operator="search", operand="magic")])
        )
-        self.assertFalse(
-            is_spectator_compatible([{"operator": "dm", "operand": "hamlet@zulip.com"}])
-        )
-        self.assertFalse(
-            is_spectator_compatible([{"operator": "dm-including", "operand": "hamlet@zulip.com"}])
-        )
-        self.assertTrue(is_spectator_compatible([{"operator": "channel", "operand": "Denmark"}]))
+        self.assertTrue(is_spectator_compatible([NarrowParameter(operator="near", operand="15")]))
        self.assertTrue(
            is_spectator_compatible(
                [
-                    {"operator": "channel", "operand": "Denmark"},
-                    {"operator": "topic", "operand": "logic"},
+                    NarrowParameter(operator="id", operand="15"),
+                    NarrowParameter(operator="has", operand="attachment"),
                ]
            )
        )
-        self.assertFalse(is_spectator_compatible([{"operator": "is", "operand": "starred"}]))
-        self.assertFalse(is_spectator_compatible([{"operator": "is", "operand": "dm"}]))
-        self.assertTrue(is_spectator_compatible([{"operator": "channels", "operand": "public"}]))
-
-        # Malformed input not allowed
-        self.assertFalse(is_spectator_compatible([{"operator": "has"}]))
-
-        # "is:private" is a legacy alias for "is:dm".
-        self.assertFalse(is_spectator_compatible([{"operator": "is", "operand": "private"}]))
-        # "pm-with:"" is a legacy alias for "dm:"
-        self.assertFalse(
-            is_spectator_compatible([{"operator": "pm-with", "operand": "hamlet@zulip.com"}])
+        self.assertTrue(
+            is_spectator_compatible(
+                [NarrowParameter(operator="sender", operand="hamlet@zulip.com")]
            )
-        # "group-pm-with:" was deprecated by the addition of "dm-including:"
-        self.assertFalse(
-            is_spectator_compatible([{"operator": "group-pm-with", "operand": "hamlet@zulip.com"}])
        )
-        # "stream" is a legacy alias for "channel" operator
-        self.assertTrue(is_spectator_compatible([{"operator": "stream", "operand": "Denmark"}]))
+        self.assertFalse(
+            is_spectator_compatible([NarrowParameter(operator="dm", operand="hamlet@zulip.com")])
+        )
+        self.assertFalse(
+            is_spectator_compatible(
+                [NarrowParameter(operator="dm-including", operand="hamlet@zulip.com")]
+            )
+        )
+        self.assertTrue(
+            is_spectator_compatible([NarrowParameter(operator="channel", operand="Denmark")])
+        )
        self.assertTrue(
            is_spectator_compatible(
                [
-                    {"operator": "stream", "operand": "Denmark"},
-                    {"operator": "topic", "operand": "logic"},
+                    NarrowParameter(operator="channel", operand="Denmark"),
+                    NarrowParameter(operator="topic", operand="logic"),
+                ]
+            )
+        )
+        self.assertFalse(
+            is_spectator_compatible([NarrowParameter(operator="is", operand="starred")])
+        )
+        self.assertFalse(is_spectator_compatible([NarrowParameter(operator="is", operand="dm")]))
+        self.assertTrue(
+            is_spectator_compatible([NarrowParameter(operator="channels", operand="public")])
+        )
+
+        # "is:private" is a legacy alias for "is:dm".
+        self.assertFalse(
+            is_spectator_compatible([NarrowParameter(operator="is", operand="private")])
+        )
+        # "pm-with:"" is a legacy alias for "dm:"
+        self.assertFalse(
+            is_spectator_compatible(
+                [NarrowParameter(operator="pm-with", operand="hamlet@zulip.com")]
+            )
+        )
+        # "group-pm-with:" was deprecated by the addition of "dm-including:"
+        self.assertFalse(
+            is_spectator_compatible(
+                [NarrowParameter(operator="group-pm-with", operand="hamlet@zulip.com")]
+            )
+        )
+        # "stream" is a legacy alias for "channel" operator
+        self.assertTrue(
+            is_spectator_compatible([NarrowParameter(operator="stream", operand="Denmark")])
+        )
+        self.assertTrue(
+            is_spectator_compatible(
+                [
+                    NarrowParameter(operator="stream", operand="Denmark"),
+                    NarrowParameter(operator="topic", operand="logic"),
                ]
            )
        )
        # "streams" is a legacy alias for "channels" operator
-        self.assertTrue(is_spectator_compatible([{"operator": "streams", "operand": "public"}]))
+        self.assertTrue(
+            is_spectator_compatible([NarrowParameter(operator="streams", operand="public")])
+        )


 class IncludeHistoryTest(ZulipTestCase):
@@ -1042,19 +1069,19 @@ class IncludeHistoryTest(ZulipTestCase):

        # Negated channel searches should not include history.
        narrow = [
-            dict(operator="channel", operand="public_channel", negated=True),
+            NarrowParameter(operator="channel", operand="public_channel", negated=True),
        ]
        self.assertFalse(ok_to_include_history(narrow, user_profile, False))

        # channels:public searches should include history for non-guest members.
        narrow = [
-            dict(operator="channels", operand="public"),
+            NarrowParameter(operator="channels", operand="public"),
        ]
        self.assertTrue(ok_to_include_history(narrow, user_profile, False))

        # Negated -channels:public searches should not include history.
        narrow = [
-            dict(operator="channels", operand="public", negated=True),
+            NarrowParameter(operator="channels", operand="public", negated=True),
        ]
        self.assertFalse(ok_to_include_history(narrow, user_profile, False))

@@ -1063,7 +1090,7 @@ class IncludeHistoryTest(ZulipTestCase):
        subscribed_user_profile = self.example_user("cordelia")
        self.subscribe(subscribed_user_profile, "private_channel")
        narrow = [
-            dict(operator="channel", operand="private_channel"),
+            NarrowParameter(operator="channel", operand="private_channel"),
        ]
        self.assertFalse(ok_to_include_history(narrow, user_profile, False))

@@ -1078,69 +1105,69 @@ class IncludeHistoryTest(ZulipTestCase):
        subscribed_user_profile = self.example_user("cordelia")
        self.subscribe(subscribed_user_profile, "private_channel_2")
        narrow = [
-            dict(operator="channel", operand="private_channel_2"),
+            NarrowParameter(operator="channel", operand="private_channel_2"),
        ]
        self.assertFalse(ok_to_include_history(narrow, user_profile, False))
        self.assertTrue(ok_to_include_history(narrow, subscribed_user_profile, False))

        # History doesn't apply to direct messages.
        narrow = [
-            dict(operator="is", operand="dm"),
+            NarrowParameter(operator="is", operand="dm"),
        ]
        self.assertFalse(ok_to_include_history(narrow, user_profile, False))
        # "is:private" is a legacy alias for "is:dm".
        narrow = [
-            dict(operator="is", operand="private"),
+            NarrowParameter(operator="is", operand="private"),
        ]
        self.assertFalse(ok_to_include_history(narrow, user_profile, False))

        # History doesn't apply to unread messages.
        narrow = [
-            dict(operator="is", operand="unread"),
+            NarrowParameter(operator="is", operand="unread"),
        ]
        self.assertFalse(ok_to_include_history(narrow, user_profile, False))

        # If we are looking for something like starred messages, there is
        # no point in searching historical messages.
        narrow = [
-            dict(operator="channel", operand="public_channel"),
-            dict(operator="is", operand="starred"),
+            NarrowParameter(operator="channel", operand="public_channel"),
+            NarrowParameter(operator="is", operand="starred"),
        ]
        self.assertFalse(ok_to_include_history(narrow, user_profile, False))

        # No point in searching history for is operator even if included with
        # channels:public
        narrow = [
-            dict(operator="channels", operand="public"),
-            dict(operator="is", operand="mentioned"),
+            NarrowParameter(operator="channels", operand="public"),
+            NarrowParameter(operator="is", operand="mentioned"),
        ]
        self.assertFalse(ok_to_include_history(narrow, user_profile, False))
        narrow = [
-            dict(operator="channels", operand="public"),
-            dict(operator="is", operand="unread"),
+            NarrowParameter(operator="channels", operand="public"),
+            NarrowParameter(operator="is", operand="unread"),
        ]
        self.assertFalse(ok_to_include_history(narrow, user_profile, False))
        narrow = [
-            dict(operator="channels", operand="public"),
-            dict(operator="is", operand="alerted"),
+            NarrowParameter(operator="channels", operand="public"),
+            NarrowParameter(operator="is", operand="alerted"),
        ]
        self.assertFalse(ok_to_include_history(narrow, user_profile, False))
        narrow = [
-            dict(operator="channels", operand="public"),
-            dict(operator="is", operand="resolved"),
+            NarrowParameter(operator="channels", operand="public"),
+            NarrowParameter(operator="is", operand="resolved"),
        ]
        self.assertTrue(ok_to_include_history(narrow, user_profile, False))

        # simple True case
        narrow = [
-            dict(operator="channel", operand="public_channel"),
+            NarrowParameter(operator="channel", operand="public_channel"),
        ]
        self.assertTrue(ok_to_include_history(narrow, user_profile, False))

        narrow = [
-            dict(operator="channel", operand="public_channel"),
-            dict(operator="topic", operand="whatever"),
-            dict(operator="search", operand="needle in haystack"),
+            NarrowParameter(operator="channel", operand="public_channel"),
+            NarrowParameter(operator="topic", operand="whatever"),
+            NarrowParameter(operator="search", operand="needle in haystack"),
        ]
        self.assertTrue(ok_to_include_history(narrow, user_profile, False))

@@ -1151,14 +1178,14 @@ class IncludeHistoryTest(ZulipTestCase):

        # channels:public searches should not include history for guest members.
        narrow = [
-            dict(operator="channels", operand="public"),
+            NarrowParameter(operator="channels", operand="public"),
        ]
        self.assertFalse(ok_to_include_history(narrow, guest_user_profile, False))

        # Guest user can't access public channel
        self.subscribe(subscribed_user_profile, "public_channel_2")
        narrow = [
-            dict(operator="channel", operand="public_channel_2"),
+            NarrowParameter(operator="channel", operand="public_channel_2"),
        ]
        self.assertFalse(ok_to_include_history(narrow, guest_user_profile, False))
        self.assertTrue(ok_to_include_history(narrow, subscribed_user_profile, False))
@@ -1166,7 +1193,7 @@ class IncludeHistoryTest(ZulipTestCase):
        # Definitely, a guest user can't access the unsubscribed private channel
        self.subscribe(subscribed_user_profile, "private_channel_3")
        narrow = [
-            dict(operator="channel", operand="private_channel_3"),
+            NarrowParameter(operator="channel", operand="private_channel_3"),
        ]
        self.assertFalse(ok_to_include_history(narrow, guest_user_profile, False))
        self.assertTrue(ok_to_include_history(narrow, subscribed_user_profile, False))
@@ -1175,7 +1202,7 @@ class IncludeHistoryTest(ZulipTestCase):
        self.subscribe(guest_user_profile, "private_channel_4")
        self.subscribe(subscribed_user_profile, "private_channel_4")
        narrow = [
-            dict(operator="channel", operand="private_channel_4"),
+            NarrowParameter(operator="channel", operand="private_channel_4"),
        ]
        self.assertTrue(ok_to_include_history(narrow, guest_user_profile, False))
        self.assertTrue(ok_to_include_history(narrow, subscribed_user_profile, False))
@@ -4123,8 +4150,8 @@ class GetOldMessagesTest(ZulipTestCase):

        # If nothing relevant is muted, then exclude_muting_conditions()
        # should return an empty list.
-        narrow: List[Dict[str, object]] = [
-            dict(operator="channel", operand="Scotland"),
+        narrow: List[NarrowParameter] = [
+            NarrowParameter(operator="channel", operand="Scotland"),
        ]
        muting_conditions = exclude_muting_conditions(user_profile, narrow)
        self.assertEqual(muting_conditions, [])
@@ -4132,7 +4159,7 @@ class GetOldMessagesTest(ZulipTestCase):
        # Also test that passing channel ID works
        channel_id = get_stream("Scotland", realm).id
        narrow = [
-            dict(operator="channel", operand=channel_id),
+            NarrowParameter(operator="channel", operand=channel_id),
        ]
        muting_conditions = exclude_muting_conditions(user_profile, narrow)
        self.assertEqual(muting_conditions, [])
@@ -4146,7 +4173,7 @@ class GetOldMessagesTest(ZulipTestCase):

        # And verify that our query will exclude them.
        narrow = [
-            dict(operator="channel", operand="Scotland"),
+            NarrowParameter(operator="channel", operand="Scotland"),
        ]

        muting_conditions = exclude_muting_conditions(user_profile, narrow)
@@ -4173,7 +4200,7 @@ WHERE NOT (recipient_id = %(recipient_id_1)s AND upper(subject) = upper(%(param_
        # Using a bogus channel name should be similar to using no narrow at
        # all, and we'll exclude all mutes.
        narrow = [
-            dict(operator="channel", operand="bogus-channel-name"),
+            NarrowParameter(operator="channel", operand="bogus-channel-name"),
        ]

        muting_conditions = exclude_muting_conditions(user_profile, narrow)
--- a/zerver/views/message_fetch.py
+++ b/zerver/views/message_fetch.py
@@ -16,7 +16,6 @@ from zerver.lib.exceptions import JsonableError, MissingAuthenticationError
 from zerver.lib.message import get_first_visible_message_id, messages_for_ids
 from zerver.lib.narrow import (
    NarrowParameter,
-    OptionalNarrowListT,
    add_narrow_conditions,
    fetch_messages,
    is_spectator_compatible,
@@ -83,7 +82,9 @@ def get_search_fields(
    }


-def clean_narrow_for_web_public_api(narrow: OptionalNarrowListT) -> OptionalNarrowListT:
+def clean_narrow_for_web_public_api(
+    narrow: Optional[List[NarrowParameter]],
+) -> Optional[List[NarrowParameter]]:
    if narrow is None:
        return None

@@ -93,7 +94,7 @@ def clean_narrow_for_web_public_api(narrow: OptionalNarrowListT) -> OptionalNarr
    return [
        term
        for term in narrow
-        if not (term["operator"] == "in" and term["operand"] == "home" and not term["negated"])
+        if not (term.operator == "in" and term.operand == "home" and not term.negated)
    ]


@@ -123,11 +124,6 @@ def get_messages_backend(
    if num_before > 0 and num_after > 0 and not include_anchor:
        raise JsonableError(_("The anchor can only be excluded at an end of the range"))

-    if narrow is not None and len(narrow) > 0:
-        narrow_parameter_list: OptionalNarrowListT = [x.model_dump() for x in narrow]
-    else:
-        narrow_parameter_list = None
-
    realm = get_valid_realm_from_request(request)
    if not maybe_user_profile.is_authenticated:
        # If user is not authenticated, clients must include
@@ -142,11 +138,11 @@ def get_messages_backend(
        # non-web-public stream messages) via this path.
        if not realm.allow_web_public_streams_access():
            raise MissingAuthenticationError
-        narrow_parameter_list = clean_narrow_for_web_public_api(narrow_parameter_list)
-        if not is_web_public_narrow(narrow_parameter_list):
+        narrow = clean_narrow_for_web_public_api(narrow)
+        if not is_web_public_narrow(narrow):
            raise MissingAuthenticationError
-        assert narrow_parameter_list is not None
-        if not is_spectator_compatible(narrow_parameter_list):
+        assert narrow is not None
+        if not is_spectator_compatible(narrow):
            raise MissingAuthenticationError

        # We use None to indicate unauthenticated requests as it's more
@@ -168,14 +164,14 @@ def get_messages_backend(
        # email_address_visibility setting.
        client_gravatar = False

-    if narrow_parameter_list is not None:
+    if narrow is not None:
        # Add some metadata to our logging data for narrows
        verbose_operators = []
-        for term in narrow_parameter_list:
-            if term["operator"] == "is":
-                verbose_operators.append("is:" + term["operand"])
+        for term in narrow:
+            if term.operator == "is":
+                verbose_operators.append("is:" + term.operand)
            else:
-                verbose_operators.append(term["operator"])
+                verbose_operators.append(term.operator)
        log_data = RequestNotes.get_notes(request).log_data
        assert log_data is not None
        log_data["extra"] = "[{}]".format(",".join(verbose_operators))
@@ -206,7 +202,7 @@ def get_messages_backend(
            cursor.execute("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ READ ONLY")

        query_info = fetch_messages(
-            narrow=narrow_parameter_list,
+            narrow=narrow,
            user_profile=user_profile,
            realm=realm,
            is_web_public_query=is_web_public_query,
@@ -297,8 +293,6 @@ def messages_in_narrow_backend(
    msg_ids: Json[List[int]],
    narrow: Json[List[NarrowParameter]],
 ) -> HttpResponse:
-    narrow_parameter_list: OptionalNarrowListT = [x.model_dump() for x in narrow]
-
    first_visible_message_id = get_first_visible_message_id(user_profile.realm)
    msg_ids = [message_id for message_id in msg_ids if message_id >= first_visible_message_id]
    # This query is limited to messages the user has access to because they
@@ -326,7 +320,7 @@ def messages_in_narrow_backend(
        user_profile=user_profile,
        inner_msg_id_col=inner_msg_id_col,
        query=query,
-        narrow=narrow_parameter_list,
+        narrow=narrow,
        is_web_public_query=False,
        realm=user_profile.realm,
    )
--- a/zerver/views/message_flags.py
+++ b/zerver/views/message_flags.py
@@ -92,13 +92,8 @@ def update_message_flags_for_narrow(
    )
    num_after = min(num_after, MAX_MESSAGES_PER_UPDATE - num_before)

-    if narrow is not None and len(narrow) > 0:
-        narrow_dict = [x.model_dump() for x in narrow]
-    else:
-        narrow_dict = None
-
    query_info = fetch_messages(
-        narrow=narrow_dict,
+        narrow=narrow,
        user_profile=user_profile,
        realm=user_profile.realm,
        is_web_public_query=False,