-
Notifications
You must be signed in to change notification settings - Fork 241
Expand file tree
/
Copy pathabstract_doc_store.py
More file actions
111 lines (96 loc) · 3.54 KB
/
abstract_doc_store.py
File metadata and controls
111 lines (96 loc) · 3.54 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
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
# Licensed to the LF AI & Data foundation under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from abc import ABC, abstractmethod
from typing import Dict, Iterator, List, Type
from typing_extensions import TYPE_CHECKING
if TYPE_CHECKING:
from docarray import BaseDoc, DocList
class AbstractDocStore(ABC):
@staticmethod
@abstractmethod
def list(namespace: str, show_table: bool) -> List[str]:
"""List all DocLists in the specified backend at the namespace.
:param namespace: The namespace to list
:param show_table: If true, a table is printed to the console
:return: A list of DocList names
"""
...
@staticmethod
@abstractmethod
def delete(name: str, missing_ok: bool) -> bool:
"""Delete the DocList object at the specified name
:param name: The name of the DocList to delete
:param missing_ok: If true, no error will be raised if the DocList does not exist.
:return: True if the DocList was deleted, False if it did not exist.
"""
...
@staticmethod
@abstractmethod
def push(
docs: 'DocList',
name: str,
show_progress: bool,
) -> Dict:
"""Push this DocList to the specified name.
:param docs: The DocList to push
:param name: The name to push to
:param show_progress: If true, a progress bar will be displayed.
"""
...
@staticmethod
@abstractmethod
def push_stream(
docs: Iterator['BaseDoc'],
url: str,
show_progress: bool = False,
) -> Dict:
"""Push a stream of documents to the specified name.
:param docs: a stream of documents
:param url: The name to push to
:param show_progress: If true, a progress bar will be displayed.
"""
...
@staticmethod
@abstractmethod
def pull(
docs_cls: Type['DocList'],
name: str,
show_progress: bool,
local_cache: bool,
) -> 'DocList':
"""Pull a DocList from the specified name.
:param docs_cls: The DocList class to instantiate
:param name: The name to pull from
:param show_progress: If true, a progress bar will be displayed.
:param local_cache: If true, the DocList will be cached locally
:return: A DocList
"""
...
@staticmethod
@abstractmethod
def pull_stream(
docs_cls: Type['DocList'],
name: str,
show_progress: bool,
local_cache: bool,
) -> Iterator['BaseDoc']:
"""Pull a stream of documents from the specified name.
:param docs_cls: The DocList class to instantiate
:param name: The name to pull from
:param show_progress: If true, a progress bar will be displayed.
:param local_cache: If true, the DocList will be cached locally
:return: An iterator of documents"""
...