plantdb_client

PlantDB Client ModuleLink

A client library for interacting with the PlantDB API, providing a streamlined interface for managing plant-related data including scans, filesets, and associated metadata.

Key FeaturesLink

Scan Management: Create new scans and manage scan metadata
Fileset Operations: Create and manage collections of files associated with scans
File Handling: Upload and manage individual files within filesets
Metadata Management: Comprehensive CRUD operations for scan, fileset, and file metadata
RESTful Interface: Implements standard HTTP methods for API communication

Usage ExamplesLink

>>> # Start a test REST API server first:
>>> # $ fsdb_rest_api --test
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> # Create a new scan
>>> scan_id = client.create_scan(
...     name="Plant Sample 001",
...     description="Arabidopsis specimen under controlled conditions"
... )
>>> # Create a fileset for the scan
>>> fileset_id = client.create_fileset(
...     scan_id=scan_id,
...     fileset_id="RGB Images",
...     description="Top view RGB images"
... )

PlantDBClient Link

PlantDBClient(base_url, prefix=None, api_token=None)

Client for interacting with the PlantDB REST API.

This class provides methods to interact with a PlantDB REST API, allowing operations on scans, filesets, and files. It handles authentication, error processing, and provides a consistent interface for all API endpoints.

Parameters:

Name	Type	Description	Default
`base_url` Link	`str`	The base URL of the PlantDB REST API.	required
`prefix` Link	`str`	The URL prefix used by the PlantDB REST API.	`None`

Attributes:

Name	Type	Description
`base_url`	`str`	The base URL of the PlantDB REST API.
`_session`	`Session`	HTTP session that maintains cookies and connection pooling.
`_access_token`	`str`	The JSON Web Token to authenticate with the PlantDB REST API.
`_refresh_token`	`str`	The refresh token to obtain new access tokens.
`_api_token`	`str`	The long-lived API token granting permission to specific datasets.
`_username`	`str`	The login username.
`logger`	`Logger`	The logger to use.

Notes

This client automatically handles HTTP errors and extracts meaningful error messages from the API responses. All methods will raise appropriate exceptions with descriptive messages when API requests fail.

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Use the client against the server
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> scans = client.list_scans()
>>> print(scans)
['virtual_plant', 'real_plant_analyzed', 'real_plant', 'virtual_plant_analyzed', 'arabidopsis000']
>>> client.login('admin', 'admin')
>>> print(client.base_url)
http://localhost:5000
>>> print(client._access_token)
eyJhbG...BsovwQ
>>> # Get an API token for a set of dataset (matching the 'dataset_*'):
>>> api_token = client.create_api_token(3600, {'dataset_*': ('read', 'write', 'create')})
>>> # Create a new client with the API token:
>>> client2 = PlantDBClient(plantdb_url('localhost', port=5000), api_token=api_token)
>>> resp = client2.create_scan('dataset_A', {'description': "Test dataset"})
>>> print(resp['id'])  # id of the created dataset
dataset_A
>>> resp = client2.create_scan('dataset_B', {'description': "Test dataset"})
>>> print(resp['id'])  # id of the created dataset
dataset_B
>>> # Finally, stop the server
>>> server.stop()

Initialize the PlantDBClient with a base URL.

Source code in plantdb/client/plantdb_client.py

def __init__(self, base_url, prefix=None, api_token=None):
    """Initialize the PlantDBClient with a base URL."""
    if prefix is None:
        prefix = api_prefix()
    self.base_url = f"{base_url}{prefix}"

    self._access_token = None
    self._refresh_token = None
    self._api_token = api_token
    self._username = None

    # Dedicated session for access‑token requests
    self._session = requests.Session()
    if self._api_token:
        # Validate provided API token:
        url = join_url(self.base_url, api_endpoints.token_validation())
        response = self._session.request("POST", url, headers={"Authorization": f"Bearer {self._api_token}"})
        if response.ok:
            self._session.headers.update({"Authorization": f"Bearer {self._api_token}"})
        else:
            raise ValueError(f"Invalid API token: {self._api_token}")

    self.logger = get_logger(__class__.__name__)

create_api_token Link

create_api_token(token_exp, dataset_permissions)

Creates an API token with a specified expiration time and dataset permissions.

This method generates an API token that is tied to the permissions of specific datasets. The permissions for each dataset must be provided, and they will be validated against the defined Permission type. This allows fine-grained control over dataset access via the generated token. The token can only be created successfully if the server accepts the provided data.

Parameters:

Name	Type	Description	Default
`token_exp` Link	`int`	The expiration time for the API token in seconds.	required
`dataset_permissions` Link	`dict of str to tuple[Permission \| str, ...] or Permission or str`	A dictionary where each key is a dataset name (unix globbing possible), and the corresponding value is the permission(s) for that dataset. Permissions can be a single `Permission` instance, a string, or a tuple of `Permission` instances or strings.	required

Returns:

Type	Description
`str or None`	Returns the generated API token as a string if successful. Returns None if the token creation fails due to server error or other issues.

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> from plantdb.commons.auth.models import Permission
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> # Use it to log in as 'admin'
>>> client.login('admin', 'admin')
>>> # Create an API token with 'WRITE' permission to create a new scan dataset 'new_scan'
>>> api_token = client.create_api_token(3600, {'new_scan': ('read', 'write', 'create')})
>>> # Create a new scan dataset using the API token (with a new client initialized with the API token):
>>> client2 = PlantDBClient(plantdb_url('localhost', port=5000), api_token=api_token)
>>> client2.create_scan('new_scan', {'description': "Test API token dataset"}, use_api_token=True)
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def create_api_token(
        self,
        token_exp: int,
        dataset_permissions: dict[str, tuple[Permission | str, ...] | Permission | str]
) -> str | None:
    """
    Creates an API token with a specified expiration time and dataset permissions.

    This method generates an API token that is tied to the permissions of specific
    datasets. The permissions for each dataset must be provided, and they will be
    validated against the defined `Permission` type. This allows fine-grained
    control over dataset access via the generated token. The token can only be
    created successfully if the server accepts the provided data.

    Parameters
    ----------
    token_exp : int
        The expiration time for the API token in seconds.
    dataset_permissions : dict of str to tuple[Permission | str, ...] or Permission or str
        A dictionary where each key is a dataset name (unix globbing possible),
        and the corresponding value is the permission(s) for that dataset.
        Permissions can be a single `Permission` instance, a string, or a
        tuple of `Permission` instances or strings.

    Returns
    -------
    str or None
        Returns the generated API token as a string if successful. Returns None
        if the token creation fails due to server error or other issues.

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> from plantdb.commons.auth.models import Permission
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> # Use it to log in as 'admin'
    >>> client.login('admin', 'admin')
    >>> # Create an API token with 'WRITE' permission to create a new scan dataset 'new_scan'
    >>> api_token = client.create_api_token(3600, {'new_scan': ('read', 'write', 'create')})
    >>> # Create a new scan dataset using the API token (with a new client initialized with the API token):
    >>> client2 = PlantDBClient(plantdb_url('localhost', port=5000), api_token=api_token)
    >>> client2.create_scan('new_scan', {'description': "Test API token dataset"}, use_api_token=True)
    >>> # Finally, stop the server
    >>> server.stop()
    """
    url = join_url(self.base_url, api_endpoints.create_api_token())

    # Validate dataset permissions
    datasets = {}
    for dataset, permissions in dataset_permissions.items():
        if not isinstance(permissions, (set, list, tuple)):
            permissions = (permissions,)
        datasets[dataset] = []
        for permission in permissions:
            if isinstance(permission, str) and permission not in Permission:
                raise ValueError(f"Invalid permission: {permission}. "
                                 f"Must be one of {[p.value for p in Permission]}.")
            datasets[dataset].append(str(permission))

    data = {
        "token_exp": token_exp,
        "datasets": datasets,
    }
    try:
        response = self._request("POST", url, json=data)
        if response.ok:
            self._api_token = response.json().get('api_token')
            return self._api_token
        else:
            error_msg = response.json().get('message', 'Unknown server error.')
            self.logger.error(f"Failed to create API token: {error_msg}")
            return None
    except RequestException as e:
        self.logger.error(f"Failed to create API token: {e}")
        return None

