Source code for docarray.array.document

from typing import Optional, overload, TYPE_CHECKING, Dict, Union

from .base import BaseDocumentArray
from .mixins import AllMixins

if TYPE_CHECKING:
    from ..typing import DocumentArraySourceType
    from .memory import DocumentArrayInMemory
    from .sqlite import DocumentArraySqlite
    from .annlite import DocumentArrayAnnlite
    from .weaviate import DocumentArrayWeaviate
    from .elastic import DocumentArrayElastic
    from .storage.sqlite import SqliteConfig
    from .storage.annlite import AnnliteConfig
    from .storage.weaviate import WeaviateConfig
    from .storage.elastic import ElasticConfig


[docs]class DocumentArray(AllMixins, BaseDocumentArray): @overload def __new__( cls, _docs: Optional['DocumentArraySourceType'] = None, copy: bool = False ) -> 'DocumentArrayInMemory': """Create an in-memory DocumentArray object.""" ... @overload def __new__( cls, _docs: Optional['DocumentArraySourceType'] = None, storage: str = 'sqlite', config: Optional[Union['SqliteConfig', Dict]] = None, ) -> 'DocumentArraySqlite': """Create a SQLite-powered DocumentArray object.""" ... @overload def __new__( cls, _docs: Optional['DocumentArraySourceType'] = None, storage: str = 'weaviate', config: Optional[Union['WeaviateConfig', Dict]] = None, ) -> 'DocumentArrayWeaviate': """Create a Weaviate-powered DocumentArray object.""" ... @overload def __new__( cls, _docs: Optional['DocumentArraySourceType'] = None, storage: str = 'annlite', config: Optional[Union['AnnliteConfig', Dict]] = None, ) -> 'DocumentArrayAnnlite': """Create a AnnLite-powered DocumentArray object.""" ... @overload def __new__( cls, _docs: Optional['DocumentArraySourceType'] = None, storage: str = 'elasticsearch', config: Optional[Union['ElasticConfig', Dict]] = None, ) -> 'DocumentArrayElastic': """Create a Elastic-powered DocumentArray object.""" ... def __enter__(self): return self def __exit__(self, *args, **kwargs): """ Ensures that offset2ids are stored in the db after operations in the DocumentArray are performed. """ self._save_offset2ids() def __new__(cls, *args, storage: str = 'memory', **kwargs): if cls is DocumentArray: if storage == 'memory': from .memory import DocumentArrayInMemory instance = super().__new__(DocumentArrayInMemory) elif storage == 'sqlite': from .sqlite import DocumentArraySqlite instance = super().__new__(DocumentArraySqlite) elif storage == 'annlite': from .annlite import DocumentArrayAnnlite instance = super().__new__(DocumentArrayAnnlite) elif storage == 'weaviate': from .weaviate import DocumentArrayWeaviate instance = super().__new__(DocumentArrayWeaviate) elif storage == 'qdrant': from .qdrant import DocumentArrayQdrant instance = super().__new__(DocumentArrayQdrant) elif storage == 'elasticsearch': from .elastic import DocumentArrayElastic instance = super().__new__(DocumentArrayElastic) else: raise ValueError(f'storage=`{storage}` is not supported.') else: instance = super().__new__(cls) return instance