Skip to content

Client Distribution API

VectorDBClient

This class determines whether it is local or remote based on the URI, thereby implementing the distribution of local and remote client communication. The data storage location is also determined accordingly.

  • When uri is None: The local client is used by default, and the data is stored under the default root path.
  • When uri is a local path: The local client is used, and the data is stored under the specified path.
  • When uri is a remote URL: The remote client is used, and the data is stored on the remote server.
Source code in lynse/__init__.py 
 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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
class VectorDBClient:
    """
    This class determines whether it is local or remote based on the URI, thereby implementing the distribution of local and remote client communication.
    The data storage location is also determined accordingly.

    - **When `uri` is None**: The local client is used by default, and the data is stored under the default root path.
    - **When `uri` is a local path**: The local client is used, and the data is stored under the specified path.
    - **When `uri` is a remote URL**: The remote client is used, and the data is stored on the remote server.
    """
    def __init__(self, uri: Union[str, None, Path] = None):
        """
        Initialize the LynseDB client.

        Parameters:
            uri (Pathlike or str or None): The URI of the LynseDB server. It can be either a local path or a remote URL.

               - If it is a remote URL, the client will use the HTTP API.
               - If it is a local path, the client will use the native API.
                    The path refers to the root path of the LynseDB storage.
               - If set to None, the client will use the native API,
                    and the database will be stored in the default root path,
                    when you need to change the default root path,
                    you can set the environment variable LYNSE_DEFAULT_ROOT_PATH or change the config file.

        """
        self._is_remote = uri is not None and (uri.startswith('http://') or uri.startswith('https://'))

        if self._is_remote:
            from spinesUtils.asserts import raise_if
            import httpx

            raise_if(ValueError, not isinstance(uri, (str, Path)),
                     'uri must be a string, Pathlike or None')
            if isinstance(uri, Path):
                uri = uri.as_posix()

            if uri.startswith('http://') or uri.startswith('https://'):
                try:
                    response = httpx.get(uri)
                    if response.status_code != 200:
                        raise ConnectionError(f'Failed to connect to the server at {uri}.')

                    try:
                        rj = response.json()
                        if rj != {'status': 'success', 'message': 'LynseDB HTTP API'}:
                            raise ConnectionError(f'Failed to connect to the server at {uri}.')
                    except Exception as e:
                        print(e)
                        raise ConnectionError(f'Failed to connect to the server at {uri}.')

                except httpx.RequestError:
                    raise ConnectionError(f'Failed to connect to the server at {uri}.')
        else:
            uri = Path(uri) if uri is not None else None

        self._uri = uri

    def create_database(self, database_name: str, drop_if_exists: bool = False):
        """
        Create the database using a lazy mode, where entities are only created when they are actually used.

        Parameters:
            database_name (str): The name of the database to create.
            drop_if_exists (bool): Whether to drop the database if it already exists.
                If set to True, the existing database will be immediately deleted before creating a new one.

        Returns:
            None
        """
        from .api.http_api.client_api import raise_error_response
        from .configs.config import config
        import httpx

        if not self._is_remote:
            from .api.native_api.database_manager import DatabaseManager

            db_manager = DatabaseManager(root_path=self._uri or config.LYNSE_DEFAULT_ROOT_PATH)
            db_manager.register(db_name=database_name)

            if drop_if_exists:
                _InstanceDistributor(root_path=self._uri or config.LYNSE_DEFAULT_ROOT_PATH,
                                     database_name=database_name).drop_database()
        else:
            try:
                rj = httpx.post(f'{self._uri}/create_database', json={'database_name': database_name,
                                                                      'drop_if_exists': drop_if_exists})
                if rj.status_code != 200:
                    raise_error_response(rj)
            except httpx.RequestError:
                raise ConnectionError(f'Failed to connect to the server at {self._uri}.')

        return _InstanceDistributor(root_path=self._uri or config.LYNSE_DEFAULT_ROOT_PATH, database_name=database_name)

    def get_database(self, database_name: str):
        """
        Get an existing database.

        Parameters:
            database_name (str): The name of the database to get.

        Returns:
            LynseDB: (LocalClient, HTTPClient):
                The appropriate LynseDB client instance based on the root path.
                If the root path is a local path, return a LocalClient instance,
                otherwise return a HTTPClient instance.
        """
        from spinesUtils.asserts import raise_if

        from .api.http_api.client_api import raise_error_response
        from .configs.config import config
        import httpx

        if not self._is_remote:
            from .api.native_api.database_manager import DatabaseManager

            db_manager = DatabaseManager(root_path=self._uri or config.LYNSE_DEFAULT_ROOT_PATH)
            databases = db_manager.list_database()
            raise_if(ValueError, database_name not in databases, f'{database_name} does not exist.')
        else:
            try:
                rj = httpx.get(f'{self._uri}/list_databases')
                if rj.status_code != 200:
                    raise_error_response(rj)

                databases = rj.json()['params']['databases']
                raise_if(ValueError, database_name not in databases, f'{database_name} does not exist.')
            except httpx.RequestError:
                raise ConnectionError(f'Failed to connect to the server at {self._uri}.')

        return _InstanceDistributor(root_path=self._uri or config.LYNSE_DEFAULT_ROOT_PATH, database_name=database_name)

    def list_databases(self):
        """
        List all databases.

        Returns:
            List: A list of all databases.
        """
        from .api.http_api.client_api import raise_error_response
        import httpx

        if not self._is_remote:
            from .api.native_api.database_manager import DatabaseManager
            from .configs.config import config

            db_manager = DatabaseManager(root_path=self._uri or config.LYNSE_DEFAULT_ROOT_PATH)
            return db_manager.list_database()
        else:
            try:
                rj = httpx.get(f'{self._uri}/list_databases')
                if rj.status_code != 200:
                    raise_error_response(rj)

                return rj.json()['params']['databases']
            except httpx.RequestError:
                raise ConnectionError(f'Failed to connect to the server at {self._uri}.')

    def drop_database(self, database_name: str):
        """
        Delete a database.

        Parameters:
            database_name (str): The name of the database to delete.

        Returns:
            None
        """
        from .api.http_api.client_api import raise_error_response
        import httpx

        if not self._is_remote:
            from .api.native_api.database_manager import DatabaseManager
            from .configs.config import config

            db_manager = DatabaseManager(root_path=self._uri or config.LYNSE_DEFAULT_ROOT_PATH)
            databases = db_manager.list_database()
            if database_name not in databases:
                return
            db_manager.delete(database_name)
        else:
            try:
                rj = httpx.post(f'{self._uri}/delete_database', json={'database_name': database_name})
                if rj.status_code != 200:
                    raise_error_response(rj)
            except httpx.RequestError:
                raise ConnectionError(f'Failed to connect to the server at {self._uri}.')

    def __repr__(self):
        from .configs.config import config
        return f'{self.__class__.__name__}(uri={self._uri or "DefaultRootPath"})'

    def __str__(self):
        return self.__repr__()

