Skip to content

Indexes

Indexation documents of collection.

IndexMixin

Indexation documents in collection.

Source code in src/ramifice/commons/indexes.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
class IndexMixin:
    """Indexation documents in collection."""

    @classmethod
    @abstractmethod
    async def indexing(cls) -> None:
        """To set up and start indexing."""

    @classmethod
    async def create_index(  # type: ignore[no-untyped-def]
        cls,
        keys: Any,
        session: Any | None = None,
        comment: Any | None = None,
        **kwargs,
    ) -> str:
        """Creates an index on this collection."""
        metadata = cls.META
        # Get collection for current model.
        collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
        # Create index.
        result: str = await collection.create_index(
            keys=keys,
            session=session,
            comment=comment,
            **kwargs,
        )
        return result

    @classmethod
    async def drop_index(  # type: ignore[no-untyped-def]
        cls,
        index_or_name: Any,
        session: Any | None = None,
        comment: Any | None = None,
        **kwargs,
    ) -> None:
        """Drops the specified index on this collection."""
        metadata = cls.META
        # Get collection for current model.
        collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
        # Delete index.
        await collection.drop_index(
            index_or_name=index_or_name,
            session=session,
            comment=comment,
            **kwargs,
        )

    @classmethod
    async def create_indexes(  # type: ignore[no-untyped-def]
        cls,
        indexes: Any,
        session: Any | None = None,
        comment: Any | None = None,
        **kwargs,
    ) -> list[str]:
        """Create one or more indexes on this collection."""
        metadata = cls.META
        # Get collection for current model.
        collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
        # Create indexes.
        result: list[str] = await collection.create_indexes(
            indexes=indexes,
            session=session,
            comment=comment,
            **kwargs,
        )
        return result

    @classmethod
    async def drop_indexes(  # type: ignore[no-untyped-def]
        cls,
        session: Any | None = None,
        comment: Any | None = None,
        **kwargs,
    ) -> None:
        """Drops all indexes on this collection."""
        metadata = cls.META
        # Get collection for current model.
        collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
        # Delete indexes.
        await collection.drop_indexes(session=session, comment=comment, **kwargs)

    @classmethod
    async def index_information(
        cls,
        session: Any | None = None,
        comment: Any | None = None,
    ) -> Any:
        """Get information on this collection’s indexes."""
        metadata = cls.META
        # Get collection for current model.
        collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
        #
        return await collection.index_information(session=session, comment=comment)

    @classmethod
    async def list_indexes(
        cls,
        session: Any | None = None,
        comment: Any | None = None,
    ) -> Any:
        """Get a cursor over the index documents for this collection."""
        metadata = cls.META
        # Get collection for current model.
        collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
        #
        return await collection.list_indexes(session=session, comment=comment)

create_index(keys, session=None, comment=None, **kwargs) async classmethod

Creates an index on this collection.

Source code in src/ramifice/commons/indexes.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@classmethod
async def create_index(  # type: ignore[no-untyped-def]
    cls,
    keys: Any,
    session: Any | None = None,
    comment: Any | None = None,
    **kwargs,
) -> str:
    """Creates an index on this collection."""
    metadata = cls.META
    # Get collection for current model.
    collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
    # Create index.
    result: str = await collection.create_index(
        keys=keys,
        session=session,
        comment=comment,
        **kwargs,
    )
    return result

create_indexes(indexes, session=None, comment=None, **kwargs) async classmethod

Create one or more indexes on this collection.

Source code in src/ramifice/commons/indexes.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@classmethod
async def create_indexes(  # type: ignore[no-untyped-def]
    cls,
    indexes: Any,
    session: Any | None = None,
    comment: Any | None = None,
    **kwargs,
) -> list[str]:
    """Create one or more indexes on this collection."""
    metadata = cls.META
    # Get collection for current model.
    collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
    # Create indexes.
    result: list[str] = await collection.create_indexes(
        indexes=indexes,
        session=session,
        comment=comment,
        **kwargs,
    )
    return result

drop_index(index_or_name, session=None, comment=None, **kwargs) async classmethod

Drops the specified index on this collection.

Source code in src/ramifice/commons/indexes.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@classmethod
async def drop_index(  # type: ignore[no-untyped-def]
    cls,
    index_or_name: Any,
    session: Any | None = None,
    comment: Any | None = None,
    **kwargs,
) -> None:
    """Drops the specified index on this collection."""
    metadata = cls.META
    # Get collection for current model.
    collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
    # Delete index.
    await collection.drop_index(
        index_or_name=index_or_name,
        session=session,
        comment=comment,
        **kwargs,
    )

drop_indexes(session=None, comment=None, **kwargs) async classmethod

Drops all indexes on this collection.

Source code in src/ramifice/commons/indexes.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
@classmethod
async def drop_indexes(  # type: ignore[no-untyped-def]
    cls,
    session: Any | None = None,
    comment: Any | None = None,
    **kwargs,
) -> None:
    """Drops all indexes on this collection."""
    metadata = cls.META
    # Get collection for current model.
    collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
    # Delete indexes.
    await collection.drop_indexes(session=session, comment=comment, **kwargs)

index_information(session=None, comment=None) async classmethod

Get information on this collection’s indexes.

Source code in src/ramifice/commons/indexes.py 
116
117
118
119
120
121
122
123
124
125
126
127
@classmethod
async def index_information(
    cls,
    session: Any | None = None,
    comment: Any | None = None,
) -> Any:
    """Get information on this collection’s indexes."""
    metadata = cls.META
    # Get collection for current model.
    collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
    #
    return await collection.index_information(session=session, comment=comment)

indexing() abstractmethod async classmethod

To set up and start indexing.

Source code in src/ramifice/commons/indexes.py 
35
36
37
38
@classmethod
@abstractmethod
async def indexing(cls) -> None:
    """To set up and start indexing."""

list_indexes(session=None, comment=None) async classmethod

Get a cursor over the index documents for this collection.

Source code in src/ramifice/commons/indexes.py 
129
130
131
132
133
134
135
136
137
138
139
140
@classmethod
async def list_indexes(
    cls,
    session: Any | None = None,
    comment: Any | None = None,
) -> Any:
    """Get a cursor over the index documents for this collection."""
    metadata = cls.META
    # Get collection for current model.
    collection: AsyncCollection = Config.MONGO_DATABASE[metadata["collection_name"]]
    #
    return await collection.list_indexes(session=session, comment=comment)