create_file Link

create_file(file_data, file_id, ext, scan_id, fileset_id, metadata=None)

Create a new file in a fileset and upload its data.

Parameters:

Name	Type	Description	Default
`file_data` Link	`str, pathlib.Path, or BytesIO`	Path to the file to upload or BytesIO object containing file data	required
`file_id` Link	`str`	The ID of the file in the database	required
`ext` Link	`str`	File extension (must be one of the valid extensions)	required
`scan_id` Link	`str`	The ID of the scan containing the fileset	required
`fileset_id` Link	`str`	The ID of the fileset to create the file in	required
`metadata` Link	`dict`	Additional metadata for the file	`None`

Returns:

Type	Description
`dict`	Server response containing creation confirmation message

Raises:

Type	Description
`RequestException`	If the request fails
`ValueError`	If required parameters are missing or invalid

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> # Log in as admin to get sufficient rights
>>> client.login('admin', 'admin')
>>> # Example 1 - Existing YAML file path as string
>>> metadata = {'description': 'Test document', 'author': 'John Doe'}
>>> dummy_data = {'name': 'Test Plant', 'species': 'Arabidopsis thaliana'}
>>> with tempfile.NamedTemporaryFile(suffix='.yaml', mode='w', delete=False) as f: temp_file_name = f.name; yaml.dump(dummy_data, f)
>>> response = client.create_file(temp_file_name, file_id='new_file',ext='yaml',scan_id='real_plant',fileset_id='images',metadata=metadata)
>>> print(response)
{'message': "File 'new_file.yaml' created and written successfully in fileset 'images'.", 'id': 'new_file'}
>>> # Example 2 - RGB Image with BytesIO
>>> import numpy as np
>>> from PIL import Image
>>> from io import BytesIO
>>> # Generate random RGB data (values from 0-255)
>>> rgb_data = np.random.randint(0, 256, (200, 150, 3), dtype=np.uint8)
>>> # Create PIL Image from NumPy array
>>> img = Image.fromarray(rgb_data, 'RGB')
>>> # Save image to BytesIO object
>>> image_data = BytesIO()
>>> img.save(image_data, format='PNG')
>>> image_data.seek(0)  # Move to the beginning of the BytesIO object
>>> metadata = {'description': 'Random RGB test image', 'author': 'John Doe'}
>>> response = client.create_file(image_data, file_id='random_image', ext='png', scan_id='real_plant', fileset_id='images', metadata=metadata)
>>> print(response)
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def create_file(self, file_data, file_id, ext, scan_id, fileset_id, metadata=None):
    """Create a new file in a fileset and upload its data.

    Parameters
    ----------
    file_data : str, pathlib.Path, or BytesIO
        Path to the file to upload or BytesIO object containing file data
    file_id : str
        The ID of the file in the database
    ext : str
        File extension (must be one of the valid extensions)
    scan_id : str
        The ID of the scan containing the fileset
    fileset_id : str
        The ID of the fileset to create the file in
    metadata : dict, optional
        Additional metadata for the file

    Returns
    -------
    dict
        Server response containing creation confirmation message

    Raises
    ------
    requests.exceptions.RequestException
        If the request fails
    ValueError
        If required parameters are missing or invalid

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> # Log in as admin to get sufficient rights
    >>> client.login('admin', 'admin')
    >>> # Example 1 - Existing YAML file path as string
    >>> metadata = {'description': 'Test document', 'author': 'John Doe'}
    >>> dummy_data = {'name': 'Test Plant', 'species': 'Arabidopsis thaliana'}
    >>> with tempfile.NamedTemporaryFile(suffix='.yaml', mode='w', delete=False) as f: temp_file_name = f.name; yaml.dump(dummy_data, f)
    >>> response = client.create_file(temp_file_name, file_id='new_file',ext='yaml',scan_id='real_plant',fileset_id='images',metadata=metadata)
    >>> print(response)
    {'message': "File 'new_file.yaml' created and written successfully in fileset 'images'.", 'id': 'new_file'}
    >>> # Example 2 - RGB Image with BytesIO
    >>> import numpy as np
    >>> from PIL import Image
    >>> from io import BytesIO
    >>> # Generate random RGB data (values from 0-255)
    >>> rgb_data = np.random.randint(0, 256, (200, 150, 3), dtype=np.uint8)
    >>> # Create PIL Image from NumPy array
    >>> img = Image.fromarray(rgb_data, 'RGB')
    >>> # Save image to BytesIO object
    >>> image_data = BytesIO()
    >>> img.save(image_data, format='PNG')
    >>> image_data.seek(0)  # Move to the beginning of the BytesIO object
    >>> metadata = {'description': 'Random RGB test image', 'author': 'John Doe'}
    >>> response = client.create_file(image_data, file_id='random_image', ext='png', scan_id='real_plant', fileset_id='images', metadata=metadata)
    >>> print(response)
    >>> # Finally, stop the server
    >>> server.stop()
    """
    import os
    import json
    from io import BytesIO
    from pathlib import Path

    url = join_url(self.base_url, api_endpoints.file(scan_id, fileset_id, file_id))

    # Prepare data
    ext = ext.lstrip('.').lower()  # Remove the leading dot if present
    data = {'ext': ext}

    # Add metadata if provided
    if metadata:
        if isinstance(metadata, dict):
            data['metadata'] = json.dumps(metadata)
        elif isinstance(metadata, str):
            data['metadata'] = metadata
        else:
            raise TypeError("Invalid metadata type. Must be a dictionary or string.")

    # Prepare file data based on the type of file_data
    if isinstance(file_data, BytesIO):
        # If it's already a BytesIO object, use it directly
        filename = f"{file_id}.{ext}"
        files = {
            'file': (filename, file_data, get_mime_type(ext))
        }
        response = self._request("POST", url, files=files, data=data)
    else:
        # Convert to a Path object if it's a string
        file_path = Path(file_data) if isinstance(file_data, str) else file_data

        # Handle file from a path
        with open(file_path, 'rb') as file_handle:
            filename = os.path.basename(str(file_path))
            files = {
                'file': (filename, file_handle, 'application/octet-stream')
            }
            response = self._request("POST", url, files=files, data=data)

    # Handle HTTP errors with explicit messages
    self._handle_http_errors(response)
    return response.json()

