collections

numerous.collections

The Python SDK for numerous collections.

CollectionReference dataclass

Source code in numerous/collections/collection_reference.py

@dataclass
class CollectionReference:
    id: str
    key: str
    _client: Client

    @property
    def tags(self) -> dict[str, str]:
        """Get the tags for the collection."""
        tags = self._client.collection_tags(self.id)
        return {tag.key: tag.value for tag in tags}

    def collection(self, collection_key: str) -> CollectionReference:
        """
        Get or create a child collection of this collection by key.

        Args:
            collection_key: Key of the nested collection. A key uniquely identifies a
                collection within the parent collection. Keys are case sensitive, and
                can be used as human-readable identifiers for collections

        Returns:
            NumerousCollection: The child collection identified by the given key.

        """
        ref = self._client.collection_reference(
            collection_key=collection_key, parent_collection_id=self.id
        )

        return CollectionReference(ref.id, ref.key, self._client)

    def document(self, key: str) -> DocumentReference:
        """
        Get or create a document by key.

        Args:
            key: Key of the document. A key uniquely identifies a document within its
                collection. Keys are case sensitive.

        Returns:
            The document in the collection with the given key.

        Raises:
            ParentCollectionNotFoundError: If the collection does not exist, for example
                if it was deleted.

        """
        doc_ref = self._client.document_reference(self.id, key)
        if doc_ref is None:
            return self._document_reference(doc_id=None, doc_key=key)
        return self._document_reference_from_identifier(doc_ref)

    def _document_reference_from_identifier(
        self, doc_ref: CollectionDocumentIdentifier
    ) -> DocumentReference:
        return self._document_reference(doc_ref.id, doc_ref.key)

    def _document_reference(
        self,
        doc_id: str | None,
        doc_key: str,
    ) -> DocumentReference:
        return DocumentReference(
            id=doc_id,
            key=doc_key,
            collection_id=self.id,
            collection_key=self.key,
            _client=self._client,
        )

    def file(self, key: str) -> FileReference:
        """
        Get or create a file by key.

        Args:
            key: The key of the file.

        """
        file_identifier = self._client.file_reference(self.id, key)
        if file_identifier is None:
            msg = "Failed to retrieve or create the file."
            raise ValueError(msg)

        return self._file_reference_from_identifier(file_identifier)

    def save_file(self, file_key: str, file_data: str) -> None:
        """
        Save data to a file in the collection.

        If the file with the specified key already exists,
        it will be overwritten with the new data.

        Args:
            file_key: The key of the file to save or update.
            file_data: The data to be written to the file.

        Raises:
            ValueError: If the file cannot be created or saved.

        """
        file = self.file(file_key)
        file.save(file_data)

    def files(
        self, tag_key: str | None = None, tag_value: str | None = None
    ) -> Generator[FileReference, None, None]:
        """
        Retrieve files from the collection, filtered by a tag key and value.

        Args:
            tag_key: The key of the tag used to filter files.
            tag_value: The value of the tag used to filter files.

        Yields:
            File references from the collection.

        """
        end_cursor = ""
        tag = None
        if tag_key is not None and tag_value is not None:
            tag = Tag(key=tag_key, value=tag_value)
        has_next_page = True
        while has_next_page:
            result = self._client.collection_files(self.id, end_cursor, tag)
            if result is None:
                break
            file_identifiers, has_next_page, end_cursor = result
            for file_identifier in file_identifiers:
                if file_identifier is None:
                    continue
                yield self._file_reference_from_identifier(file_identifier)

    def documents(
        self, tag_key: str | None = None, tag_value: str | None = None
    ) -> Generator[DocumentReference, None, None]:
        """
        Retrieve documents from the collection, filtered by a tag key and value.

        Args:
            tag_key: If this and `tag_value` is specified, filter documents with this
                tag.
            tag_value: If this and `tag_key` is specified, filter documents with this
                tag.

        Yields:
            Documents from the collection.

        """
        end_cursor = ""
        tag_input = None
        if tag_key is not None and tag_value is not None:
            tag_input = Tag(key=tag_key, value=tag_value)
        has_next_page = True
        while has_next_page:
            result = self._client.collection_documents(self.id, end_cursor, tag_input)
            doc_refs, has_next_page, end_cursor = result
            if doc_refs is None:
                break
            for doc_ref in doc_refs:
                if doc_ref is None:
                    continue
                yield self._document_reference_from_identifier(doc_ref)

    def collections(
        self, tag_key: str | None = None, tag_value: str | None = None
    ) -> Generator[CollectionReference, None, None]:
        """
        Retrieve collections from the collection, filtered by a tag key and value.

        Args:
            tag_key: If this and `tag_value` is specified, filter collections with this
                tag.
            tag_value: If this and `tag_key` is specified, filter collections with this
                tag.

        Yields:
            Nested collections of this collection.

        """
        end_cursor = ""
        tag = None
        if tag_key is not None and tag_value is not None:
            tag = Tag(key=tag_key, value=tag_value)
        has_next_page = True
        while has_next_page:
            result = self._client.collection_collections(self.id, end_cursor, tag)
            if result is None:
                break
            refs, has_next_page, end_cursor = result
            if refs is None:
                break
            for ref in refs:
                yield CollectionReference(ref.id, ref.key, self._client)

    def _file_reference_from_identifier(
        self, identifier: CollectionFileIdentifier
    ) -> FileReference:
        return FileReference(id=identifier.id, key=identifier.key, _client=self._client)

    def tag(self, key: str, value: str) -> None:
        """
        Add a tag to the collection.

        Args:
            key: The tag key.
            value: The tag value.

        """
        self._client.collection_tag_add(self.id, Tag(key=key, value=value))

    def tag_delete(self, key: str) -> None:
        """
        Delete a tag from the collection.

        Args:
            key: The key of the tag to delete.

        """
        self._client.collection_tag_delete(self.id, key)

