This repository was archived by the owner on Sep 9, 2026. It is now read-only.
Repository navigation
Expand file tree
/
Copy pathdocument.py
More file actions
77 lines (65 loc) · 2.56 KB
/
Copy pathdocument.py
File metadata and controls
77 lines (65 loc) · 2.56 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
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
71
72
73
74
75
76
77
from typing import Optional, overload, TYPE_CHECKING, Dict, Union
from .base import BaseDocumentArray
from .mixins import AllMixins
if TYPE_CHECKING:
from ..types import DocumentArraySourceType
from .memory import DocumentArrayInMemory
from .sqlite import DocumentArraySqlite
from .pqlite import DocumentArrayPqlite
from .weaviate import DocumentArrayWeaviate
from .storage.sqlite import SqliteConfig
from .storage.pqlite import PqliteConfig
from .storage.weaviate import WeaviateConfig
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 = 'pqlite',
config: Optional[Union['PqliteConfig', Dict]] = None,
) -> 'DocumentArrayPqlite':
"""Create a PQLite-powered DocumentArray object."""
...
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 == 'pqlite':
from .pqlite import DocumentArrayPqlite
instance = super().__new__(DocumentArrayPqlite)
elif storage == 'weaviate':
from .weaviate import DocumentArrayWeaviate
instance = super().__new__(DocumentArrayWeaviate)
else:
raise ValueError(f'storage=`{storage}` is not supported.')
else:
instance = super().__new__(cls)
return instance