m2_g2_fs/usr/local/lib/python3.10/dist-packages/redis/commands/redismodules.py

from __future__ import annotations

from json import JSONDecoder, JSONEncoder
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from .bf import BFBloom, CFBloom, CMSBloom, TDigestBloom, TOPKBloom
    from .json import JSON
    from .search import AsyncSearch, Search
    from .timeseries import TimeSeries
    from .vectorset import VectorSet


class RedisModuleCommands:
    """This class contains the wrapper functions to bring supported redis
    modules into the command namespace.
    """

    def json(self, encoder=JSONEncoder(), decoder=JSONDecoder()) -> JSON:
        """Access the json namespace, providing support for redis json."""

        from .json import JSON

        jj = JSON(client=self, encoder=encoder, decoder=decoder)
        return jj

    def ft(self, index_name="idx") -> Search:
        """Access the search namespace, providing support for redis search."""

        from .search import Search

        s = Search(client=self, index_name=index_name)
        return s

    def ts(self) -> TimeSeries:
        """Access the timeseries namespace, providing support for
        redis timeseries data.
        """

        from .timeseries import TimeSeries

        s = TimeSeries(client=self)
        return s

    def bf(self) -> BFBloom:
        """Access the bloom namespace."""

        from .bf import BFBloom

        bf = BFBloom(client=self)
        return bf

    def cf(self) -> CFBloom:
        """Access the bloom namespace."""

        from .bf import CFBloom

        cf = CFBloom(client=self)
        return cf

    def cms(self) -> CMSBloom:
        """Access the bloom namespace."""

        from .bf import CMSBloom

        cms = CMSBloom(client=self)
        return cms

    def topk(self) -> TOPKBloom:
        """Access the bloom namespace."""

        from .bf import TOPKBloom

        topk = TOPKBloom(client=self)
        return topk

    def tdigest(self) -> TDigestBloom:
        """Access the bloom namespace."""

        from .bf import TDigestBloom

        tdigest = TDigestBloom(client=self)
        return tdigest

    def vset(self) -> VectorSet:
        """Access the VectorSet commands namespace."""

        from .vectorset import VectorSet

        vset = VectorSet(client=self)
        return vset


class AsyncRedisModuleCommands(RedisModuleCommands):
    def ft(self, index_name="idx") -> AsyncSearch:
        """Access the search namespace, providing support for redis search."""

        from .search import AsyncSearch

        s = AsyncSearch(client=self, index_name=index_name)
        return s