tags: dict[str, str] property

Get the tags for the collection.

collection(collection_key)

Get or create a child collection of this collection by key.

Parameters:

Name	Type	Description	Default
collection_key	str	Key of the nested collection. A key uniquely identifies a collection within the parent collection. Keys are case sensitive, and can be used as human-readable identifiers for collections	required

Returns:

Name	Type	Description
NumerousCollection	CollectionReference	The child collection identified by the given key.

Source code in numerous/collections/collection_reference.py

def collection(self, collection_key: str) -> CollectionReference:
    """
    Get or create a child collection of this collection by key.

    Args:
        collection_key: Key of the nested collection. A key uniquely identifies a
            collection within the parent collection. Keys are case sensitive, and
            can be used as human-readable identifiers for collections

    Returns:
        NumerousCollection: The child collection identified by the given key.

    """
    ref = self._client.collection_reference(
        collection_key=collection_key, parent_collection_id=self.id
    )

    return CollectionReference(ref.id, ref.key, self._client)

collections(tag_key=None, tag_value=None)

Retrieve collections from the collection, filtered by a tag key and value.

Parameters:

Name	Type	Description	Default
tag_key	str | None	If this and tag_value is specified, filter collections with this tag.	None
tag_value	str | None	If this and tag_key is specified, filter collections with this tag.	None

Yields:

Type	Description
CollectionReference	Nested collections of this collection.

Source code in numerous/collections/collection_reference.py

def collections(
    self, tag_key: str | None = None, tag_value: str | None = None
) -> Generator[CollectionReference, None, None]:
    """
    Retrieve collections from the collection, filtered by a tag key and value.

    Args:
        tag_key: If this and `tag_value` is specified, filter collections with this
            tag.
        tag_value: If this and `tag_key` is specified, filter collections with this
            tag.

    Yields:
        Nested collections of this collection.

    """
    end_cursor = ""
    tag = None
    if tag_key is not None and tag_value is not None:
        tag = Tag(key=tag_key, value=tag_value)
    has_next_page = True
    while has_next_page:
        result = self._client.collection_collections(self.id, end_cursor, tag)
        if result is None:
            break
        refs, has_next_page, end_cursor = result
        if refs is None:
            break
        for ref in refs:
            yield CollectionReference(ref.id, ref.key, self._client)

document(key)

Get or create a document by key.

Parameters:

Name	Type	Description	Default
key	str	Key of the document. A key uniquely identifies a document within its collection. Keys are case sensitive.	required

Returns:

Type	Description
DocumentReference	The document in the collection with the given key.

Raises:

Type	Description
ParentCollectionNotFoundError	If the collection does not exist, for example if it was deleted.

Source code in numerous/collections/collection_reference.py

def document(self, key: str) -> DocumentReference:
    """
    Get or create a document by key.

    Args:
        key: Key of the document. A key uniquely identifies a document within its
            collection. Keys are case sensitive.

    Returns:
        The document in the collection with the given key.

    Raises:
        ParentCollectionNotFoundError: If the collection does not exist, for example
            if it was deleted.

    """
    doc_ref = self._client.document_reference(self.id, key)
    if doc_ref is None:
        return self._document_reference(doc_id=None, doc_key=key)
    return self._document_reference_from_identifier(doc_ref)

