diff --git a/redis/commands/core.py b/redis/commands/core.py
index 4e1e242dc..a56b3d2cb 100644
--- a/redis/commands/core.py
+++ b/redis/commands/core.py
@@ -5094,6 +5094,374 @@ def hstrlen(self, name: str, key: str) -> Union[Awaitable[int], int]:
         """
         return self.execute_command("HSTRLEN", name, key, keys=[name])
 
+    def hexpire(
+        self,
+        name: KeyT,
+        seconds: ExpiryT,
+        *fields: str,
+        nx: bool = False,
+        xx: bool = False,
+        gt: bool = False,
+        lt: bool = False,
+    ) -> ResponseT:
+        """
+        Sets or updates the expiration time for fields within a hash key, using relative
+        time in seconds.
+
+        If a field already has an expiration time, the behavior of the update can be
+        controlled using the `nx`, `xx`, `gt`, and `lt` parameters.
+
+        The return value provides detailed information about the outcome for each field.
+
+        For more information, see https://redis.io/commands/hexpire
+
+        Args:
+            name: The name of the hash key.
+            seconds: Expiration time in seconds, relative. Can be an integer, or a
+                     Python `timedelta` object.
+            fields: List of fields within the hash to apply the expiration time to.
+            nx: Set expiry only when the field has no expiry.
+            xx: Set expiry only when the field has an existing expiry.
+            gt: Set expiry only when the new expiry is greater than the current one.
+            lt: Set expiry only when the new expiry is less than the current one.
+
+        Returns:
+            If the key does not exist, returns an empty list. If the key exists, returns
+            a list which contains for each field in the request:
+                - `-2` if the field does not exist.
+                - `0` if the specified NX | XX | GT | LT condition was not met.
+                - `1` if the expiration time was set or updated.
+                - `2` if the field was deleted because the specified expiration time is
+                  in the past.
+        """
+        conditions = [nx, xx, gt, lt]
+        if sum(conditions) > 1:
+            raise ValueError("Only one of 'nx', 'xx', 'gt', 'lt' can be specified.")
+
+        if isinstance(seconds, datetime.timedelta):
+            seconds = int(seconds.total_seconds())
+
+        options = []
+        if nx:
+            options.append("NX")
+        if xx:
+            options.append("XX")
+        if gt:
+            options.append("GT")
+        if lt:
+            options.append("LT")
+
+        return self.execute_command(
+            "HEXPIRE", name, seconds, *options, "FIELDS", len(fields), *fields
+        )
+
+    def hpexpire(
+        self,
+        name: KeyT,
+        milliseconds: ExpiryT,
+        *fields: str,
+        nx: bool = False,
+        xx: bool = False,
+        gt: bool = False,
+        lt: bool = False,
+    ) -> ResponseT:
+        """
+        Sets or updates the expiration time for fields within a hash key, using relative
+        time in milliseconds.
+
+        If a field already has an expiration time, the behavior of the update can be
+        controlled using the `nx`, `xx`, `gt`, and `lt` parameters.
+
+        The return value provides detailed information about the outcome for each field.
+
+        For more information, see https://redis.io/commands/hpexpire
+
+        Args:
+            name: The name of the hash key.
+            milliseconds: Expiration time in milliseconds, relative. Can be an integer,
+                          or a Python `timedelta` object.
+            fields: List of fields within the hash to apply the expiration time to.
+            nx: Set expiry only when the field has no expiry.
+            xx: Set expiry only when the field has an existing expiry.
+            gt: Set expiry only when the new expiry is greater than the current one.
+            lt: Set expiry only when the new expiry is less than the current one.
+
+        Returns:
+            If the key does not exist, returns an empty list. If the key exists, returns
+            a list which contains for each field in the request:
+                - `-2` if the field does not exist.
+                - `0` if the specified NX | XX | GT | LT condition was not met.
+                - `1` if the expiration time was set or updated.
+                - `2` if the field was deleted because the specified expiration time is
+                  in the past.
+        """
+        conditions = [nx, xx, gt, lt]
+        if sum(conditions) > 1:
+            raise ValueError("Only one of 'nx', 'xx', 'gt', 'lt' can be specified.")
+
+        if isinstance(milliseconds, datetime.timedelta):
+            milliseconds = int(milliseconds.total_seconds() * 1000)
+
+        options = []
+        if nx:
+            options.append("NX")
+        if xx:
+            options.append("XX")
+        if gt:
+            options.append("GT")
+        if lt:
+            options.append("LT")
+
+        return self.execute_command(
+            "HPEXPIRE", name, milliseconds, *options, "FIELDS", len(fields), *fields
+        )
+
+    def hexpireat(
+        self,
+        name: KeyT,
+        unix_time_seconds: AbsExpiryT,
+        *fields: str,
+        nx: bool = False,
+        xx: bool = False,
+        gt: bool = False,
+        lt: bool = False,
+    ) -> ResponseT:
+        """
+        Sets or updates the expiration time for fields within a hash key, using an
+        absolute Unix timestamp in seconds.
+
+        If a field already has an expiration time, the behavior of the update can be
+        controlled using the `nx`, `xx`, `gt`, and `lt` parameters.
+
+        The return value provides detailed information about the outcome for each field.
+
+        For more information, see https://redis.io/commands/hexpireat
+
+        Args:
+            name: The name of the hash key.
+            unix_time_seconds: Expiration time as Unix timestamp in seconds. Can be an
+                               integer or a Python `datetime` object.
+            fields: List of fields within the hash to apply the expiration time to.
+            nx: Set expiry only when the field has no expiry.
+            xx: Set expiry only when the field has an existing expiration time.
+            gt: Set expiry only when the new expiry is greater than the current one.
+            lt: Set expiry only when the new expiry is less than the current one.
+
+        Returns:
+            If the key does not exist, returns an empty list. If the key exists, returns
+            a list which contains for each field in the request:
+                - `-2` if the field does not exist.
+                - `0` if the specified NX | XX | GT | LT condition was not met.
+                - `1` if the expiration time was set or updated.
+                - `2` if the field was deleted because the specified expiration time is
+                  in the past.
+        """
+        conditions = [nx, xx, gt, lt]
+        if sum(conditions) > 1:
+            raise ValueError("Only one of 'nx', 'xx', 'gt', 'lt' can be specified.")
+
+        if isinstance(unix_time_seconds, datetime.datetime):
+            unix_time_seconds = int(unix_time_seconds.timestamp())
+
+        options = []
+        if nx:
+            options.append("NX")
+        if xx:
+            options.append("XX")
+        if gt:
+            options.append("GT")
+        if lt:
+            options.append("LT")
+
+        return self.execute_command(
+            "HEXPIREAT",
+            name,
+            unix_time_seconds,
+            *options,
+            "FIELDS",
+            len(fields),
+            *fields,
+        )
+
+    def hpexpireat(
+        self,
+        name: KeyT,
+        unix_time_milliseconds: AbsExpiryT,
+        *fields: str,
+        nx: bool = False,
+        xx: bool = False,
+        gt: bool = False,
+        lt: bool = False,
+    ) -> ResponseT:
+        """
+        Sets or updates the expiration time for fields within a hash key, using an
+        absolute Unix timestamp in milliseconds.
+
+        If a field already has an expiration time, the behavior of the update can be
+        controlled using the `nx`, `xx`, `gt`, and `lt` parameters.
+
+        The return value provides detailed information about the outcome for each field.
+
+        For more information, see https://redis.io/commands/hpexpireat
+
+        Args:
+            name: The name of the hash key.
+            unix_time_milliseconds: Expiration time as Unix timestamp in milliseconds.
+                                    Can be an integer or a Python `datetime` object.
+            fields: List of fields within the hash to apply the expiry.
+            nx: Set expiry only when the field has no expiry.
+            xx: Set expiry only when the field has an existing expiry.
+            gt: Set expiry only when the new expiry is greater than the current one.
+            lt: Set expiry only when the new expiry is less than the current one.
+
+        Returns:
+            If the key does not exist, returns an empty list. If the key exists, returns
+            a list which contains for each field in the request:
+                - `-2` if the field does not exist.
+                - `0` if the specified NX | XX | GT | LT condition was not met.
+                - `1` if the expiration time was set or updated.
+                - `2` if the field was deleted because the specified expiration time is
+                  in the past.
+        """
+        conditions = [nx, xx, gt, lt]
+        if sum(conditions) > 1:
+            raise ValueError("Only one of 'nx', 'xx', 'gt', 'lt' can be specified.")
+
+        if isinstance(unix_time_milliseconds, datetime.datetime):
+            unix_time_milliseconds = int(unix_time_milliseconds.timestamp() * 1000)
+
+        options = []
+        if nx:
+            options.append("NX")
+        if xx:
+            options.append("XX")
+        if gt:
+            options.append("GT")
+        if lt:
+            options.append("LT")
+
+        return self.execute_command(
+            "HPEXPIREAT",
+            name,
+            unix_time_milliseconds,
+            *options,
+            "FIELDS",
+            len(fields),
+            *fields,
+        )
+
+    def hpersist(self, name: KeyT, *fields: str) -> ResponseT:
+        """
+        Removes the expiration time for each specified field in a hash.
+
+        For more information, see https://redis.io/commands/hpersist
+
+        Args:
+            name: The name of the hash key.
+            fields: A list of fields within the hash from which to remove the
+                    expiration time.
+
+        Returns:
+            If the key does not exist, returns an empty list. If the key exists, returns
+            a list which contains for each field in the request:
+                - `-2` if the field does not exist.
+                - `-1` if the field exists but has no associated expiration time.
+                - `1` if the expiration time was successfully removed from the field.
+        """
+        return self.execute_command("HPERSIST", name, "FIELDS", len(fields), *fields)
+
+    def hexpiretime(self, key: KeyT, *fields: str) -> ResponseT:
+        """
+        Returns the expiration times of hash fields as Unix timestamps in seconds.
+
+        For more information, see https://redis.io/commands/hexpiretime
+
+        Args:
+            key: The hash key.
+            fields: A list of fields within the hash for which to get the expiration
+                    time.
+
+        Returns:
+            If the key does not exist, returns an empty list. If the key exists, returns
+            a list which contains for each field in the request:
+                - `-2` if the field does not exist.
+                - `-1` if the field exists but has no associated expire time.
+                - A positive integer representing the expiration Unix timestamp in
+                  seconds, if the field has an associated expiration time.
+        """
+        return self.execute_command(
+            "HEXPIRETIME", key, "FIELDS", len(fields), *fields, keys=[key]
+        )
+
+    def hpexpiretime(self, key: KeyT, *fields: str) -> ResponseT:
+        """
+        Returns the expiration times of hash fields as Unix timestamps in milliseconds.
+
+        For more information, see https://redis.io/commands/hpexpiretime
+
+        Args:
+            key: The hash key.
+            fields: A list of fields within the hash for which to get the expiration
+                    time.
+
+        Returns:
+            If the key does not exist, returns an empty list. If the key exists, returns
+            a list which contains for each field in the request:
+                - `-2` if the field does not exist.
+                - `-1` if the field exists but has no associated expire time.
+                - A positive integer representing the expiration Unix timestamp in
+                  milliseconds, if the field has an associated expiration time.
+        """
+        return self.execute_command(
+            "HPEXPIRETIME", key, "FIELDS", len(fields), *fields, keys=[key]
+        )
+
+    def httl(self, key: KeyT, *fields: str) -> ResponseT:
+        """
+        Returns the TTL (Time To Live) in seconds for each specified field within a hash
+        key.
+
+        For more information, see https://redis.io/commands/httl
+
+        Args:
+            key: The hash key.
+            fields: A list of fields within the hash for which to get the TTL.
+
+        Returns:
+            If the key does not exist, returns an empty list. If the key exists, returns
+            a list which contains for each field in the request:
+                - `-2` if the field does not exist.
+                - `-1` if the field exists but has no associated expire time.
+                - A positive integer representing the TTL in seconds if the field has
+                  an associated expiration time.
+        """
+        return self.execute_command(
+            "HTTL", key, "FIELDS", len(fields), *fields, keys=[key]
+        )
+
+    def hpttl(self, key: KeyT, *fields: str) -> ResponseT:
+        """
+        Returns the TTL (Time To Live) in milliseconds for each specified field within a
+        hash key.
+
+        For more information, see https://redis.io/commands/hpttl
+
+        Args:
+            key: The hash key.
+            fields: A list of fields within the hash for which to get the TTL.
+
+        Returns:
+            If the key does not exist, returns an empty list. If the key exists, returns
+            a list which contains for each field in the request:
+                - `-2` if the field does not exist.
+                - `-1` if the field exists but has no associated expire time.
+                - A positive integer representing the TTL in milliseconds if the field
+                  has an associated expiration time.
+        """
+        return self.execute_command(
+            "HPTTL", key, "FIELDS", len(fields), *fields, keys=[key]
+        )
+
 
 AsyncHashCommands = HashCommands
 
diff --git a/tests/test_asyncio/test_hash.py b/tests/test_asyncio/test_hash.py
new file mode 100644
index 000000000..8428f124e
--- /dev/null
+++ b/tests/test_asyncio/test_hash.py
@@ -0,0 +1,300 @@
+import asyncio
+from datetime import datetime, timedelta
+
+from tests.conftest import skip_if_server_version_lt
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hexpire_basic(r):
+    await r.delete("test:hash")
+    await r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    assert await r.hexpire("test:hash", 1, "field1") == [1]
+    await asyncio.sleep(1.1)
+    assert await r.hexists("test:hash", "field1") is False
+    assert await r.hexists("test:hash", "field2") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hexpire_with_timedelta(r):
+    await r.delete("test:hash")
+    await r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    assert await r.hexpire("test:hash", timedelta(seconds=1), "field1") == [1]
+    await asyncio.sleep(1.1)
+    assert await r.hexists("test:hash", "field1") is False
+    assert await r.hexists("test:hash", "field2") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hexpire_conditions(r):
+    await r.delete("test:hash")
+    await r.hset("test:hash", mapping={"field1": "value1"})
+    assert await r.hexpire("test:hash", 2, "field1", xx=True) == [0]
+    assert await r.hexpire("test:hash", 2, "field1", nx=True) == [1]
+    assert await r.hexpire("test:hash", 1, "field1", xx=True) == [1]
+    assert await r.hexpire("test:hash", 2, "field1", nx=True) == [0]
+    await asyncio.sleep(1.1)
+    assert await r.hexists("test:hash", "field1") is False
+    await r.hset("test:hash", "field1", "value1")
+    await r.hexpire("test:hash", 2, "field1")
+    assert await r.hexpire("test:hash", 1, "field1", gt=True) == [0]
+    assert await r.hexpire("test:hash", 1, "field1", lt=True) == [1]
+    await asyncio.sleep(1.1)
+    assert await r.hexists("test:hash", "field1") is False
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hexpire_nonexistent_key_or_field(r):
+    await r.delete("test:hash")
+    assert await r.hexpire("test:hash", 1, "field1") == []
+    await r.hset("test:hash", "field1", "value1")
+    assert await r.hexpire("test:hash", 1, "nonexistent_field") == [-2]
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hexpire_multiple_fields(r):
+    await r.delete("test:hash")
+    await r.hset(
+        "test:hash",
+        mapping={"field1": "value1", "field2": "value2", "field3": "value3"},
+    )
+    assert await r.hexpire("test:hash", 1, "field1", "field2") == [1, 1]
+    await asyncio.sleep(1.1)
+    assert await r.hexists("test:hash", "field1") is False
+    assert await r.hexists("test:hash", "field2") is False
+    assert await r.hexists("test:hash", "field3") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hpexpire_basic(r):
+    await r.delete("test:hash")
+    await r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    assert await r.hpexpire("test:hash", 500, "field1") == [1]
+    await asyncio.sleep(0.6)
+    assert await r.hexists("test:hash", "field1") is False
+    assert await r.hexists("test:hash", "field2") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hpexpire_with_timedelta(r):
+    await r.delete("test:hash")
+    await r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    assert await r.hpexpire("test:hash", timedelta(milliseconds=500), "field1") == [1]
+    await asyncio.sleep(0.6)
+    assert await r.hexists("test:hash", "field1") is False
+    assert await r.hexists("test:hash", "field2") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hpexpire_conditions(r):
+    await r.delete("test:hash")
+    await r.hset("test:hash", mapping={"field1": "value1"})
+    assert await r.hpexpire("test:hash", 1500, "field1", xx=True) == [0]
+    assert await r.hpexpire("test:hash", 1500, "field1", nx=True) == [1]
+    assert await r.hpexpire("test:hash", 500, "field1", xx=True) == [1]
+    assert await r.hpexpire("test:hash", 1500, "field1", nx=True) == [0]
+    await asyncio.sleep(0.6)
+    assert await r.hexists("test:hash", "field1") is False
+    await r.hset("test:hash", "field1", "value1")
+    await r.hpexpire("test:hash", 1000, "field1")
+    assert await r.hpexpire("test:hash", 500, "field1", gt=True) == [0]
+    assert await r.hpexpire("test:hash", 500, "field1", lt=True) == [1]
+    await asyncio.sleep(0.6)
+    assert await r.hexists("test:hash", "field1") is False
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hpexpire_nonexistent_key_or_field(r):
+    await r.delete("test:hash")
+    assert await r.hpexpire("test:hash", 500, "field1") == []
+    await r.hset("test:hash", "field1", "value1")
+    assert await r.hpexpire("test:hash", 500, "nonexistent_field") == [-2]
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hpexpire_multiple_fields(r):
+    await r.delete("test:hash")
+    await r.hset(
+        "test:hash",
+        mapping={"field1": "value1", "field2": "value2", "field3": "value3"},
+    )
+    assert await r.hpexpire("test:hash", 500, "field1", "field2") == [1, 1]
+    await asyncio.sleep(0.6)
+    assert await r.hexists("test:hash", "field1") is False
+    assert await r.hexists("test:hash", "field2") is False
+    assert await r.hexists("test:hash", "field3") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hexpireat_basic(r):
+    await r.delete("test:hash")
+    await r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    exp_time = int((datetime.now() + timedelta(seconds=1)).timestamp())
+    assert await r.hexpireat("test:hash", exp_time, "field1") == [1]
+    await asyncio.sleep(1.1)
+    assert await r.hexists("test:hash", "field1") is False
+    assert await r.hexists("test:hash", "field2") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hexpireat_with_datetime(r):
+    await r.delete("test:hash")
+    await r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    exp_time = datetime.now() + timedelta(seconds=1)
+    assert await r.hexpireat("test:hash", exp_time, "field1") == [1]
+    await asyncio.sleep(1.1)
+    assert await r.hexists("test:hash", "field1") is False
+    assert await r.hexists("test:hash", "field2") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hexpireat_conditions(r):
+    await r.delete("test:hash")
+    await r.hset("test:hash", mapping={"field1": "value1"})
+    future_exp_time = int((datetime.now() + timedelta(seconds=2)).timestamp())
+    past_exp_time = int((datetime.now() - timedelta(seconds=1)).timestamp())
+    assert await r.hexpireat("test:hash", future_exp_time, "field1", xx=True) == [0]
+    assert await r.hexpireat("test:hash", future_exp_time, "field1", nx=True) == [1]
+    assert await r.hexpireat("test:hash", past_exp_time, "field1", gt=True) == [0]
+    assert await r.hexpireat("test:hash", past_exp_time, "field1", lt=True) == [2]
+    assert await r.hexists("test:hash", "field1") is False
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hexpireat_nonexistent_key_or_field(r):
+    await r.delete("test:hash")
+    future_exp_time = int((datetime.now() + timedelta(seconds=1)).timestamp())
+    assert await r.hexpireat("test:hash", future_exp_time, "field1") == []
+    await r.hset("test:hash", "field1", "value1")
+    assert await r.hexpireat("test:hash", future_exp_time, "nonexistent_field") == [-2]
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hexpireat_multiple_fields(r):
+    await r.delete("test:hash")
+    await r.hset(
+        "test:hash",
+        mapping={"field1": "value1", "field2": "value2", "field3": "value3"},
+    )
+    exp_time = int((datetime.now() + timedelta(seconds=1)).timestamp())
+    assert await r.hexpireat("test:hash", exp_time, "field1", "field2") == [1, 1]
+    await asyncio.sleep(1.1)
+    assert await r.hexists("test:hash", "field1") is False
+    assert await r.hexists("test:hash", "field2") is False
+    assert await r.hexists("test:hash", "field3") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hpexpireat_basic(r):
+    await r.delete("test:hash")
+    await r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    exp_time = int((datetime.now() + timedelta(milliseconds=400)).timestamp() * 1000)
+    assert await r.hpexpireat("test:hash", exp_time, "field1") == [1]
+    await asyncio.sleep(0.5)
+    assert await r.hexists("test:hash", "field1") is False
+    assert await r.hexists("test:hash", "field2") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hpexpireat_with_datetime(r):
+    await r.delete("test:hash")
+    await r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    exp_time = datetime.now() + timedelta(milliseconds=400)
+    assert await r.hpexpireat("test:hash", exp_time, "field1") == [1]
+    await asyncio.sleep(0.5)
+    assert await r.hexists("test:hash", "field1") is False
+    assert await r.hexists("test:hash", "field2") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hpexpireat_conditions(r):
+    await r.delete("test:hash")
+    await r.hset("test:hash", mapping={"field1": "value1"})
+    future_exp_time = int(
+        (datetime.now() + timedelta(milliseconds=500)).timestamp() * 1000
+    )
+    past_exp_time = int(
+        (datetime.now() - timedelta(milliseconds=500)).timestamp() * 1000
+    )
+    assert await r.hpexpireat("test:hash", future_exp_time, "field1", xx=True) == [0]
+    assert await r.hpexpireat("test:hash", future_exp_time, "field1", nx=True) == [1]
+    assert await r.hpexpireat("test:hash", past_exp_time, "field1", gt=True) == [0]
+    assert await r.hpexpireat("test:hash", past_exp_time, "field1", lt=True) == [2]
+    assert await r.hexists("test:hash", "field1") is False
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hpexpireat_nonexistent_key_or_field(r):
+    await r.delete("test:hash")
+    future_exp_time = int(
+        (datetime.now() + timedelta(milliseconds=500)).timestamp() * 1000
+    )
+    assert await r.hpexpireat("test:hash", future_exp_time, "field1") == []
+    await r.hset("test:hash", "field1", "value1")
+    assert await r.hpexpireat("test:hash", future_exp_time, "nonexistent_field") == [-2]
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hpexpireat_multiple_fields(r):
+    await r.delete("test:hash")
+    await r.hset(
+        "test:hash",
+        mapping={"field1": "value1", "field2": "value2", "field3": "value3"},
+    )
+    exp_time = int((datetime.now() + timedelta(milliseconds=400)).timestamp() * 1000)
+    assert await r.hpexpireat("test:hash", exp_time, "field1", "field2") == [1, 1]
+    await asyncio.sleep(0.5)
+    assert await r.hexists("test:hash", "field1") is False
+    assert await r.hexists("test:hash", "field2") is False
+    assert await r.hexists("test:hash", "field3") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hpersist_multiple_fields_mixed_conditions(r):
+    await r.delete("test:hash")
+    await r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    await r.hexpire("test:hash", 5000, "field1")
+    assert await r.hpersist("test:hash", "field1", "field2", "field3") == [1, -1, -2]
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hexpiretime_multiple_fields_mixed_conditions(r):
+    await r.delete("test:hash")
+    await r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    future_time = int((datetime.now() + timedelta(minutes=30)).timestamp())
+    await r.hexpireat("test:hash", future_time, "field1")
+    result = await r.hexpiretime("test:hash", "field1", "field2", "field3")
+    assert future_time - 10 < result[0] <= future_time
+    assert result[1:] == [-1, -2]
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_hpexpiretime_multiple_fields_mixed_conditions(r):
+    await r.delete("test:hash")
+    await r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    future_time = int((datetime.now() + timedelta(minutes=30)).timestamp())
+    await r.hexpireat("test:hash", future_time, "field1")
+    result = await r.hpexpiretime("test:hash", "field1", "field2", "field3")
+    assert future_time * 1000 - 10000 < result[0] <= future_time * 1000
+    assert result[1:] == [-1, -2]
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_ttl_multiple_fields_mixed_conditions(r):
+    await r.delete("test:hash")
+    await r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    future_time = int((datetime.now() + timedelta(minutes=30)).timestamp())
+    await r.hexpireat("test:hash", future_time, "field1")
+    result = await r.httl("test:hash", "field1", "field2", "field3")
+    assert 30 * 60 - 10 < result[0] <= 30 * 60
+    assert result[1:] == [-1, -2]
+
+
+@skip_if_server_version_lt("7.4.0")
+async def test_pttl_multiple_fields_mixed_conditions(r):
+    await r.delete("test:hash")
+    await r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    future_time = int((datetime.now() + timedelta(minutes=30)).timestamp())
+    await r.hexpireat("test:hash", future_time, "field1")
+    result = await r.hpttl("test:hash", "field1", "field2", "field3")
+    assert 30 * 60000 - 10000 < result[0] <= 30 * 60000
+    assert result[1:] == [-1, -2]
diff --git a/tests/test_hash.py b/tests/test_hash.py
new file mode 100644
index 000000000..547a339af
--- /dev/null
+++ b/tests/test_hash.py
@@ -0,0 +1,370 @@
+import time
+from datetime import datetime, timedelta
+
+import pytest
+
+from tests.conftest import skip_if_server_version_lt
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hexpire_basic(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    assert r.hexpire("test:hash", 1, "field1") == [1]
+    time.sleep(1.1)
+    assert r.hexists("test:hash", "field1") is False
+    assert r.hexists("test:hash", "field2") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hexpire_with_timedelta(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    assert r.hexpire("test:hash", timedelta(seconds=1), "field1") == [1]
+    time.sleep(1.1)
+    assert r.hexists("test:hash", "field1") is False
+    assert r.hexists("test:hash", "field2") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hexpire_conditions(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1"})
+    assert r.hexpire("test:hash", 2, "field1", xx=True) == [0]
+    assert r.hexpire("test:hash", 2, "field1", nx=True) == [1]
+    assert r.hexpire("test:hash", 1, "field1", xx=True) == [1]
+    assert r.hexpire("test:hash", 2, "field1", nx=True) == [0]
+    time.sleep(1.1)
+    assert r.hexists("test:hash", "field1") is False
+    r.hset("test:hash", "field1", "value1")
+    r.hexpire("test:hash", 2, "field1")
+    assert r.hexpire("test:hash", 1, "field1", gt=True) == [0]
+    assert r.hexpire("test:hash", 1, "field1", lt=True) == [1]
+    time.sleep(1.1)
+    assert r.hexists("test:hash", "field1") is False
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hexpire_nonexistent_key_or_field(r):
+    r.delete("test:hash")
+    assert r.hexpire("test:hash", 1, "field1") == []
+    r.hset("test:hash", "field1", "value1")
+    assert r.hexpire("test:hash", 1, "nonexistent_field") == [-2]
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hexpire_multiple_fields(r):
+    r.delete("test:hash")
+    r.hset(
+        "test:hash",
+        mapping={"field1": "value1", "field2": "value2", "field3": "value3"},
+    )
+    assert r.hexpire("test:hash", 1, "field1", "field2") == [1, 1]
+    time.sleep(1.1)
+    assert r.hexists("test:hash", "field1") is False
+    assert r.hexists("test:hash", "field2") is False
+    assert r.hexists("test:hash", "field3") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hexpire_multiple_condition_flags_error(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1"})
+    with pytest.raises(ValueError) as e:
+        r.hexpire("test:hash", 1, "field1", nx=True, xx=True)
+    assert "Only one of" in str(e)
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpexpire_basic(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    assert r.hpexpire("test:hash", 500, "field1") == [1]
+    time.sleep(0.6)
+    assert r.hexists("test:hash", "field1") is False
+    assert r.hexists("test:hash", "field2") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpexpire_with_timedelta(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    assert r.hpexpire("test:hash", timedelta(milliseconds=500), "field1") == [1]
+    time.sleep(0.6)
+    assert r.hexists("test:hash", "field1") is False
+    assert r.hexists("test:hash", "field2") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpexpire_conditions(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1"})
+    assert r.hpexpire("test:hash", 1500, "field1", xx=True) == [0]
+    assert r.hpexpire("test:hash", 1500, "field1", nx=True) == [1]
+    assert r.hpexpire("test:hash", 500, "field1", xx=True) == [1]
+    assert r.hpexpire("test:hash", 1500, "field1", nx=True) == [0]
+    time.sleep(0.6)
+    assert r.hexists("test:hash", "field1") is False
+    r.hset("test:hash", "field1", "value1")
+    r.hpexpire("test:hash", 1000, "field1")
+    assert r.hpexpire("test:hash", 500, "field1", gt=True) == [0]
+    assert r.hpexpire("test:hash", 500, "field1", lt=True) == [1]
+    time.sleep(0.6)
+    assert r.hexists("test:hash", "field1") is False
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpexpire_nonexistent_key_or_field(r):
+    r.delete("test:hash")
+    assert r.hpexpire("test:hash", 500, "field1") == []
+    r.hset("test:hash", "field1", "value1")
+    assert r.hpexpire("test:hash", 500, "nonexistent_field") == [-2]
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpexpire_multiple_fields(r):
+    r.delete("test:hash")
+    r.hset(
+        "test:hash",
+        mapping={"field1": "value1", "field2": "value2", "field3": "value3"},
+    )
+    assert r.hpexpire("test:hash", 500, "field1", "field2") == [1, 1]
+    time.sleep(0.6)
+    assert r.hexists("test:hash", "field1") is False
+    assert r.hexists("test:hash", "field2") is False
+    assert r.hexists("test:hash", "field3") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpexpire_multiple_condition_flags_error(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1"})
+    with pytest.raises(ValueError) as e:
+        r.hpexpire("test:hash", 500, "field1", nx=True, xx=True)
+    assert "Only one of" in str(e)
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hexpireat_basic(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    exp_time = int((datetime.now() + timedelta(seconds=1)).timestamp())
+    assert r.hexpireat("test:hash", exp_time, "field1") == [1]
+    time.sleep(1.1)
+    assert r.hexists("test:hash", "field1") is False
+    assert r.hexists("test:hash", "field2") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hexpireat_with_datetime(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    exp_time = datetime.now() + timedelta(seconds=1)
+    assert r.hexpireat("test:hash", exp_time, "field1") == [1]
+    time.sleep(1.1)
+    assert r.hexists("test:hash", "field1") is False
+    assert r.hexists("test:hash", "field2") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hexpireat_conditions(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1"})
+    future_exp_time = int((datetime.now() + timedelta(seconds=2)).timestamp())
+    past_exp_time = int((datetime.now() - timedelta(seconds=1)).timestamp())
+    assert r.hexpireat("test:hash", future_exp_time, "field1", xx=True) == [0]
+    assert r.hexpireat("test:hash", future_exp_time, "field1", nx=True) == [1]
+    assert r.hexpireat("test:hash", past_exp_time, "field1", gt=True) == [0]
+    assert r.hexpireat("test:hash", past_exp_time, "field1", lt=True) == [2]
+    assert r.hexists("test:hash", "field1") is False
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hexpireat_nonexistent_key_or_field(r):
+    r.delete("test:hash")
+    future_exp_time = int((datetime.now() + timedelta(seconds=1)).timestamp())
+    assert r.hexpireat("test:hash", future_exp_time, "field1") == []
+    r.hset("test:hash", "field1", "value1")
+    assert r.hexpireat("test:hash", future_exp_time, "nonexistent_field") == [-2]
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hexpireat_multiple_fields(r):
+    r.delete("test:hash")
+    r.hset(
+        "test:hash",
+        mapping={"field1": "value1", "field2": "value2", "field3": "value3"},
+    )
+    exp_time = int((datetime.now() + timedelta(seconds=1)).timestamp())
+    assert r.hexpireat("test:hash", exp_time, "field1", "field2") == [1, 1]
+    time.sleep(1.1)
+    assert r.hexists("test:hash", "field1") is False
+    assert r.hexists("test:hash", "field2") is False
+    assert r.hexists("test:hash", "field3") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hexpireat_multiple_condition_flags_error(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1"})
+    exp_time = int((datetime.now() + timedelta(seconds=1)).timestamp())
+    with pytest.raises(ValueError) as e:
+        r.hexpireat("test:hash", exp_time, "field1", nx=True, xx=True)
+    assert "Only one of" in str(e)
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpexpireat_basic(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    exp_time = int((datetime.now() + timedelta(milliseconds=400)).timestamp() * 1000)
+    assert r.hpexpireat("test:hash", exp_time, "field1") == [1]
+    time.sleep(0.5)
+    assert r.hexists("test:hash", "field1") is False
+    assert r.hexists("test:hash", "field2") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpexpireat_with_datetime(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    exp_time = datetime.now() + timedelta(milliseconds=400)
+    assert r.hpexpireat("test:hash", exp_time, "field1") == [1]
+    time.sleep(0.5)
+    assert r.hexists("test:hash", "field1") is False
+    assert r.hexists("test:hash", "field2") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpexpireat_conditions(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1"})
+    future_exp_time = int(
+        (datetime.now() + timedelta(milliseconds=500)).timestamp() * 1000
+    )
+    past_exp_time = int(
+        (datetime.now() - timedelta(milliseconds=500)).timestamp() * 1000
+    )
+    assert r.hpexpireat("test:hash", future_exp_time, "field1", xx=True) == [0]
+    assert r.hpexpireat("test:hash", future_exp_time, "field1", nx=True) == [1]
+    assert r.hpexpireat("test:hash", past_exp_time, "field1", gt=True) == [0]
+    assert r.hpexpireat("test:hash", past_exp_time, "field1", lt=True) == [2]
+    assert r.hexists("test:hash", "field1") is False
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpexpireat_nonexistent_key_or_field(r):
+    r.delete("test:hash")
+    future_exp_time = int(
+        (datetime.now() + timedelta(milliseconds=500)).timestamp() * 1000
+    )
+    assert r.hpexpireat("test:hash", future_exp_time, "field1") == []
+    r.hset("test:hash", "field1", "value1")
+    assert r.hpexpireat("test:hash", future_exp_time, "nonexistent_field") == [-2]
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpexpireat_multiple_fields(r):
+    r.delete("test:hash")
+    r.hset(
+        "test:hash",
+        mapping={"field1": "value1", "field2": "value2", "field3": "value3"},
+    )
+    exp_time = int((datetime.now() + timedelta(milliseconds=400)).timestamp() * 1000)
+    assert r.hpexpireat("test:hash", exp_time, "field1", "field2") == [1, 1]
+    time.sleep(0.5)
+    assert r.hexists("test:hash", "field1") is False
+    assert r.hexists("test:hash", "field2") is False
+    assert r.hexists("test:hash", "field3") is True
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpexpireat_multiple_condition_flags_error(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1"})
+    exp_time = int((datetime.now() + timedelta(milliseconds=500)).timestamp())
+    with pytest.raises(ValueError) as e:
+        r.hpexpireat("test:hash", exp_time, "field1", nx=True, xx=True)
+    assert "Only one of" in str(e)
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpersist_multiple_fields(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    r.hexpire("test:hash", 5000, "field1")
+    assert r.hpersist("test:hash", "field1", "field2", "field3") == [1, -1, -2]
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpersist_nonexistent_key(r):
+    r.delete("test:hash")
+    assert r.hpersist("test:hash", "field1", "field2", "field3") == []
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hexpiretime_multiple_fields_mixed_conditions(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    future_time = int((datetime.now() + timedelta(minutes=30)).timestamp())
+    r.hexpireat("test:hash", future_time, "field1")
+    result = r.hexpiretime("test:hash", "field1", "field2", "field3")
+    assert future_time - 10 < result[0] <= future_time
+    assert result[1:] == [-1, -2]
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hexpiretime_nonexistent_key(r):
+    r.delete("test:hash")
+    assert r.hexpiretime("test:hash", "field1", "field2", "field3") == []
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpexpiretime_multiple_fields_mixed_conditions(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    future_time = int((datetime.now() + timedelta(minutes=30)).timestamp())
+    r.hexpireat("test:hash", future_time, "field1")
+    result = r.hpexpiretime("test:hash", "field1", "field2", "field3")
+    assert future_time * 1000 - 10000 < result[0] <= future_time * 1000
+    assert result[1:] == [-1, -2]
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpexpiretime_nonexistent_key(r):
+    r.delete("test:hash")
+    assert r.hpexpiretime("test:hash", "field1", "field2", "field3") == []
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_httl_multiple_fields_mixed_conditions(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    future_time = int((datetime.now() + timedelta(minutes=30)).timestamp())
+    r.hexpireat("test:hash", future_time, "field1")
+    result = r.httl("test:hash", "field1", "field2", "field3")
+    assert 30 * 60 - 10 < result[0] <= 30 * 60
+    assert result[1:] == [-1, -2]
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_httl_nonexistent_key(r):
+    r.delete("test:hash")
+    assert r.httl("test:hash", "field1", "field2", "field3") == []
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpttl_multiple_fields_mixed_conditions(r):
+    r.delete("test:hash")
+    r.hset("test:hash", mapping={"field1": "value1", "field2": "value2"})
+    future_time = int((datetime.now() + timedelta(minutes=30)).timestamp())
+    r.hexpireat("test:hash", future_time, "field1")
+    result = r.hpttl("test:hash", "field1", "field2", "field3")
+    assert 30 * 60000 - 10000 < result[0] <= 30 * 60000
+    assert result[1:] == [-1, -2]
+
+
+@skip_if_server_version_lt("7.4.0")
+def test_hpttl_nonexistent_key(r):
+    r.delete("test:hash")
+    assert r.hpttl("test:hash", "field1", "field2", "field3") == []