__init__(uri=None)

Initialize the LynseDB client.

Parameters:

Name Type Description Default
uri Pathlike or str or None

The URI of the LynseDB server. It can be either a local path or a remote URL.

  • If it is a remote URL, the client will use the HTTP API.
  • If it is a local path, the client will use the native API. The path refers to the root path of the LynseDB storage.
  • If set to None, the client will use the native API, and the database will be stored in the default root path, when you need to change the default root path, you can set the environment variable LYNSE_DEFAULT_ROOT_PATH or change the config file.
 None
Source code in lynse/__init__.py 
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
def __init__(self, uri: Union[str, None, Path] = None):
    """
    Initialize the LynseDB client.

    Parameters:
        uri (Pathlike or str or None): The URI of the LynseDB server. It can be either a local path or a remote URL.

           - If it is a remote URL, the client will use the HTTP API.
           - If it is a local path, the client will use the native API.
                The path refers to the root path of the LynseDB storage.
           - If set to None, the client will use the native API,
                and the database will be stored in the default root path,
                when you need to change the default root path,
                you can set the environment variable LYNSE_DEFAULT_ROOT_PATH or change the config file.

    """
    self._is_remote = uri is not None and (uri.startswith('http://') or uri.startswith('https://'))

    if self._is_remote:
        from spinesUtils.asserts import raise_if
        import httpx

        raise_if(ValueError, not isinstance(uri, (str, Path)),
                 'uri must be a string, Pathlike or None')
        if isinstance(uri, Path):
            uri = uri.as_posix()

        if uri.startswith('http://') or uri.startswith('https://'):
            try:
                response = httpx.get(uri)
                if response.status_code != 200:
                    raise ConnectionError(f'Failed to connect to the server at {uri}.')

                try:
                    rj = response.json()
                    if rj != {'status': 'success', 'message': 'LynseDB HTTP API'}:
                        raise ConnectionError(f'Failed to connect to the server at {uri}.')
                except Exception as e:
                    print(e)
                    raise ConnectionError(f'Failed to connect to the server at {uri}.')

            except httpx.RequestError:
                raise ConnectionError(f'Failed to connect to the server at {uri}.')
    else:
        uri = Path(uri) if uri is not None else None

    self._uri = uri

create_database(database_name, drop_if_exists=False)

Create the database using a lazy mode, where entities are only created when they are actually used.

Parameters:

Name Type Description Default
database_name str

The name of the database to create.

 required
drop_if_exists bool

Whether to drop the database if it already exists. If set to True, the existing database will be immediately deleted before creating a new one.

 False

Returns:

Type Description

None
Source code in lynse/__init__.py 
 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