documents(tag_key=None, tag_value=None)

Retrieve documents from the collection, filtered by a tag key and value.

Parameters:

Name	Type	Description	Default
tag_key	str | None	If this and tag_value is specified, filter documents with this tag.	None
tag_value	str | None	If this and tag_key is specified, filter documents with this tag.	None

Yields:

Type	Description
DocumentReference	Documents from the collection.

Source code in numerous/collections/collection_reference.py

def documents(
    self, tag_key: str | None = None, tag_value: str | None = None
) -> Generator[DocumentReference, None, None]:
    """
    Retrieve documents from the collection, filtered by a tag key and value.

    Args:
        tag_key: If this and `tag_value` is specified, filter documents with this
            tag.
        tag_value: If this and `tag_key` is specified, filter documents with this
            tag.

    Yields:
        Documents from the collection.

    """
    end_cursor = ""
    tag_input = None
    if tag_key is not None and tag_value is not None:
        tag_input = Tag(key=tag_key, value=tag_value)
    has_next_page = True
    while has_next_page:
        result = self._client.collection_documents(self.id, end_cursor, tag_input)
        doc_refs, has_next_page, end_cursor = result
        if doc_refs is None:
            break
        for doc_ref in doc_refs:
            if doc_ref is None:
                continue
            yield self._document_reference_from_identifier(doc_ref)

file(key)

Get or create a file by key.

Parameters:

Name	Type	Description	Default
key	str	The key of the file.	required

Source code in numerous/collections/collection_reference.py

def file(self, key: str) -> FileReference:
    """
    Get or create a file by key.

    Args:
        key: The key of the file.

    """
    file_identifier = self._client.file_reference(self.id, key)
    if file_identifier is None:
        msg = "Failed to retrieve or create the file."
        raise ValueError(msg)

    return self._file_reference_from_identifier(file_identifier)

files(tag_key=None, tag_value=None)

Retrieve files from the collection, filtered by a tag key and value.

Parameters:

Name	Type	Description	Default
tag_key	str | None	The key of the tag used to filter files.	None
tag_value	str | None	The value of the tag used to filter files.	None

Yields:

Type	Description
FileReference	File references from the collection.

Source code in numerous/collections/collection_reference.py

def files(
    self, tag_key: str | None = None, tag_value: str | None = None
) -> Generator[FileReference, None, None]:
    """
    Retrieve files from the collection, filtered by a tag key and value.

    Args:
        tag_key: The key of the tag used to filter files.
        tag_value: The value of the tag used to filter files.

    Yields:
        File references from the collection.

    """
    end_cursor = ""
    tag = None
    if tag_key is not None and tag_value is not None:
        tag = Tag(key=tag_key, value=tag_value)
    has_next_page = True
    while has_next_page:
        result = self._client.collection_files(self.id, end_cursor, tag)
        if result is None:
            break
        file_identifiers, has_next_page, end_cursor = result
        for file_identifier in file_identifiers:
            if file_identifier is None:
                continue
            yield self._file_reference_from_identifier(file_identifier)

save_file(file_key, file_data)

Save data to a file in the collection.

If the file with the specified key already exists, it will be overwritten with the new data.

Parameters:

Name	Type	Description	Default
file_key	str	The key of the file to save or update.	required
file_data	str	The data to be written to the file.	required

Raises:

Type	Description
ValueError	If the file cannot be created or saved.

Source code in numerous/collections/collection_reference.py

def save_file(self, file_key: str, file_data: str) -> None:
    """
    Save data to a file in the collection.

    If the file with the specified key already exists,
    it will be overwritten with the new data.

    Args:
        file_key: The key of the file to save or update.
        file_data: The data to be written to the file.

    Raises:
        ValueError: If the file cannot be created or saved.

    """
    file = self.file(file_key)
    file.save(file_data)

tag(key, value)

Add a tag to the collection.

Parameters:

Name	Type	Description	Default
key	str	The tag key.	required
value	str	The tag value.	required

Source code in numerous/collections/collection_reference.py

def tag(self, key: str, value: str) -> None:
    """
    Add a tag to the collection.

    Args:
        key: The tag key.
        value: The tag value.

    """
    self._client.collection_tag_add(self.id, Tag(key=key, value=value))