create_fileset Link

create_fileset(fileset_id, scan_id, metadata=None)

Create a new fileset associated with a scan.

Parameters:

Name	Type	Description	Default
`fileset_id` Link	`str`	The ID of the fileset to create	required
`scan_id` Link	`str`	The ID of the scan to associate the fileset with	required
`metadata` Link	`dict`	Additional metadata for the fileset	`None`

Returns:

Type	Description
`dict`	Server response containing a creation confirmation message

Raises:

Type	Description
`RequestException`	If the request fails

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> # Log in as admin to get sufficient rights
>>> client.login('admin', 'admin')
>>> metadata = {'description': 'This is a test fileset'}
>>> response = client.create_fileset('my_fileset', 'real_plant', metadata=metadata)
>>> print(response)
{'message': "Fileset created successfully in 'real_plant'.", 'id': 'my_fileset'}
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def create_fileset(self, fileset_id, scan_id, metadata=None):
    """Create a new fileset associated with a scan.

    Parameters
    ----------
    fileset_id : str
        The ID of the fileset to create
    scan_id : str
        The ID of the scan to associate the fileset with
    metadata : dict, optional
        Additional metadata for the fileset

    Returns
    -------
    dict
        Server response containing a creation confirmation message

    Raises
    ------
    requests.exceptions.RequestException
        If the request fails

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> # Log in as admin to get sufficient rights
    >>> client.login('admin', 'admin')
    >>> metadata = {'description': 'This is a test fileset'}
    >>> response = client.create_fileset('my_fileset', 'real_plant', metadata=metadata)
    >>> print(response)
    {'message': "Fileset created successfully in 'real_plant'.", 'id': 'my_fileset'}
    >>> # Finally, stop the server
    >>> server.stop()
    """
    url = join_url(self.base_url, api_endpoints.fileset(scan_id, fileset_id))
    data = {'fileset_id': fileset_id, 'scan_id': scan_id}
    if metadata:
        data['metadata'] = metadata
    response = self._request("POST", url, json=data)

    # Handle HTTP errors with explicit messages
    self._handle_http_errors(response)
    return response.json()

create_scan Link

create_scan(name, metadata=None)

Create a new scan in the database.

Parameters:

Name	Type	Description	Default
`name` Link	`str`	Name of the scan to create	required
`metadata` Link	`dict`	Additional metadata for the scan	`None`

Returns:

Type	Description
`dict`	Server response containing creation confirmation message

Raises:

Type	Description
`RequestException`	If the request fails

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> # Scan creation requires authentication
>>> response = client.create_scan('test_plant')
WARNING  [PlantDBClient] Client error 401: UNAUTHORIZED
requests.exceptions.RequestException: No authenticated user!
>>> # Log in as admin to get sufficient rights to create scan
>>> client.login('admin', 'admin')
>>> response = client.create_scan('test_plant', metadata={'description': 'Test plant scan'})
>>> print(response['message'])
Scan created successfully
>>> print(response['id'])
test_plant
>>> # Or create an API token with 'WRITE' permission to create a new scan dataset 'new_scan'
>>> print(client.create_api_token(3600, {'new_scan': ('read', 'write', 'create')}))
>>> # Create a new scan dataset using the API token:
>>> client.create_scan('new_scan', {'description': "Test API token dataset"}, use_api_token=True)
{'message': 'Scan created successfully', 'id': 'new_scan'}
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def create_scan(self, name, metadata=None):
    """Create a new scan in the database.

    Parameters
    ----------
    name : str
        Name of the scan to create
    metadata : dict, optional
        Additional metadata for the scan

    Returns
    -------
    dict
        Server response containing creation confirmation message

    Raises
    ------
    requests.exceptions.RequestException
        If the request fails

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> # Scan creation requires authentication
    >>> response = client.create_scan('test_plant')
    WARNING  [PlantDBClient] Client error 401: UNAUTHORIZED
    requests.exceptions.RequestException: No authenticated user!
    >>> # Log in as admin to get sufficient rights to create scan
    >>> client.login('admin', 'admin')
    >>> response = client.create_scan('test_plant', metadata={'description': 'Test plant scan'})
    >>> print(response['message'])
    Scan created successfully
    >>> print(response['id'])
    test_plant
    >>> # Or create an API token with 'WRITE' permission to create a new scan dataset 'new_scan'
    >>> print(client.create_api_token(3600, {'new_scan': ('read', 'write', 'create')}))
    >>> # Create a new scan dataset using the API token:
    >>> client.create_scan('new_scan', {'description': "Test API token dataset"}, use_api_token=True)
    {'message': 'Scan created successfully', 'id': 'new_scan'}
    >>> # Finally, stop the server
    >>> server.stop()
    """
    url = join_url(self.base_url, api_endpoints.scan(name))

    data = {}
    if metadata:
        data['metadata'] = metadata

    # Send the request:
    response = self._request("POST", url, json=data)
    # Handle HTTP errors with explicit messages
    self._handle_http_errors(response)

    return response.json()

create_user Link

create_user(username, password, fullname)

Create a new user in the PlantDB API.

Parameters:

Name	Type	Description	Default
`username` Link	`str`	New username to create.	required
`password` Link	`str`	Password for authentication.	required
`fullname` Link	`str`	The full name of the user.	required

Returns:

Type	Description
`bool`	`True` if user creation is successful, `False` otherwise

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> # Use it to log in as 'admin'
>>> _ = client.login('admin', 'admin')
>>> # Create a new user
>>> _ = client.create_user('batman', 'JokerInArkham', 'Bruce Wayne')
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def create_user(self, username: str, password: str, fullname: str) -> bool:
    """Create a new user in the PlantDB API.

    Parameters
    ----------
    username : str
        New username to create.
    password : str
        Password for authentication.
    fullname : str
        The full name of the user.

    Returns
    -------
    bool
        ``True`` if user creation is successful, ``False`` otherwise

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> # Use it to log in as 'admin'
    >>> _ = client.login('admin', 'admin')
    >>> # Create a new user
    >>> _ = client.create_user('batman', 'JokerInArkham', 'Bruce Wayne')
    >>> # Finally, stop the server
    >>> server.stop()
    """
    url = join_url(self.base_url, api_endpoints.create_user())
    data = {
        'username': username,
        'password': password,
        'fullname': fullname,
    }

    try:
        # create_user usually requires admin, use _request_with_refresh
        response = self._request("POST", url, json=data)
        if response.ok:
            return True
        else:
            error_msg = response.json().get('message', 'Unknown server error.')
            self.logger.error(f"Failed to create user: {error_msg}")
            return False

    except RequestException as e:
        self.logger.error(f"User registration request failed: {e}")
        return False

get_file_metadata Link

get_file_metadata(scan_id, fileset_id, file_id, key=None)

Retrieve metadata for a specified file.

Parameters:

Name	Type	Description	Default
`scan_id` Link	`str`	The ID of the scan containing the fileset	required
`fileset_id` Link	`str`	The ID of the fileset containing the file	required
`file_id` Link	`str`	The ID of the file	required
`key` Link	`str`	If provided, returns only the value for this specific metadata key	`None`

Returns:

Type	Description
`dict`	Server response containing the metadata or specific key value

Raises:

Type	Description
`RequestException`	If the request fails

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> # Get all metadata
>>> metadata = client.get_file_metadata('test_plant', 'images', 'image_001')
>>> print(metadata)
{'metadata': {'description': 'Test file'}}
>>> # Get a specific metadata key
>>> value = client.get_file_metadata('test_plant', 'images', 'image_001', key='description')
>>> print(value)
{'metadata': 'Test file'}
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def get_file_metadata(self, scan_id, fileset_id, file_id, key=None):
    """Retrieve metadata for a specified file.

    Parameters
    ----------
    scan_id : str
        The ID of the scan containing the fileset
    fileset_id : str
        The ID of the fileset containing the file
    file_id : str
        The ID of the file
    key : str, optional
        If provided, returns only the value for this specific metadata key

    Returns
    -------
    dict
        Server response containing the metadata or specific key value

    Raises
    ------
    requests.exceptions.RequestException
        If the request fails

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> # Get all metadata
    >>> metadata = client.get_file_metadata('test_plant', 'images', 'image_001')
    >>> print(metadata)
    {'metadata': {'description': 'Test file'}}
    >>> # Get a specific metadata key
    >>> value = client.get_file_metadata('test_plant', 'images', 'image_001', key='description')
    >>> print(value)
    {'metadata': 'Test file'}
    >>> # Finally, stop the server
    >>> server.stop()
    """
    url = join_url(self.base_url, api_endpoints.file_metadata(scan_id, fileset_id, file_id))
    params = {'key': key} if key else None
    response = self._request("GET", url, params=params)

    # Handle HTTP errors with explicit messages
    self._handle_http_errors(response)
    return response.json()

get_fileset_metadata Link

get_fileset_metadata(scan_id, fileset_id, key=None)

Retrieve metadata for a specified fileset.

Parameters:

Name	Type	Description	Default
`scan_id` Link	`str`	The ID of the scan containing the fileset	required
`fileset_id` Link	`str`	The ID of the fileset	required
`key` Link	`str`	If provided, returns only the value for this specific metadata key	`None`

Returns:

Type	Description
`dict`	Server response containing the metadata or specific key value

Raises:

Type	Description
`RequestException`	If the request fails

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> # Get all metadata
>>> metadata = client.get_fileset_metadata('real_plant', 'images')
>>> print(metadata['metadata']['channels'])
['rgb']
>>> # Get a specific metadata key
>>> value = client.get_fileset_metadata('real_plant', 'images', key='channels')
>>> print(value['metadata'])
['rgb']
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def get_fileset_metadata(self, scan_id, fileset_id, key=None):
    """Retrieve metadata for a specified fileset.

    Parameters
    ----------
    scan_id : str
        The ID of the scan containing the fileset
    fileset_id : str
        The ID of the fileset
    key : str, optional
        If provided, returns only the value for this specific metadata key

    Returns
    -------
    dict
        Server response containing the metadata or specific key value

    Raises
    ------
    requests.exceptions.RequestException
        If the request fails

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> # Get all metadata
    >>> metadata = client.get_fileset_metadata('real_plant', 'images')
    >>> print(metadata['metadata']['channels'])
    ['rgb']
    >>> # Get a specific metadata key
    >>> value = client.get_fileset_metadata('real_plant', 'images', key='channels')
    >>> print(value['metadata'])
    ['rgb']
    >>> # Finally, stop the server
    >>> server.stop()
    """
    url = join_url(self.base_url, api_endpoints.fileset_metadata(scan_id, fileset_id))
    params = {'key': key} if key else None
    response = self._request("GET", url, params=params)

    # Handle HTTP errors with explicit messages
    self._handle_http_errors(response)
    return response.json()

get_scan_metadata Link

get_scan_metadata(scan_id, key=None)

Retrieve metadata for a specified scan.

Parameters:

Name	Type	Description	Default
`scan_id` Link	`str`	The ID of the scan	required
`key` Link	`str`	If provided, returns only the value for this specific metadata key	`None`

Returns:

Type	Description
`dict`	Server response containing the metadata or specific key value

Raises:

Type	Description
`RequestException`	If the request fails

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> # Get all metadata
>>> metadata = client.get_scan_metadata('real_plant')
>>> print(metadata['metadata']['owner'])
guest
>>> # Get a specific metadata key
>>> value = client.get_scan_metadata('real_plant', key='owner')
>>> print(value['metadata'])
guest
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def get_scan_metadata(self, scan_id, key=None):
    """Retrieve metadata for a specified scan.

    Parameters
    ----------
    scan_id : str
        The ID of the scan
    key : str, optional
        If provided, returns only the value for this specific metadata key

    Returns
    -------
    dict
        Server response containing the metadata or specific key value

    Raises
    ------
    requests.exceptions.RequestException
        If the request fails

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> # Get all metadata
    >>> metadata = client.get_scan_metadata('real_plant')
    >>> print(metadata['metadata']['owner'])
    guest
    >>> # Get a specific metadata key
    >>> value = client.get_scan_metadata('real_plant', key='owner')
    >>> print(value['metadata'])
    guest
    >>> # Finally, stop the server
    >>> server.stop()
    """
    url = join_url(self.base_url, api_endpoints.scan_metadata(scan_id))
    params = {'key': key} if key else None
    response = self._request("GET", url, params=params)

    # Handle HTTP errors with explicit messages
    self._handle_http_errors(response)
    return response.json()

