GeezerVisonThe World Wide Web, restored to 800 pixels of glory. |
| Home|About|Gaming cabinet|Archives|How it works|Starter sites |
GeezerVison is showing github.com — docarray/docarray/array/doc_list/pushpull.py at main · docarray/docarray
This repository was archived by the owner on Sep 9, 2026. It is now read-only.
Expand file tree/ Copy pathpushpull.pyMore file actions 166 lines (138 loc) · 5.84 KB / Copy pathpushpull.pyFile metadata and controls166 lines (138 loc) · 5.84 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 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 import logging from abc import abstractmethod from typing import ( TYPE_CHECKING, Dict, Iterable, Iterator, Tuple, Type, TypeVar, cast, ) from typing_extensions import Literal from typing_inspect import get_args PUSH_PULL_PROTOCOL = Literal['s3', 'file'] SUPPORTED_PUSH_PULL_PROTOCOLS = get_args(PUSH_PULL_PROTOCOL) if TYPE_CHECKING: # pragma: no cover from docarray import BaseDoc, DocList from docarray.store.abstract_doc_store import AbstractDocStore SelfPushPullMixin = TypeVar('SelfPushPullMixin', bound='PushPullMixin') class PushPullMixin(Iterable['BaseDoc']): """Mixin class for push/pull functionality.""" __backends__: Dict[str, Type['AbstractDocStore']] = {} doc_type: Type['BaseDoc'] @abstractmethod def __len__(self) -> int: ... @staticmethod def resolve_url(url: str) -> Tuple[PUSH_PULL_PROTOCOL, str]: """Resolve the URL to the correct protocol and name. :param url: url to resolve """ protocol, name = url.split('://', 2) if protocol in SUPPORTED_PUSH_PULL_PROTOCOLS: protocol = cast(PUSH_PULL_PROTOCOL, protocol) return protocol, name else: raise ValueError(f'Unsupported protocol {protocol}') @classmethod def get_pushpull_backend( cls: Type[SelfPushPullMixin], protocol: PUSH_PULL_PROTOCOL ) -> Type['AbstractDocStore']: """ Get the backend for the given protocol. :param protocol: the protocol to use, e.g. 'file', 's3' :return: the backend class """ if protocol in cls.__backends__: return cls.__backends__[protocol] if protocol == 'file': from docarray.store.file import FileDocStore cls.__backends__[protocol] = FileDocStore logging.debug('Loaded Local Filesystem backend') elif protocol == 's3': from docarray.store.s3 import S3DocStore cls.__backends__[protocol] = S3DocStore logging.debug('Loaded S3 backend') else: raise NotImplementedError(f'protocol {protocol} not supported') return cls.__backends__[protocol] def push( self, url: str, show_progress: bool = False, **kwargs, ) -> Dict: """Push this `DocList` object to the specified url. :param url: url specifying the protocol and save name of the `DocList`. Should be of the form ``protocol://namespace/name``. e.g. ``s3://bucket/path/to/namespace/name``, ``file:///path/to/folder/name`` :param show_progress: If true, a progress bar will be displayed. """ logging.info(f'Pushing {len(self)} docs to {url}') protocol, name = self.__class__.resolve_url(url) return self.__class__.get_pushpull_backend(protocol).push( self, name, show_progress # type: ignore ) @classmethod def push_stream( cls: Type[SelfPushPullMixin], docs: Iterator['BaseDoc'], url: str, show_progress: bool = False, ) -> Dict: """Push a stream of documents to the specified url. :param docs: a stream of documents :param url: url specifying the protocol and save name of the `DocList`. Should be of the form ``protocol://namespace/name``. e.g. ``s3://bucket/path/to/namespace/name``, ``file:///path/to/folder/name`` :param show_progress: If true, a progress bar will be displayed. """ logging.info(f'Pushing stream to {url}') protocol, name = cls.resolve_url(url) return cls.get_pushpull_backend(protocol).push_stream(docs, name, show_progress) @classmethod def pull( cls: Type[SelfPushPullMixin], url: str, show_progress: bool = False, local_cache: bool = True, ) -> 'DocList': """Pull a `DocList` from the specified url. :param url: url specifying the protocol and save name of the `DocList`. Should be of the form ``protocol://namespace/name``. e.g. ``s3://bucket/path/to/namespace/name``, ``file:///path/to/folder/name`` :param show_progress: if true, display a progress bar. :param local_cache: store the downloaded `DocList` to local folder :return: a `DocList` object """ from docarray.base_doc import AnyDoc if cls.doc_type == AnyDoc: raise TypeError( 'There is no document schema defined. ' 'Please specify the `DocList`\'s Document type using `DocList[MyDoc]`.' ) logging.info(f'Pulling {url}') protocol, name = cls.resolve_url(url) return cls.get_pushpull_backend(protocol).pull( cls, name, show_progress, local_cache # type: ignore ) @classmethod def pull_stream( cls: Type[SelfPushPullMixin], url: str, show_progress: bool = False, local_cache: bool = False, ) -> Iterator['BaseDoc']: """Pull a stream of Documents from the specified url. :param url: url specifying the protocol and save name of the `DocList`. Should be of the form ``protocol://namespace/name``. e.g. ``s3://bucket/path/to/namespace/name``, ``file:///path/to/folder/name`` :param show_progress: if true, display a progress bar. :param local_cache: store the downloaded `DocList` to local folder :return: Iterator of Documents """ from docarray.base_doc import AnyDoc if cls.doc_type == AnyDoc: raise TypeError( 'There is no document schema defined. ' 'Please specify the `DocList`\'s Document type using `DocList[MyDoc]`.' ) logging.info(f'Pulling Document stream from {url}') protocol, name = cls.resolve_url(url) return cls.get_pushpull_backend(protocol).pull_stream( cls, name, show_progress, local_cache # type: ignore ) |
| You are visitor 01402503 — GeezerVison does not store the pages you look up. Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required. |