tag_delete(key)

Delete a tag from the collection.

Parameters:

Name	Type	Description	Default
key	str	The key of the tag to delete.	required

Source code in numerous/collections/collection_reference.py

def tag_delete(self, key: str) -> None:
    """
    Delete a tag from the collection.

    Args:
        key: The key of the tag to delete.

    """
    self._client.collection_tag_delete(self.id, key)

DocumentReference dataclass

Source code in numerous/collections/document_reference.py

@dataclass
class DocumentReference:
    id: str | None
    key: str
    collection_id: str
    collection_key: str
    _client: Client

    @property
    def exists(self) -> bool:
        """True if the document exists, False otherwise."""
        self._set_id_if_reference_exists()
        if self.id is None:
            return False

        return self._client.document_exists(self.id)

    def _set_id_if_reference_exists(self) -> None:
        if self.id is not None:
            return

        ref = self._client.document_reference(self.collection_id, self.key)
        if ref is not None:
            self.id = ref.id

    @property
    def tags(self) -> dict[str, str]:
        """Get the tags for the document."""
        self._set_id_if_reference_exists()
        if self.id is None:
            return {}

        return self._client.document_tags(self.id) or {}

    def set(self, data: dict[str, Any]) -> None:
        """
        Set the data for the document.

        Args:
            data: The data to set for the document.

        Raises:
            TypeError: If the data is not JSON serializable.

        """
        base64_data = dict_to_base64(data)
        self._client.document_set(self.collection_id, self.key, base64_data)

    def get(self) -> dict[str, Any] | None:
        """
        Get the data of the document.

        Returns:
            The data of the document if it is set.

        """
        self._set_id_if_reference_exists()
        if self.id is None:
            return None

        data = self._client.document_get(self.id)
        if data is None:
            return None

        return base64_to_dict(data)

    def delete(self) -> None:
        """Delete the referenced document."""
        self._set_id_if_reference_exists()
        if self.id is None:
            raise DocumentDoesNotExistError(
                collection_id=self.collection_id, key=self.key
            )

        self._client.document_delete(self.id)

    def tag(self, key: str, value: str) -> None:
        """
        Add a tag to the document.

        Args:
            key: The tag key.
            value: The tag value.

        """
        self._set_id_if_reference_exists()
        if self.id is None:
            raise DocumentDoesNotExistError(
                collection_id=self.collection_id, key=self.key
            )

        self._client.document_tag_add(self.id, Tag(key=key, value=value))

    def tag_delete(self, tag_key: str) -> None:
        """
        Delete a tag from the document.

        Args:
            tag_key: The key of the tag to delete.

        """
        self._set_id_if_reference_exists()
        if self.id is None:
            raise DocumentDoesNotExistError(
                collection_id=self.collection_id, key=self.key
            )

        self._client.document_tag_delete(self.id, tag_key)

exists: bool property

True if the document exists, False otherwise.

tags: dict[str, str] property

Get the tags for the document.

delete()

Delete the referenced document.

Source code in numerous/collections/document_reference.py

def delete(self) -> None:
    """Delete the referenced document."""
    self._set_id_if_reference_exists()
    if self.id is None:
        raise DocumentDoesNotExistError(
            collection_id=self.collection_id, key=self.key
        )

    self._client.document_delete(self.id)

get()

Get the data of the document.

Returns:

Type	Description
dict[str, Any] | None	The data of the document if it is set.

Source code in numerous/collections/document_reference.py

def get(self) -> dict[str, Any] | None:
    """
    Get the data of the document.

    Returns:
        The data of the document if it is set.

    """
    self._set_id_if_reference_exists()
    if self.id is None:
        return None

    data = self._client.document_get(self.id)
    if data is None:
        return None

    return base64_to_dict(data)

set(data)

Set the data for the document.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	The data to set for the document.	required

Raises:

Type	Description
TypeError	If the data is not JSON serializable.

Source code in numerous/collections/document_reference.py

def set(self, data: dict[str, Any]) -> None:
    """
    Set the data for the document.

    Args:
        data: The data to set for the document.

    Raises:
        TypeError: If the data is not JSON serializable.

    """
    base64_data = dict_to_base64(data)
    self._client.document_set(self.collection_id, self.key, base64_data)

tag(key, value)

Add a tag to the document.

Parameters:

Name	Type	Description	Default
key	str	The tag key.	required
value	str	The tag value.	required

Source code in numerous/collections/document_reference.py