list_fileset_files Link

list_fileset_files(scan_id, fileset_id, query=None, fuzzy=False)

List all files in a specified fileset.

Parameters:

Name	Type	Description	Default
`scan_id` Link	`str`	The ID of the scan containing the fileset	required
`fileset_id` Link	`str`	The ID of the fileset	required
`query` Link	`str`	Query string to filter files	`None`
`fuzzy` Link	`bool`	Whether to use fuzzy matching for the query	`False`

Returns:

Type	Description
`dict`	Server response containing the list of files

Raises:

Type	Description
`RequestException`	If the request fails

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> response = client.fileset_files_list('real_plant','images')
>>> print(response)
{'files': ['00000_rgb', '00001_rgb', '00002_rgb', ...]}
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def list_fileset_files(self, scan_id, fileset_id, query=None, fuzzy=False):
    """List all files in a specified fileset.

    Parameters
    ----------
    scan_id : str
        The ID of the scan containing the fileset
    fileset_id : str
        The ID of the fileset
    query : str, optional
        Query string to filter files
    fuzzy : bool, optional
        Whether to use fuzzy matching for the query

    Returns
    -------
    dict
        Server response containing the list of files

    Raises
    ------
    requests.exceptions.RequestException
        If the request fails

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> response = client.fileset_files_list('real_plant','images')
    >>> print(response)
    {'files': ['00000_rgb', '00001_rgb', '00002_rgb', ...]}
    >>> # Finally, stop the server
    >>> server.stop()
    """
    url = join_url(self.base_url, api_endpoints.fileset_files_list(scan_id, fileset_id))
    params = {}
    if query is not None:
        params['query'] = query
    if fuzzy:
        params['fuzzy'] = fuzzy
    response = self._request("GET", url, params=params)

    # Handle HTTP errors with explicit messages
    self._handle_http_errors(response)
    return response.json()

list_scan_filesets Link

list_scan_filesets(scan_id, query=None, fuzzy=False)

List all filesets in a specified scan.

Parameters:

Name	Type	Description	Default
`scan_id` Link	`str`	The ID of the scan	required
`query` Link	`str`	Query string to filter filesets	`None`
`fuzzy` Link	`bool`	Whether to use fuzzy matching for the query	`False`

Returns:

Type	Description
`dict`	Server response containing the list of fileset IDs

Raises:

Type	Description
`RequestException`	If the request fails

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> response = client.list_scan_filesets('real_plant')
>>> print(response)
{'filesets': ['images']}
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def list_scan_filesets(self, scan_id, query=None, fuzzy=False):
    """List all filesets in a specified scan.

    Parameters
    ----------
    scan_id : str
        The ID of the scan
    query : str, optional
        Query string to filter filesets
    fuzzy : bool, optional
        Whether to use fuzzy matching for the query

    Returns
    -------
    dict
        Server response containing the list of fileset IDs

    Raises
    ------
    requests.exceptions.RequestException
        If the request fails

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> response = client.list_scan_filesets('real_plant')
    >>> print(response)
    {'filesets': ['images']}
    >>> # Finally, stop the server
    >>> server.stop()
    """
    url = join_url(self.base_url, api_endpoints.scan_filesets_list(scan_id))
    params = {}
    if query is not None:
        params['query'] = query
    if fuzzy:
        params['fuzzy'] = fuzzy
    response = self._request("GET", url, params=params)

    # Handle HTTP errors with explicit messages
    self._handle_http_errors(response)
    return response.json()

list_scans Link

list_scans(query=None, fuzzy=False)

List all scans in the database.

Parameters:

Name	Type	Description	Default
`query` Link	`str`	Query string to filter scans	`None`
`fuzzy` Link	`bool`	Whether to use fuzzy matching for the query (default: False)	`False`

Returns:

Type	Description
`dict`	Server response containing the list of scan IDs

Raises:

Type	Description
`RequestException`	If the request fails

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> response = client.list_scans()
>>> print(sorted(response))
['arabidopsis000', 'real_plant', 'real_plant_analyzed', 'virtual_plant', 'virtual_plant_analyzed']
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def list_scans(self, query=None, fuzzy=False):
    """List all scans in the database.

    Parameters
    ----------
    query : str, optional
        Query string to filter scans
    fuzzy : bool, optional
        Whether to use fuzzy matching for the query (default: False)

    Returns
    -------
    dict
        Server response containing the list of scan IDs

    Raises
    ------
    requests.exceptions.RequestException
        If the request fails

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> response = client.list_scans()
    >>> print(sorted(response))
    ['arabidopsis000', 'real_plant', 'real_plant_analyzed', 'virtual_plant', 'virtual_plant_analyzed']
    >>> # Finally, stop the server
    >>> server.stop()
    """
    url = join_url(self.base_url, api_endpoints.scans())
    params = {}
    if query is not None:
        params['query'] = query
    if fuzzy:
        params['fuzzy'] = fuzzy
    response = self._request('GET', url, params=params)

    # Handle HTTP errors with explicit messages
    self._handle_http_errors(response)
    return response.json()

list_scans_info Link

list_scans_info(query=None, fuzzy=False)

Retrieve detailed scan information dictionaries from the ScansTable resource.

Parameters:

Name	Type	Description	Default
`query` Link	`dict`	A dictionary that will be JSON‑encoded and sent as the `filterQuery` URL parameter. Use the same structure accepted by the server, e.g. `{"object": {"species": "Arabidopsis.*"}}`.	`None`
`fuzzy` Link	`bool`	When `True` the server performs fuzzy matching (default `False`).	`False`

Returns:

Type	Description
`list[dict]`	A list where each entry is a dictionary containing the scan’s `metadata`, `tasks`, `files` and other information as defined by `ScansTable`.

Raises:

