Skip to content

Collection

Methods for working with collections.

Collection

Methods for working with collections.

Source code in src/scruby/mixins/collection.py 
20
21
22
23
24
25
26
27
28
29
30
31
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
class Collection:
    """Methods for working with collections."""

    @final
    def collection_name(self) -> str:
        """Asynchronous method for getting the collection name.

        Returns:
            Collection name.
        """
        return self._class_model.__name__

    @final
    @staticmethod
    def collection_list() -> list[str] | None:
        """Synchronous method for getting collection list."""
        collections = [item.__name__ for item in ScrubyModel.__subclasses__()]
        return collections or None

    @final
    @staticmethod
    def clear_collection(collection_name: str) -> None:
        """Synchronous method to remove all documents from a collection.

        Args:
            collection_name (str): Collection name.

        Returns:
            None.
        """
        db_root = ScrubyConfig.db_root
        hash_reduce_left = ScrubyConfig.HASH_REDUCE_LEFT
        max_number_branch = ScrubyConfig.MAX_NUMBER_BRANCH

        # Delete collection on file system
        target_directory = f"{db_root}/{collection_name}"
        rmtree(target_directory)

        # Create a directory for the collection and add metadata
        Metadata.create(
            db_root,
            hash_reduce_left,
            max_number_branch,
            collection_name,
        )

        # Clear collection in cache
        if hash_reduce_left != 0:
            del DocCache.cache[collection_name]
            DocCache.create_structure(collection_name)
        return

clear_collection(collection_name) staticmethod

Synchronous method to remove all documents from a collection.

Parameters:

Name Type Description Default
collection_name str

Collection name.

 required

Returns:

Type Description
None

None.
Source code in src/scruby/mixins/collection.py 
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
@final
@staticmethod
def clear_collection(collection_name: str) -> None:
    """Synchronous method to remove all documents from a collection.

    Args:
        collection_name (str): Collection name.

    Returns:
        None.
    """
    db_root = ScrubyConfig.db_root
    hash_reduce_left = ScrubyConfig.HASH_REDUCE_LEFT
    max_number_branch = ScrubyConfig.MAX_NUMBER_BRANCH

    # Delete collection on file system
    target_directory = f"{db_root}/{collection_name}"
    rmtree(target_directory)

    # Create a directory for the collection and add metadata
    Metadata.create(
        db_root,
        hash_reduce_left,
        max_number_branch,
        collection_name,
    )

    # Clear collection in cache
    if hash_reduce_left != 0:
        del DocCache.cache[collection_name]
        DocCache.create_structure(collection_name)
    return

collection_list() staticmethod

Synchronous method for getting collection list.

Source code in src/scruby/mixins/collection.py 
32
33
34
35
36
37
@final
@staticmethod
def collection_list() -> list[str] | None:
    """Synchronous method for getting collection list."""
    collections = [item.__name__ for item in ScrubyModel.__subclasses__()]
    return collections or None

collection_name()

Asynchronous method for getting the collection name.

Returns:

Type Description
str

Collection name.
Source code in src/scruby/mixins/collection.py 
23
24
25
26
27
28
29
30
@final
def collection_name(self) -> str:
    """Asynchronous method for getting the collection name.

    Returns:
        Collection name.
    """
    return self._class_model.__name__