def tag(self, key: str, value: str) -> None:
    """
    Add a tag to the document.

    Args:
        key: The tag key.
        value: The tag value.

    """
    self._set_id_if_reference_exists()
    if self.id is None:
        raise DocumentDoesNotExistError(
            collection_id=self.collection_id, key=self.key
        )

    self._client.document_tag_add(self.id, Tag(key=key, value=value))

tag_delete(tag_key)

Delete a tag from the document.

Parameters:

Name	Type	Description	Default
tag_key	str	The key of the tag to delete.	required

Source code in numerous/collections/document_reference.py

def tag_delete(self, tag_key: str) -> None:
    """
    Delete a tag from the document.

    Args:
        tag_key: The key of the tag to delete.

    """
    self._set_id_if_reference_exists()
    if self.id is None:
        raise DocumentDoesNotExistError(
            collection_id=self.collection_id, key=self.key
        )

    self._client.document_tag_delete(self.id, tag_key)

FileReference dataclass

Represents a file in a collection.

Source code in numerous/collections/file_reference.py

@dataclass
class FileReference:
    """Represents a file in a collection."""

    id: str
    key: str
    _client: Client

    @property
    def exists(self) -> bool:
        """
        Indicate whether the file exists.

        Returns:
            True if the file exists; False otherwise.

        """
        return self._client.file_exists(self.id)

    @property
    def tags(self) -> dict[str, str]:
        """
        Return the tags associated with the file.

        Returns:
            A dictionary of tag key-value pairs.

        """
        tags = self._client.file_tags(self.id)
        if tags is None:
            raise ValueError(NO_FILE_ERROR_MSG)
        return tags

    def read_text(self) -> str:
        """
        Read the file's content as text.

        Returns:
            The text content of the file.

        """
        return self._client.file_read_text(self.id)

    def read_bytes(self) -> bytes:
        """
        Read the file's content as bytes.

        Returns:
            The byte content of the file.

        """
        return self._client.file_read_bytes(self.id)

    def open(self) -> BinaryIO:
        """
        Open the file for reading in binary mode.

        Returns:
            A binary file-like object for reading the file.

        """
        return self._client.file_open(self.id)

    def save(self, data: bytes | str) -> None:
        """
        Upload and saves data to the file on the server.

        Args:
            data: The content to save to the file, either as bytes or string.

        """
        self._client.file_save(self.id, data)

    def save_file(self, data: TextIOWrapper) -> None:
        """
        Upload and saves a text file to the server.

        Args:
            data: A file-like object containing the text content to upload.

        """
        self._client.file_save(self.id, data.read())

    def delete(self) -> None:
        """Delete the file from the server."""
        self._client.file_delete(self.id)

    def tag(self, key: str, value: str) -> None:
        """
        Add a tag to the file.

        Args:
            key: The tag key.
            value: The tag value.

        """
        self._client.file_tag_add(self.id, Tag(key=key, value=value))

    def tag_delete(self, tag_key: str) -> None:
        """
        Delete a tag from the file.

        Args:
            tag_key: The key of the tag to delete.

        Raises:
            ValueError: If the file does not exist.

        """
        self._client.file_delete_tag(self.id, tag_key)

exists: bool property

Indicate whether the file exists.

Returns:

Type	Description
bool	True if the file exists; False otherwise.

tags: dict[str, str] property

Return the tags associated with the file.

Returns:

Type	Description
dict[str, str]	A dictionary of tag key-value pairs.

delete()

Delete the file from the server.

Source code in numerous/collections/file_reference.py

def delete(self) -> None:
    """Delete the file from the server."""
    self._client.file_delete(self.id)

open()

Open the file for reading in binary mode.

Returns:

Type	Description
BinaryIO	A binary file-like object for reading the file.

Source code in numerous/collections/file_reference.py

def open(self) -> BinaryIO:
    """
    Open the file for reading in binary mode.

    Returns:
        A binary file-like object for reading the file.

    """
    return self._client.file_open(self.id)

read_bytes()

Read the file's content as bytes.

Returns:

Type	Description
bytes	The byte content of the file.

Source code in numerous/collections/file_reference.py

def read_bytes(self) -> bytes:
    """
    Read the file's content as bytes.

    Returns:
        The byte content of the file.

    """
    return self._client.file_read_bytes(self.id)

read_text()

Read the file's content as text.

Returns:

Type	Description
str	The text content of the file.

Source code in numerous/collections/file_reference.py