Type	Description
`RequestException`	If the request fails or the server returns an error status.

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> response = (client.list_scans_info())
>>> print(len(response))
5
>>> print(response[0]['id'])
real_plant_analyzed
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def list_scans_info(self, query=None, fuzzy=False):
    """Retrieve detailed scan information dictionaries from the ScansTable resource.

    Parameters
    ----------
    query : dict, optional
        A dictionary that will be JSON‑encoded and sent as the ``filterQuery`` URL
        parameter.  Use the same structure accepted by the server, _e.g._
        ``{"object": {"species": "Arabidopsis.*"}}``.
    fuzzy : bool, optional
        When ``True`` the server performs fuzzy matching (default ``False``).

    Returns
    -------
    list[dict]
        A list where each entry is a dictionary containing the scan’s
        ``metadata``, ``tasks``, ``files`` and other information as defined by `ScansTable`.

    Raises
    ------
    requests.exceptions.RequestException
        If the request fails or the server returns an error status.

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> response = (client.list_scans_info())
    >>> print(len(response))
    5
    >>> print(response[0]['id'])
    real_plant_analyzed
    >>> # Finally, stop the server
    >>> server.stop()
    """
    # Build the URL for the “scans info” endpoint - the server side class is ScansTable
    url = join_url(self.base_url, api_endpoints.scans_info())

    # Prepare query parameters exactly as the REST API expects
    params = {}
    if query is not None:
        # The API expects a JSON string in the ``filterQuery`` parameter
        params["filterQuery"] = json.dumps(query)
    if fuzzy:
        params["fuzzy"] = fuzzy

    # Perform the request; token refresh is handled automatically
    response = self._request("GET", url, params=params)

    # Turn HTTP errors into readable exceptions
    self._handle_http_errors(response)

    # Return the parsed JSON payload (list of dicts)
    return response.json()

login Link

login(username, password)

Authenticate the user with the PlantDB API.

Parameters:

Name	Type	Description	Default
`username` Link	`str`	Username for authentication.	required
`password` Link	`str`	Password for authentication.	required

Returns:

Type	Description
`bool`	`True` if login successful, `False` otherwise

Notes

Using this method with an existing API token will revoke it upon successful login.

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> # Use it to log in as 'admin'
>>> client.login('admin', 'admin')
>>> print(client._access_token)  # print the 'admin' access token
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def login(self, username: str, password: str) -> bool:
    """Authenticate the user with the PlantDB API.

    Parameters
    ----------
    username : str
        Username for authentication.
    password : str
        Password for authentication.

    Returns
    -------
    bool
        ``True`` if login successful, ``False`` otherwise

    Notes
    -----
    Using this method with an existing API token will revoke it upon successful login.

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> # Use it to log in as 'admin'
    >>> client.login('admin', 'admin')
    >>> print(client._access_token)  # print the 'admin' access token
    >>> # Finally, stop the server
    >>> server.stop()
    """
    url = join_url(self.base_url, api_endpoints.login())
    data = {
        'username': username,
        'password': password
    }

    try:
        # Use _session.request directly for login to avoid using expired tokens in headers
        response = self._session.request("POST", url, json=data)
        if response.ok:
            result = response.json()
            self._access_token = result.get('access_token')
            self._refresh_token = result.get('refresh_token')
            self._username = username
            if self._api_token:
                self.logger.warning(f"Access token will replace API token in request authentication!")
                self._api_token = None
            # Add the JWT to the header
            self._session.headers.update({'Authorization': f'Bearer {self._access_token}'})
            return True
        else:
            error_msg = response.json().get('message', 'Login failed')
            self.logger.error(f"Login failed: {error_msg}")
            return False

    except RequestException as e:
        self.logger.error(f"Login request failed: {e}")
        return False

logout Link

logout()

Logout user from the PlantDB API.

Returns:

Type	Description
`bool`	`True` if logout successful, `False` otherwise.

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> # Use it to log in as 'admin'
>>> client.login('admin', 'admin')
>>> # Then log out
>>> client.logout()
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def logout(self) -> bool:
    """Logout user from the PlantDB API.

    Returns
    -------
    bool
        ``True`` if logout successful, ``False`` otherwise.

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> # Use it to log in as 'admin'
    >>> client.login('admin', 'admin')
    >>> # Then log out
    >>> client.logout()
    >>> # Finally, stop the server
    >>> server.stop()
    """
    url = join_url(self.base_url, api_endpoints.logout())
    try:
        # Use _request_with_refresh for logout as it requires authentication
        response = self._request("POST", url)
        if response.ok:
            self._username = None
            self._access_token = None
            self._refresh_token = None
            # Remove the Authorization with the JWT from the header
            if 'Authorization' in self._session.headers:
                self._session.headers.pop('Authorization')
            return True
        return False
    except Exception:
        return False

refresh Link

refresh(scan_id=None)

Refresh the database.

Parameters:

Name	Type	Description	Default
`scan_id` Link	`str \| None`	The optional id of the scan to refresh, else refresh the whole database.	`None`

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> client.refresh()
True
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def refresh(self, scan_id: str | None = None) -> bool:
    """Refresh the database.

    Parameters
    ----------
    scan_id : str | None
        The optional id of the scan to refresh, else refresh the whole database.

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> client.refresh()
    True
    >>> # Finally, stop the server
    >>> server.stop()
    """
    url = join_url(self.base_url, api_endpoints.refresh(scan_id))
    try:
        response = self._request("GET", url)
        if response.ok:
            return True
        return False
    except Exception:
        return False

refresh_token Link

refresh_token()

Refresh the JSON Web Token.

Uses the stored refresh token to obtain a new access/refresh token pair.

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> # Use it to log in as 'admin'
>>> client.login('admin', 'admin')
>>> old_access_token = client._access_token
>>> client.refresh_token()
True
>>> client._access_token == old_access_token
False
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def refresh_token(self) -> bool:
    """Refresh the JSON Web Token.

    Uses the stored refresh token to obtain a new access/refresh token pair.

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> # Use it to log in as 'admin'
    >>> client.login('admin', 'admin')
    >>> old_access_token = client._access_token
    >>> client.refresh_token()
    True
    >>> client._access_token == old_access_token
    False
    >>> # Finally, stop the server
    >>> server.stop()
    """
    if not self._refresh_token:
        self.logger.error("No refresh token available")
        return False

    url = join_url(self.base_url, api_endpoints.token_refresh())
    data = {'refresh_token': self._refresh_token}
    try:
        # Use _session.request directly to avoid infinite recursion with _request_with_refresh
        response = self._session.request("POST", url, json=data)
        if response.ok:
            result = response.json()
            self._access_token = result.get('access_token')
            self._refresh_token = result.get('refresh_token')
            # Update the header with the new access token
            self._session.headers.update({'Authorization': f'Bearer {self._access_token}'})
            return True
        else:
            error_msg = response.json().get('message', 'Token refresh failed')
            self.logger.error(f"Token refresh failed: {error_msg}")
            self._access_token = None
            self._refresh_token = None
            self._username = None
            return False
    except Exception as e:
        self.logger.error(f"Token refresh request failed: {e}")
        return False

update_file_metadata Link

update_file_metadata(scan_id, fileset_id, file_id, metadata, replace=False)

Update metadata for a specified file.

Parameters:

Name	Type	Description	Default
`scan_id` Link	`str`	The ID of the scan containing the fileset	required
`fileset_id` Link	`str`	The ID of the fileset containing the file	required
`file_id` Link	`str`	The ID of the file	required
`metadata` Link	`dict`	The metadata to update/set	required
`replace` Link	`bool`	If `True`, replaces entire metadata. If `False` (default), updates only specified keys.	`False`

