We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Configure Distance Metric in Weaviate
As discussed over Slack, when Weaviate supports different distance metrics other than `cosine` they need to be configurable in doc array as well.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `docarray/array/storage/weaviate/backend.py`
Content:
```
1 import uuid
2 from dataclasses import dataclass, field
3 from typing import (
4     Iterable,
5     Dict,
6     Optional,
7     TYPE_CHECKING,
8     Union,
9     Tuple,
10     List,
11 )
12 
13 import numpy as np
14 import weaviate
15 
16 from .... import Document
17 from ....helper import dataclass_from_dict, filter_dict
18 from ..base.backend import BaseBackendMixin
19 from ..registry import _REGISTRY
20 
21 if TYPE_CHECKING:
22     from ....typing import ArrayType, DocumentArraySourceType
23 
24 
25 @dataclass
26 class WeaviateConfig:
27     """This class stores the config variables to initialize
28     connection to the Weaviate server"""
29 
30     host: Optional[str] = field(default='localhost')
31     port: Optional[int] = field(default=8080)
32     protocol: Optional[str] = field(default='http')
33     name: Optional[str] = None
34     serialize_config: Dict = field(default_factory=dict)
35     n_dim: Optional[int] = None  # deprecated, not used anymore since weaviate 1.10
36     # vectorIndexConfig parameters
37     ef: Optional[int] = None
38     ef_construction: Optional[int] = None
39     timeout_config: Optional[Tuple[int, int]] = None
40     max_connections: Optional[int] = None
41     dynamic_ef_min: Optional[int] = None
42     dynamic_ef_max: Optional[int] = None
43     dynamic_ef_factor: Optional[int] = None
44     vector_cache_max_objects: Optional[int] = None
45     flat_search_cutoff: Optional[int] = None
46     cleanup_interval_seconds: Optional[int] = None
47     skip: Optional[bool] = None
48 
49 
50 class BackendMixin(BaseBackendMixin):
51     """Provide necessary functions to enable this storage backend."""
52 
53     def _init_storage(
54         self,
55         _docs: Optional['DocumentArraySourceType'] = None,
56         config: Optional[Union[WeaviateConfig, Dict]] = None,
57         **kwargs,
58     ):
59         """Initialize weaviate storage.
60 
61         :param docs: the list of documents to initialize to
62         :param config: the config object used to ininitialize connection to weaviate server
63         :param kwargs: extra keyword arguments
64         :raises ValueError: only one of name or docs can be used for initialization,
65             raise an error if both are provided
66         """
67 
68         if not config:
69             config = WeaviateConfig()
70         elif isinstance(config, dict):
71             config = dataclass_from_dict(WeaviateConfig, config)
72 
73         self._serialize_config = config.serialize_config
74 
75         if config.name and config.name != config.name.capitalize():
76             raise ValueError(
77                 'Weaviate class name has to be capitalized. '
78                 'Please capitalize when declaring the name field in config.'
79             )
80 
81         self._persist = bool(config.name)
82 
83         self._client = weaviate.Client(
84             f'{config.protocol}://{config.host}:{config.port}',
85             timeout_config=config.timeout_config,
86         )
87         self._config = config
88 
89         self._schemas = self._load_or_create_weaviate_schema()
90 
91         _REGISTRY[self.__class__.__name__][self._class_name].append(self)
92 
93         super()._init_storage(_docs, **kwargs)
94 
95         # To align with Sqlite behavior; if `docs` is not `None` and table name
96         # is provided, :class:`DocumentArraySqlite` will clear the existing
97         # table and load the given `docs`
98         if _docs is None:
99             return
100         elif isinstance(_docs, Iterable):
101             self.clear()
102             self.extend(_docs)
103         else:
104             self.clear()
105             if isinstance(_docs, Document):
106                 self.append(_docs)
107 
108     def _get_weaviate_class_name(self) -> str:
109         """Generate the class/schema name using the ``uuid1`` module with some
110         formatting to tailor to weaviate class name convention
111 
112         :return: string representing the name of  weaviate class/schema name of
113             this :class:`DocumentArrayWeaviate` object
114         """
115         return f'Class{uuid.uuid4().hex}'
116 
117     def _get_schema_by_name(self, cls_name: str) -> Dict:
118         """Return the schema dictionary object with the class name
119         Content of the all dictionaries by this method are the same except the name
120         of the weaviate's ``class``
121 
122         :param cls_name: the name of the schema/class in weaviate
123         :return: the schema dictionary
124         """
125         # TODO: ideally we should only use one schema. this will allow us to deal with
126         # consistency better
127         hnsw_config = {
128             'ef': self._config.ef,
129             'efConstruction': self._config.ef_construction,
130             'maxConnections': self._config.max_connections,
131             'dynamicEfMin': self._config.dynamic_ef_min,
132             'dynamicEfMax': self._config.dynamic_ef_max,
133             'dynamicEfFactor': self._config.dynamic_ef_factor,
134             'vectorCacheMaxObjects': self._config.vector_cache_max_objects,
135             'flatSearchCutoff': self._config.flat_search_cutoff,
136             'cleanupIntervalSeconds': self._config.cleanup_interval_seconds,
137             'skip': self._config.skip,
138         }
139 
140         return {
141             'classes': [
142                 {
143                     'class': cls_name,
144                     "vectorizer": "none",
145                     'vectorIndexConfig': {'skip': False, **filter_dict(hnsw_config)},
146                     'properties': [
147                         {
148                             'dataType': ['blob'],
149                             'name': '_serialized',
150                             'indexInverted': False,
151                         },
152                     ],
153                 },
154                 {
155                     'class': cls_name + 'Meta',
156                     "vectorizer": "none",
157                     'vectorIndexConfig': {'skip': True},
158                     'properties': [
159                         {
160                             'dataType': ['string[]'],
161                             'name': '_offset2ids',
162                             'indexInverted': False,
163                         },
164                     ],
165                 },
166             ]
167         }
168 
169     def _load_or_create_weaviate_schema(self):
170         """Create a new weaviate schema for this :class:`DocumentArrayWeaviate` object
171         if not present in weaviate or if ``self._config.name`` is None. If ``self._config.name``
172         is provided and not None and schema with the specified name exists in weaviate,
173         then load the object with the given ``self._config.name``
174 
175         :return: the schemas of this :class`DocumentArrayWeaviate` object and its meta
176         """
177         if not self._config.name:
178             name_candidate = self._get_weaviate_class_name()
179             doc_schemas = self._get_schema_by_name(name_candidate)
180             while self._client.schema.contains(doc_schemas):
181                 name_candidate = self._get_weaviate_class_name()
182                 doc_schemas = self._get_schema_by_name(name_candidate)
183             self._client.schema.create(doc_schemas)
184             self._config.name = name_candidate
185             return doc_schemas
186 
187         doc_schemas = self._get_schema_by_name(self._config.name)
188         if self._client.schema.contains(doc_schemas):
189             return doc_schemas
190 
191         self._client.schema.create(doc_schemas)
192         return doc_schemas
193 
194     def _update_offset2ids_meta(self):
195         """Update the offset2ids in weaviate the the current local version"""
196         if self._offset2ids_wid is not None and self._client.data_object.exists(
197             self._offset2ids_wid
198         ):
199             self._client.data_object.update(
200                 data_object={'_offset2ids': self._offset2ids.ids},
201                 class_name=self._meta_name,
202                 uuid=self._offset2ids_wid,
203             )
204         else:
205             self._offset2ids_wid = str(uuid.uuid1())
206             self._client.data_object.create(
207                 data_object={'_offset2ids': self._offset2ids.ids},
208                 class_name=self._meta_name,
209                 uuid=self._offset2ids_wid,
210             )
211 
212     def _get_offset2ids_meta(self) -> Tuple[List, str]:
213         """Return the offset2ids stored in weaviate along with the name of the schema/class
214         in weaviate that stores meta information of this object
215 
216         :return: a tuple with first element as a list of offset2ids and second element
217                  being name of weaviate class/schema of the meta object
218 
219         :raises ValueError: error is raised if meta class name is not defined
220         """
221         if not self._meta_name:
222             raise ValueError('meta object is not defined')
223 
224         resp = (
225             self._client.query.get(self._meta_name, ['_offset2ids', '_additional {id}'])
226             .do()
227             .get('data', {})
228             .get('Get', {})
229             .get(self._meta_name, [])
230         )
231 
232         if not resp:
233             return [], None
234         elif len(resp) == 1:
235             return resp[0]['_offset2ids'], resp[0]['_additional']['id']
236         else:
237             raise ValueError('received multiple meta copies which is invalid')
238 
239     @property
240     def name(self):
241         """An alias to _class_name that returns the id/name of the class
242         in the weaviate of this :class:`DocumentArrayWeaviate`
243 
244         :return: name of weaviate class/schema of this :class:`DocumentArrayWeaviate`
245         """
246         return self._class_name
247 
248     @property
249     def _class_name(self):
250         """Return the name of the class in weaviate of this :class:`DocumentArrayWeaviate
251 
252         :return: name of weaviate class/schema of this :class:`DocumentArrayWeaviate`
253         """
254         if not self._schemas:
255             return None
256         return self._schemas['classes'][0]['class']
257 
258     @property
259     def _meta_name(self):
260         """Return the name of the class in weaviate that stores the meta information of
261         this :class:`DocumentArrayWeaviate`
262 
263         :return: name of weaviate class/schema of class that stores the meta information
264         """
265         # TODO: remove this after we combine the meta info to the DocumentArray class
266         if not self._schemas:
267             return None
268         return self._schemas['classes'][1]['class']
269 
270     @property
271     def _class_schema(self) -> Optional[Dict]:
272         """Return the schema dictionary of this :class:`DocumentArrayWeaviate`'s weaviate schema
273 
274         :return: the dictionary representing this weaviate schema
275         """
276         if not self._schemas:
277             return None
278         return self._schemas['classes'][0]
279 
280     @property
281     def _meta_schema(self):
282         """Return the schema dictionary of this weaviate schema that stores this object's meta
283 
284         :return: the dictionary representing a meta object's weaviate schema
285         """
286         if not self._schemas and len(self._schemas) < 2:
287             return None
288         return self._schemas['classes'][1]
289 
290     def _doc2weaviate_create_payload(self, value: 'Document'):
291         """Return the payload to store :class:`Document` into weaviate
292 
293         :param value: document to create a payload for
294         :return: the payload dictionary
295         """
296         return dict(
297             data_object={'_serialized': value.to_base64(**self._serialize_config)},
298             class_name=self._class_name,
299             uuid=self._map_id(value.id),
300             vector=self._map_embedding(value.embedding),
301         )
302 
303     def _map_id(self, doc_id: str):
304         """the function maps doc id to weaviate id
305 
306         :param doc_id: id of the document
307         :return: weaviate object id
308         """
309         # appending class name to doc id to handle the case:
310         # daw1 = DocumentArrayWeaviate([Document(id=str(i), text='hi') for i in range(3)])
311         # daw2 = DocumentArrayWeaviate([Document(id=str(i), text='bye') for i in range(3)])
312         # daw2[0, 'text'] == 'hi' # this will be False if we don't append class name
313         return str(uuid.uuid5(uuid.NAMESPACE_URL, doc_id + self._class_name))
314 
315     def _map_embedding(self, embedding: 'ArrayType'):
316         if embedding is not None:
317             from ....math.ndarray import to_numpy_array
318 
319             embedding = to_numpy_array(embedding)
320 
321             if embedding.ndim > 1:
322                 embedding = np.asarray(embedding).squeeze()
323 
324             # Weaviate expects vector to have dim 2 at least
325             # or get weaviate.exceptions.UnexpectedStatusCodeException:  models.C11yVector
326             # hence we cast it to list of a single element
327             if len(embedding) == 1:
328                 embedding = [embedding[0]]
329         else:
330             embedding = None
331         return embedding
332 
333     def __getstate__(self):
334         d = dict(self.__dict__)
335         del d['_client']
336         return d
337 
338     def __setstate__(self, state):
339         self.__dict__ = state
340         self._client = weaviate.Client(
341             f'{state["_config"].protocol}://{state["_config"].host}:{state["_config"].port}'
342         )
343
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/docarray/array/storage/weaviate/backend.py b/docarray/array/storage/weaviate/backend.py
--- a/docarray/array/storage/weaviate/backend.py
+++ b/docarray/array/storage/weaviate/backend.py
@@ -45,6 +45,7 @@
     flat_search_cutoff: Optional[int] = None
     cleanup_interval_seconds: Optional[int] = None
     skip: Optional[bool] = None
+    distance: Optional[str] = None
 
 
 class BackendMixin(BaseBackendMixin):
@@ -135,6 +136,7 @@
             'flatSearchCutoff': self._config.flat_search_cutoff,
             'cleanupIntervalSeconds': self._config.cleanup_interval_seconds,
             'skip': self._config.skip,
+            'distance': self._config.distance,
         }
 
         return {

{"golden_diff": "diff --git a/docarray/array/storage/weaviate/backend.py b/docarray/array/storage/weaviate/backend.py\n--- a/docarray/array/storage/weaviate/backend.py\n+++ b/docarray/array/storage/weaviate/backend.py\n@@ -45,6 +45,7 @@\n     flat_search_cutoff: Optional[int] = None\n     cleanup_interval_seconds: Optional[int] = None\n     skip: Optional[bool] = None\n+    distance: Optional[str] = None\n \n \n class BackendMixin(BaseBackendMixin):\n@@ -135,6 +136,7 @@\n             'flatSearchCutoff': self._config.flat_search_cutoff,\n             'cleanupIntervalSeconds': self._config.cleanup_interval_seconds,\n             'skip': self._config.skip,\n+            'distance': self._config.distance,\n         }\n \n         return {\n", "issue": "Configure Distance Metric in Weaviate\nAs discussed over Slack, when Weaviate supports different distance metrics other than `cosine` they need to be configurable in doc array as well.\n", "before_files": [{"content": "import uuid\nfrom dataclasses import dataclass, field\nfrom typing import (\n    Iterable,\n    Dict,\n    Optional,\n    TYPE_CHECKING,\n    Union,\n    Tuple,\n    List,\n)\n\nimport numpy as np\nimport weaviate\n\nfrom .... import Document\nfrom ....helper import dataclass_from_dict, filter_dict\nfrom ..base.backend import BaseBackendMixin\nfrom ..registry import _REGISTRY\n\nif TYPE_CHECKING:\n    from ....typing import ArrayType, DocumentArraySourceType\n\n\n@dataclass\nclass WeaviateConfig:\n    \"\"\"This class stores the config variables to initialize\n    connection to the Weaviate server\"\"\"\n\n    host: Optional[str] = field(default='localhost')\n    port: Optional[int] = field(default=8080)\n    protocol: Optional[str] = field(default='http')\n    name: Optional[str] = None\n    serialize_config: Dict = field(default_factory=dict)\n    n_dim: Optional[int] = None  # deprecated, not used anymore since weaviate 1.10\n    # vectorIndexConfig parameters\n    ef: Optional[int] = None\n    ef_construction: Optional[int] = None\n    timeout_config: Optional[Tuple[int, int]] = None\n    max_connections: Optional[int] = None\n    dynamic_ef_min: Optional[int] = None\n    dynamic_ef_max: Optional[int] = None\n    dynamic_ef_factor: Optional[int] = None\n    vector_cache_max_objects: Optional[int] = None\n    flat_search_cutoff: Optional[int] = None\n    cleanup_interval_seconds: Optional[int] = None\n    skip: Optional[bool] = None\n\n\nclass BackendMixin(BaseBackendMixin):\n    \"\"\"Provide necessary functions to enable this storage backend.\"\"\"\n\n    def _init_storage(\n        self,\n        _docs: Optional['DocumentArraySourceType'] = None,\n        config: Optional[Union[WeaviateConfig, Dict]] = None,\n        **kwargs,\n    ):\n        \"\"\"Initialize weaviate storage.\n\n        :param docs: the list of documents to initialize to\n        :param config: the config object used to ininitialize connection to weaviate server\n        :param kwargs: extra keyword arguments\n        :raises ValueError: only one of name or docs can be used for initialization,\n            raise an error if both are provided\n        \"\"\"\n\n        if not config:\n            config = WeaviateConfig()\n        elif isinstance(config, dict):\n            config = dataclass_from_dict(WeaviateConfig, config)\n\n        self._serialize_config = config.serialize_config\n\n        if config.name and config.name != config.name.capitalize():\n            raise ValueError(\n                'Weaviate class name has to be capitalized. '\n                'Please capitalize when declaring the name field in config.'\n            )\n\n        self._persist = bool(config.name)\n\n        self._client = weaviate.Client(\n            f'{config.protocol}://{config.host}:{config.port}',\n            timeout_config=config.timeout_config,\n        )\n        self._config = config\n\n        self._schemas = self._load_or_create_weaviate_schema()\n\n        _REGISTRY[self.__class__.__name__][self._class_name].append(self)\n\n        super()._init_storage(_docs, **kwargs)\n\n        # To align with Sqlite behavior; if `docs` is not `None` and table name\n        # is provided, :class:`DocumentArraySqlite` will clear the existing\n        # table and load the given `docs`\n        if _docs is None:\n            return\n        elif isinstance(_docs, Iterable):\n            self.clear()\n            self.extend(_docs)\n        else:\n            self.clear()\n            if isinstance(_docs, Document):\n                self.append(_docs)\n\n    def _get_weaviate_class_name(self) -> str:\n        \"\"\"Generate the class/schema name using the ``uuid1`` module with some\n        formatting to tailor to weaviate class name convention\n\n        :return: string representing the name of  weaviate class/schema name of\n            this :class:`DocumentArrayWeaviate` object\n        \"\"\"\n        return f'Class{uuid.uuid4().hex}'\n\n    def _get_schema_by_name(self, cls_name: str) -> Dict:\n        \"\"\"Return the schema dictionary object with the class name\n        Content of the all dictionaries by this method are the same except the name\n        of the weaviate's ``class``\n\n        :param cls_name: the name of the schema/class in weaviate\n        :return: the schema dictionary\n        \"\"\"\n        # TODO: ideally we should only use one schema. this will allow us to deal with\n        # consistency better\n        hnsw_config = {\n            'ef': self._config.ef,\n            'efConstruction': self._config.ef_construction,\n            'maxConnections': self._config.max_connections,\n            'dynamicEfMin': self._config.dynamic_ef_min,\n            'dynamicEfMax': self._config.dynamic_ef_max,\n            'dynamicEfFactor': self._config.dynamic_ef_factor,\n            'vectorCacheMaxObjects': self._config.vector_cache_max_objects,\n            'flatSearchCutoff': self._config.flat_search_cutoff,\n            'cleanupIntervalSeconds': self._config.cleanup_interval_seconds,\n            'skip': self._config.skip,\n        }\n\n        return {\n            'classes': [\n                {\n                    'class': cls_name,\n                    \"vectorizer\": \"none\",\n                    'vectorIndexConfig': {'skip': False, **filter_dict(hnsw_config)},\n                    'properties': [\n                        {\n                            'dataType': ['blob'],\n                            'name': '_serialized',\n                            'indexInverted': False,\n                        },\n                    ],\n                },\n                {\n                    'class': cls_name + 'Meta',\n                    \"vectorizer\": \"none\",\n                    'vectorIndexConfig': {'skip': True},\n                    'properties': [\n                        {\n                            'dataType': ['string[]'],\n                            'name': '_offset2ids',\n                            'indexInverted': False,\n                        },\n                    ],\n                },\n            ]\n        }\n\n    def _load_or_create_weaviate_schema(self):\n        \"\"\"Create a new weaviate schema for this :class:`DocumentArrayWeaviate` object\n        if not present in weaviate or if ``self._config.name`` is None. If ``self._config.name``\n        is provided and not None and schema with the specified name exists in weaviate,\n        then load the object with the given ``self._config.name``\n\n        :return: the schemas of this :class`DocumentArrayWeaviate` object and its meta\n        \"\"\"\n        if not self._config.name:\n            name_candidate = self._get_weaviate_class_name()\n            doc_schemas = self._get_schema_by_name(name_candidate)\n            while self._client.schema.contains(doc_schemas):\n                name_candidate = self._get_weaviate_class_name()\n                doc_schemas = self._get_schema_by_name(name_candidate)\n            self._client.schema.create(doc_schemas)\n            self._config.name = name_candidate\n            return doc_schemas\n\n        doc_schemas = self._get_schema_by_name(self._config.name)\n        if self._client.schema.contains(doc_schemas):\n            return doc_schemas\n\n        self._client.schema.create(doc_schemas)\n        return doc_schemas\n\n    def _update_offset2ids_meta(self):\n        \"\"\"Update the offset2ids in weaviate the the current local version\"\"\"\n        if self._offset2ids_wid is not None and self._client.data_object.exists(\n            self._offset2ids_wid\n        ):\n            self._client.data_object.update(\n                data_object={'_offset2ids': self._offset2ids.ids},\n                class_name=self._meta_name,\n                uuid=self._offset2ids_wid,\n            )\n        else:\n            self._offset2ids_wid = str(uuid.uuid1())\n            self._client.data_object.create(\n                data_object={'_offset2ids': self._offset2ids.ids},\n                class_name=self._meta_name,\n                uuid=self._offset2ids_wid,\n            )\n\n    def _get_offset2ids_meta(self) -> Tuple[List, str]:\n        \"\"\"Return the offset2ids stored in weaviate along with the name of the schema/class\n        in weaviate that stores meta information of this object\n\n        :return: a tuple with first element as a list of offset2ids and second element\n                 being name of weaviate class/schema of the meta object\n\n        :raises ValueError: error is raised if meta class name is not defined\n        \"\"\"\n        if not self._meta_name:\n            raise ValueError('meta object is not defined')\n\n        resp = (\n            self._client.query.get(self._meta_name, ['_offset2ids', '_additional {id}'])\n            .do()\n            .get('data', {})\n            .get('Get', {})\n            .get(self._meta_name, [])\n        )\n\n        if not resp:\n            return [], None\n        elif len(resp) == 1:\n            return resp[0]['_offset2ids'], resp[0]['_additional']['id']\n        else:\n            raise ValueError('received multiple meta copies which is invalid')\n\n    @property\n    def name(self):\n        \"\"\"An alias to _class_name that returns the id/name of the class\n        in the weaviate of this :class:`DocumentArrayWeaviate`\n\n        :return: name of weaviate class/schema of this :class:`DocumentArrayWeaviate`\n        \"\"\"\n        return self._class_name\n\n    @property\n    def _class_name(self):\n        \"\"\"Return the name of the class in weaviate of this :class:`DocumentArrayWeaviate\n\n        :return: name of weaviate class/schema of this :class:`DocumentArrayWeaviate`\n        \"\"\"\n        if not self._schemas:\n            return None\n        return self._schemas['classes'][0]['class']\n\n    @property\n    def _meta_name(self):\n        \"\"\"Return the name of the class in weaviate that stores the meta information of\n        this :class:`DocumentArrayWeaviate`\n\n        :return: name of weaviate class/schema of class that stores the meta information\n        \"\"\"\n        # TODO: remove this after we combine the meta info to the DocumentArray class\n        if not self._schemas:\n            return None\n        return self._schemas['classes'][1]['class']\n\n    @property\n    def _class_schema(self) -> Optional[Dict]:\n        \"\"\"Return the schema dictionary of this :class:`DocumentArrayWeaviate`'s weaviate schema\n\n        :return: the dictionary representing this weaviate schema\n        \"\"\"\n        if not self._schemas:\n            return None\n        return self._schemas['classes'][0]\n\n    @property\n    def _meta_schema(self):\n        \"\"\"Return the schema dictionary of this weaviate schema that stores this object's meta\n\n        :return: the dictionary representing a meta object's weaviate schema\n        \"\"\"\n        if not self._schemas and len(self._schemas) < 2:\n            return None\n        return self._schemas['classes'][1]\n\n    def _doc2weaviate_create_payload(self, value: 'Document'):\n        \"\"\"Return the payload to store :class:`Document` into weaviate\n\n        :param value: document to create a payload for\n        :return: the payload dictionary\n        \"\"\"\n        return dict(\n            data_object={'_serialized': value.to_base64(**self._serialize_config)},\n            class_name=self._class_name,\n            uuid=self._map_id(value.id),\n            vector=self._map_embedding(value.embedding),\n        )\n\n    def _map_id(self, doc_id: str):\n        \"\"\"the function maps doc id to weaviate id\n\n        :param doc_id: id of the document\n        :return: weaviate object id\n        \"\"\"\n        # appending class name to doc id to handle the case:\n        # daw1 = DocumentArrayWeaviate([Document(id=str(i), text='hi') for i in range(3)])\n        # daw2 = DocumentArrayWeaviate([Document(id=str(i), text='bye') for i in range(3)])\n        # daw2[0, 'text'] == 'hi' # this will be False if we don't append class name\n        return str(uuid.uuid5(uuid.NAMESPACE_URL, doc_id + self._class_name))\n\n    def _map_embedding(self, embedding: 'ArrayType'):\n        if embedding is not None:\n            from ....math.ndarray import to_numpy_array\n\n            embedding = to_numpy_array(embedding)\n\n            if embedding.ndim > 1:\n                embedding = np.asarray(embedding).squeeze()\n\n            # Weaviate expects vector to have dim 2 at least\n            # or get weaviate.exceptions.UnexpectedStatusCodeException:  models.C11yVector\n            # hence we cast it to list of a single element\n            if len(embedding) == 1:\n                embedding = [embedding[0]]\n        else:\n            embedding = None\n        return embedding\n\n    def __getstate__(self):\n        d = dict(self.__dict__)\n        del d['_client']\n        return d\n\n    def __setstate__(self, state):\n        self.__dict__ = state\n        self._client = weaviate.Client(\n            f'{state[\"_config\"].protocol}://{state[\"_config\"].host}:{state[\"_config\"].port}'\n        )\n", "path": "docarray/array/storage/weaviate/backend.py"}], "after_files": [{"content": "import uuid\nfrom dataclasses import dataclass, field\nfrom typing import (\n    Iterable,\n    Dict,\n    Optional,\n    TYPE_CHECKING,\n    Union,\n    Tuple,\n    List,\n)\n\nimport numpy as np\nimport weaviate\n\nfrom .... import Document\nfrom ....helper import dataclass_from_dict, filter_dict\nfrom ..base.backend import BaseBackendMixin\nfrom ..registry import _REGISTRY\n\nif TYPE_CHECKING:\n    from ....typing import ArrayType, DocumentArraySourceType\n\n\n@dataclass\nclass WeaviateConfig:\n    \"\"\"This class stores the config variables to initialize\n    connection to the Weaviate server\"\"\"\n\n    host: Optional[str] = field(default='localhost')\n    port: Optional[int] = field(default=8080)\n    protocol: Optional[str] = field(default='http')\n    name: Optional[str] = None\n    serialize_config: Dict = field(default_factory=dict)\n    n_dim: Optional[int] = None  # deprecated, not used anymore since weaviate 1.10\n    # vectorIndexConfig parameters\n    ef: Optional[int] = None\n    ef_construction: Optional[int] = None\n    timeout_config: Optional[Tuple[int, int]] = None\n    max_connections: Optional[int] = None\n    dynamic_ef_min: Optional[int] = None\n    dynamic_ef_max: Optional[int] = None\n    dynamic_ef_factor: Optional[int] = None\n    vector_cache_max_objects: Optional[int] = None\n    flat_search_cutoff: Optional[int] = None\n    cleanup_interval_seconds: Optional[int] = None\n    skip: Optional[bool] = None\n    distance: Optional[str] = None\n\n\nclass BackendMixin(BaseBackendMixin):\n    \"\"\"Provide necessary functions to enable this storage backend.\"\"\"\n\n    def _init_storage(\n        self,\n        _docs: Optional['DocumentArraySourceType'] = None,\n        config: Optional[Union[WeaviateConfig, Dict]] = None,\n        **kwargs,\n    ):\n        \"\"\"Initialize weaviate storage.\n\n        :param docs: the list of documents to initialize to\n        :param config: the config object used to ininitialize connection to weaviate server\n        :param kwargs: extra keyword arguments\n        :raises ValueError: only one of name or docs can be used for initialization,\n            raise an error if both are provided\n        \"\"\"\n\n        if not config:\n            config = WeaviateConfig()\n        elif isinstance(config, dict):\n            config = dataclass_from_dict(WeaviateConfig, config)\n\n        self._serialize_config = config.serialize_config\n\n        if config.name and config.name != config.name.capitalize():\n            raise ValueError(\n                'Weaviate class name has to be capitalized. '\n                'Please capitalize when declaring the name field in config.'\n            )\n\n        self._persist = bool(config.name)\n\n        self._client = weaviate.Client(\n            f'{config.protocol}://{config.host}:{config.port}',\n            timeout_config=config.timeout_config,\n        )\n        self._config = config\n\n        self._schemas = self._load_or_create_weaviate_schema()\n\n        _REGISTRY[self.__class__.__name__][self._class_name].append(self)\n\n        super()._init_storage(_docs, **kwargs)\n\n        # To align with Sqlite behavior; if `docs` is not `None` and table name\n        # is provided, :class:`DocumentArraySqlite` will clear the existing\n        # table and load the given `docs`\n        if _docs is None:\n            return\n        elif isinstance(_docs, Iterable):\n            self.clear()\n            self.extend(_docs)\n        else:\n            self.clear()\n            if isinstance(_docs, Document):\n                self.append(_docs)\n\n    def _get_weaviate_class_name(self) -> str:\n        \"\"\"Generate the class/schema name using the ``uuid1`` module with some\n        formatting to tailor to weaviate class name convention\n\n        :return: string representing the name of  weaviate class/schema name of\n            this :class:`DocumentArrayWeaviate` object\n        \"\"\"\n        return f'Class{uuid.uuid4().hex}'\n\n    def _get_schema_by_name(self, cls_name: str) -> Dict:\n        \"\"\"Return the schema dictionary object with the class name\n        Content of the all dictionaries by this method are the same except the name\n        of the weaviate's ``class``\n\n        :param cls_name: the name of the schema/class in weaviate\n        :return: the schema dictionary\n        \"\"\"\n        # TODO: ideally we should only use one schema. this will allow us to deal with\n        # consistency better\n        hnsw_config = {\n            'ef': self._config.ef,\n            'efConstruction': self._config.ef_construction,\n            'maxConnections': self._config.max_connections,\n            'dynamicEfMin': self._config.dynamic_ef_min,\n            'dynamicEfMax': self._config.dynamic_ef_max,\n            'dynamicEfFactor': self._config.dynamic_ef_factor,\n            'vectorCacheMaxObjects': self._config.vector_cache_max_objects,\n            'flatSearchCutoff': self._config.flat_search_cutoff,\n            'cleanupIntervalSeconds': self._config.cleanup_interval_seconds,\n            'skip': self._config.skip,\n            'distance': self._config.distance,\n        }\n\n        return {\n            'classes': [\n                {\n                    'class': cls_name,\n                    \"vectorizer\": \"none\",\n                    'vectorIndexConfig': {'skip': False, **filter_dict(hnsw_config)},\n                    'properties': [\n                        {\n                            'dataType': ['blob'],\n                            'name': '_serialized',\n                            'indexInverted': False,\n                        },\n                    ],\n                },\n                {\n                    'class': cls_name + 'Meta',\n                    \"vectorizer\": \"none\",\n                    'vectorIndexConfig': {'skip': True},\n                    'properties': [\n                        {\n                            'dataType': ['string[]'],\n                            'name': '_offset2ids',\n                            'indexInverted': False,\n                        },\n                    ],\n                },\n            ]\n        }\n\n    def _load_or_create_weaviate_schema(self):\n        \"\"\"Create a new weaviate schema for this :class:`DocumentArrayWeaviate` object\n        if not present in weaviate or if ``self._config.name`` is None. If ``self._config.name``\n        is provided and not None and schema with the specified name exists in weaviate,\n        then load the object with the given ``self._config.name``\n\n        :return: the schemas of this :class`DocumentArrayWeaviate` object and its meta\n        \"\"\"\n        if not self._config.name:\n            name_candidate = self._get_weaviate_class_name()\n            doc_schemas = self._get_schema_by_name(name_candidate)\n            while self._client.schema.contains(doc_schemas):\n                name_candidate = self._get_weaviate_class_name()\n                doc_schemas = self._get_schema_by_name(name_candidate)\n            self._client.schema.create(doc_schemas)\n            self._config.name = name_candidate\n            return doc_schemas\n\n        doc_schemas = self._get_schema_by_name(self._config.name)\n        if self._client.schema.contains(doc_schemas):\n            return doc_schemas\n\n        self._client.schema.create(doc_schemas)\n        return doc_schemas\n\n    def _update_offset2ids_meta(self):\n        \"\"\"Update the offset2ids in weaviate the the current local version\"\"\"\n        if self._offset2ids_wid is not None and self._client.data_object.exists(\n            self._offset2ids_wid\n        ):\n            self._client.data_object.update(\n                data_object={'_offset2ids': self._offset2ids.ids},\n                class_name=self._meta_name,\n                uuid=self._offset2ids_wid,\n            )\n        else:\n            self._offset2ids_wid = str(uuid.uuid1())\n            self._client.data_object.create(\n                data_object={'_offset2ids': self._offset2ids.ids},\n                class_name=self._meta_name,\n                uuid=self._offset2ids_wid,\n            )\n\n    def _get_offset2ids_meta(self) -> Tuple[List, str]:\n        \"\"\"Return the offset2ids stored in weaviate along with the name of the schema/class\n        in weaviate that stores meta information of this object\n\n        :return: a tuple with first element as a list of offset2ids and second element\n                 being name of weaviate class/schema of the meta object\n\n        :raises ValueError: error is raised if meta class name is not defined\n        \"\"\"\n        if not self._meta_name:\n            raise ValueError('meta object is not defined')\n\n        resp = (\n            self._client.query.get(self._meta_name, ['_offset2ids', '_additional {id}'])\n            .do()\n            .get('data', {})\n            .get('Get', {})\n            .get(self._meta_name, [])\n        )\n\n        if not resp:\n            return [], None\n        elif len(resp) == 1:\n            return resp[0]['_offset2ids'], resp[0]['_additional']['id']\n        else:\n            raise ValueError('received multiple meta copies which is invalid')\n\n    @property\n    def name(self):\n        \"\"\"An alias to _class_name that returns the id/name of the class\n        in the weaviate of this :class:`DocumentArrayWeaviate`\n\n        :return: name of weaviate class/schema of this :class:`DocumentArrayWeaviate`\n        \"\"\"\n        return self._class_name\n\n    @property\n    def _class_name(self):\n        \"\"\"Return the name of the class in weaviate of this :class:`DocumentArrayWeaviate\n\n        :return: name of weaviate class/schema of this :class:`DocumentArrayWeaviate`\n        \"\"\"\n        if not self._schemas:\n            return None\n        return self._schemas['classes'][0]['class']\n\n    @property\n    def _meta_name(self):\n        \"\"\"Return the name of the class in weaviate that stores the meta information of\n        this :class:`DocumentArrayWeaviate`\n\n        :return: name of weaviate class/schema of class that stores the meta information\n        \"\"\"\n        # TODO: remove this after we combine the meta info to the DocumentArray class\n        if not self._schemas:\n            return None\n        return self._schemas['classes'][1]['class']\n\n    @property\n    def _class_schema(self) -> Optional[Dict]:\n        \"\"\"Return the schema dictionary of this :class:`DocumentArrayWeaviate`'s weaviate schema\n\n        :return: the dictionary representing this weaviate schema\n        \"\"\"\n        if not self._schemas:\n            return None\n        return self._schemas['classes'][0]\n\n    @property\n    def _meta_schema(self):\n        \"\"\"Return the schema dictionary of this weaviate schema that stores this object's meta\n\n        :return: the dictionary representing a meta object's weaviate schema\n        \"\"\"\n        if not self._schemas and len(self._schemas) < 2:\n            return None\n        return self._schemas['classes'][1]\n\n    def _doc2weaviate_create_payload(self, value: 'Document'):\n        \"\"\"Return the payload to store :class:`Document` into weaviate\n\n        :param value: document to create a payload for\n        :return: the payload dictionary\n        \"\"\"\n        return dict(\n            data_object={'_serialized': value.to_base64(**self._serialize_config)},\n            class_name=self._class_name,\n            uuid=self._map_id(value.id),\n            vector=self._map_embedding(value.embedding),\n        )\n\n    def _map_id(self, doc_id: str):\n        \"\"\"the function maps doc id to weaviate id\n\n        :param doc_id: id of the document\n        :return: weaviate object id\n        \"\"\"\n        # appending class name to doc id to handle the case:\n        # daw1 = DocumentArrayWeaviate([Document(id=str(i), text='hi') for i in range(3)])\n        # daw2 = DocumentArrayWeaviate([Document(id=str(i), text='bye') for i in range(3)])\n        # daw2[0, 'text'] == 'hi' # this will be False if we don't append class name\n        return str(uuid.uuid5(uuid.NAMESPACE_URL, doc_id + self._class_name))\n\n    def _map_embedding(self, embedding: 'ArrayType'):\n        if embedding is not None:\n            from ....math.ndarray import to_numpy_array\n\n            embedding = to_numpy_array(embedding)\n\n            if embedding.ndim > 1:\n                embedding = np.asarray(embedding).squeeze()\n\n            # Weaviate expects vector to have dim 2 at least\n            # or get weaviate.exceptions.UnexpectedStatusCodeException:  models.C11yVector\n            # hence we cast it to list of a single element\n            if len(embedding) == 1:\n                embedding = [embedding[0]]\n        else:\n            embedding = None\n        return embedding\n\n    def __getstate__(self):\n        d = dict(self.__dict__)\n        del d['_client']\n        return d\n\n    def __setstate__(self, state):\n        self.__dict__ = state\n        self._client = weaviate.Client(\n            f'{state[\"_config\"].protocol}://{state[\"_config\"].host}:{state[\"_config\"].port}'\n        )\n", "path": "docarray/array/storage/weaviate/backend.py"}]}