def read_text(self) -> str:
    """
    Read the file's content as text.

    Returns:
        The text content of the file.

    """
    return self._client.file_read_text(self.id)

save(data)

Upload and saves data to the file on the server.

Parameters:

Name	Type	Description	Default
data	bytes | str	The content to save to the file, either as bytes or string.	required

Source code in numerous/collections/file_reference.py

def save(self, data: bytes | str) -> None:
    """
    Upload and saves data to the file on the server.

    Args:
        data: The content to save to the file, either as bytes or string.

    """
    self._client.file_save(self.id, data)

save_file(data)

Upload and saves a text file to the server.

Parameters:

Name	Type	Description	Default
data	TextIOWrapper	A file-like object containing the text content to upload.	required

Source code in numerous/collections/file_reference.py

def save_file(self, data: TextIOWrapper) -> None:
    """
    Upload and saves a text file to the server.

    Args:
        data: A file-like object containing the text content to upload.

    """
    self._client.file_save(self.id, data.read())

tag(key, value)

Add a tag to the file.

Parameters:

Name	Type	Description	Default
key	str	The tag key.	required
value	str	The tag value.	required

Source code in numerous/collections/file_reference.py

def tag(self, key: str, value: str) -> None:
    """
    Add a tag to the file.

    Args:
        key: The tag key.
        value: The tag value.

    """
    self._client.file_tag_add(self.id, Tag(key=key, value=value))

tag_delete(tag_key)

Delete a tag from the file.

Parameters:

Name	Type	Description	Default
tag_key	str	The key of the tag to delete.	required

Raises:

Type	Description
ValueError	If the file does not exist.

Source code in numerous/collections/file_reference.py

def tag_delete(self, tag_key: str) -> None:
    """
    Delete a tag from the file.

    Args:
        tag_key: The key of the tag to delete.

    Raises:
        ValueError: If the file does not exist.

    """
    self._client.file_delete_tag(self.id, tag_key)

bulk_download(collection_ref, local_base_path=Path('collections'), document_suffix=DEFAULT_DOCUMENT_SUFFIX)

Download an entire collection hierarchy recursively to the local filesystem.

Downloads all files and documents from the given collection and all its nested sub-collections. Files are saved as binary, and documents are saved as JSON files with the specified suffix.

The local directory structure will mirror the collection hierarchy, using collection keys as folder names. Existing local files will be overwritten.

Parameters:

Name	Type	Description	Default
collection_ref	CollectionReference	The CollectionReference of the root collection to download.	required
local_base_path	Path	Local base directory where collection data will be saved. Defaults to "collections" in the current working directory. The actual download will be inside a subdirectory named after the root collection's key, e.g., local_base_path/collection_key/...	Path('collections')
document_suffix	str	Suffix to append to document filenames when saving locally. Defaults to ".collection-doc.json".	DEFAULT_DOCUMENT_SUFFIX

Source code in numerous/collections/bulk/main.py