Returns:

Type	Description
`dict`	Server response containing the updated metadata

Raises:

Type	Description
`RequestException`	If the request fails

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> # Log in as admin to get sufficient rights
>>> client.login('admin', 'admin')
>>> # Update metadata
>>> new_metadata = {'description': 'Updated description'}
>>> response = client.update_file_metadata('test_plant', 'images', 'image_001', new_metadata)
>>> print(response)
{'metadata': {'description': 'Updated description'}}
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def update_file_metadata(self, scan_id, fileset_id, file_id, metadata, replace=False):
    """Update metadata for a specified file.

    Parameters
    ----------
    scan_id : str
        The ID of the scan containing the fileset
    fileset_id : str
        The ID of the fileset containing the file
    file_id : str
        The ID of the file
    metadata : dict
        The metadata to update/set
    replace : bool, optional
        If ``True``, replaces entire metadata. If ``False`` (default),
        updates only specified keys.

    Returns
    -------
    dict
        Server response containing the updated metadata

    Raises
    ------
    requests.exceptions.RequestException
        If the request fails

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> # Log in as admin to get sufficient rights
    >>> client.login('admin', 'admin')
    >>> # Update metadata
    >>> new_metadata = {'description': 'Updated description'}
    >>> response = client.update_file_metadata('test_plant', 'images', 'image_001', new_metadata)
    >>> print(response)
    {'metadata': {'description': 'Updated description'}}
    >>> # Finally, stop the server
    >>> server.stop()
    """
    url = join_url(self.base_url, api_endpoints.file_metadata(scan_id, fileset_id, file_id))
    data = {'metadata': metadata, 'replace': replace}
    response = self._request("POST", url, json=data)

    # Handle HTTP errors with explicit messages
    self._handle_http_errors(response)
    return response.json()

update_fileset_metadata Link

update_fileset_metadata(scan_id, fileset_id, metadata, replace=False)

Update metadata for a specified fileset.

Parameters:

Name	Type	Description	Default
`scan_id` Link	`str`	The ID of the scan containing the fileset	required
`fileset_id` Link	`str`	The ID of the fileset	required
`metadata` Link	`dict`	The metadata to update/set	required
`replace` Link	`bool`	If `True`, replaces entire metadata. If `False` (default), updates only specified keys.	`False`

Returns:

Type	Description
`dict`	Server response containing the updated metadata

Raises:

Type	Description
`RequestException`	If the request fails

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> # Log in as admin to get sufficient rights
>>> client.login('admin', 'admin')
>>> # Replace the fileset metadata
>>> new_metadata = {'description': 'Updated fileset description', 'owner': 'John Doe'}
>>> response = client.update_fileset_metadata('real_plant', 'images', new_metadata, replace=True)
>>> print(response['metadata'])
{'metadata': {'description': 'Updated fileset description', 'author': 'John Doe'}}
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def update_fileset_metadata(self, scan_id, fileset_id, metadata, replace=False):
    """Update metadata for a specified fileset.

    Parameters
    ----------
    scan_id : str
        The ID of the scan containing the fileset
    fileset_id : str
        The ID of the fileset
    metadata : dict
        The metadata to update/set
    replace : bool, optional
        If ``True``, replaces entire metadata. If ``False`` (default),
        updates only specified keys.

    Returns
    -------
    dict
        Server response containing the updated metadata

    Raises
    ------
    requests.exceptions.RequestException
        If the request fails

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> # Log in as admin to get sufficient rights
    >>> client.login('admin', 'admin')
    >>> # Replace the fileset metadata
    >>> new_metadata = {'description': 'Updated fileset description', 'owner': 'John Doe'}
    >>> response = client.update_fileset_metadata('real_plant', 'images', new_metadata, replace=True)
    >>> print(response['metadata'])
    {'metadata': {'description': 'Updated fileset description', 'author': 'John Doe'}}
    >>> # Finally, stop the server
    >>> server.stop()
    """
    url = join_url(self.base_url, api_endpoints.fileset_metadata(scan_id, fileset_id))
    data = {'metadata': metadata, 'replace': replace}
    response = self._request("POST", url, json=data)

    # Handle HTTP errors with explicit messages
    self._handle_http_errors(response)
    return response.json()

update_scan_metadata Link

update_scan_metadata(scan_id, metadata, replace=False)

Update metadata for a specified scan.

Parameters:

Name	Type	Description	Default
`scan_id` Link	`str`	The ID of the scan to update metadata for	required
`metadata` Link	`dict`	The metadata to update/set	required
`replace` Link	`bool`	If `True`, replaces entire metadata. If `False` (default), updates only specified keys.	`False`

Returns:

Type	Description
`dict`	Server response containing the updated metadata

Raises:

Type	Description
`RequestException`	If the request fails

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> # Log in as admin to get sufficient rights
>>> client.login('admin', 'admin')
>>> new_metadata = {'description': 'Updated scan description'}
>>> response = client.update_scan_metadata('real_plant', new_metadata)
>>> print(response['metadata']['description'])
Updated scan description
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def update_scan_metadata(self, scan_id, metadata, replace=False):
    """Update metadata for a specified scan.

    Parameters
    ----------
    scan_id : str
        The ID of the scan to update metadata for
    metadata : dict
        The metadata to update/set
    replace : bool, optional
        If ``True``, replaces entire metadata. If ``False`` (default),
        updates only specified keys.

    Returns
    -------
    dict
        Server response containing the updated metadata

    Raises
    ------
    requests.exceptions.RequestException
        If the request fails

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> # Log in as admin to get sufficient rights
    >>> client.login('admin', 'admin')
    >>> new_metadata = {'description': 'Updated scan description'}
    >>> response = client.update_scan_metadata('real_plant', new_metadata)
    >>> print(response['metadata']['description'])
    Updated scan description
    >>> # Finally, stop the server
    >>> server.stop()
    """
    url = join_url(self.base_url, api_endpoints.scan_metadata(scan_id))
    data = {'metadata': metadata, 'replace': replace}
    response = self._request("POST", url, json=data)

    # Handle HTTP errors with explicit messages
    self._handle_http_errors(response)
    return response.json()

validate_token Link

validate_token(token)

Validate an authentication token against the remote service.

This method sends a POST request to the token‑validation endpoint using the supplied token in the Authorization header.

Parameters:

Name	Type	Description	Default
`token` Link	`str`	The bearer token to be validated.	required

Returns:

Type	Description
``True`` if the token is accepted by the server, otherwise ``False``.

Examples:

>>> from plantdb.server.test_rest_api import TestRestApiServer
>>> # Start a test PlantDB REST API server first:
>>> server = TestRestApiServer(test=True, port=5000)
>>> server.start()
>>> # Create a client
>>> from plantdb.client.plantdb_client import PlantDBClient
>>> from plantdb.client.rest_api import plantdb_url
>>> client = PlantDBClient(plantdb_url('localhost', port=5000))
>>> # Use it to log in as 'admin'
>>> client.login('admin', 'admin')
>>> client.validate_token(client._access_token)
True
>>> # Finally, stop the server
>>> server.stop()

Source code in plantdb/client/plantdb_client.py

def validate_token(self, token) -> bool:
    """Validate an authentication token against the remote service.

    This method sends a ``POST`` request to the token‑validation endpoint
    using the supplied ``token`` in the ``Authorization`` header.

    Parameters
    ----------
    token : str
        The bearer token to be validated.

    Returns
    -------
    ``True`` if the token is accepted by the server, otherwise ``False``.

    Examples
    --------
    >>> from plantdb.server.test_rest_api import TestRestApiServer
    >>> # Start a test PlantDB REST API server first:
    >>> server = TestRestApiServer(test=True, port=5000)
    >>> server.start()
    >>> # Create a client
    >>> from plantdb.client.plantdb_client import PlantDBClient
    >>> from plantdb.client.rest_api import plantdb_url
    >>> client = PlantDBClient(plantdb_url('localhost', port=5000))
    >>> # Use it to log in as 'admin'
    >>> client.login('admin', 'admin')
    >>> client.validate_token(client._access_token)
    True
    >>> # Finally, stop the server
    >>> server.stop()
    """
    url = join_url(self.base_url, api_endpoints.token_validation())
    response = self._request("POST", url, headers={"Authorization": f"Bearer {token}"})
    if response.ok:
        resp_username = response.json()['user']['username']
        if not self._username:
            self._username = resp_username
            self.refresh_token()
        if self._username and resp_username != self._username:
            self.logger.warning(f"Given token correspond to a different username")
        return True
    else:
        return False

api_prefix Link

api_prefix(prefix='')

Set the API prefix for all URL generation functions.

Parameters:

Name	Type	Description	Default
`prefix` Link	`str`	The prefix to add to all API URLs, e.g., '/plantdb'. Defaults to empty string.	`''`

Examples:

>>> import os
>>> from plantdb.client.plantdb_client import api_prefix
>>> api_prefix()
''
>>> os.environ['PLANTDB_PREFIX'] = "/plantdb"
>>> api_prefix()
'/plantdb'

Source code in plantdb/client/plantdb_client.py

def api_prefix(prefix=""):
    """Set the API prefix for all URL generation functions.

    Parameters
    ----------
    prefix : str, optional
        The prefix to add to all API URLs, e.g., '/plantdb'. Defaults to empty string.

    Examples
    --------
    >>> import os
    >>> from plantdb.client.plantdb_client import api_prefix
    >>> api_prefix()
    ''
    >>> os.environ['PLANTDB_PREFIX'] = "/plantdb"
    >>> api_prefix()
    '/plantdb'
    """
    if prefix is None or prefix == "":
        prefix = os.getenv("PLANTDB_PREFIX", "")  # Default to no prefix

    prefix = prefix.rstrip('/')  # Remove the trailing slash if present
    os.environ['PLANTDB_PREFIX'] = prefix
    return prefix

get_mime_type Link

get_mime_type(extension)

Determine the MIME type from a file extension.

Parameters:

Name	Type	Description	Default
`extension` Link	`str`	File extension (with or without a leading dot)	required

Returns:

Type	Description
`str`	The MIME type string or 'application/octet-stream' if not found

Source code in plantdb/client/plantdb_client.py

def get_mime_type(extension):
    """Determine the MIME type from a file extension.

    Parameters
    ----------
    extension : str
        File extension (with or without a leading dot)

    Returns
    -------
    str
        The MIME type string or 'application/octet-stream' if not found
    """
    # Ensure the extension starts with a dot
    if not extension.startswith('.'):
        extension = f'.{extension}'

    mime_type, _ = mimetypes.guess_type(f'file{extension}')

    # Return a default for unknown types
    if mime_type is None:
        return 'application/octet-stream'

    return mime_type

plantdb_client

PlantDB Client ModuleLink

Key FeaturesLink

Usage ExamplesLink

PlantDBClient Link

base_url Link

prefix Link

create_api_token Link

token_exp Link

dataset_permissions Link

create_file Link

file_data Link

file_id Link

ext Link

scan_id Link

fileset_id Link

metadata Link

create_fileset Link

fileset_id Link

scan_id Link

metadata Link

create_scan Link

name Link

metadata Link

create_user Link

username Link

password Link

fullname Link

get_file_metadata Link

scan_id Link

fileset_id Link

file_id Link

key Link

get_fileset_metadata Link

scan_id Link

fileset_id Link

key Link

get_scan_metadata Link

scan_id Link

key Link

list_fileset_files Link

scan_id Link

fileset_id Link

query Link

fuzzy Link

list_scan_filesets Link

scan_id Link

query Link

fuzzy Link

list_scans Link

query Link

fuzzy Link

list_scans_info Link

query Link

fuzzy Link

login Link

username Link

password Link

logout Link

refresh Link

scan_id Link

refresh_token Link

update_file_metadata Link

scan_id Link

fileset_id Link

file_id Link

metadata Link

replace Link

update_fileset_metadata Link

scan_id Link

fileset_id Link

metadata Link

replace Link

update_scan_metadata Link

scan_id Link

metadata Link

replace Link

validate_token Link

token Link

api_prefix Link

`base_url` Link

`prefix` Link

`token_exp` Link

`dataset_permissions` Link

`file_data` Link

`file_id` Link

`ext` Link

`scan_id` Link

`fileset_id` Link

`metadata` Link

`fileset_id` Link

`scan_id` Link

`metadata` Link

`name` Link

`metadata` Link

`username` Link

`password` Link

`fullname` Link

`scan_id` Link

`fileset_id` Link

`file_id` Link

`key` Link

`scan_id` Link

`fileset_id` Link

`key` Link

`scan_id` Link

`key` Link

`scan_id` Link

`fileset_id` Link

`query` Link

`fuzzy` Link

`scan_id` Link

`query` Link

`fuzzy` Link

`query` Link

`fuzzy` Link

`query` Link

`fuzzy` Link

`username` Link

`password` Link

`scan_id` Link

`scan_id` Link

`fileset_id` Link

`file_id` Link

`metadata` Link

`replace` Link

`scan_id` Link

`fileset_id` Link

`metadata` Link

`replace` Link

`scan_id` Link

`metadata` Link

`replace` Link

`token` Link

`prefix` Link

`extension` Link