def create_database(self, database_name: str, drop_if_exists: bool = False):
    """
    Create the database using a lazy mode, where entities are only created when they are actually used.

    Parameters:
        database_name (str): The name of the database to create.
        drop_if_exists (bool): Whether to drop the database if it already exists.
            If set to True, the existing database will be immediately deleted before creating a new one.

    Returns:
        None
    """
    from .api.http_api.client_api import raise_error_response
    from .configs.config import config
    import httpx

    if not self._is_remote:
        from .api.native_api.database_manager import DatabaseManager

        db_manager = DatabaseManager(root_path=self._uri or config.LYNSE_DEFAULT_ROOT_PATH)
        db_manager.register(db_name=database_name)

        if drop_if_exists:
            _InstanceDistributor(root_path=self._uri or config.LYNSE_DEFAULT_ROOT_PATH,
                                 database_name=database_name).drop_database()
    else:
        try:
            rj = httpx.post(f'{self._uri}/create_database', json={'database_name': database_name,
                                                                  'drop_if_exists': drop_if_exists})
            if rj.status_code != 200:
                raise_error_response(rj)
        except httpx.RequestError:
            raise ConnectionError(f'Failed to connect to the server at {self._uri}.')

    return _InstanceDistributor(root_path=self._uri or config.LYNSE_DEFAULT_ROOT_PATH, database_name=database_name)

drop_database(database_name)

Delete a database.

Parameters:

Name Type Description Default
database_name str

The name of the database to delete.

 required

Returns:

Type Description

None
Source code in lynse/__init__.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def drop_database(self, database_name: str):
    """
    Delete a database.

    Parameters:
        database_name (str): The name of the database to delete.

    Returns:
        None
    """
    from .api.http_api.client_api import raise_error_response
    import httpx

    if not self._is_remote:
        from .api.native_api.database_manager import DatabaseManager
        from .configs.config import config

        db_manager = DatabaseManager(root_path=self._uri or config.LYNSE_DEFAULT_ROOT_PATH)
        databases = db_manager.list_database()
        if database_name not in databases:
            return
        db_manager.delete(database_name)
    else:
        try:
            rj = httpx.post(f'{self._uri}/delete_database', json={'database_name': database_name})
            if rj.status_code != 200:
                raise_error_response(rj)
        except httpx.RequestError:
            raise ConnectionError(f'Failed to connect to the server at {self._uri}.')

get_database(database_name)

Get an existing database.

Parameters:

Name Type Description Default
database_name str

The name of the database to get.

 required

Returns:

Name Type Description
LynseDB

(LocalClient, HTTPClient): The appropriate LynseDB client instance based on the root path. If the root path is a local path, return a LocalClient instance, otherwise return a HTTPClient instance.
Source code in lynse/__init__.py 
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
167
168
169
170
def get_database(self, database_name: str):
    """
    Get an existing database.

    Parameters:
        database_name (str): The name of the database to get.

    Returns:
        LynseDB: (LocalClient, HTTPClient):
            The appropriate LynseDB client instance based on the root path.
            If the root path is a local path, return a LocalClient instance,
            otherwise return a HTTPClient instance.
    """
    from spinesUtils.asserts import raise_if

    from .api.http_api.client_api import raise_error_response
    from .configs.config import config
    import httpx

    if not self._is_remote:
        from .api.native_api.database_manager import DatabaseManager

        db_manager = DatabaseManager(root_path=self._uri or config.LYNSE_DEFAULT_ROOT_PATH)
        databases = db_manager.list_database()
        raise_if(ValueError, database_name not in databases, f'{database_name} does not exist.')
    else:
        try:
            rj = httpx.get(f'{self._uri}/list_databases')
            if rj.status_code != 200:
                raise_error_response(rj)

            databases = rj.json()['params']['databases']
            raise_if(ValueError, database_name not in databases, f'{database_name} does not exist.')
        except httpx.RequestError:
            raise ConnectionError(f'Failed to connect to the server at {self._uri}.')

    return _InstanceDistributor(root_path=self._uri or config.LYNSE_DEFAULT_ROOT_PATH, database_name=database_name)

list_databases()

List all databases.

Returns:

Name Type Description
List

A list of all databases.
Source code in lynse/__init__.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def list_databases(self):
    """
    List all databases.

    Returns:
        List: A list of all databases.
    """
    from .api.http_api.client_api import raise_error_response
    import httpx

    if not self._is_remote:
        from .api.native_api.database_manager import DatabaseManager
        from .configs.config import config

        db_manager = DatabaseManager(root_path=self._uri or config.LYNSE_DEFAULT_ROOT_PATH)
        return db_manager.list_database()
    else:
        try:
            rj = httpx.get(f'{self._uri}/list_databases')
            if rj.status_code != 200:
                raise_error_response(rj)

            return rj.json()['params']['databases']
        except httpx.RequestError:
            raise ConnectionError(f'Failed to connect to the server at {self._uri}.')