def bulk_download(  # noqa: C901
    collection_ref: CollectionReference,
    local_base_path: Path = Path("collections"),
    document_suffix: str = DEFAULT_DOCUMENT_SUFFIX,
) -> None:
    """
    Download an entire collection hierarchy recursively to the local filesystem.

    Downloads all files and documents from the given collection and all its
    nested sub-collections. Files are saved as binary, and documents are
    saved as JSON files with the specified suffix.

    The local directory structure will mirror the collection hierarchy, using
    collection keys as folder names. Existing local files will be overwritten.

    Args:
        collection_ref: The `CollectionReference` of the root collection to download.
        local_base_path: Local base directory where collection data will be saved.
                         Defaults to "collections" in the current working directory.
                         The actual download will be inside a subdirectory named
                         after the root collection's key,
                         e.g., `local_base_path/collection_key/...`
        document_suffix: Suffix to append to document filenames when saving locally.
                         Defaults to ".collection-doc.json".

    """
    root_collection_download_path = local_base_path / collection_ref.key

    _ensure_local_directory(root_collection_download_path)
    logger.info(
        "Starting bulk download for collection '%s' into '%s'",
        collection_ref.key,
        root_collection_download_path,
    )

    def _download_recursive(
        current_col: CollectionReference, target_local_dir: Path
    ) -> None:
        """Download a given collection recursively to the target local directory."""
        logger.info(
            "Processing collection '%s' into '%s'",
            current_col.key,
            target_local_dir,
        )
        _ensure_local_directory(target_local_dir)

        try:
            for file_item in current_col.files():
                try:
                    local_file_path = target_local_dir / file_item.key
                    logger.debug(
                        "Downloading file: '%s/%s' to '%s'",
                        current_col.key,
                        file_item.key,
                        local_file_path,
                    )
                    file_content = file_item.read_bytes()
                    with local_file_path.open("wb") as f_out:
                        f_out.write(file_content)
                except Exception:  # noqa: PERF203
                    logger.exception(
                        "Error downloading file '%s' from collection '%s'",
                        file_item.key,
                        current_col.key,
                    )
        except Exception:
            logger.exception("Error listing files for collection '%s'", current_col.key)

        try:
            for doc_item in current_col.documents():
                try:
                    local_doc_path = (
                        target_local_dir / f"{doc_item.key}{document_suffix}"
                    )
                    logger.debug(
                        "Downloading document: '%s/%s' to '%s'",
                        current_col.key,
                        doc_item.key,
                        local_doc_path,
                    )
                    document_data = doc_item.get()
                    if document_data is not None:
                        _save_document_as_json(document_data, local_doc_path)
                    else:
                        logger.warning(
                            "Document '%s' in collection '%s' is empty, skipping save.",
                            doc_item.key,
                            current_col.key,
                        )
                except Exception:  # noqa: PERF203
                    logger.exception(
                        "Error downloading document '%s' from collection '%s'",
                        doc_item.key,
                        current_col.key,
                    )
        except Exception:
            logger.exception(
                "Error listing documents for collection '%s'", current_col.key
            )

        try:
            for sub_col in current_col.collections():
                sub_collection_local_path = target_local_dir / sub_col.key
                _download_recursive(sub_col, sub_collection_local_path)
        except Exception:
            logger.exception(
                "Error listing sub-collections for collection '%s'",
                current_col.key,
            )

    try:
        _download_recursive(collection_ref, root_collection_download_path)
        logger.info(
            "Bulk download for collection '%s' completed successfully.",
            collection_ref.key,
        )
    except Exception as e:
        logger.critical(
            "Bulk download for collection '%s' failed critically: %s",
            collection_ref.key,
            e,
        )
        raise

bulk_upload(collection_ref, local_base_path=Path('collections'), ignore_file_name=DEFAULT_IGNORE_FILE, document_suffix=DEFAULT_DOCUMENT_SUFFIX)

Upload a local directory structure recursively to a Numerous collection.

Scans the specified local directory (expected to be local_base_path/collection_ref.key) and uploads files and documents. It creates nested collections as needed.

Files with names matching patterns in an ignore file (default: .numerous-ignore) will be skipped. Files with the specified document suffix will be uploaded as documents. Existing remote files and documents will be overwritten.

Parameters:

Name	Type	Description	Default
collection_ref	CollectionReference	The CollectionReference of the root collection to upload into.	required
local_base_path	Path	Local base directory from which to upload. The function expects content to be uploaded to be inside a subdirectory named after the root collection's key, e.g., local_base_path/collection_key/...	Path('collections')
ignore_file_name	str	Name of the ignore file (e.g., ".numerous-ignore") located in the root of the directory being uploaded (i.e., inside local_base_path/collection_key/).	DEFAULT_IGNORE_FILE
document_suffix	str	Suffix that identifies files to be uploaded as documents. These files will have the suffix removed from their key. Defaults to ".collection-doc.json".	DEFAULT_DOCUMENT_SUFFIX

Source code in numerous/collections/bulk/main.py

def bulk_upload(  # noqa: C901, PLR0915
    collection_ref: CollectionReference,
    local_base_path: Path = Path("collections"),
    ignore_file_name: str = DEFAULT_IGNORE_FILE,
    document_suffix: str = DEFAULT_DOCUMENT_SUFFIX,
) -> None:
    """
    Upload a local directory structure recursively to a Numerous collection.

    Scans the specified local directory (expected to be
    `local_base_path/collection_ref.key`) and uploads files and documents.
    It creates nested collections as needed.

    Files with names matching patterns in an ignore file
    (default: .numerous-ignore) will be skipped. Files with the specified
    document suffix will be uploaded as documents. Existing remote files and
    documents will be overwritten.

    Args:
        collection_ref: The `CollectionReference` of the root collection to upload into.
        local_base_path: Local base directory from which to upload. The function
                         expects content to be uploaded to be inside a subdirectory
                         named after the root collection's key,
                         e.g., `local_base_path/collection_key/...`
        ignore_file_name: Name of the ignore file (e.g., ".numerous-ignore")
                          located in the root of the directory being uploaded
                          (i.e., inside `local_base_path/collection_key/`).
        document_suffix: Suffix that identifies files to be uploaded as documents.
                         These files will have the suffix removed from their key.
                         Defaults to ".collection-doc.json".

    """
    source_directory_root = local_base_path / collection_ref.key
    if not source_directory_root.is_dir():
        msg = (
            f"Source directory '{source_directory_root}' not found or not a directory."
        )
        logger.error(
            "Source directory '%s' does not exist or is not a directory. "
            "Aborting upload.",
            source_directory_root,
        )
        raise FileNotFoundError(msg)

    ignore_file_path = source_directory_root / ignore_file_name
    ignore_patterns = _load_ignore_patterns(ignore_file_path)
    if ignore_patterns:
        logger.info(
            "Loaded %d ignore patterns from '%s'.",
            len(ignore_patterns),
            ignore_file_path,
        )
    else:
        logger.info(
            "No ignore patterns loaded (file '%s' not found or empty).",
            ignore_file_path,
        )

    logger.info(
        "Starting bulk upload from '%s' to collection '%s'.",
        source_directory_root,
        collection_ref.key,
    )

    def _upload_recursive(  # noqa: C901
        current_local_dir: Path,
        target_col: CollectionReference,
        effective_ignore_file_path: Path,
    ) -> None:
        """Upload contents of current_local_dir recursively to target_col."""
        logger.info(
            "Processing directory '%s' for upload to collection '%s'.",
            current_local_dir,
            target_col.key,
        )

        for item_path in current_local_dir.iterdir():
            if item_path == effective_ignore_file_path:
                logger.debug("Skipping the ignore file itself: %s", item_path)
                continue

            if _should_ignore_path(item_path, ignore_patterns, source_directory_root):
                logger.info("Ignoring '%s' due to ignore rules.", item_path)
                continue

            item_key = item_path.name
            if not _is_valid_collection_key(item_key):
                _warn_invalid_key(
                    item_path, "Name contains invalid characters or is a reserved name."
                )
                continue

            if item_path.is_dir():
                logger.debug(
                    "Found directory: '%s'. Creating/getting sub-collection '%s'.",
                    item_path,
                    item_key,
                )
                try:
                    sub_collection = target_col.collection(item_key)
                    _upload_recursive(
                        item_path, sub_collection, effective_ignore_file_path
                    )
                except Exception:
                    logger.exception(
                        "Error creating or accessing sub-collection '%s' from '%s'",
                        item_key,
                        target_col.key,
                    )

            elif item_path.is_file():
                logger.debug(
                    "Found file: '%s'. Preparing for upload as '%s'.",
                    item_path,
                    item_key,
                )
                try:
                    if _is_document_file(item_path, document_suffix):
                        logger.debug("Treating '%s' as a document.", item_key)
                        with item_path.open("r", encoding="utf-8") as f_in:
                            try:
                                doc_data = json.load(f_in)
                            except json.JSONDecodeError:
                                logger.exception(
                                    "Error decoding JSON from '%s'. Skipping.",
                                    item_path,
                                )
                                continue

                        doc_key = _get_document_key_from_filename(
                            item_key, document_suffix
                        )
                        doc_ref = target_col.document(doc_key)
                        doc_ref.set(doc_data)
                        logger.info(
                            "Uploaded document '%s' (key: '%s') to collection '%s'.",
                            item_key,
                            doc_key,
                            target_col.key,
                        )
                    else:
                        file_ref = target_col.file(item_key)
                        with item_path.open("rb") as f_in:
                            file_content = f_in.read()
                        file_ref.save(file_content)
                        logger.info(
                            "Uploaded file '%s' to collection '%s'.",
                            item_key,
                            target_col.key,
                        )
                except Exception:
                    logger.exception(
                        "Error uploading file/document '%s' to collection '%s'",
                        item_key,
                        target_col.key,
                    )
            else:
                logger.warning(
                    "Skipping '%s' as it is not a file or directory.", item_path
                )

    try:
        _upload_recursive(source_directory_root, collection_ref, ignore_file_path)
        logger.info(
            "Bulk upload from '%s' to collection '%s' completed.",
            source_directory_root,
            collection_ref.key,
        )
    except Exception as e:
        logger.critical(
            "Bulk upload from '%s' failed critically: %s",
            source_directory_root,
            e,
        )
        raise