core

File Structure DataBase

This module implements a lightweight, filesystem‑based data store that offers user authentication, role‑based access control, and convenient management of hierarchical scan objects. It is useful for applications that need to organise, share, and version data collections (e.g. scientific scans) without requiring a full‑blown relational database.

Key Features

• Authentication & RBAC: Functions such as require_token, require_authentication and the rbac_manager attribute enforce user permissions and token‑based access.
• Scan management: The FSDB class provides methods to create, list, delete and lock scans (create_scan, list_scans, is_scan_locked), as well as to retrieve scan‑specific metadata.
• Fileset handling: Within each scan, Fileset objects group related files; they support creation, deletion, metadata handling and existence checks.
• File abstraction: The File class offers transparent read/write of raw bytes or structured data, along with per‑file metadata storage.
• Session & lock management: Built‑in session handling and lock utilities prevent concurrent modifications and ensure data integrity.
• Convenient utilities: Helper functions like get_logged_username, use_guest_as_default and requires_permission simplify common workflow steps.

Implementation

Assuming that the FSDB root database directory is dbroot/, there is a Scan with 'myscan_001' as Scan.id and there are some metadata (see below), you should have the following file structure:

    dbroot/                            # base directory of the database
    ├── myscan_001/                    # scan dataset directory, id=`myscan_001`
    │   ├── files.json                 # JSON file referencing the all files for the dataset
    │   ├── images/                    # `Fileset` gathering the 'images'
    │   │   ├── scan_img_01.jpg        # 'image' `File` 01
    │   │   ├── scan_img_02.jpg        # 'image' `File` 02
    │   │   ├── [...]                  #
    │   │   └── scan_img_99.jpg        # 'image' `File` 99
    │   ├── metadata/                  # metadata directory
    │   │   ├── images                 # 'images' metadata directory
    │   │   │   ├── scan_img_01.json   # JSON metadata attached to this file
    │   │   │   ├── scan_img_02.json   #
    │   │   │   [...]                  #
    │   │   │   └── scan_img_99.json   #
    │   │   ├── Task_A/                # optional, only present if metadata attached to one of the outputs from `Task_A`
    │   │   │   └── outfile.json       # optional metadata attached to the output file from `Task_A`
    │   │   └── (metadata.json)        # optional metadata attached to the dataset
    │   ├── task_A/                    # `Fileset` gathering the outputs of `Task_A`
    │   │   └── outfile.ext            # output file from `Task_A`
    │   └── (measures.json)            # optional manual measurements file
    ├── myscan_002/                    # scan dataset directory, id=`myscan_002`
    :
    ├── users.json                     # user registry
    └── MARKER_FILE_NAME               # ROMI DB marker file

The myscan_001/files.json file then contains the following structure:

{
    "filesets": [
        {
            "id": "images",
            "files": [
                {
                    "id": "scan_img_01",
                    "file": "scan_img_01.jpg"
                },
                {
                    "id": "scan_img_02",
                    "file": "scan_img_02.jpg"
                },
                [...]
                {
                    "id": "scan_img_99",
                    "file": "scan_img_99.jpg"
                }
            ]
        }
    ]
}

The metadata of the scan (metadata.json), of the set of 'images' files (<Fileset.id>.json) and of each 'image' files (<File.id>.json) are all stored as JSON files in a separate directory:

myscan_001/metadata/
myscan_001/metadata/metadata.json
myscan_001/metadata/images.json
myscan_001/metadata/images/scan_img_01.json
myscan_001/metadata/images/scan_img_02.json
[...]
myscan_001/metadata/images/scan_img_99.json

Usage Examples

>>> from plantdb.commons.test_database import test_database
>>> # Initialize the database (creates base directory if needed)
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> # Create a new scan named "experiment‑001"
>>> scan = db.create_scan("experiment-001")
>>> # Add a fileset to the scan
>>> fileset = scan.create_fileset("raw-data")
>>> # Store a file inside the fileset
>>> file = fileset.create_file("sensor")
>>> file.write_raw(b"timestamp,value\n0,12.3\n1,13.7", ext="csv")
>>> # Retrieve metadata later
>>> meta = file.get_metadata()
>>> print(meta.get("size"))

FSDB

FSDB(basedir, extra_dirs=['configs'], logger=None, session_manager=None, session_timeout=3600, max_login_attempts=3, lockout_duration=900, max_concurrent_sessions=10, **kwargs)

Bases: DB

Implement a local File System DataBase version of abstract class db.DB.

Implement as a simple local file structure with the following directory structure and marker files: * directory ${FSDB.basedir} as the database root directory; * marker file MARKER_FILE_NAME at the database root directory;

Attributes:

Name	Type	Description
basedir	Path	The absolute path to the base directory hosting the database.
scans	dict[str, Scan]	The dictionary of Scan instances attached to the database, indexed by their identifier.
is_connected	bool	True if the database is connected (locked directory), else False.
logger	Logger	An instance to use for logging. Defaults to the module logger.
session_manager	Union[SingleSessionManager, SessionManager, JWTSessionManager]	The session manager to use for session authentication.
lock_manager	ScanLockManager	Manager for scanning lock operations.
rbac_manager	RBACManager	Manager for role-based access control.

Notes

Requires the marker file MARKER_FILE_NAME at the given basedir.

See Also

plantdb.commons.db.DB plantdb.commons.fsdb.core.MARKER_FILE_NAME

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> # EXAMPLE 1: Use a temporary dummy local database:
>>> db = dummy_db()
>>> print(type(db))
<class 'plantdb.commons.fsdb.core.FSDB'>
>>> print(db.path())
/tmp/romidb_...
>>> # Create a new `Scan`:
>>> new_scan = db.create_scan("007")
>>> print(type(new_scan))
<class 'plantdb.commons.fsdb.core.Scan'>
>>> db.disconnect()  # clean up (delete) the temporary dummy database

>>> # EXAMPLE 2: Use a local database with a :
>>> import os
>>> from plantdb.commons.fsdb.core import FSDB
>>> db = FSDB(os.environ.get('ROMI_DB', "/data/ROMI/DB/"), log_level="DEBUG")
>>> db.connect()
>>> token = db.login('guest', 'guest')
>>> scan_ids = db.list_scans(owner_only=False, token=token)
>>> [scan.id for scan in db.get_scans(token=token)]  # list scan ids found in database
>>> scan = db.get_scans()[1]
>>> [fs.id for fs in scan.get_filesets()]  # list fileset ids found in scan
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Database constructor.

Check the given basedir directory exists and load accessible Scan objects.

Parameters:

Name	Type	Description	Default
basedir	str or Path	The path to the root directory of the database.	required
extra_dirs	list of str	A list of directory names that can reside at the root of the filesystem database without trowing an error. Defaults to ['configs'].	['configs']
logger	Logger	Logger instance to use for logging. Defaults to the module logger.	None
session_manager	SessionManager	The session manager to use for session authentication. Defaults to NoAuthSessionManager if 'no_auth' kwarg is set to `True, elseSingleSessionManager``.	None
session_timeout	int	Session timeout in seconds. Defaults to 3600 (1 hour).	3600
lockout_duration	int	The number of seconds to wait for a lockout before giving up. Defaults to 900 seconds (15 minutes).	900
max_login_attempts	int	The maximum number of attempts to login before locking up. Defaults to 3.	3
max_concurrent_sessions	int	The maximum number of concurrent sessions to login before locking up. Defaults to 10.	10

Other Parameters:

Name	Type	Description
log_level	str	A logging level to set for the logger. Defaults to INFO.
no_auth	bool	A boolean flag to switch between session managers. If True, use NoAuthSessionManager else use SingleSessionManager.

Raises:

Type	Description
NotADirectoryError	If the given basedir is not an existing directory.
NotAnFSDBError	If the MARKER_FILE_NAME is missing from the basedir.

See Also

plantdb.commons.fsdb.core.MARKER_FILE_NAME

Returned by:

• Reference API commons commons test_database
  •  dummy_db
  •  test_database

Used by:

• Reference API
  • client client sync  FSDBSync
  • commons commons
    • cli fsdb_healthcheck  fix_missing_scans_reference
    • fsdb
      • core  Scan
      • fsdb_tools  add_metadata_to_scan
  • server server
    • api
      • assets
        •  Archive
        •  CurveSkeleton
        •  FilePath
        •  Image
        •  Mesh
        •  PointCloud
        •  Sequence
      • auth
        •  CreateApiToken
        •  Login
        •  Logout
        •  Register
        •  TokenRefresh
        •  TokenValidation
      • base
        •  HealthCheck
        •  Home
        •  Refresh
      • file
        •  File
        •  FileMetadata
      • fileset
        •  Fileset
        •  FilesetFiles
        •  FilesetMetadata
      • scan
        •  Scan
        •  ScanFilesets
        •  ScanMetadata
        •  ScansList
        •  ScansTable
      • utils  resource_file
    • services
      • assets  get_scan_path
      • auth
        •  logout_user
        •  register_user
    • webcache
      •  __file_path
      •  __image_cache
      •  __image_cached_path
      •  __mesh_cache
      •  __mesh_cached_path
      •  __pointcloud_cache
      •  __pointcloud_cached_path
      •  __webcache_path
      •  image_path
      •  mesh_path
      •  pointcloud_path

Source code in plantdb/commons/fsdb/core.py

def __init__(self, basedir: Union[str, Path],
             extra_dirs: list[str] = ['configs'],
             logger: Optional[logging.Logger] = None,
             session_manager: SessionManager = None,
             session_timeout: int = 3600, max_login_attempts: int = 3,
             lockout_duration: int = 900, max_concurrent_sessions: int = 10, **kwargs):
    """Database constructor.

    Check the given `` basedir `` directory exists and load accessible ``Scan`` objects.

    Parameters
    ----------
    basedir : str or pathlib.Path
        The path to the root directory of the database.
    extra_dirs : list of str
        A list of directory names that can reside at the root of the filesystem database without trowing an error.
        Defaults to ``['configs']``.
    logger : logging.Logger, optional
        Logger instance to use for logging. Defaults to the module logger.
    session_manager : SessionManager, optional
        The session manager to use for session authentication.
        Defaults to ``NoAuthSessionManager`` if ``'no_auth'`` kwarg is set to `True``, else ``SingleSessionManager``.
    session_timeout : int, optional
        Session timeout in seconds.
        Defaults to 3600 (1 hour).
    lockout_duration : int, optional
        The number of seconds to wait for a lockout before giving up.
        Defaults to 900 seconds (15 minutes).
    max_login_attempts : int, optional
        The maximum number of attempts to login before locking up.
        Defaults to 3.
    max_concurrent_sessions : int, optional
        The maximum number of concurrent sessions to login before locking up.
        Defaults to 10.

    Other Parameters
    ----------------
    log_level : str
        A logging level to set for the logger. Defaults to ``INFO``.
    no_auth : bool
        A boolean flag to switch between session managers.
        If ``True``, use `NoAuthSessionManager` else use `SingleSessionManager`.

    Raises
    ------
    NotADirectoryError
        If the given `basedir` is not an existing directory.
    NotAnFSDBError
        If the `MARKER_FILE_NAME` is missing from the `basedir`.

    See Also
    --------
    plantdb.commons.fsdb.core.MARKER_FILE_NAME
    """
    super().__init__()
    self.logger = logger or get_logger(__class__.__name__,
                                       log_level=kwargs.get('log_level', DEFAULT_LOG_LEVEL))

    basedir = Path(basedir)
    # Check the given path to the root directory of the database is a directory:
    if not basedir.is_dir():
        raise NotADirectoryError(f"Directory {basedir} does not exists!")

    self.basedir = Path(basedir).resolve()
    self.extra_dirs = extra_dirs
    self.scans = {}
    self.is_connected: bool = False

    # Initialize lock manager
    self.lock_manager = LockManager(basedir)

    # Initialize the session manager
    if session_manager is None:
        if kwargs.get('no_auth', False):
            self.session_manager = NoAuthSessionManager(session_timeout=session_timeout)
        else:
            self.session_manager = SingleSessionManager(session_timeout=session_timeout)
    elif isinstance(session_manager,
                    (SingleSessionManager, SessionManager, JWTSessionManager, NoAuthSessionManager)):
        self.session_manager = session_manager
    else:
        raise ValueError(
            f"session_manager must be an instance of 'SingleSessionManager', 'SessionManager', "
            f" 'JWTSessionManager' or 'NoAuthSessionManager', got {type(session_manager)}"
        )

    if isinstance(self.session_manager, NoAuthSessionManager):
        # Create a temporary directory with empty to avoid enabling access to real user & groups data
        manager_dir = Path(tempfile.mkdtemp(prefix='plantdb-'))
    else:
        # Will use/create the real users & groups JSON files in basedir
        manager_dir = basedir

    # Initialize RBAC manager with users & groups files
    users_file = manager_dir / "users.json"
    users_file.touch(exist_ok=True)  # create the file if missing
    groups_file = manager_dir / "groups.json"
    groups_file.touch(exist_ok=True)  # create the file if missing
    self.rbac_manager = RBACManager(users_file, groups_file, max_login_attempts, lockout_duration)

add_user_to_group

add_user_to_group(group_name, user, current_user=None, **kwargs)

Add a user to a group.

Parameters:

Name	Type	Description	Default
group_name	str	Name of the group to add the user to.	required
user	str	Name of the user to add to the group.	required

Other Parameters:

Name	Type	Description
username	str	The username formulating the request.
token	str	A token referring to the username formulating the request.

Returns:

Type	Description
bool	True if the user was successfully added to the group, False otherwise.

Raises:

Type	Description
PermissionError	If no user is authenticated. If the authenticated user lacks permission to modify the group.

Examples:

>>> from plantdb.commons.auth.models import Role
>>> from plantdb.commons.test_database import test_database
>>> db = test_database()
>>> db.connect()
>>> token = db.login('admin', 'admin')
>>> db.create_user('batman', 'Bruce Wayne', 'joker', roles=Role.CONTRIBUTOR)
>>> group_a = db.create_group('groupA', ['batman'], description="The group A.")
>>> print(group_a.users)
{'admin', 'batman'}
>>> db.create_user('hquinn', 'Harley Quinn', 'joker', roles=Role.READER)
>>> db.add_user_to_group('groupA', 'hquinn')
>>> print(group_a.users)
{'hquinn', 'batman', 'admin'}
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
def add_user_to_group(self, group_name, user, current_user=None, **kwargs) -> bool:
    """Add a user to a group.

    Parameters
    ----------
    group_name : str
        Name of the group to add the user to.
    user : str
        Name of the user to add to the group.

    Other Parameters
    ----------------
    username : str
        The username formulating the request.
    token : str
        A token referring to the username formulating the request.

    Returns
    -------
    bool
        ``True`` if the `user` was successfully added to the group, ``False`` otherwise.

    Raises
    ------
    PermissionError
        If no user is authenticated.
        If the authenticated user lacks permission to modify the group.

    Examples
    --------
    >>> from plantdb.commons.auth.models import Role
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database()
    >>> db.connect()
    >>> token = db.login('admin', 'admin')
    >>> db.create_user('batman', 'Bruce Wayne', 'joker', roles=Role.CONTRIBUTOR)
    >>> group_a = db.create_group('groupA', ['batman'], description="The group A.")
    >>> print(group_a.users)
    {'admin', 'batman'}
    >>> db.create_user('hquinn', 'Harley Quinn', 'joker', roles=Role.READER)
    >>> db.add_user_to_group('groupA', 'hquinn')
    >>> print(group_a.users)
    {'hquinn', 'batman', 'admin'}
    >>> db.disconnect()
    """
    if not self.rbac_manager.add_user_to_group(current_user, group_name, user):
        raise PermissionError("Insufficient permissions or operation failed")
    return True

cleanup_scan_locks

cleanup_scan_locks()

Emergency cleanup of all scan locks. Use with caution - only call when you're sure no operations are in progress.

Source code in plantdb/commons/fsdb/core.py

def cleanup_scan_locks(self) -> None:
    """Emergency cleanup of all scan locks.
    Use with caution - only call when you're sure no operations are in progress.
    """
    self.lock_manager.cleanup_all_locks()
    self.logger.warning("All scan locks have been cleaned up")

connect

connect()

Connect the database by loading the scans' dataset.

Source code in plantdb/commons/fsdb/core.py

def connect(self) -> None:
    """Connect the database by loading the scans' dataset."""
    if not _is_fsdb(self.basedir, extra_dirs=self.extra_dirs):
        raise NotAnFSDBError(f"Directory `{self.basedir}` is not a valid path to an FSDB!")
    try:
        # Initialize scan discovery
        self.scans = _load_scans(self)
        self.is_connected = True
        self.logger.info("Successfully connected to the database")
    except Exception as e:
        self.logger.error(f"Failed to connect to database: {e}")
        raise

create_api_token

create_api_token(token_exp, datasets, current_user=None, **kwargs)

Creates an API token for the current user with optional dataset permissions.

This method generates a new API token using the session manager, which must be an instance of JWTSessionManager. The token is valid for a specified expiration duration and optionally includes permissions for datasets. The generated token can be used for authenticated API access.

Parameters:

Name	Type	Description	Default
token_exp	int	The expiration duration of the API token in seconds.	required
datasets	dict[str, list[Permission] | Permission]	A dictionary where the keys are dataset names, and the values are either a tuple of Permission instances or a single Permission instance defining the access levels for each dataset.	required
current_user	User	The user object representing the currently authenticated user. The username from this object is used to associate the API token.	None

Other Parameters:

Name	Type	Description
token	str	The token to use to authenticate against the JWTSessionManager.

Returns:

Type	Description
str	The generated API token.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> from plantdb.commons.auth.session import JWTSessionManager
>>> from plantdb.commons.auth.models import Permission
>>> db = test_database(session_manager=JWTSessionManager())
>>> db.connect()
>>> access_token, refresh_token = db.login('admin', 'admin')
>>> api_token = db.create_api_token(3600, {'real_plant_analyzed': [Permission.WRITE]}, token=access_token)
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
@get_authentication
@require_authentication
def create_api_token(
        self, token_exp: int,
        datasets: dict[str, list[Permission] | Permission],
        current_user: User | None = None, **kwargs
) -> str:
    """
    Creates an API token for the current user with optional dataset permissions.

    This method generates a new API token using the session manager, which must
    be an instance of `JWTSessionManager`. The token is valid for a specified
    expiration duration and optionally includes permissions for datasets. The
    generated token can be used for authenticated API access.

    Parameters
    ----------
    token_exp : int
        The expiration duration of the API token in seconds.
    datasets : dict[str, list[Permission] | Permission]
        A dictionary where the keys are dataset names, and the values are either
        a tuple of `Permission` instances or a single `Permission` instance
        defining the access levels for each dataset.
    current_user : plantdb.commons.fsdb.auth.models.User, optional
        The user object representing the currently authenticated user. The username
        from this object is used to associate the API token.

    Other Parameters
    ----------------
    token : str
        The token to use to authenticate against the ``JWTSessionManager``.

    Returns
    -------
    str
        The generated API token.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> from plantdb.commons.auth.session import JWTSessionManager
    >>> from plantdb.commons.auth.models import Permission
    >>> db = test_database(session_manager=JWTSessionManager())
    >>> db.connect()
    >>> access_token, refresh_token = db.login('admin', 'admin')
    >>> api_token = db.create_api_token(3600, {'real_plant_analyzed': [Permission.WRITE]}, token=access_token)
    >>> db.disconnect()
    """
    if not isinstance(self.session_manager, JWTSessionManager):
        raise RuntimeError("Session manager must be an instance of JWTSessionManager.")

    return self.session_manager.create_api_token(current_user.username, token_exp, datasets)

create_group

create_group(name, users=None, description=None, current_user=None, **kwargs)

Create a new group.

Parameters:

Name	Type	Description	Default
name	str	Unique name for the group.	required
users	set	Initial set of users to add to the group.	None
description	str	An optional description of the group.	None

Other Parameters:

Name	Type	Description
username	str	The username formulating the request.
token	str	A token referring to the username formulating the request.

Returns:

Type	Description
Optional[Group]	The created group object if successful, None otherwise.

Raises:

Type	Description
PermissionError	If no user is authenticated. If the user lacks permission to create groups.
ValueError	If the group already exists.

Examples:

>>> from plantdb.commons.auth.models import Role
>>> from plantdb.commons.test_database import test_database
>>> db = test_database()
>>> db.connect()
>>> token = db.login('admin', 'admin')
>>> db.create_user('batman', 'Bruce Wayne', 'joker', roles=Role.CONTRIBUTOR)
>>> group_a = db.create_group('groupA', ['batman'], description="The group A.")
>>> print(group_a.users)
{'admin', 'batman'}
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
def create_group(self, name, users=None, description=None, current_user=None, **kwargs) -> Group | None:
    """Create a new group.

    Parameters
    ----------
    name : str
        Unique name for the group.
    users : set, optional
        Initial set of users to add to the group.
    description : str, optional
        An optional description of the group.

    Other Parameters
    ----------------
    username : str
        The username formulating the request.
    token : str
        A token referring to the username formulating the request.

    Returns
    -------
    Optional[Group]
        The created group object if successful, ``None`` otherwise.

    Raises
    ------
    PermissionError
        If no user is authenticated.
        If the user lacks permission to create groups.
    ValueError
        If the group already exists.

    Examples
    --------
    >>> from plantdb.commons.auth.models import Role
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database()
    >>> db.connect()
    >>> token = db.login('admin', 'admin')
    >>> db.create_user('batman', 'Bruce Wayne', 'joker', roles=Role.CONTRIBUTOR)
    >>> group_a = db.create_group('groupA', ['batman'], description="The group A.")
    >>> print(group_a.users)
    {'admin', 'batman'}
    >>> db.disconnect()
    """
    if isinstance(users, str):
        users = [users]
    if isinstance(users, Iterable):
        users = set(users)

    return self.rbac_manager.create_group(current_user, name, users, description)

create_scan

create_scan(scan_id, metadata=None, current_user=None, **kwargs)

Create a new Scan instance in the local database.

Parameters:

Name	Type	Description	Default
scan_id	str	The identifier of the scan to create. It should contain only alphanumeric characters, underscores, dashes and dots. It should be non-empty and not longer than 255 characters It should not exist in the local database	required
metadata	dict	A dictionary of metadata to append to the new Scan instance. The key 'owner' will be added/updated using the currently connected user. Default is None.	None

Returns:

Type	Description
Scan	The Scan instance created in the local database.

Raises:

Type	Description
PermissionError	If no user is authenticated. If the user lacks permission to create groups.
ScanExistsError	If the scan_id already exists in the local database.
ValueError	If the given scan_id is invalid.

See Also

plantdb.commons.fsdb.validation._is_valid_id plantdb.commons.fsdb.file_ops._make_scan

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db()
>>> new_scan = db.create_scan('007', metadata={'project': 'GoldenEye'})  # create a new scan dataset
>>> print(new_scan.get_metadata('owner'))  # default user 'admin' for dummy database
admin
>>> print(new_scan.get_metadata('project'))
GoldenEye
>>> scan = db.create_scan('007')  # attempt to create an existing scan dataset
plantdb.commons.fsdb.exceptions.ScanExistsError: Scan id '007' already exists in database '/tmp/ROMI_DB_bx519w11'!
>>> scan = db.create_scan('0/07')  # attempt to create a scan dataset using invalid characters
ValueError: Invalid scan identifier '0/07'!
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
@get_authentication
@require_authentication
@requires_permission(Permission.CREATE, check_scan_access=False)
def create_scan(self, scan_id, metadata=None, current_user=None, **kwargs) -> "Scan":
    """Create a new ``Scan`` instance in the local database.

    Parameters
    ----------
    scan_id : str
        The identifier of the scan to create.
        It should contain only alphanumeric characters, underscores, dashes and dots.
        It should be non-empty and not longer than 255 characters
        It should not exist in the local database
    metadata : dict, optional
        A dictionary of metadata to append to the new ``Scan`` instance.
        The key 'owner' will be added/updated using the currently connected user.
        Default is ``None``.

    Returns
    -------
    plantdb.commons.fsdb.core.Scan
        The ``Scan`` instance created in the local database.

    Raises
    ------
    PermissionError
        If no user is authenticated.
        If the user lacks permission to create groups.
    ScanExistsError
        If the ``scan_id`` already exists in the local database.
    ValueError
        If the given ``scan_id`` is invalid.

    See Also
    --------
    plantdb.commons.fsdb.validation._is_valid_id
    plantdb.commons.fsdb.file_ops._make_scan

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db()
    >>> new_scan = db.create_scan('007', metadata={'project': 'GoldenEye'})  # create a new scan dataset
    >>> print(new_scan.get_metadata('owner'))  # default user 'admin' for dummy database
    admin
    >>> print(new_scan.get_metadata('project'))
    GoldenEye
    >>> scan = db.create_scan('007')  # attempt to create an existing scan dataset
    plantdb.commons.fsdb.exceptions.ScanExistsError: Scan id '007' already exists in database '/tmp/ROMI_DB_bx519w11'!
    >>> scan = db.create_scan('0/07')  # attempt to create a scan dataset using invalid characters
    ValueError: Invalid scan identifier '0/07'!
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    # Verify if the given `fs_id` is valid
    if not _is_valid_id(scan_id):
        raise ValueError(f"Invalid scan identifier '{scan_id}'!")
    # Verify if the given `scan_id` already exists in the local database
    if self.scan_exists(scan_id):
        raise ScanExistsError(self, scan_id)

    # Try to determine the timelaps ID from the metadata, if any
    tl_id = None
    if metadata and isinstance(metadata, dict):
        tl_meta = metadata.get("timelapse")
        if isinstance(tl_meta, dict):
            tl_id = tl_meta.get("id")

    if tl_id:
        # Validate the timelapse (ID and JSON file):
        if not _is_valid_id(tl_id):
            raise ValueError(f"Invalid timelapse identifier '{tl_id}'!")
        tl_path = _timelapse_path(self, tl_id)
        tl_marker = _timelapse_marker(tl_path)
        if not tl_marker.is_file():
            raise TimeLapseNotFoundError(self, tl_id)
    else:
        # Standalone scan: reject collision with timelapse container of same name
        if (self.path() / scan_id / TIMELAPSE_MARKER_FILE_NAME).is_file():
            raise ScanExistsError(self, scan_id)

    # Prepare metadata with ownership
    if metadata is None:
        metadata = {}

    # Set basic metadata
    metadata['owner'] = current_user.username
    now = iso_date_now()
    metadata['created'] = now  # creation timestamp
    metadata['last_modified'] = now  # modification timestamp
    metadata['created_by'] = current_user.fullname

    # Validate sharing groups if specified
    if 'sharing' in metadata:
        sharing_groups = metadata['sharing']
        if isinstance(sharing_groups, str):
            sharing_groups = [sharing_groups]
        if not isinstance(sharing_groups, list):
            raise ValueError("Sharing field must be a list of group names")
        if not self.rbac_manager.validate_sharing_groups(sharing_groups):
            raise ValueError("One or more sharing groups do not exist")

    # Validate the MIAPPE biological block if provided (optional on scans)
    validate_biological_metadata(metadata)

    # Use exclusive lock for scan creation
    self.logger.debug(f"Creating a scan '{scan_id}' as user '{current_user.username}'...")
    with self.lock_manager.acquire_lock(scan_id, LockType.EXCLUSIVE, current_user.username, LockLevel.SCAN):
        # Initialize scan object
        scan = Scan(self, scan_id)  # Initialize a new Scan instance
        # Set metadata dictionary before making scan so _scan_path resolves nested path if tl_id is present
        _set_metadata(scan.metadata, metadata, None)
        scan_path = _make_scan(scan)  # Create directory structure
        _store_scan_metadata(scan)
        scan.store()  # store the new scan in the local database
        self.scans[scan_id] = scan  # Update scans dictionary with the new one

        if tl_id:
            self._register_scan_in_timelapse(tl_id, scan_id, current_user.username)

    self.logger.debug(f"Done creating scan.")
    return scan

create_timelapse

create_timelapse(tl_id, metadata=None, current_user=None, **kwargs)

Create a new timelapse container in the local database.

Parameters:

Name	Type	Description	Default
tl_id	str	The identifier of the timelapse to create.	required
metadata	dict	A dictionary of metadata for the timelapse container.	None

Returns:

Type	Description
TimeLapse	The created TimeLapse object.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> db.list_timelapses()
['mytl_001']
>>> tl.path().exists()
True
>>> tl.list_scan()
[]
>>> scan = tl.create_scan('mytl_001_01')
>>> scan.path().exists()
True
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
@get_authentication
@require_authentication
@requires_permission(Permission.CREATE, check_scan_access=False)
def create_timelapse(self, tl_id: str, metadata: dict = None, current_user=None, **kwargs) -> "TimeLapse":
    """Create a new timelapse container in the local database.

    Parameters
    ----------
    tl_id : str
        The identifier of the timelapse to create.
    metadata : dict, optional
        A dictionary of metadata for the timelapse container.

    Returns
    -------
    plantdb.commons.fsdb.core.TimeLapse
        The created TimeLapse object.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> tl = db.create_timelapse('mytl_001')
    >>> db.list_timelapses()
    ['mytl_001']
    >>> tl.path().exists()
    True
    >>> tl.list_scan()
    []
    >>> scan = tl.create_scan('mytl_001_01')
    >>> scan.path().exists()
    True
    >>> db.disconnect()
    """
    if not _is_valid_id(tl_id):
        raise ValueError(f"Invalid timelapse identifier '{tl_id}'!")
    if self.scan_exists(tl_id) or (self.path() / tl_id).exists():
        raise TimeLapseExistsError(self, tl_id)

    tl_path = _timelapse_path(self, tl_id)
    marker_path = _timelapse_marker(tl_path)
    now = iso_date_now()
    owner = current_user.username if current_user else "guest"
    meta = dict(metadata) if metadata else {}
    meta["last_modified"] = now

    with self.lock_manager.acquire_lock(tl_id, LockType.EXCLUSIVE, owner, LockLevel.SCAN):
        tl_path.mkdir(parents=True, exist_ok=True)
        tl_data = {
            "id": tl_id,
            "created_at": now,
            "owner": owner,
            "scans": [],
            "metadata": meta,
        }
        with marker_path.open("w") as f:
            json.dump(tl_data, f, indent=4)

    return TimeLapse(self, tl_id, metadata=meta, created_at=now, owner=owner, scans=[])

create_user

create_user(new_username, fullname, password, roles=None, current_user=None, **kwargs)

Create a new user with the specified details.

Parameters:

Name	Type	Description	Default
new_username	str	The unique username for the new user.	required
fullname	str	The full name of the new user.	required
password	str	The password for the new user.	required
roles	list[str]	A list of roles to assign to the new user. Default is None.	None
current_user	User | TokenUser | None	The current user, based on logged status or provided API token. Default is None.	None

Raises:

Type	Description
PermissionError	If no user is authenticated. If the current_user lacks permission to create new users.

See Also

RBACManager.create_user : Method used to actually create the user.

Examples:

>>> from plantdb.commons.auth.models import Role
>>> from plantdb.commons.test_database import test_database
>>> db = test_database()
>>> db.connect()
>>> token = db.login('admin', 'admin')
>>> db.create_user('batman', 'Bruce Wayne', 'joker', roles=Role.CONTRIBUTOR)
INFO     [RBACManager] User 'admin' is creating new user 'batman' with roles=<Role.CONTRIBUTOR: 'contributor'>.
INFO     [UserManager] Welcome Bruce Wayne, please log in...'
>>> db.logout()  # log out from 'admin' session
(True, 'admin')
>>> token = db.login('batman', 'joker')
>>> print(len(token))
43
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
def create_user(self, new_username, fullname, password, roles=None,
                current_user: User | TokenUser | None = None, **kwargs) -> None:
    """Create a new user with the specified details.

    Parameters
    ----------
    new_username : str
        The unique username for the new user.
    fullname : str
        The full name of the new user.
    password : str
        The password for the new user.
    roles : list[str], optional
        A list of roles to assign to the new user. Default is ``None``.
    current_user : User | TokenUser | None
        The current user, based on logged status or provided API token.
        Default is ``None``.

    Raises
    ------
    PermissionError
        If no user is authenticated.
        If the `current_user` lacks permission to create new users.

    See Also
    --------
    RBACManager.create_user : Method used to actually create the user.

    Examples
    --------
    >>> from plantdb.commons.auth.models import Role
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database()
    >>> db.connect()
    >>> token = db.login('admin', 'admin')
    >>> db.create_user('batman', 'Bruce Wayne', 'joker', roles=Role.CONTRIBUTOR)
    INFO     [RBACManager] User 'admin' is creating new user 'batman' with roles=<Role.CONTRIBUTOR: 'contributor'>.
    INFO     [UserManager] Welcome Bruce Wayne, please log in...'
    >>> db.logout()  # log out from 'admin' session
    (True, 'admin')
    >>> token = db.login('batman', 'joker')
    >>> print(len(token))
    43
    >>> db.disconnect()
    """
    _ = self.rbac_manager.create_user(current_user, new_username, fullname, password, roles)
    return

delete_group

delete_group(group_name, current_user=None, **kwargs)

Delete a group.

Parameters:

Name	Type	Description	Default
group_name	str	Name of the group to delete	required

Other Parameters:

Name	Type	Description
username	str	The username formulating the request.
token	str	A token referring to the username formulating the request.

Returns:

Type	Description
bool	True if the group was deleted successfully

Raises:

Type	Description
PermissionError	If no user is authenticated. If the authenticated user lacks permission to delete this group.

Examples:

>>> from plantdb.commons.auth.models import Role
>>> from plantdb.commons.test_database import test_database
>>> db = test_database()
>>> db.connect()
>>> token = db.login('admin', 'admin')
>>> db.create_user('batman', 'Bruce Wayne', 'joker', roles=Role.CONTRIBUTOR)
>>> group_a = db.create_group('groupA', ['batman'], description="The group A.")
>>> print(group_a.users)
{'admin', 'batman'}
>>> db.logout()
>>> token = db.login('batman', 'joker')
ERROR    [RBACManager] Insufficient permission to delete group 'groupA' by user 'batman!
PermissionError: Insufficient permissions or group 'groupA' not found
>>> db.delete_group('groupA')
>>> db.logout()
>>> token = db.login('admin', 'admin')
>>> db.delete_group('groupA')
WARNING  [RBACManager] Deleting group 'groupA' by user 'admin'!
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
def delete_group(self, group_name, current_user=None, **kwargs) -> bool:
    """Delete a group.

    Parameters
    ----------
    group_name : str
        Name of the group to delete

    Other Parameters
    ----------------
    username : str
        The username formulating the request.
    token : str
        A token referring to the username formulating the request.

    Returns
    -------
    bool
        True if the group was deleted successfully

    Raises
    ------
    PermissionError
        If no user is authenticated.
        If the authenticated user lacks permission to delete this group.

    Examples
    --------
    >>> from plantdb.commons.auth.models import Role
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database()
    >>> db.connect()
    >>> token = db.login('admin', 'admin')
    >>> db.create_user('batman', 'Bruce Wayne', 'joker', roles=Role.CONTRIBUTOR)
    >>> group_a = db.create_group('groupA', ['batman'], description="The group A.")
    >>> print(group_a.users)
    {'admin', 'batman'}
    >>> db.logout()
    >>> token = db.login('batman', 'joker')
    ERROR    [RBACManager] Insufficient permission to delete group 'groupA' by user 'batman!
    PermissionError: Insufficient permissions or group 'groupA' not found
    >>> db.delete_group('groupA')
    >>> db.logout()
    >>> token = db.login('admin', 'admin')
    >>> db.delete_group('groupA')
    WARNING  [RBACManager] Deleting group 'groupA' by user 'admin'!
    >>> db.disconnect()
    """
    if not self.rbac_manager.delete_group(current_user, group_name):
        raise PermissionError(f"Insufficient permissions or group '{group_name}' not found")
    return True

delete_scan

delete_scan(scan_id, current_user=None, **kwargs)

Delete an existing Scan from the local database.

Parameters:

Name	Type	Description	Default
scan_id	str	The name of the scan to delete from the local database.	required

Returns:

Type	Description
bool	A boolean value indicating whether the scan was successfully deleted.

Raises:

Type	Description
PermissionError	If no user is authenticated. If the user lacks permission to create groups.
ValueError	If the scan_id does not exist in the local database.
IOError	If the scan is locked by another user.

See Also

plantdb.commons.fsdb.file_ops._delete_scan

Examples:

>>> from plantdb.commons.fsdb.core import FSDB
>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db()
>>> new_scan = db.create_scan('007')
>>> print(new_scan)
<plantdb.commons.fsdb.core.Scan object at 0x7f0730b1e390>
>>> db.delete_scan('007')
>>> scan = db.get_scan('007')
>>> print(scan)
None
>>> db.delete_scan('008')
OSError: Invalid id
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
@get_authentication
@require_authentication
@requires_permission(Permission.DELETE, check_scan_access=True)
def delete_scan(self, scan_id, current_user=None, **kwargs) -> bool:
    """Delete an existing `Scan` from the local database.

    Parameters
    ----------
    scan_id : str
        The name of the scan to delete from the local database.

    Returns
    -------
    bool
        A boolean value indicating whether the scan was successfully deleted.

    Raises
    ------
    PermissionError
        If no user is authenticated.
        If the user lacks permission to create groups.
    ValueError
        If the ``scan_id`` does not exist in the local database.
    IOError
        If the scan is locked by another user.

    See Also
    --------
    plantdb.commons.fsdb.file_ops._delete_scan

    Examples
    --------
    >>> from plantdb.commons.fsdb.core import FSDB
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db()
    >>> new_scan = db.create_scan('007')
    >>> print(new_scan)
    <plantdb.commons.fsdb.core.Scan object at 0x7f0730b1e390>
    >>> db.delete_scan('007')
    >>> scan = db.get_scan('007')
    >>> print(scan)
    None
    >>> db.delete_scan('008')
    OSError: Invalid id
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    # Use exclusive lock for scan deletion
    self.logger.debug(f"Deleting scan '{scan_id}' as '{current_user.username}' user...")
    with self.lock_manager.acquire_lock(scan_id, LockType.EXCLUSIVE, current_user.username, LockLevel.SCAN):
        # Get the Scan instance from the database
        scan = self.scans[scan_id]
        tl_id = None
        if isinstance(scan.metadata, dict):
            tl_meta = scan.metadata.get("timelapse")
            if isinstance(tl_meta, dict):
                tl_id = tl_meta.get("id")
        _delete_scan(scan)  # delete the scan directory
        self.scans.pop(scan_id)  # remove the scan from the scan list
        if tl_id:
            self._unregister_scan_from_timelapse(tl_id, scan_id)

    self.logger.debug(f"Done deleting scan.")
    return True

delete_timelapse

delete_timelapse(tl_id, recursive=False, current_user=None, **kwargs)

Delete a timelapse container from the local database.

Parameters:

Name	Type	Description	Default
tl_id	str	Identifier of the timelapse to delete.	required
recursive	bool	Whether to recursively delete member scans. Defaults to False.	False

Returns:

Type	Description
bool	True on successful deletion.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> db.list_timelapses()
['mytl_001']
>>> db.delete_timelapse('mytl_001')
>>> db.list_timelapses()
[]
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
@get_authentication
@require_authentication
@requires_permission(Permission.DELETE, check_scan_access=False)
def delete_timelapse(self, tl_id: str, recursive: bool = False, current_user=None, **kwargs) -> bool:
    """Delete a timelapse container from the local database.

    Parameters
    ----------
    tl_id : str
        Identifier of the timelapse to delete.
    recursive : bool
        Whether to recursively delete member scans. Defaults to False.

    Returns
    -------
    bool
        True on successful deletion.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> tl = db.create_timelapse('mytl_001')
    >>> db.list_timelapses()
    ['mytl_001']
    >>> db.delete_timelapse('mytl_001')
    >>> db.list_timelapses()
    []
    >>> db.disconnect()
    """
    tl_path = _timelapse_path(self, tl_id)
    if not _timelapse_marker(tl_path).is_file() and not tl_path.is_dir():
        raise TimeLapseNotFoundError(self, tl_id)

    with self.lock_manager.acquire_lock(tl_id, LockType.EXCLUSIVE, current_user.username, LockLevel.SCAN):
        _delete_timelapse(self, tl_id, recursive=recursive)

    return True

disconnect

disconnect()

Disconnect from the database.

This method disconnects from the database, if currently connected, by erasing all scans (from memory) and reseting the connection status.

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db()
>>> print(db.is_connected)
True
>>> print(db.path().exists())
True
>>> db.disconnect()  # clean up (delete) the temporary dummy database
>>> print(db.is_connected)
False
>>> print(db.path().exists())
False

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
def disconnect(self) -> None:
    """Disconnect from the database.

    This method disconnects from the database, if currently connected, by erasing all scans (from memory)
    and reseting the connection status.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db()
    >>> print(db.is_connected)
    True
    >>> print(db.path().exists())
    True
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    >>> print(db.is_connected)
    False
    >>> print(db.path().exists())
    False
    """
    for s_id, scan in self.scans.items():
        scan._erase()
    self.scans = {}
    self.is_connected = False

    # If this FSDB instance was created by dummy_db, clean up the temp directory
    if getattr(self, "_is_dummy", False):
        import shutil
        try:
            shutil.rmtree(self.basedir)
            self.logger.info(f"Removed temporary database directory {self.basedir}")
        except Exception as e:
            self.logger.warning(f"Failed to remove temporary directory {self.basedir}: {e}")
    return

get_guest_user

get_guest_user()

Retrieve the guest user information from the RBAC manager.

Returns:

Type	Description
User	The User object corresponding to the guest username.

Notes

This method interacts with the underlying RBAC manager to fetch guest user information.

See Also

rbac_manager.get_guest_user : The underlying method used by this function.

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db()  # automatic login as 'admin'
>>> db.get_guest_user()
User(username='guest', fullname='PlantDB Guest', password_hash='$argon2id$v=19$m=65536,t=3,p=4$2++/KY75t4qvt5x1fO4dJA$MDmREeceXOJhcupT1G6yuRFvPUJ3SjNpuSga5wkUEYw', roles={<Role.READER: 'reader'>}, created_at=datetime.datetime(2026, 1, 29, 17, 19, 13, 313677), permissions=None, last_login=datetime.datetime(2026, 1, 29, 17, 19, 20, 177023), is_active=True, failed_attempts=0, last_failed_attempt=None, locked_until=None, password_last_change=datetime.datetime(2026, 1, 29, 17, 19, 13, 313677))
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

def get_guest_user(self) -> User:
    """Retrieve the guest user information from the RBAC manager.

    Returns
    -------
    plantdb.commons.auth.models.User
        The User object corresponding to the guest username.

    Notes
    -----
    This method interacts with the underlying RBAC manager to fetch guest user information.

    See Also
    --------
    rbac_manager.get_guest_user : The underlying method used by this function.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db()  # automatic login as 'admin'
    >>> db.get_guest_user()
    User(username='guest', fullname='PlantDB Guest', password_hash='$argon2id$v=19$m=65536,t=3,p=4$2++/KY75t4qvt5x1fO4dJA$MDmREeceXOJhcupT1G6yuRFvPUJ3SjNpuSga5wkUEYw', roles={<Role.READER: 'reader'>}, created_at=datetime.datetime(2026, 1, 29, 17, 19, 13, 313677), permissions=None, last_login=datetime.datetime(2026, 1, 29, 17, 19, 20, 177023), is_active=True, failed_attempts=0, last_failed_attempt=None, locked_until=None, password_last_change=datetime.datetime(2026, 1, 29, 17, 19, 13, 313677))
    >>> db.disconnect()
    """
    return self.rbac_manager.get_guest_user()

get_scan

get_scan(scan_id, current_user=None, **kwargs)

Get a Scan instance in the local database.

Parameters:

Name	Type	Description	Default
scan_id	str	The name of the scan dataset to get/create. It should exist if create is False.	required

Raises:

Type	Description
ScanNotFoundError	If the scan_id does not exist in the local database and create is False.

Returns:

Type	Description
Scan	The Scan identified by the scan_id.

Examples:

>>> # Example #1 - API demo
>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_scan=True)
>>> scan = db.get_scan('myscan_001')
>>> print(scan)
<plantdb.commons.fsdb.core.Scan object at ...>
>>> db.list_scans()
['007']
>>> unknown_scan = db.get_scan('unknown')
plantdb.commons.fsdb.ScanNotFoundError: Unknown scan id 'unknown'!
>>> db.disconnect()  # clean up (delete) the temporary dummy database

>>> # Example #2 - API demo with authentication
>>> from plantdb.commons.test_database import test_database
>>> db = test_database()
>>> db.connect()
>>> scan = db.get_scan('real_plant_analyzed')
plantdb.commons.fsdb.exceptions.NoAuthUserError: No authenticated user!
>>> token = db.login('admin', 'admin')
>>> scan = db.get_scan('real_plant_analyzed')
>>> print(scan.id)
real_plant_analyzed
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
@get_authentication
@require_authentication
@requires_permission(Permission.READ, check_scan_access=True)
def get_scan(self, scan_id, current_user=None, **kwargs) -> "Scan":
    """Get a ` Scan ` instance in the local database.

    Parameters
    ----------
    scan_id : str
        The name of the scan dataset to get/create.
        It should exist if `create` is `False`.

    Raises
    ------
    plantdb.commons.fsdb.ScanNotFoundError
        If the `scan_id` does not exist in the local database and `create` is ``False``.

    Returns
    -------
    plantdb.commons.fsdb.core.Scan
        The `Scan` identified by the `scan_id`.

    Examples
    --------
    >>> # Example #1 - API demo
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_scan=True)
    >>> scan = db.get_scan('myscan_001')
    >>> print(scan)
    <plantdb.commons.fsdb.core.Scan object at ...>
    >>> db.list_scans()
    ['007']
    >>> unknown_scan = db.get_scan('unknown')
    plantdb.commons.fsdb.ScanNotFoundError: Unknown scan id 'unknown'!
    >>> db.disconnect()  # clean up (delete) the temporary dummy database

    >>> # Example #2 - API demo with authentication
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database()
    >>> db.connect()
    >>> scan = db.get_scan('real_plant_analyzed')
    plantdb.commons.fsdb.exceptions.NoAuthUserError: No authenticated user!
    >>> token = db.login('admin', 'admin')
    >>> scan = db.get_scan('real_plant_analyzed')
    >>> print(scan.id)
    real_plant_analyzed
    >>> db.disconnect()
    """
    with self.lock_manager.acquire_lock(scan_id, LockType.SHARED,
                                        current_user.username if current_user else "guest", LockLevel.SCAN):
        if not self.scan_exists(scan_id):
            raise ScanNotFoundError(self, scan_id)
        return self.scans[scan_id]

get_scan_access_summary

get_scan_access_summary(scan_id, current_user=None, **kwargs)

Get access summary for the current user on a scan.

Parameters:

Name	Type	Description	Default
scan_id	str	The scan identifier	required

Other Parameters:

Name	Type	Description
username	str	The username formulating the request.
token	str	A token referring to the username formulating the request.

Returns:

Type	Description
dict or None	An access summary dictionary if scan exists and is accessible

Raises:

Type	Description
PermissionError	If no user is authenticated.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(dataset="all")
>>> db.connect()
>>> token = db.login('guest', 'guest')
>>> scan_access = db.get_scan_access_summary("real_plant")
>>> print(scan_access["effective_role"])
contributor
>>> print(scan_access["permissions"])
['write', 'create', 'read']
>>> print(scan_access["is_owner"])
True
>>> db.logout()
>>> token = db.login('admin', 'admin')
>>> scan_access = db.get_scan_access_summary("real_plant")
>>> print(scan_access["effective_role"])
admin
>>> print(scan_access["is_owner"])
False
>>> print(scan_access["access_reason"])
['admin_role']
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
def get_scan_access_summary(self, scan_id, current_user=None, **kwargs) -> dict | None:
    """Get access summary for the current user on a scan.

    Parameters
    ----------
    scan_id : str
        The scan identifier

    Other Parameters
    ----------------
    username : str
        The username formulating the request.
    token : str
        A token referring to the username formulating the request.

    Returns
    -------
    dict or None
        An access summary dictionary if scan exists and is accessible

    Raises
    ------
    PermissionError
        If no user is authenticated.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(dataset="all")
    >>> db.connect()
    >>> token = db.login('guest', 'guest')
    >>> scan_access = db.get_scan_access_summary("real_plant")
    >>> print(scan_access["effective_role"])
    contributor
    >>> print(scan_access["permissions"])
    ['write', 'create', 'read']
    >>> print(scan_access["is_owner"])
    True
    >>> db.logout()
    >>> token = db.login('admin', 'admin')
    >>> scan_access = db.get_scan_access_summary("real_plant")
    >>> print(scan_access["effective_role"])
    admin
    >>> print(scan_access["is_owner"])
    False
    >>> print(scan_access["access_reason"])
    ['admin_role']
    >>> db.disconnect()
    """
    if scan_id not in self.scans:
        return None

    scan = self.scans[scan_id]
    try:
        metadata = scan.get_metadata()
        return self.rbac_manager.get_user_scan_role_summary(current_user, metadata)
    except Exception as e:
        self.logger.warning(f"Error getting access summary for scan {scan_id}: {e}")
        return None

get_scan_lock_status

get_scan_lock_status(scan_id)

Get the current lock status for a specific scan.

Parameters:

Name	Type	Description	Default
scan_id	str	The name of the scan in the local database.	required

Returns:

Type	Description
dict	Dictionary with lock status information

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
def get_scan_lock_status(self, scan_id: str) -> dict:
    """Get the current lock status for a specific scan.

    Parameters
    ----------
    scan_id : str
        The name of the scan in the local database.

    Returns
    -------
    dict
        Dictionary with lock status information
    """
    return self.lock_manager.get_lock_status(scan_id, LockLevel.SCAN)

get_scans

get_scans(query=None, current_user=None, **kwargs)

Get a list of Scan instances defined in the local database, possibly filtered using a query.

Parameters:

Name	Type	Description	Default
query	dict	A query to use to filter the returned list of scans. The metadata must match given key and value from the query dictionary.	None

Other Parameters:

Name	Type	Description
fuzzy	bool	Whether to use fuzzy matching or not, that is the use of regular expressions.
owner_only	bool	Whether to filter the returned list of scans to only include scans owned by the current user. Default is True.

Returns:

Type	Description
list of plantdb.commons.fsdb.core.Scan	A list of Scans, filtered by the query if any.

See Also

plantdb.commons.fsdb._filter_query

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_file=True)
>>> db.get_scans()
[<plantdb.commons.fsdb.core.Scan at ...>]
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
@get_authentication
@require_authentication
def get_scans(self, query=None, current_user=None, **kwargs) -> list["Scan"]:
    """Get a list of `Scan` instances defined in the local database, possibly filtered using a `query`.

    Parameters
    ----------
    query : dict, optional
        A query to use to filter the returned list of scans.
        The metadata must match given ``key`` and ``value`` from the `query` dictionary.

    Other Parameters
    ----------------
    fuzzy : bool, optional
        Whether to use fuzzy matching or not, that is the use of regular expressions.
    owner_only : bool, optional
        Whether to filter the returned list of scans to only include scans owned by the current user.
        Default is ``True``.

    Returns
    -------
    list of plantdb.commons.fsdb.core.Scan
        A list of `Scan`s, filtered by the `query` if any.

    See Also
    --------
    plantdb.commons.fsdb._filter_query

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_file=True)
    >>> db.get_scans()
    [<plantdb.commons.fsdb.core.Scan at ...>]
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    # Get all scans and filter by access permissions
    accessible_scans = {}

    for scan_id, scan in self.scans.items():
        try:
            # Access metadata directly to avoid nested lock acquisition
            metadata = _get_metadata(scan.metadata, None, {})
            if self.rbac_manager.can_access_scan(current_user, metadata, Permission.READ):
                accessible_scans[scan_id] = scan
        except Exception as e:
            self.logger.warning(f"Error checking access for scan {scan_id}: {e}")
            continue

    if kwargs.get('owner_only', False):
        if query is None:
            query = {'owner': current_user.username}
        else:
            query.update({'owner': current_user.username})

    # Apply query filter if provided
    if query:
        accessible_scans = _filter_query(list(accessible_scans.values()), query, kwargs.get('fuzzy', False))
    else:
        accessible_scans = list(accessible_scans.values())

    sort = kwargs.get('sort', None)
    accessible_scans = _sort_scans(accessible_scans, sort)

    return accessible_scans

get_timelapse

get_timelapse(tl_id, current_user=None, **kwargs)

Get a timelapse container by its identifier.

Parameters:

Name	Type	Description	Default
tl_id	str	The identifier of the timelapse.	required

Returns:

Type	Description
TimeLapse	The TimeLapse object.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> db.list_timelapses()
['mytl_001']
>>> tl = db.get_timelapse('mytl_001')
>>> print(tl.id)
mytl_001
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
@get_authentication
@require_authentication
def get_timelapse(self, tl_id: str, current_user=None, **kwargs) -> "TimeLapse":
    """Get a timelapse container by its identifier.

    Parameters
    ----------
    tl_id : str
        The identifier of the timelapse.

    Returns
    -------
    plantdb.commons.fsdb.core.TimeLapse
        The TimeLapse object.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> tl = db.create_timelapse('mytl_001')
    >>> db.list_timelapses()
    ['mytl_001']
    >>> tl = db.get_timelapse('mytl_001')
    >>> print(tl.id)
    mytl_001
    >>> db.disconnect()
    """
    tl_path = _timelapse_path(self, tl_id)
    marker_path = _timelapse_marker(tl_path)
    if not marker_path.is_file():
        raise TimeLapseNotFoundError(self, tl_id)

    username = current_user.username if current_user else "guest"
    with self.lock_manager.acquire_lock(tl_id, LockType.SHARED, username, LockLevel.SCAN):
        with marker_path.open("r") as f:
            tl_data = json.load(f)

    return TimeLapse(self, tl_id, metadata=tl_data.get("metadata", {}), created_at=tl_data.get("created_at"),
                     owner=tl_data.get("owner"), scans=tl_data.get("scans"))

get_timelapses

get_timelapses(current_user=None, **kwargs)

Get the list of all TimeLapse objects in the local database.

Returns:

Type	Description
list of plantdb.commons.fsdb.core.TimeLapse	List of TimeLapse instances.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> tl = db.create_timelapse('mytl_002')
>>> [tl.id for tl in db.get_timelapses()]
['mytl_001', 'mytl_002']
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
@get_authentication
@require_authentication
def get_timelapses(self, current_user=None, **kwargs) -> list["TimeLapse"]:
    """Get the list of all TimeLapse objects in the local database.

    Returns
    -------
    list of plantdb.commons.fsdb.core.TimeLapse
        List of TimeLapse instances.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> tl = db.create_timelapse('mytl_001')
    >>> tl = db.create_timelapse('mytl_002')
    >>> [tl.id for tl in db.get_timelapses()]
    ['mytl_001', 'mytl_002']
    >>> db.disconnect()
    """
    return [self.get_timelapse(tl_id, current_user=current_user, **kwargs) for tl_id in self.list_timelapses()]

get_user_data

get_user_data(username=None, token=None)

Get the user data.

Parameters:

Name	Type	Description	Default
username	str | None	The username to retrieve the user data from.	None
token	str | None	The token provided by the RBAC manager.	None

Returns:

Type	Description
User | TokenUser | None	A User instance corresponding to the currently authenticated user, if any, None otherwise.

Notes

If both username and token are provided, prefer token for accessing user data.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database()
>>> db.connect()
>>> token = db.login('admin', 'admin')
>>> db.get_user_data(username='admin')
User(username='admin', fullname='PlantDB Admin', password_hash='$argon2id$v=19$m=65536,t=3,p=4$zMr0ZhclnHHdOgwWKv3Hbg$SZshbPdNiCdBONb8vgzZAKyWPl5sNIUwB8mQWkzGYOQ', roles={<Role.ADMIN: 'admin'>}, created_at=datetime.datetime(2026, 1, 29, 17, 22, 49, 683163), permissions=None, last_login=datetime.datetime(2026, 1, 29, 17, 22, 49, 770793), is_active=True, failed_attempts=0, last_failed_attempt=None, locked_until=None, password_last_change=datetime.datetime(2026, 1, 29, 17, 22, 49, 683163))
>>> user_data = db.get_user_data(username='guest')
>>> print(user_data.fullname)
PlantDB Guest
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

def get_user_data(self, username=None, token=None) -> User | TokenUser | None:
    """Get the user data.

    Parameters
    ----------
    username : str | None
        The username to retrieve the user data from.
    token : str | None
        The token provided by the RBAC manager.

    Returns
    -------
    plantdb.commons.auth.models.User | plantdb.commons.auth.models.TokenUser | None
        A ``User`` instance corresponding to the currently authenticated user, if any, ``None`` otherwise.

    Notes
    -----
    If both `username` and `token` are provided, prefer `token` for accessing user data.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database()
    >>> db.connect()
    >>> token = db.login('admin', 'admin')
    >>> db.get_user_data(username='admin')
    User(username='admin', fullname='PlantDB Admin', password_hash='$argon2id$v=19$m=65536,t=3,p=4$zMr0ZhclnHHdOgwWKv3Hbg$SZshbPdNiCdBONb8vgzZAKyWPl5sNIUwB8mQWkzGYOQ', roles={<Role.ADMIN: 'admin'>}, created_at=datetime.datetime(2026, 1, 29, 17, 22, 49, 683163), permissions=None, last_login=datetime.datetime(2026, 1, 29, 17, 22, 49, 770793), is_active=True, failed_attempts=0, last_failed_attempt=None, locked_until=None, password_last_change=datetime.datetime(2026, 1, 29, 17, 22, 49, 683163))
    >>> user_data = db.get_user_data(username='guest')
    >>> print(user_data.fullname)
    PlantDB Guest
    >>> db.disconnect()
    """
    if username and token:
        self.logger.warning("Trying to retrieve user data from both 'username' and token!")
        self.logger.debug("Using 'token' to access user data.")
        username = None

    if username:
        return self.rbac_manager.users.get_user(username)
    elif token:
        if isinstance(self.session_manager, SingleSessionManager):
            return self.rbac_manager.users.get_user(self.session_manager.validate_session(token)['username'])
        else:
            return self.rbac_manager.users.get_user_from_decoded_token(self.session_manager.validate_session(token))
    else:
        self.logger.error("No username or token provided")
        return None

get_user_groups

get_user_groups(user=None, current_user=None, **kwargs)

Get groups for a user.

Parameters:

Name	Type	Description	Default
user	str	Username to query. If None, uses the currently authenticated user.	None

Other Parameters:

Name	Type	Description
username	str	The username formulating the request.
token	str	A token referring to the username formulating the request.

Returns:

Type	Description
list[Groups]	A list of Group objects the user belongs to.

Raises:

Type	Description
PermissionError	If no user is authenticated.

Examples:

>>> from plantdb.commons.auth.models import Role
>>> from plantdb.commons.test_database import test_database
>>> db = test_database()
>>> db.connect()
>>> token = db.login('admin', 'admin')
>>> db.create_user('batman', 'Bruce Wayne', 'joker', roles=Role.CONTRIBUTOR)
>>> group_a = db.create_group('groupA', ['batman'], description="The group A.")
>>> print([g.name for g in db.get_user_groups('batman')])
['groupA']
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
def get_user_groups(self, user=None, current_user=None, **kwargs) -> list[Group]:
    """Get groups for a user.

    Parameters
    ----------
    user : str, optional
        Username to query.
        If None, uses the currently authenticated user.

    Other Parameters
    ----------------
    username : str
        The username formulating the request.
    token : str
        A token referring to the username formulating the request.

    Returns
    -------
    list[Groups]
        A list of Group objects the user belongs to.

    Raises
    ------
    PermissionError
        If no user is authenticated.

    Examples
    --------
    >>> from plantdb.commons.auth.models import Role
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database()
    >>> db.connect()
    >>> token = db.login('admin', 'admin')
    >>> db.create_user('batman', 'Bruce Wayne', 'joker', roles=Role.CONTRIBUTOR)
    >>> group_a = db.create_group('groupA', ['batman'], description="The group A.")
    >>> print([g.name for g in db.get_user_groups('batman')])
    ['groupA']
    >>> db.disconnect()
    """
    if user is None:
        user = current_user.username
    return self.rbac_manager.get_user_groups(user)

get_username

get_username(token)

Get the username.

Parameters:

Name	Type	Description	Default
token	str	The token provided by the RBAC manager.	required

Returns:

Type	Description
Optional[str]	The User.username if the token is valid, None otherwise.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database()
>>> db.connect()
>>> token = db.login('admin', 'admin')
>>> db.get_username(token)
'admin'
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

def get_username(self, token) -> str | None:
    """Get the username.

    Parameters
    ----------
    token : str
        The token provided by the RBAC manager.

    Returns
    -------
    Optional[str]
        The ``User.username`` if the token is valid, ``None`` otherwise.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database()
    >>> db.connect()
    >>> token = db.login('admin', 'admin')
    >>> db.get_username(token)
    'admin'
    >>> db.disconnect()
    """
    return self.session_manager.session_username(token)

is_scan_locked

is_scan_locked(scan_id)

Check if a scan is locked in the system.

This method determines whether the specified scan is currently locked by fetching its lock status. A scan is considered locked if it does not have an exclusive lock and there are no shared locks. This can be useful for ensuring that certain operations are not performed on scans that are not yet locked.

Parameters:

Name	Type	Description	Default
scan_id	str	The unique identifier of the scan whose lock status is to be checked.	required

Returns:

Type	Description
bool	True if the scan is locked (having neither an exclusive lock nor any shared locks), False otherwise.

See Also

ScanManager.lock_manager : Component responsible for managing scan locks.

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
def is_scan_locked(self, scan_id: str) -> bool:
    """Check if a scan is locked in the system.

    This method determines whether the specified scan is currently locked by
    fetching its lock status. A scan is considered locked if it does not have
    an exclusive lock and there are no shared locks. This can be useful for
    ensuring that certain operations are not performed on scans that are not
    yet locked.

    Parameters
    ----------
    scan_id : str
        The unique identifier of the scan whose lock status is to be checked.

    Returns
    -------
    bool
        True if the scan is locked (having neither an exclusive lock nor any
        shared locks), False otherwise.

    See Also
    --------
    ScanManager.lock_manager : Component responsible for managing scan locks.
    """
    lock_status = self.lock_manager.get_lock_status(scan_id)
    if lock_status['exclusive'] is None and len(lock_status['shared']) == 0:
        return True
    return False

list_active_locks

list_active_locks()

List all currently active locks across all scans.

Returns:

Type	Description
dict	Dictionary mapping scan IDs to their lock status

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
def list_active_locks(self) -> dict[str, dict]:
    """List all currently active locks across all scans.

    Returns
    -------
    dict
        Dictionary mapping scan IDs to their lock status
    """
    active_locks = {}
    for scan_id in self.scans.keys():
        lock_status = self.get_scan_lock_status(scan_id)
        if lock_status['exclusive'] or lock_status['shared']:
            active_locks[scan_id] = lock_status

    return active_locks

list_groups

list_groups(current_user=None, **kwargs)

List all groups.

Returns:

Type	Description
list[Group]	A list of Group objects

Raises:

Type	Description
PermissionError	If no user is authenticated.

Examples:

>>> from plantdb.commons.auth.models import Role
>>> from plantdb.commons.test_database import test_database
>>> db = test_database()
>>> db.connect()
>>> token = db.login('admin', 'admin')
>>> db.create_user('batman', 'Bruce Wayne', 'joker', roles=Role.CONTRIBUTOR)
>>> group_a = db.create_group('groupA', ['batman'], description="The group A.")
>>> print([g.name for g in db.list_groups()])
['groupA']
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
def list_groups(self, current_user=None, **kwargs) -> list[Group]:
    """List all groups.

    Returns
    -------
    list[Group]
        A list of Group objects

    Raises
    ------
    PermissionError
        If no user is authenticated.

    Examples
    --------
    >>> from plantdb.commons.auth.models import Role
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database()
    >>> db.connect()
    >>> token = db.login('admin', 'admin')
    >>> db.create_user('batman', 'Bruce Wayne', 'joker', roles=Role.CONTRIBUTOR)
    >>> group_a = db.create_group('groupA', ['batman'], description="The group A.")
    >>> print([g.name for g in db.list_groups()])
    ['groupA']
    >>> db.disconnect()
    """
    groups = self.rbac_manager.list_groups(current_user)
    return groups if groups is not None else []

list_scans

list_scans(query=None, fuzzy=False, owner_only=True, current_user=None, **kwargs)

Get the list of scan identifiers from the local database.

Parameters:

Name	Type	Description	Default
query	dict	A query to use to filter the returned list of scans. The metadata must match given key and value from the query dictionary.	None
fuzzy	bool	Whether to use fuzzy matching or not, that is the use of regular expressions.	False

Returns:

Type	Description
list[str]	The list of scan identifiers from the local database.

See Also

plantdb.commons.fsdb.core._filter_query

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_scan=True)
>>> db.list_scans()  # list scans owned by the current user
['myscan_001']
>>> db.create("batman", "Bruce Wayne", 'joker')
>>> db.connect('batman', 'joker')
>>> db.list_scans()  # list scans owned by the current user
>>> []
>>> db.list_scans(owner_only=False)  # list all scans
['myscan_001']
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
@get_authentication
def list_scans(self, query=None, fuzzy=False, owner_only=True, current_user=None, **kwargs) -> list[str]:
    """Get the list of scan identifiers from the local database.

    Parameters
    ----------
    query : dict, optional
        A query to use to filter the returned list of scans.
        The metadata must match given ``key`` and ``value`` from the `query` dictionary.
    fuzzy : bool
        Whether to use fuzzy matching or not, that is the use of regular expressions.

    Returns
    -------
    list[str]
        The list of scan identifiers from the local database.

    See Also
    --------
    plantdb.commons.fsdb.core._filter_query

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_scan=True)
    >>> db.list_scans()  # list scans owned by the current user
    ['myscan_001']
    >>> db.create("batman", "Bruce Wayne", 'joker')
    >>> db.connect('batman', 'joker')
    >>> db.list_scans()  # list scans owned by the current user
    >>> []
    >>> db.list_scans(owner_only=False)  # list all scans
    ['myscan_001']
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    if not current_user:
        owner_only = False

    if owner_only:
        if query is None:
            query = {'owner': current_user.username}
        else:
            query.update({'owner': current_user.username})

    sort = kwargs.get('sort', None)
    if query is None and not sort:
        return list(self.scans.keys())
    else:
        scans_list = _filter_query(list(self.scans.values()), query, fuzzy) if query else list(self.scans.values())
        scans_list = _sort_scans(scans_list, sort)
        return [scan.id for scan in scans_list]

list_timelapses

list_timelapses(current_user=None, **kwargs)

Get the list of timelapse identifiers from the local database.

Returns:

Type	Description
list[str]	List of timelapse container IDs.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> db.list_timelapses()
['mytl_001']
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
@get_authentication
def list_timelapses(self, current_user=None, **kwargs) -> list[str]:
    """Get the list of timelapse identifiers from the local database.

    Returns
    -------
    list[str]
        List of timelapse container IDs.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> tl = db.create_timelapse('mytl_001')
    >>> db.list_timelapses()
    ['mytl_001']
    >>> db.disconnect()
    """
    timelapses = []
    if self.path().is_dir():
        for d in self.path().iterdir():
            if d.is_dir() and not d.name.startswith('.') and _timelapse_marker(d).is_file():
                timelapses.append(d.name)
    return sorted(timelapses)

login

login(username, password, **kwargs)

Authenticate a user and create a session.

Parameters:

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

Returns:

Type	Description
str or None	Returns the user session ID if successful, None otherwise.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database()
>>> db.connect()
>>> token = db.login('admin', 'admin')
>>> token = db.login('guest', 'guest')
ERROR    [FSDB] Failed to login as 'guest'! Another user is logged in.
>>> db.logout()  # log out from 'admin' session
(True, 'admin')
>>> token = db.login('guest', 'guest')
>>> print(len(token))
43
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
def login(self, username: str, password: str, **kwargs) -> str | None:
    """Authenticate a user and create a session.

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

    Returns
    -------
    str or None
        Returns the user session ID if successful, ``None`` otherwise.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database()
    >>> db.connect()
    >>> token = db.login('admin', 'admin')
    >>> token = db.login('guest', 'guest')
    ERROR    [FSDB] Failed to login as 'guest'! Another user is logged in.
    >>> db.logout()  # log out from 'admin' session
    (True, 'admin')
    >>> token = db.login('guest', 'guest')
    >>> print(len(token))
    43
    >>> db.disconnect()
    """
    if self.validate_user(username, password):

        # If a SingleSessionManager and a currently logged user, abort
        if isinstance(self.session_manager,
                      (NoAuthSessionManager, SingleSessionManager)) and self.session_manager.has_logged_user():
            current_username = get_logged_username(self)
            if current_username:
                if current_username != username:
                    self.logger.error(f"Failed to login as '{username}'! Another user is logged in.")
                    return None
                else:
                    self.logger.warning(f"Already logged in as '{username}'.")
                    return

        # Else try to create a new session:
        session_token = self.session_manager.create_session(username)
        try:
            assert session_token is not None
        except AssertionError:
            self.logger.warning(f"User '{username}' has reached max concurrent sessions")
        else:
            self.logger.debug(f"Successfully logged in as '{username}'.")
        return session_token
    else:
        self.logger.error(f"Failed to login as '{username}'!")
        return None

logout

logout(**kwargs)

Log out a user by invalidating its session.

Returns:

Type	Description
bool	Indicate successfull log-out.
str	The logged out username.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database()
>>> db.connect()
>>> token = db.login('admin', 'admin')
>>> db.logout()  # log out from 'admin' session
(True, 'admin')
>>> db.logout()  # log out from NO session
ERROR    [FSDB] No logged user!
WARNING  [FSDB] Failed to logout!
(False, None)
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@require_token
def logout(self, **kwargs) -> tuple[bool, str]:
    """Log out a user by invalidating its session.

    Returns
    -------
    bool
        Indicate successfull log-out.
    str
        The logged out username.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database()
    >>> db.connect()
    >>> token = db.login('admin', 'admin')
    >>> db.logout()  # log out from 'admin' session
    (True, 'admin')
    >>> db.logout()  # log out from NO session
    ERROR    [FSDB] No logged user!
    WARNING  [FSDB] Failed to logout!
    (False, None)
    >>> db.disconnect()
    """
    success, username = self.session_manager.invalidate_session(kwargs.get('token', None))
    if success:
        self.logger.debug(f"Successfully logged out from '{username}'.")
        return success, username
    else:
        self.logger.warning(f"Failed to logout!")
        return success, username

path

path()

Get the path to the local database root directory.

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db()
>>> print(db.path())
/tmp/romidb_...
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

def path(self) -> pathlib.Path:
    """Get the path to the local database root directory.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db()
    >>> print(db.path())
    /tmp/romidb_...
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    return copy.deepcopy(self.basedir)

reload

reload(scan_id=None)

Reload the database by scanning datasets.

Parameters:

Name	Type	Description	Default
scan_id	str or list of str	The name of the scan(s) to reload.	None

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
def reload(self, scan_id: str | list[str] | None = None) -> None:
    """Reload the database by scanning datasets.

    Parameters
    ----------
    scan_id : str or list of str, optional
        The name of the scan(s) to reload.
    """
    if scan_id is None:
        self.logger.info("Reloading the database...")
        self.scans = _load_scans(self)
    elif isinstance(scan_id, str):
        self.logger.info(f"Reloading scan '{scan_id}'...")
        scan = _load_scan(self, scan_id)
        if not scan:
            self.logger.error(f"Failed to reload scan '{scan_id}'!")
        self.scans[scan_id] = scan
    elif isinstance(scan_id, Iterable):
        [self.reload(scan_i) for scan_i in scan_id]
    else:
        self.logger.error(f"Wrong parameter `scan_name`, expected a string or list of string but got '{scan_id}'!")
    self.logger.info("Done!")
    return

remove_user_from_group

remove_user_from_group(group_name, user, current_user=None, **kwargs)

Remove a user from a group.

Parameters:

Name	Type	Description	Default
group_name	str	Name of the group to remove the user from.	required
user	str	Name of the user to remove from the group.	required

Other Parameters:

Name	Type	Description
username	str	The username formulating the request.
token	str	A token referring to the username formulating the request.

Returns:

Type	Description
bool	True if the user was successfully removed from the group, False otherwise.

Raises:

Type	Description
PermissionError	If no user is authenticated. If the authenticated user lacks permission to remove the user from the group.

Examples:

>>> from plantdb.commons.auth.models import Role
>>> from plantdb.commons.test_database import test_database
>>> db = test_database()
>>> db.connect()
>>> token = db.login('admin', 'admin')
>>> db.create_user('batman', 'Bruce Wayne', 'joker', roles=Role.CONTRIBUTOR)
>>> group_a = db.create_group('groupA', ['batman'], description="The group A.")
>>> print(group_a.users)
{'admin', 'batman'}
>>> db.remove_user_from_group('groupA', 'batman')
>>> print(group_a.users)
{'admin'}
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
def remove_user_from_group(self, group_name, user, current_user=None, **kwargs) -> bool:
    """Remove a user from a group.

    Parameters
    ----------
    group_name : str
        Name of the group to remove the user from.
    user : str
        Name of the user to remove from the group.

    Other Parameters
    ----------------
    username : str
        The username formulating the request.
    token : str
        A token referring to the username formulating the request.

    Returns
    -------
    bool
        ``True`` if the `user` was successfully removed from the group, ``False`` otherwise.

    Raises
    ------
    PermissionError
        If no user is authenticated.
        If the authenticated user lacks permission to remove the user from the group.

    Examples
    --------
    >>> from plantdb.commons.auth.models import Role
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database()
    >>> db.connect()
    >>> token = db.login('admin', 'admin')
    >>> db.create_user('batman', 'Bruce Wayne', 'joker', roles=Role.CONTRIBUTOR)
    >>> group_a = db.create_group('groupA', ['batman'], description="The group A.")
    >>> print(group_a.users)
    {'admin', 'batman'}
    >>> db.remove_user_from_group('groupA', 'batman')
    >>> print(group_a.users)
    {'admin'}
    >>> db.disconnect()
    """
    if not self.rbac_manager.remove_user_from_group(current_user, group_name, user):
        raise PermissionError("Insufficient permissions or operation failed")
    return True

scan_exists

scan_exists(scan_id)

Check if a given scan ID exists in the database.

Parameters:

Name	Type	Description	Default
scan_id	str	The ID of the scan to check.	required

Returns:

Type	Description
bool	True if the scan exists, False otherwise.

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_scan=True)
>>> db.scan_exists("myscan_001")
True
>>> db.scan_exists("nonexistent_id")
False
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
def scan_exists(self, scan_id: str) -> bool:
    """Check if a given scan ID exists in the database.

    Parameters
    ----------
    scan_id : str
        The ID of the scan to check.

    Returns
    -------
    bool
        ``True`` if the scan exists, ``False`` otherwise.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_scan=True)
    >>> db.scan_exists("myscan_001")
    True
    >>> db.scan_exists("nonexistent_id")
    False
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    return scan_id in self.scans

timelapse_exists

timelapse_exists(tl_id)

Check if a timelapse exists in the database.

Parameters:

Name	Type	Description	Default
tl_id	str	The ID of the timelapse to check.	required

Returns:

Type	Description
bool	True if the timelapse exists, False otherwise.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
 >>> tl = db.create_timelapse('mytl_001')
>>> db.timelapse_exists('mytl_001')
True
>>> db.timelapse_exists('mytl_002')
False
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
def timelapse_exists(self, tl_id: str) -> bool:
    """Check if a timelapse exists in the database.

    Parameters
    ----------
    tl_id : str
        The ID of the timelapse to check.

    Returns
    -------
    bool
        True if the timelapse exists, False otherwise.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
     >>> tl = db.create_timelapse('mytl_001')
    >>> db.timelapse_exists('mytl_001')
    True
    >>> db.timelapse_exists('mytl_002')
    False
    >>> db.disconnect()
    """
    tl_path = _timelapse_path(self, tl_id)
    return _timelapse_marker(tl_path).is_file()

update_user_password

update_user_password(old_password, new_password, current_user=None, **kwargs)

Create a new user with the specified details.

Parameters:

Name	Type	Description	Default
old_password	str	The password for the user.	required
new_password	str	The new password for the user.	required
current_user	User | TokenUser | None	The current user, based on logged status or provided API token. Default is None.	None

Raises:

Type	Description
PermissionError	If no user is authenticated, that is logged in or provided an API token. If the user lacks permissions.

See Also

RBACManager.users.update_password : Method used to actually update the user's password.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database()
>>> db.connect()
>>> token = db.login('admin', 'admin')
>>> db.create_user('batman', 'Bruce Wayne', 'joker', roles=Role.CONTRIBUTOR)
INFO     [UserManager] Welcome Bruce Wayne, please log in...'
>>> db.logout()  # log out from 'admin' session
(True, 'admin')
>>> token = db.login('batman', 'joker')
>>> db.update_user_password('joker', 'alfred')
INFO     [UserManager] Password updated for user 'batman'...
>>> db.logout()  # log out from 'batman' session
(True, 'batman')
>>> token = db.login('batman', 'alfred')  # login with new password
>>> print(len(token))
43
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
def update_user_password(self, old_password: str, new_password: str,
                         current_user: User | TokenUser | None = None, **kwargs) -> None:
    """Create a new user with the specified details.

    Parameters
    ----------
    old_password : str
        The password for the user.
    new_password : str
        The new password for the user.
    current_user : User | TokenUser | None
        The current user, based on logged status or provided API token.
        Default is ``None``.

    Raises
    ------
    PermissionError
        If no user is authenticated, that is logged in or provided an API token.
        If the user lacks permissions.

    See Also
    --------
    RBACManager.users.update_password : Method used to actually update the user's password.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database()
    >>> db.connect()
    >>> token = db.login('admin', 'admin')
    >>> db.create_user('batman', 'Bruce Wayne', 'joker', roles=Role.CONTRIBUTOR)
    INFO     [UserManager] Welcome Bruce Wayne, please log in...'
    >>> db.logout()  # log out from 'admin' session
    (True, 'admin')
    >>> token = db.login('batman', 'joker')
    >>> db.update_user_password('joker', 'alfred')
    INFO     [UserManager] Password updated for user 'batman'...
    >>> db.logout()  # log out from 'batman' session
    (True, 'batman')
    >>> token = db.login('batman', 'alfred')  # login with new password
    >>> print(len(token))
    43
    >>> db.disconnect()
    """
    return self.rbac_manager.users.update_password(current_user.username, old_password, new_password)

validate_user

validate_user(username, password)

Validate the user credentials.

Parameters:

Name	Type	Description	Default
username	str	The username provided by the user attempting to log in.	required
password	str	The password provided by the user attempting to log in.	required

Returns:

Type	Description
bool	True if the login attempt is successful, False otherwise.

Raises:

Type	Description
KeyError	If there is an issue accessing necessary user data.

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db()  # NoAuthSessionManager with automatic login as 'admin'
>>> db.validate_user('guest', 'guest')
True
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@require_connected_db
def validate_user(self, username: str, password: str) -> bool:
    """Validate the user credentials.

    Parameters
    ----------
    username : str
        The username provided by the user attempting to log in.
    password : str
        The password provided by the user attempting to log in.

    Returns
    -------
    bool
        ``True`` if the login attempt is successful, ``False`` otherwise.

    Raises
    ------
    KeyError
        If there is an issue accessing necessary user data.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db()  # NoAuthSessionManager with automatic login as 'admin'
    >>> db.validate_user('guest', 'guest')
    True
    >>> db.disconnect()
    """
    return self.rbac_manager.users.validate(username, password)

File

File(fileset, f_id, **kwargs)

Bases: File, MetadataManager

Implement File for the local File System DataBase from abstract class db.File.

Attributes:

Name	Type	Description
db	FSDB	Database where to find the fileset.
fileset	Fileset	Set of files containing the file.
id	str	Name of the file in the FSDB local database.
filename	str	File name.
metadata	dict	Dictionary of metadata attached to the file.

See Also

plantdb.commons.db.File

Notes

File must be written using write_raw or write methods to exist on disk. Else they are just referenced in the database!

Contrary to other classes (Scan & Fileset) the uniqueness is not checked!

Returned by:

• Reference API commons commons
  • fsdb core  Fileset
    •  create_file
    •  get_file
  • utils  fsdb_file_from_local_file

Used by:

• Reference API commons commons utils  to_file

Source code in plantdb/commons/fsdb/core.py

def __init__(self, fileset, f_id, **kwargs):
    super().__init__(fileset, f_id, **kwargs)
    self.metadata = {}

    self.session_manager = self.db.session_manager
    self.logger = self.db.logger

get_db

get_db()

Get parent database instance.

Returns:

Type	Description
DB	The parent database instance.

Source code in plantdb/commons/db.py

def get_db(self):
    """Get parent database instance.

    Returns
    -------
    plantdb.commons.db.DB
        The parent database instance.
    """
    return self.db

get_fileset

get_fileset()

Get parent fileset.

Returns:

Type	Description
Fileset	The parent fileset instance.

Source code in plantdb/commons/db.py

def get_fileset(self):
    """Get parent fileset.

    Returns
    -------
    plantdb.commons.db.Fileset
        The parent fileset instance.
    """
    return self.fileset

get_id

get_id()

Get file id.

Returns:

Type	Description
str	The id of the file instance.

Source code in plantdb/commons/db.py

def get_id(self):
    """Get file id.

    Returns
    -------
    str
        The id of the file instance.
    """
    return deepcopy(self.id)

get_metadata

get_metadata(key=None, default={}, current_user=None, **kwargs)

Get the metadata associated with a file.

Parameters:

Name	Type	Description	Default
key	str	A key that should exist in the file's metadata.	None
default	Any	The default value to return if the key does not exist in the metadata. Default is an empty dictionary{}.	{}

Returns:

Type	Description
any	If key is None, returns a dictionary. Else, returns the value attached to this key.

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> from plantdb.commons.fsdb.path_helpers import _file_metadata_path
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan("myscan_001")
>>> fs = scan.get_fileset('fileset_001')
>>> f = fs.get_file("test_json")
>>> print(f.get_metadata())
{'random json': True}
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@use_guest_as_default
@requires_permission(Permission.READ, check_scan_access=True)
def get_metadata(self, key=None, default={}, current_user=None, **kwargs) -> Any:
    """Get the metadata associated with a file.

    Parameters
    ----------
    key : str
        A key that should exist in the file's metadata.
    default : Any, optional
        The default value to return if the key does not exist in the metadata.
        Default is an empty dictionary``{}``.

    Returns
    -------
    any
        If `key` is ``None``, returns a dictionary.
        Else, returns the value attached to this key.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> from plantdb.commons.fsdb.path_helpers import _file_metadata_path
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan("myscan_001")
    >>> fs = scan.get_fileset('fileset_001')
    >>> f = fs.get_file("test_json")
    >>> print(f.get_metadata())
    {'random json': True}
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    # Use shared lock for read operations
    with self.db.lock_manager.acquire_lock(f"{self.scan.id}/{self.fileset.id}/{self.id}", LockType.SHARED,
                                           current_user.username, LockLevel.FILE):
        return _get_metadata(self.metadata, key, default)

get_scan

get_scan()

Get parent scan instance.

Returns:

Type	Description
Scan	The parent scan instance.

Source code in plantdb/commons/db.py

def get_scan(self):
    """Get parent scan instance.

    Returns
    -------
    plantdb.commons.db.Scan
        The parent scan instance.
    """
    return self.scan

import_file

import_file(path, current_user=None, **kwargs)

Import the file from its local path to the current fileset.

Parameters:

Name	Type	Description	Default
path	str or Path	The path to the file to import.	required

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> from plantdb.commons.fsdb.path_helpers import _file_metadata_path
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan("myscan_001")
>>> fs = scan.get_fileset('fileset_001')
>>> file = fs.get_file("test_json")
>>> new_file = fs.create_file('test_json2')
>>> new_file.import_file(file.path())
>>> print(new_file.path().exists())
True
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
@requires_permission(Permission.WRITE, check_scan_access=True)
def import_file(self, path, current_user=None, **kwargs) -> None:
    """Import the file from its local path to the current fileset.

    Parameters
    ----------
    path : str or pathlib.Path
        The path to the file to import.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> from plantdb.commons.fsdb.path_helpers import _file_metadata_path
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan("myscan_001")
    >>> fs = scan.get_fileset('fileset_001')
    >>> file = fs.get_file("test_json")
    >>> new_file = fs.create_file('test_json2')
    >>> new_file.import_file(file.path())
    >>> print(new_file.path().exists())
    True
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    # Check if the path is a file
    if isinstance(path, str):
        path = Path(path)
    if not os.path.isfile(path):
        raise ValueError(f"The provided path is not a file: {path}.")

    # Use fileset-level exclusive lock, as `store()` serializes the whole fileset.
    self.logger.debug(
        f"Importing file '{self.id}' in '{self.scan.id}/{self.fileset.id}' as user '{current_user.username}'...")
    with self.db.lock_manager.acquire_lock(f"{self.scan.id}/{self.fileset.id}", LockType.EXCLUSIVE,
                                           current_user.username, LockLevel.FILESET):
        # Get the file name and extension
        ext = path.suffix[1:]
        self.filename = _get_filename(self, ext)
        # Get the path to the new `File` instance
        newpath = _file_path(self)
        # Copy the file to its new destination
        copyfile(path, newpath)
        self.store()  # register it to the scan main JSON FILE

    self.logger.debug(f"Done importing file.")
    return

path

path()

Get the path to the local file.

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_scan=True, with_file=True)
>>> scan = db.get_scan("myscan_001")
>>> fs = scan.get_fileset("fileset_001")
>>> fs.list_files()
['dummy_image', 'test_image', 'test_json']
>>> f = fs.get_file('dummy_image')
>>> f.path()
/tmp/romidb_.../myscan_001/fileset_001/dummy_image.png
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

def path(self) -> pathlib.Path:
    """Get the path to the local file.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_scan=True, with_file=True)
    >>> scan = db.get_scan("myscan_001")
    >>> fs = scan.get_fileset("fileset_001")
    >>> fs.list_files()
    ['dummy_image', 'test_image', 'test_json']
    >>> f = fs.get_file('dummy_image')
    >>> f.path()
    /tmp/romidb_.../myscan_001/fileset_001/dummy_image.png
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    return _file_path(self)

read

read()

Read the file and return its contents.

Returns:

Type	Description
str	The contents of the file.

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan("myscan_001")
>>> fs = scan.get_fileset("fileset_001")
>>> f = fs.get_file("test_json")
>>> js = f.read()
>>> print(js)  # print the content of the file
{
    "Who you gonna call?": "Ghostbuster"
}
>>> # Convert this raw json into a dictionary with dedicated method from `json` library:
>>> import json
>>> js_dict = json.loads(js)
>>> print(js_dict)
{'Who you gonna call?': 'Ghostbuster'}
>>> print(type(js_dict))
<class 'dict'>
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

def read(self) -> str:
    """Read the file and return its contents.

    Returns
    -------
    str
        The contents of the file.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan("myscan_001")
    >>> fs = scan.get_fileset("fileset_001")
    >>> f = fs.get_file("test_json")
    >>> js = f.read()
    >>> print(js)  # print the content of the file
    {
        "Who you gonna call?": "Ghostbuster"
    }
    >>> # Convert this raw json into a dictionary with dedicated method from `json` library:
    >>> import json
    >>> js_dict = json.loads(js)
    >>> print(js_dict)
    {'Who you gonna call?': 'Ghostbuster'}
    >>> print(type(js_dict))
    <class 'dict'>
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    path = _file_path(self)
    with path.open(mode="r") as f:
        return f.read()

read_raw

read_raw()

Read the file and return its contents.

Returns:

Type	Description
bytes	The contents of the file.

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan("myscan_001")
>>> fs = scan.get_fileset("fileset_001")
>>> f = fs.get_file("test_json")
>>> js = f.read_raw()
>>> print(js)  # print the raw bytes content
>>> # Convert this raw json into a dictionary with dedicated method from `json` library:
>>> import json
>>> js_dict = json.loads(js)
>>> print(js_dict)
{'Who you gonna call?': 'Ghostbuster'}
>>> print(type(js_dict))
<class 'dict'>
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

def read_raw(self) -> bytes:
    """Read the file and return its contents.

    Returns
    -------
    bytes
        The contents of the file.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan("myscan_001")
    >>> fs = scan.get_fileset("fileset_001")
    >>> f = fs.get_file("test_json")
    >>> js = f.read_raw()
    >>> print(js)  # print the raw bytes content
    >>> # Convert this raw json into a dictionary with dedicated method from `json` library:
    >>> import json
    >>> js_dict = json.loads(js)
    >>> print(js_dict)
    {'Who you gonna call?': 'Ghostbuster'}
    >>> print(type(js_dict))
    <class 'dict'>
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    path = _file_path(self)
    with path.open(mode="rb") as f:
        return f.read()

set_metadata

set_metadata(data, value=None, current_user=None, **kwargs)

Add a new metadata to the file.

Parameters:

Name	Type	Description	Default
data	str or dict	If a string, a key to address the value. If a dictionary, update the metadata dictionary with data (value is then unused).	required
value	any	The value to assign to data if the latest is not a dictionary.	None

Examples:

>>> import json
>>> from plantdb.commons.test_database import dummy_db
>>> from plantdb.commons.fsdb.path_helpers import _file_metadata_path
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan("myscan_001")
>>> fs = scan.get_fileset('fileset_001')
>>> file = fs.get_file("test_json")
>>> file.set_metadata("test", "value")
>>> p = _file_metadata_path(file)
>>> print(p.exists())
True
>>> print(json.load(p.open(mode='r')))
{'random json': True, 'test': 'value'}
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
@requires_permission(Permission.WRITE, check_scan_access=True)
def set_metadata(self, data, value=None, current_user=None, **kwargs) -> None:
    """Add a new metadata to the file.

    Parameters
    ----------
    data : str or dict
        If a string, a key to address the `value`.
        If a dictionary, update the metadata dictionary with `data` (`value` is then unused).
    value : any, optional
        The value to assign to `data` if the latest is not a dictionary.

    Examples
    --------
    >>> import json
    >>> from plantdb.commons.test_database import dummy_db
    >>> from plantdb.commons.fsdb.path_helpers import _file_metadata_path
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan("myscan_001")
    >>> fs = scan.get_fileset('fileset_001')
    >>> file = fs.get_file("test_json")
    >>> file.set_metadata("test", "value")
    >>> p = _file_metadata_path(file)
    >>> print(p.exists())
    True
    >>> print(json.load(p.open(mode='r')))
    {'random json': True, 'test': 'value'}
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    # The locking mechanism is handled by the `MetadataManager._update_metadata` method
    self._update_metadata(data, value, current_user, _store_file_metadata, cls_name="File")

store

store()

Save changes to the scan main JSON FILE (files.json).

Source code in plantdb/commons/fsdb/core.py

def store(self) -> None:
    """Save changes to the scan main JSON FILE (``files.json``)."""
    self.fileset.store()
    return

write

write(data, ext='', current_user=None, **kwargs)

Write a file from data.

Parameters:

Name	Type	Description	Default
data	str	A string representation of the content to write.	required
ext	str	The extension to use to save the file.	''

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan("myscan_001")
>>> fs = scan.get_fileset("fileset_001")
>>> new_f = fs.create_file('file_007')
>>> md = {"Name": "Bond, James Bond"}  # Create an example dictionary to save as JSON
>>> import json
>>> data = json.dumps(md)
>>> print(data)
{"Name": "Bond, James Bond"}
>>> print(type(data))
<class 'str'>
>>> new_f.write(data, 'json')
>>> print([f.name for f in fs.path().iterdir()])
['dummy_image.png', 'test_json.json', 'test_image.png', 'file_007.json']
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
@requires_permission(Permission.WRITE, check_scan_access=True)
def write(self, data, ext="", current_user=None, **kwargs) -> None:
    """Write a file from data.

    Parameters
    ----------
    data : str
        A string representation of the content to write.
    ext : str, optional
        The extension to use to save the file.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan("myscan_001")
    >>> fs = scan.get_fileset("fileset_001")
    >>> new_f = fs.create_file('file_007')
    >>> md = {"Name": "Bond, James Bond"}  # Create an example dictionary to save as JSON
    >>> import json
    >>> data = json.dumps(md)
    >>> print(data)
    {"Name": "Bond, James Bond"}
    >>> print(type(data))
    <class 'str'>
    >>> new_f.write(data, 'json')
    >>> print([f.name for f in fs.path().iterdir()])
    ['dummy_image.png', 'test_json.json', 'test_image.png', 'file_007.json']
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    # Use exclusive lock for this operation
    self.logger.debug(
        f"Writing file '{self.id}' in '{self.scan.id}/{self.fileset.id}' as user '{current_user.username}'...")
    with self.db.lock_manager.acquire_lock(f"{self.scan.id}/{self.fileset.id}/{self.id}", LockType.EXCLUSIVE,
                                           current_user.username, LockLevel.FILE):
        self.filename = _get_filename(self, ext)
        path = _file_path(self)
        with path.open(mode="w") as f:
            f.write(data)
        self.store()

    self.logger.debug(f"Done writing file.")
    return

write_raw

write_raw(data, ext='', current_user=None, **kwargs)

Write a file from raw byte data.

Parameters:

Name	Type	Description	Default
data	bytes	The raw byte content to write.	required
ext	str	The extension to use to save the file.	''

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan("myscan_001")
>>> fs = scan.get_fileset("fileset_001")
>>> new_f = fs.create_file('file_007')
>>> md = {"Name": "Bond, James Bond"}  # Create an example dictionary to save as JSON
>>> import json
>>> data = json.dumps(md).encode()
>>> print(data)
b'{"Name": "Bond, James Bond"}'
>>> new_f.write_raw(data, 'json')
>>> print([f.name for f in fs.path().iterdir()])
['dummy_image.png', 'test_json.json', 'test_image.png', 'file_007.json']
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
@requires_permission(Permission.WRITE, check_scan_access=True)
def write_raw(self, data, ext="", current_user=None, **kwargs) -> None:
    """Write a file from raw byte data.

    Parameters
    ----------
    data : bytes
        The raw byte content to write.
    ext : str, optional
        The extension to use to save the file.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan("myscan_001")
    >>> fs = scan.get_fileset("fileset_001")
    >>> new_f = fs.create_file('file_007')
    >>> md = {"Name": "Bond, James Bond"}  # Create an example dictionary to save as JSON
    >>> import json
    >>> data = json.dumps(md).encode()
    >>> print(data)
    b'{"Name": "Bond, James Bond"}'
    >>> new_f.write_raw(data, 'json')
    >>> print([f.name for f in fs.path().iterdir()])
    ['dummy_image.png', 'test_json.json', 'test_image.png', 'file_007.json']
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    # Use file-level exclusive lock for this operation
    self.logger.debug(
        f"Writing raw file '{self.id}' in '{self.scan.id}/{self.fileset.id}' as user '{current_user.username}'...")
    with self.db.lock_manager.acquire_lock(f"{self.scan.id}/{self.fileset.id}/{self.id}", LockType.EXCLUSIVE,
                                           current_user.username, LockLevel.FILE):
        self.filename = _get_filename(self, ext)
        path = _file_path(self)
        with path.open(mode="wb") as f:
            f.write(data)
        self.store()

    self.logger.debug(f"Done writing raw file.")
    return

Fileset

Fileset(scan, fs_id)

Bases: Fileset, MetadataManager

Implement Fileset for the local File System DataBase from the abstract class db.Fileset.

Implementation of a fileset as a simple files structure with: * directory ${FSDB.basedir}/${FSDB.scan.id}/${Fileset.id} containing a set of files; * directory ${FSDB.basedir}/${FSDB.scan.id}/metadata containing JSON metadata associated with files; * JSON file files.json containing the list of files from fileset;

Attributes:

Name	Type	Description
db	FSDB	A local database instance hosting the Scan instance.
scan	Scan	A scan instance hosting this Fileset instance.
id	str	The identifier of this Fileset instance in the scan.
metadata	dict	A metadata dictionary.
files	dict[str, File]	A dictionary of File instances attached to the fileset, indexed by their identifier.

See Also

plantdb.commons.db.Fileset

Constructor.

Parameters:

Name	Type	Description	Default
scan	Scan	A scan instance containing the fileset.	required
fs_id	str	The identifier of the fileset instance.	required

Returned by:

• Reference API commons commons
  • fsdb core  Scan
    •  create_fileset
    •  get_fileset
  • testing  FSDBTestCase  get_task_fileset

Used by:

• Reference API commons commons utils  tmpdir_from_fileset

Source code in plantdb/commons/fsdb/core.py

def __init__(self, scan, fs_id):
    """Constructor.

    Parameters
    ----------
    scan : plantdb.commons.fsdb.core.Scan
        A scan instance containing the fileset.
    fs_id : str
        The identifier of the fileset instance.
    """
    super().__init__(scan, fs_id)
    # Defines attributes:
    self.metadata = {}
    self.files = {}

    self.session_manager = self.db.session_manager
    self.logger = self.db.logger

create_file

create_file(f_id, metadata=None, current_user=None, **kwargs)

Create a new File instance in the local database attached to the current Fileset instance.

Parameters:

Name	Type	Description	Default
f_id	str	The name of the file to create.	required

Returns:

Type	Description
File	The File instance created in the current Fileset instance.

Raises:

Type	Description
PermissionError	If no user is authenticated. If the user lacks permission to create a fileset.
FileExistsError	If the f_id already exists in the local database.
ValueError	If the given f_id is invalid.

See Also

plantdb.commons.fsdb.validation._is_valid_id

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan('myscan_001')
>>> fs = scan.get_fileset('fileset_001')
>>> fs.list_files()
['dummy_image', 'test_image', 'test_json']
>>> new_f = fs.create_file('file_007')
>>> fs.list_files()
['dummy_image', 'test_image', 'test_json', 'file_007']
>>> print([f.name for f in fs.path().iterdir()])  # the file only exists in the database, not on drive!
['dummy_image.png', 'test_json.json', 'test_image.png']
>>> md = {"Name": "Bond, James Bond"}  # Create an example dictionary to save as JSON
>>> from plantdb.commons import io
>>> io.write_json(new_f, md, "json")  # write the file on drive
>>> print([f.name for f in fs.path().iterdir()])
['file_007.json', 'test_image.png', 'test_json.json', 'dummy_image.png']
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
@requires_permission(Permission.WRITE, check_scan_access=True)
def create_file(self, f_id, metadata=None, current_user=None, **kwargs) -> "File":
    """Create a new `File` instance in the local database attached to the current `Fileset` instance.

    Parameters
    ----------
    f_id : str
        The name of the file to create.

    Returns
    -------
    plantdb.commons.fsdb.core.File
        The `File` instance created in the current `Fileset` instance.

    Raises
    ------
    PermissionError
        If no user is authenticated.
        If the user lacks permission to create a fileset.
    FileExistsError
        If the ``f_id`` already exists in the local database.
    ValueError
        If the given ``f_id`` is invalid.

    See Also
    --------
    plantdb.commons.fsdb.validation._is_valid_id

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan('myscan_001')
    >>> fs = scan.get_fileset('fileset_001')
    >>> fs.list_files()
    ['dummy_image', 'test_image', 'test_json']
    >>> new_f = fs.create_file('file_007')
    >>> fs.list_files()
    ['dummy_image', 'test_image', 'test_json', 'file_007']
    >>> print([f.name for f in fs.path().iterdir()])  # the file only exists in the database, not on drive!
    ['dummy_image.png', 'test_json.json', 'test_image.png']
    >>> md = {"Name": "Bond, James Bond"}  # Create an example dictionary to save as JSON
    >>> from plantdb.commons import io
    >>> io.write_json(new_f, md, "json")  # write the file on drive
    >>> print([f.name for f in fs.path().iterdir()])
    ['file_007.json', 'test_image.png', 'test_json.json', 'dummy_image.png']
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """

    # Verify if the given `fs_id` is valid
    if not _is_valid_id(f_id):
        raise ValueError(f"Invalid file identifier '{f_id}'!")
    # Verify if the given `f_id` already exists in the local database
    if self.file_exists(f_id):
        raise FileExistsError(self, f_id)

    # Use fileset-level exclusive lock for file creation, as `store()` serializes the whole fileset.
    self.logger.debug(
        f"Creating a file '{f_id}' in '{self.scan.id}/{self.id}' as '{current_user.username}' user...")
    with self.db.lock_manager.acquire_lock(f"{self.scan.id}/{self.id}", LockType.EXCLUSIVE,
                                           current_user.username, LockLevel.FILESET):
        # Create the new File
        file = File(self, f_id)  # Initialize a new File instance

        # Set initial metadata
        initial_metadata = metadata or {}
        now = iso_date_now()
        initial_metadata['created'] = now  # creation timestamp
        initial_metadata['last_modified'] = now  # modification timestamp
        initial_metadata['created_by'] = current_user.fullname

        # Cannot use fileset.set_metadata(initial_metadata) here as ownership is not granted yet!
        _set_metadata(file.metadata, initial_metadata, None)  # add metadata dictionary to the new scan
        _store_file_metadata(file)

        self.files.update({f_id: file})  # Update filesets's file dictionary
        self.store()  # Store fileset instance to the JSON

    self.logger.debug(f"Done creating the file.")
    return file

delete_file

delete_file(f_id, current_user=None, **kwargs)

Delete a given file from the current fileset.

Parameters:

Name	Type	Description	Default
f_id	str	Name of the file to delete.	required

Raises:

Type	Description
PermissionError	If no user is authenticated. If the user does not have permission to delete this file.
ValueError	If the file does not exist.

See Also

plantdb.commons.fsdb.file_ops._delete_file

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan('myscan_001')
>>> fs = scan.get_fileset('fileset_001')
>>> fs.list_files()
['dummy_image', 'test_image', 'test_json']
>>> fs.delete_file('dummy_image')
INFO     [plantdb.commons.fsdb] Deleted JSON metadata file for file 'dummy_image' from 'myscan_001/fileset_001'.
INFO     [plantdb.commons.fsdb] Deleted file 'dummy_image' from 'myscan_001/fileset_001'.
>>> fs.list_files()
['test_image', 'test_json']
>>> print([f.name for f in fs.path().iterdir()])  # the file has been removed from the drive and the database
['test_json.json', 'test_image.png']
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
@requires_permission(Permission.DELETE, check_scan_access=True)
def delete_file(self, f_id, current_user=None, **kwargs) -> None:
    """Delete a given file from the current fileset.

    Parameters
    ----------
    f_id : str
        Name of the file to delete.

    Raises
    ------
    PermissionError
        If no user is authenticated.
        If the user does not have permission to delete this file.
    ValueError
        If the file does not exist.

    See Also
    --------
    plantdb.commons.fsdb.file_ops._delete_file

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan('myscan_001')
    >>> fs = scan.get_fileset('fileset_001')
    >>> fs.list_files()
    ['dummy_image', 'test_image', 'test_json']
    >>> fs.delete_file('dummy_image')
    INFO     [plantdb.commons.fsdb] Deleted JSON metadata file for file 'dummy_image' from 'myscan_001/fileset_001'.
    INFO     [plantdb.commons.fsdb] Deleted file 'dummy_image' from 'myscan_001/fileset_001'.
    >>> fs.list_files()
    ['test_image', 'test_json']
    >>> print([f.name for f in fs.path().iterdir()])  # the file has been removed from the drive and the database
    ['test_json.json', 'test_image.png']
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    # Verify if the given `fs_id` exists in the local database
    if not self.file_exists(f_id):
        raise ValueError(f"File '{f_id}' does not exist in '{self.scan.id}/{self.id}'")

    # Use fileset-level exclusive lock for file deletion, as `store()` serializes the whole fileset.
    self.logger.debug(
        f"Deleting file '{f_id}' from '{self.scan.id}/{self.id}' as '{current_user.username}' user...")
    with self.db.lock_manager.acquire_lock(f"{self.scan.id}/{self.id}", LockType.EXCLUSIVE,
                                           current_user.username, LockLevel.FILESET):
        f = self.files[f_id]
        _delete_file(f)  # delete the file
        self.files.pop(f_id)  # remove the File instance from the fileset
        self.store()  # save the changes to the scan main JSON FILE (``files.json``)

    self.logger.debug(f"Done deleting file.")
    return

file_exists

file_exists(file_id)

Check if a given file ID exists in the database.

Parameters:

Name	Type	Description	Default
file_id	str	The ID of the file to check.	required

Returns:

Type	Description
bool	True if the file exists, False otherwise.

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_scan=True)
>>> scan = db.get_scan('myscan_001')
>>> scan.file_exists("myfile_001")
False
>>> scan.create_file("myfile_001")
>>> scan.file_exists("myfile_001")
True
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

def file_exists(self, file_id: str) -> bool:
    """Check if a given file ID exists in the database.

    Parameters
    ----------
    file_id : str
        The ID of the file to check.

    Returns
    -------
    bool
        ``True`` if the file exists, ``False`` otherwise.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_scan=True)
    >>> scan = db.get_scan('myscan_001')
    >>> scan.file_exists("myfile_001")
    False
    >>> scan.create_file("myfile_001")
    >>> scan.file_exists("myfile_001")
    True
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    return file_id in self.files

get_db

get_db()

Get parent database instance.

Returns:

Type	Description
DB	The parent database instance.

Source code in plantdb/commons/db.py

def get_db(self):
    """Get parent database instance.

    Returns
    -------
    plantdb.commons.db.DB
        The parent database instance.
    """
    return self.db

get_file

get_file(f_id)

Get a File instance, of given f_id, in the current fileset.

Parameters:

Name	Type	Description	Default
f_id	str	Name of the file to get.	required

Returns:

Type	Description
File	The retrieved file.

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan("myscan_001")
>>> fs = scan.get_fileset("fileset_001")
>>> f = fs.get_file("test_image")
>>> # To read the file you need to load the right reader from plantdb.commons.io
>>> from plantdb.commons.io import read_image
>>> img = read_image(f)
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

def get_file(self, f_id) -> "File":
    """Get a `File` instance, of given `f_id`, in the current fileset.

    Parameters
    ----------
    f_id : str
        Name of the file to get.

    Returns
    -------
    plantdb.commons.fsdb.core.File
        The retrieved file.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan("myscan_001")
    >>> fs = scan.get_fileset("fileset_001")
    >>> f = fs.get_file("test_image")
    >>> # To read the file you need to load the right reader from plantdb.commons.io
    >>> from plantdb.commons.io import read_image
    >>> img = read_image(f)
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    if not self.file_exists(f_id):
        raise FileNotFoundError(self, f_id)

    return self.files[f_id]

get_files

get_files(query=None, fuzzy=False)

Get the list of File instances defined in the current fileset, possibly filtered using a query.

Parameters:

Name	Type	Description	Default
query	dict	Query to use to get a list of files.	None
fuzzy	bool	Whether to use fuzzy matching or not, that is the use of regular expressions.	False

Returns:

Type	Description
list of plantdb.commons.fsdb.core.File	List of Files, filtered by the query if any.

See Also

plantdb.commons.fsdb._filter_query

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan('myscan_001')
>>> fs = scan.get_fileset('fileset_001')
>>> fs.get_files()
[<plantdb.commons.fsdb.core.File at ...>,
 <plantdb.commons.fsdb.core.File at ...>,
 <plantdb.commons.fsdb.core.File at ...>]
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

def get_files(self, query=None, fuzzy=False) -> list["File"]:
    """Get the list of `File` instances defined in the current fileset, possibly filtered using a `query`.

    Parameters
    ----------
    query : dict, optional
        Query to use to get a list of files.
    fuzzy : bool
        Whether to use fuzzy matching or not, that is the use of regular expressions.

    Returns
    -------
    list of plantdb.commons.fsdb.core.File
        List of `File`s, filtered by the query if any.

    See Also
    --------
    plantdb.commons.fsdb._filter_query

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan('myscan_001')
    >>> fs = scan.get_fileset('fileset_001')
    >>> fs.get_files()
    [<plantdb.commons.fsdb.core.File at ...>,
     <plantdb.commons.fsdb.core.File at ...>,
     <plantdb.commons.fsdb.core.File at ...>]
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    return _filter_query(list(self.files.values()), query, fuzzy)

get_id

get_id()

Get the fileset instance id.

Returns:

Type	Description
str	The id of the fileset instance.

Source code in plantdb/commons/db.py

def get_id(self):
    """Get the fileset instance id.

    Returns
    -------
    str
        The id of the fileset instance.
    """
    return deepcopy(self.id)

get_metadata

get_metadata(key=None, default={}, current_user=None, **kwargs)

Get the metadata associated with a fileset.

Parameters:

Name	Type	Description	Default
key	str	A key that should exist in the fileset's metadata.	None
default	Any	The default value to return if the key does not exist in the metadata. Default is an empty dictionary{}.	{}

Returns:

Type	Description
any	If key is None, returns a dictionary. Else, returns the value attached to this key.

Examples:

>>> import json
>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan("myscan_001")
>>> fs = scan.get_fileset('fileset_001')
>>> fs.set_metadata("test", "value")
>>> print(fs.get_metadata("test"))
'value'
>>> db._is_dummy=False  # to avoid cleaning up the temporary dummy database
>>> db.disconnect()
>>> db.connect()
>>> scan = db.get_scan("myscan_001")
>>> fs = scan.get_fileset('fileset_001')
>>> print(fs.get_metadata("test"))
'value'
>>> db._is_dummy=True
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@use_guest_as_default
@requires_permission(Permission.READ, check_scan_access=True)
def get_metadata(self, key=None, default={}, current_user=None, **kwargs) -> Any:
    """Get the metadata associated with a fileset.

    Parameters
    ----------
    key : str
        A key that should exist in the fileset's metadata.
    default : Any, optional
        The default value to return if the key does not exist in the metadata.
        Default is an empty dictionary``{}``.

    Returns
    -------
    any
        If `key` is ``None``, returns a dictionary.
        Else, returns the value attached to this key.

    Examples
    --------
    >>> import json
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan("myscan_001")
    >>> fs = scan.get_fileset('fileset_001')
    >>> fs.set_metadata("test", "value")
    >>> print(fs.get_metadata("test"))
    'value'
    >>> db._is_dummy=False  # to avoid cleaning up the temporary dummy database
    >>> db.disconnect()
    >>> db.connect()
    >>> scan = db.get_scan("myscan_001")
    >>> fs = scan.get_fileset('fileset_001')
    >>> print(fs.get_metadata("test"))
    'value'
    >>> db._is_dummy=True
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    # Use shared lock for read operations
    with self.db.lock_manager.acquire_lock(f"{self.scan.id}/{self.id}", LockType.SHARED, current_user.username,
                                           LockLevel.FILESET):
        return _get_metadata(self.metadata, key, default)

get_scan

get_scan()

Get parent scan instance.

Returns:

Type	Description
Scan	The parent scan instance.

Source code in plantdb/commons/db.py

def get_scan(self):
    """Get parent scan instance.

    Returns
    -------
    plantdb.commons.db.Scan
        The parent scan instance.
    """
    return self.scan

list_files

list_files(query=None, fuzzy=False)

Get the list of files identifiers in the fileset.

Parameters:

Name	Type	Description	Default
query	dict	A query to use to filter the returned list of files. The metadata must match given key and value from the query dictionary.	None
fuzzy	bool	Whether to use fuzzy matching or not, that is the use of regular expressions.	False

Returns:

Type	Description
list[str]	The list of file identifiers in the fileset.

See Also

plantdb.commons.fsdb._filter_query

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_scan=True, with_file=True)
>>> scan = db.get_scan("myscan_001")
>>> fs = scan.get_fileset("fileset_001")
>>> fs.list_files()
['dummy_image', 'test_image', 'test_json']
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

def list_files(self, query=None, fuzzy=False) -> list:
    """Get the list of files identifiers in the fileset.

    Parameters
    ----------
    query : dict, optional
        A query to use to filter the returned list of files.
        The metadata must match given ``key`` and ``value`` from the `query` dictionary.
    fuzzy : bool
        Whether to use fuzzy matching or not, that is the use of regular expressions.

    Returns
    -------
    list[str]
        The list of file identifiers in the fileset.

    See Also
    --------
    plantdb.commons.fsdb._filter_query

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_scan=True, with_file=True)
    >>> scan = db.get_scan("myscan_001")
    >>> fs = scan.get_fileset("fileset_001")
    >>> fs.list_files()
    ['dummy_image', 'test_image', 'test_json']
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    if query is None:
        return list(self.files.keys())
    else:
        return [f.id for f in _filter_query(list(self.files.values()), query, fuzzy)]

path

path()

Get the path to the local fileset.

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_scan=True, with_file=True)
>>> [scan.id for scan in db.get_scans()]  # list scan ids found in database
['myscan_001']
>>> scan = db.get_scan("myscan_001")
>>> print(scan.path())
/tmp/romidb_.../myscan_001
>>> [fs.id for fs in scan.get_filesets()]  # list fileset ids found in scan
['fileset_001']
>>> fs = scan.get_fileset("fileset_001")
>>> print(fs.path())
/tmp/romidb_.../myscan_001/fileset_001
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

def path(self) -> pathlib.Path:
    """Get the path to the local fileset.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_scan=True, with_file=True)
    >>> [scan.id for scan in db.get_scans()]  # list scan ids found in database
    ['myscan_001']
    >>> scan = db.get_scan("myscan_001")
    >>> print(scan.path())
    /tmp/romidb_.../myscan_001
    >>> [fs.id for fs in scan.get_filesets()]  # list fileset ids found in scan
    ['fileset_001']
    >>> fs = scan.get_fileset("fileset_001")
    >>> print(fs.path())
    /tmp/romidb_.../myscan_001/fileset_001
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    return _fileset_path(self)

set_metadata

set_metadata(data, value=None, current_user=None, **kwargs)

Add a new metadata to the fileset.

Parameters:

Name	Type	Description	Default
data	str or dict	If a string, a key to address the value. If a dictionary, update the metadata dictionary with data (value is then unused).	required
value	any	The value to assign to data if the latest is not a dictionary.	None

Raises:

Type	Description
PermissionError	If no user is authenticated. If the user lacks permission to modify the metadata.

Examples:

>>> import json
>>> from plantdb.commons.test_database import dummy_db
>>> from plantdb.commons.fsdb.path_helpers import _fileset_metadata_json_path
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan("myscan_001")
>>> fs = scan.get_fileset('fileset_001')
>>> fs.set_metadata("test", "value")
>>> p = _fileset_metadata_json_path(fs)
>>> print(p.exists())
True
>>> print(json.load(p.open(mode='r')))
{'test': 'value'}
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
@requires_permission(Permission.WRITE, check_scan_access=True)
def set_metadata(self, data, value=None, current_user=None, **kwargs) -> None:
    """Add a new metadata to the fileset.

    Parameters
    ----------
    data : str or dict
        If a string, a key to address the `value`.
        If a dictionary, update the metadata dictionary with `data` (`value` is then unused).
    value : any, optional
        The value to assign to `data` if the latest is not a dictionary.

    Raises
    ------
    PermissionError
        If no user is authenticated.
        If the user lacks permission to modify the metadata.

    Examples
    --------
    >>> import json
    >>> from plantdb.commons.test_database import dummy_db
    >>> from plantdb.commons.fsdb.path_helpers import _fileset_metadata_json_path
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan("myscan_001")
    >>> fs = scan.get_fileset('fileset_001')
    >>> fs.set_metadata("test", "value")
    >>> p = _fileset_metadata_json_path(fs)
    >>> print(p.exists())
    True
    >>> print(json.load(p.open(mode='r')))
    {'test': 'value'}
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    # The locking mechanism is handled by the `MetadataManager._update_metadata` method
    self._update_metadata(data, value, current_user, _store_fileset_metadata, cls_name="Fileset")

store

store()

Save changes to the scan main JSON FILE (files.json).

Source code in plantdb/commons/fsdb/core.py

def store(self) -> None:
    """Save changes to the scan main JSON FILE (``files.json``)."""
    self.scan.store()
    return

Scan

Scan(db, scan_id, timelapse_id=None)

Bases: Scan, MetadataManager

Implement Scan for the local File System DataBase from the abstract class db.Scan.

Implementation of a scan as a simple file structure with: * directory ${Scan.db.basedir}/${Scan.db.id} as scan root directory; * (OPTIONAL) directory ${Scan.db.basedir}/${Scan.db.id}/metadata containing JSON metadata file * (OPTIONAL) JSON file metadata.json with Scan metadata

Attributes:

Name	Type	Description
db	FSDB	A local database instance hosting this Scan instance.
id	str	The identifier of this Scan instance in the local database db.
metadata	dict	A metadata dictionary.
configs	dict	A dictionary listing the paths to the TOML configuration files. Indexed by configuration file stem, typically 'scan' or 'pipeline'.
filesets	dict[str, Fileset]	A dictionary of Fileset instances, indexed by their identifier.

Notes

Optional directory metadata & JSON file metadata.json are found when using method set_metadata().

See Also

plantdb.commons.db.Scan

Examples:

>>> import os
>>> from plantdb.commons.fsdb.core import Scan
>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db()
>>> # Example #1: Initialize a `Scan` object using an `FSBD` object:
>>> scan = Scan(db, '007')
>>> print(type(scan))
<class 'plantdb.commons.fsdb.core.Scan'>
>>> print(scan.path())  # the obtained path should be different as the path to the created `dummy_db` change...
/tmp/romidb_j0pbkoo0/007
>>> print(db.get_scan('007'))  # Note that it did NOT create this `Scan` in the database!
None
>>> print(os.listdir(db.path()))  # And it is NOT found under the `basedir` directory
['romidb']
>>> # HOWEVER if you add metadata to the `Scan` object:
>>> scan.set_metadata({'Name': "Bond... James Bond!"})
>>> print(scan.metadata)
{'Name': 'Bond... James Bond!'}
>>> print(db.get_scan('007'))  # The `Scan` is still not found in the database!
None
>>> print(os.listdir(db.path()))  # BUT it is now found under the `basedir` directory
['007', 'romidb']
>>> print(os.listdir(os.path.join(db.path(), scan.id)))  # Same goes for the metadata
['metadata']
>>> print(os.listdir(os.path.join(db.path(), scan.id, "metadata")))  # Same goes for the metadata
>>> db.disconnect()  # clean up (delete) the temporary dummy database

>>> # Example #2: Initialize a `Scan` object that belongs to a `Timelapse` object
>>> db = dummy_db()
>>> db.timelapse_exists("Star_Wars")
False
>>> scan = Scan(db, 'A_New_Hope', timelapse_id='Star_Wars')
>>> db.timelapse_exists("Star_Wars")
True
>>> scan.path()
PosixPath('/tmp/ROMI_DB_.../Star_Wars/A_New_Hope')

>>> # Example #3: Get it from an `FSDB` object:
>>> db = dummy_db()
>>> scan = db.create_scan('007')
>>> print(type(scan))
<class 'plantdb.commons.fsdb.core.Scan'>
>>> print(db.get_scan('007'))  # This time the `Scan` object is found in the `FSBD`
<plantdb.commons.fsdb.core.Scan object at 0x7f34fc860fd0>
>>> print(os.listdir(db.path()))  # And it is found under the `basedir` directory
['007', 'romidb']
>>> print(os.listdir(os.path.join(db.path(), scan.id)))  # Same goes for the metadata
['metadata']
>>> db._is_dummy = False  # to avoid cleaning up the
>>> db.disconnect()
>>> # When reconnecting to db, if created scan is EMPTY (no Fileset & File) it is not found!
>>> db.connect()
>>> print(db.get_scan('007'))
None
>>> db._is_dummy = True  # to clean up the temporary dummy database
>>> db.disconnect()  # clean up (delete) the temporary dummy database

>>> # Example #4: Use an existing database:
>>> from os import environ
>>> from plantdb.commons.fsdb.core import FSDB
>>> db = FSDB(environ.get('ROMI_DB', "/data/ROMI/DB/"))
>>> db.connect()
>>> scan = db.get_scan('sango_90_300_36')
>>> scan.get_metadata()

Scan dataset constructor.

Parameters:

Name	Type	Description	Default
db	FSDB	The database to put/find the scan dataset.	required
scan_id	str	The scan dataset name, should be unique in the db.	required

Returned by:

• Reference API commons commons
  • fsdb
    • core
      •  FSDB
        •  create_scan
        •  get_scan
      •  TimeLapse
        •  create_scan
        •  get_scan
    • fsdb_tools  add_metadata_to_scan
  • testing  FSDBTestCase  get_test_scan

Used by:

• Reference API
  • client client metadata_app db_ops
    •  scan_needs_migration
    •  write_scan_metadata
  • commons commons
    • fsdb core  Fileset
    • utils  locate_task_filesets
  • server server
    • core utils
      •  compute_fileset_matches
      •  get_scan_date
    • services scan
      •  get_scan_data
      •  get_scan_info

Source code in plantdb/commons/fsdb/core.py

def __init__(self, db, scan_id, timelapse_id=None):
    """Scan dataset constructor.

    Parameters
    ----------
    db : plantdb.commons.fsdb.core.FSDB
        The database to put/find the scan dataset.
    scan_id : str
        The scan dataset name, should be unique in the `db`.
    """
    super().__init__(db, scan_id)
    # Defines attributes:
    self.metadata = {}
    self.filesets = {}
    self.configs = {}
    self.measures = None

    self.session_manager = self.db.session_manager
    self.logger = self.db.logger

    if timelapse_id is not None:
        if not self.db.timelapse_exists(timelapse_id):
            self.db.create_timelapse(timelapse_id)
        self.metadata["timelapse"] = {"id": timelapse_id}

owner property

owner

A property method to retrieve or set the owner of a resource.

If the owner is not already defined in the resource's metadata, this property ensures that a default owner is assigned (using the 'guest' user). The updated metadata is then stored in the database, and the resource is reloaded.

Returns:

Type	Description
str	The owner of the resource, as defined in the metadata.

See Also

rbac_manager.ensure_scan_owner : Ensures the presence of a valid owner in the metadata. _store_scan_metadata : Saves updated metadata to the database. db.reload : Reloads the resource from the database after updates.

timelapse property

timelapse

Property to get parent TimeLapse instance if defined, else None.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> scan = tl.create_scan('mytl_001_01')
>>> scan.timelapse.id
'mytl_001'
>>> db.disconnect()

change_owner

change_owner(new_owner, current_user=None, **kwargs)

Change the owner of the scan.

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db()
>>> new_scan = db.create_scan('007', metadata={'project': 'GoldenEye'})  # create a new scan dataset
>>> print(new_scan.get_metadata('owner'))
admin
>>> new_scan.change_owner('guest')
>>> print(new_scan.get_metadata('owner'))
guest
>>> new_scan.change_owner('admin')

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
@requires_permission(Permission.DELETE, check_scan_access=True)
def change_owner(self, new_owner, current_user=None, **kwargs) -> None:
    """Change the owner of the scan.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db()
    >>> new_scan = db.create_scan('007', metadata={'project': 'GoldenEye'})  # create a new scan dataset
    >>> print(new_scan.get_metadata('owner'))
    admin
    >>> new_scan.change_owner('guest')
    >>> print(new_scan.get_metadata('owner'))
    guest
    >>> new_scan.change_owner('admin')
    """
    # Verify that the new owner exists:
    if not self.db.rbac_manager.users.exists(new_owner):
        self.logger.error(f"Owner '{new_owner}' does not exist, can not change ownership!")
        return

    # Use exclusive lock for metadata updates
    self.logger.debug(f"Updating '{self.id}' scan owner to '{new_owner}' user...")
    with self.db.lock_manager.acquire_lock(self.id, LockType.EXCLUSIVE, current_user.username, LockLevel.SCAN):
        # Update metadata
        _set_metadata(self.metadata, 'owner', new_owner)
        # Ensure modification timestamp
        _set_metadata(self.metadata, 'last_modified', iso_date_now())
        _store_scan_metadata(self)

    self.logger.debug(f"Done updating the scan owner.")
    return

create_fileset

create_fileset(fs_id, metadata=None, current_user=None, **kwargs)

Create a new Fileset instance in the local database attached to the current Scan instance.

Parameters:

Name	Type	Description	Default
fs_id	str	The name of the fileset to create. It should not exist in the current Scan instance.	required
metadata	dict	A dictionary with the initial metadata for this new fileset.	None

Returns:

Type	Description
Fileset	The Fileset instance created in the current Scan instance.

Raises:

Type	Description
PermissionError	If no user is authenticated. If the user lacks permission to create a fileset.
FilesetExistsError	If the fs_id already exists in the local database.
ValueError	If the given fs_id is invalid.

See Also

plantdb.commons.fsdb.validation._is_valid_id plantdb.commons.fsdb.file_ops._make_fileset

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_fileset=True)
>>> scan = db.get_scan('myscan_001')
>>> scan.list_filesets()
['fileset_001']
>>> new_fs = scan.create_fileset('fs_007')
>>> scan.list_filesets()
['fileset_001', 'fs_007']
>>> wrong_fs = scan.create_fileset('fileset_001')
OSError: Given fileset identifier 'fileset_001' already exists!
>>> wrong_fs = scan.create_fileset('fileset/001')
OSError: Invalid fileset identifier 'fileset/001'!
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
@requires_permission(Permission.WRITE, check_scan_access=True)
def create_fileset(self, fs_id, metadata=None, current_user=None, **kwargs) -> "Fileset":
    """Create a new `Fileset` instance in the local database attached to the current `Scan` instance.

    Parameters
    ----------
    fs_id : str
        The name of the fileset to create. It should not exist in the current `Scan` instance.
    metadata : dict, optional
        A dictionary with the initial metadata for this new fileset.

    Returns
    -------
    plantdb.commons.fsdb.core.Fileset
        The `Fileset` instance created in the current `Scan` instance.

    Raises
    ------
    PermissionError
        If no user is authenticated.
        If the user lacks permission to create a fileset.
    FilesetExistsError
        If the ``fs_id`` already exists in the local database.
    ValueError
        If the given ``fs_id`` is invalid.

    See Also
    --------
    plantdb.commons.fsdb.validation._is_valid_id
    plantdb.commons.fsdb.file_ops._make_fileset

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_fileset=True)
    >>> scan = db.get_scan('myscan_001')
    >>> scan.list_filesets()
    ['fileset_001']
    >>> new_fs = scan.create_fileset('fs_007')
    >>> scan.list_filesets()
    ['fileset_001', 'fs_007']
    >>> wrong_fs = scan.create_fileset('fileset_001')
    OSError: Given fileset identifier 'fileset_001' already exists!
    >>> wrong_fs = scan.create_fileset('fileset/001')
    OSError: Invalid fileset identifier 'fileset/001'!
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    # Access scan metadata directly to avoid nested lock acquisition
    metadata = _get_metadata(self.metadata, None, {})

    # Verify if the given `fs_id` is valid
    if not _is_valid_id(fs_id):
        raise ValueError(f"Invalid fileset identifier '{fs_id}'!")
    # Verify if the given `fs_id` already exists in the local database
    if self.fileset_exists(fs_id):
        raise FilesetExistsError(self, fs_id)

    # Use fileset-level exclusive lock for fileset creation
    self.logger.debug(f"Creating a fileset '{fs_id}' in scan '{self.id}' as '{current_user.username}' user...")
    with self.db.lock_manager.acquire_lock(f"{self.id}/{fs_id}", LockType.EXCLUSIVE, current_user.username,
                                           LockLevel.FILESET):
        # Create the new Fileset
        fileset = Fileset(self, fs_id)  # Initialize a new Fileset instance
        _make_fileset(fileset)  # Create directory structure

        # Set initial metadata
        initial_metadata = metadata or {}
        now = iso_date_now()
        initial_metadata['created'] = now  # creation timestamp
        initial_metadata['last_modified'] = now  # modification timestamp
        initial_metadata['created_by'] = current_user.fullname

        # Cannot use fileset.set_metadata(initial_metadata) here as ownership is not granted yet!
        _set_metadata(fileset.metadata, initial_metadata, None)  # add metadata dictionary to the new fileset
        _store_fileset_metadata(fileset)

        self.filesets.update({fs_id: fileset})  # Update scan's filesets dictionary
        self.store()  # Store fileset instance to the JSON

    self.logger.debug(f"Done creating the fileset.")
    return fileset

delete_fileset

delete_fileset(fs_id, current_user=None, **kwargs)

Delete a given fileset from the scan dataset.

Parameters:

Name	Type	Description	Default
fs_id	str	Name of the fileset to delete.	required

Raises:

Type	Description
PermissionError	If no user is authenticated. If the user does not have permission to delete this fileset.
ValueError	If the fileset does not exist.

See Also

plantdb.commons.fsdb.file_ops._delete_fileset

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan('myscan_001')
>>> scan.list_filesets()
['fileset_001']
>>> scan.delete_fileset('fileset_001')
>>> scan.list_filesets()
[]
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
@requires_permission(Permission.DELETE, check_scan_access=True)
def delete_fileset(self, fs_id, current_user=None, **kwargs) -> None:
    """Delete a given fileset from the scan dataset.

    Parameters
    ----------
    fs_id : str
        Name of the fileset to delete.

    Raises
    ------
    PermissionError
        If no user is authenticated.
        If the user does not have permission to delete this fileset.
    ValueError
        If the fileset does not exist.

    See Also
    --------
    plantdb.commons.fsdb.file_ops._delete_fileset

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan('myscan_001')
    >>> scan.list_filesets()
    ['fileset_001']
    >>> scan.delete_fileset('fileset_001')
    >>> scan.list_filesets()
    []
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    # Verify if the given `fs_id` exists in the local database
    if not self.fileset_exists(fs_id):
        raise ValueError(f"Fileset '{fs_id}' does not exist in scan '{self.id}'")

    # Use exclusive lock for fileset deletion
    self.logger.debug(f"Deleting fileset '{fs_id}' from scan '{self.id}' as '{current_user.username}' user...")
    with self.db.lock_manager.acquire_lock(f"{self.id}/{fs_id}", LockType.EXCLUSIVE, current_user.username,
                                           LockLevel.FILESET):
        fs = self.filesets[fs_id]
        _delete_fileset(fs)  # delete the fileset
        self.filesets.pop(fs_id)  # remove the Fileset instance from the scan
        self.store()  # save the changes to the scan main JSON FILE (``files.json``)

    self.logger.debug(f"Done deleting fileset.")
    return

fileset_exists

fileset_exists(fileset_id)

Check if a given fileset ID exists in the database.

Parameters:

Name	Type	Description	Default
fileset_id	str	The ID of the fileset to check.	required

Returns:

Type	Description
bool	True if the fileset exists, False otherwise.

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_scan=True)
>>> scan = db.get_scan('myscan_001')
>>> scan.fileset_exists("myfileset_001")
False
>>> scan.create_fileset("myfileset_001")
>>> scan.fileset_exists("myfileset_001")
True
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

def fileset_exists(self, fileset_id: str) -> bool:
    """Check if a given fileset ID exists in the database.

    Parameters
    ----------
    fileset_id : str
        The ID of the fileset to check.

    Returns
    -------
    bool
        ``True`` if the fileset exists, ``False`` otherwise.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_scan=True)
    >>> scan = db.get_scan('myscan_001')
    >>> scan.fileset_exists("myfileset_001")
    False
    >>> scan.create_fileset("myfileset_001")
    >>> scan.fileset_exists("myfileset_001")
    True
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    return fileset_id in self.filesets

get_configuration

get_configuration(key, current_user=None, **kwargs)

Get the configurations associated with a scan.

Parameters:

Name	Type	Description	Default
key	str	The scan's configuration name to recover, usually 'scan' or 'pipeline'.	required

Returns:

Type	Description
dict	The configuration dictionary.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> scan = db.get_scan('real_plant_analyzed')
>>> # Get the scan configuration
>>> scan_cfg = scan.get_configuration('scan')
>>> print(scan_cfg['ScanPath']['class_name'])
Circle
>>> # Get the reconstruction pipeline configuration:
>>> pipe_cfg = scan.get_configuration('pipeline')
>>> print(pipe_cfg['PointCloud'])
{'upstream_task': 'Voxels', 'level_set_value': 1.0}
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@use_guest_as_default
@requires_permission(Permission.READ, check_scan_access=True)
def get_configuration(self, key, current_user=None, **kwargs) -> dict[str, Any]:
    """Get the configurations associated with a scan.

    Parameters
    ----------
    key : str
        The scan's configuration name to recover, usually `'scan'` or `'pipeline'`.

    Returns
    -------
    dict
        The configuration dictionary.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> scan = db.get_scan('real_plant_analyzed')
    >>> # Get the scan configuration
    >>> scan_cfg = scan.get_configuration('scan')
    >>> print(scan_cfg['ScanPath']['class_name'])
    Circle
    >>> # Get the reconstruction pipeline configuration:
    >>> pipe_cfg = scan.get_configuration('pipeline')
    >>> print(pipe_cfg['PointCloud'])
    {'upstream_task': 'Voxels', 'level_set_value': 1.0}
    >>> db.disconnect()
    """
    from plantdb.commons.io import read_toml
    if key not in self.configs:
        self.logger.warning(f"No configuration found for {key}.")
        return {}

    # Use shared lock for read operations
    with self.db.lock_manager.acquire_lock(self.id, LockType.SHARED, current_user.username, LockLevel.SCAN):
        cfg_path = self.configs[key]
        return read_toml(cfg_path)

get_db

get_db()

Get parent database instance.

Returns:

Type	Description
DB	Database instance where to find the scan.

Source code in plantdb/commons/db.py

def get_db(self):
    """Get parent database instance.

    Returns
    -------
    plantdb.commons.db.DB
        Database instance where to find the scan.
    """
    return self.db

get_fileset

get_fileset(fs_id)

Get a Fileset instance, of given id, in the current scan dataset.

Parameters:

Name	Type	Description	Default
fs_id	str	The name of the fileset to get.	required

Returns:

Type	Description
Fileset	The retrieved or created fileset.

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_fileset=True)
>>> scan = db.get_scan('myscan_001')
>>> scan.list_filesets()
['fileset_001']
>>> new_fileset = scan.create_fileset('007')
>>> print(new_fileset)
<plantdb.commons.fsdb.core.Fileset object at ...>
>>> scan.list_filesets()
['fileset_001', '007']
>>> unknown_fs = scan.get_fileset('unknown')
plantdb.commons.fsdb.core.FilesetNotFoundError: Unknown fileset id 'unknown'!
>>> print(unknown_fs)
None
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

def get_fileset(self, fs_id) -> "Fileset":
    """Get a `Fileset` instance, of given `id`, in the current scan dataset.

    Parameters
    ----------
    fs_id : str
        The name of the fileset to get.

    Returns
    -------
    plantdb.commons.fsdb.core.Fileset
        The retrieved or created fileset.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_fileset=True)
    >>> scan = db.get_scan('myscan_001')
    >>> scan.list_filesets()
    ['fileset_001']
    >>> new_fileset = scan.create_fileset('007')
    >>> print(new_fileset)
    <plantdb.commons.fsdb.core.Fileset object at ...>
    >>> scan.list_filesets()
    ['fileset_001', '007']
    >>> unknown_fs = scan.get_fileset('unknown')
    plantdb.commons.fsdb.core.FilesetNotFoundError: Unknown fileset id 'unknown'!
    >>> print(unknown_fs)
    None
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    if not self.fileset_exists(fs_id):
        raise FilesetNotFoundError(self, fs_id)

    return self.filesets[fs_id]

get_filesets

get_filesets(query=None, fuzzy=False)

Get the list of Fileset instances defined in the current scan dataset, possibly filtered using a query.

Parameters:

Name	Type	Description	Default
query	dict	A query to use to filter the returned list of files. The metadata must match given key and value from the query dictionary.	None
fuzzy	bool	Whether to use fuzzy matching or not, that is the use of regular expressions.	False

Returns:

Type	Description
list of plantdb.commons.fsdb.core.Fileset	List of Filesets, filtered by the query if any.

See Also

plantdb.commons.fsdb._filter_query

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_fileset=True)
>>> scan = db.get_scan('myscan_001')
>>> scan.get_filesets()
[<plantdb.commons.fsdb.core.Fileset at ...>]
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

def get_filesets(self, query=None, fuzzy=False) -> list["Fileset"]:
    """Get the list of `Fileset` instances defined in the current scan dataset, possibly filtered using a `query`.

    Parameters
    ----------
    query : dict, optional
        A query to use to filter the returned list of files.
        The metadata must match given ``key`` and ``value`` from the `query` dictionary.
    fuzzy : bool
        Whether to use fuzzy matching or not, that is the use of regular expressions.

    Returns
    -------
    list of plantdb.commons.fsdb.core.Fileset
        List of `Fileset`s, filtered by the `query` if any.

    See Also
    --------
    plantdb.commons.fsdb._filter_query

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_fileset=True)
    >>> scan = db.get_scan('myscan_001')
    >>> scan.get_filesets()
    [<plantdb.commons.fsdb.core.Fileset at ...>]
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    return [self.get_fileset(fs.id) for fs in _filter_query(list(self.filesets.values()), query, fuzzy)]

get_id

get_id()

Get the scan instance id.

Returns:

Type	Description
str	Id of the scan instance.

Source code in plantdb/commons/db.py

def get_id(self):
    """Get the scan instance id.

    Returns
    -------
    str
        Id of the scan instance.
    """
    return deepcopy(self.id)

get_measures

get_measures(key=None, current_user=None, **kwargs)

Get the manual measurements associated with a scan.

Parameters:

Name	Type	Description	Default
key	str	A key that should exist in the scan's manual measurements.	None

Returns:

Type	Description
any	If key is None, returns a dictionary. Else, returns the value attached to this key.

Notes

These manual measurements should be a JSON file named measures.json. It is located at the root folder of the scan dataset.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> scan = db.get_scan('real_plant_analyzed')
>>> measures = scan.get_measures()
>>> print(list(measures.keys()))
['angles', 'internodes']
>>> print(measures['angles'][:3])  # print the first 3 measures angles
[2.356194490192345, 0.9599310885968813, 2.1816615649929116]
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@use_guest_as_default
@requires_permission(Permission.READ, check_scan_access=True)
def get_measures(self, key=None, current_user=None, **kwargs):
    """Get the manual measurements associated with a scan.

    Parameters
    ----------
    key : str
        A key that should exist in the scan's manual measurements.

    Returns
    -------
    any
        If `key` is ``None``, returns a dictionary.
        Else, returns the value attached to this key.

    Notes
    -----
    These manual measurements should be a JSON file named `measures.json`.
    It is located at the root folder of the scan dataset.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> scan = db.get_scan('real_plant_analyzed')
    >>> measures = scan.get_measures()
    >>> print(list(measures.keys()))
    ['angles', 'internodes']
    >>> print(measures['angles'][:3])  # print the first 3 measures angles
    [2.356194490192345, 0.9599310885968813, 2.1816615649929116]
    >>> db.disconnect()
    """
    # Use shared lock for read operations
    with self.db.lock_manager.acquire_lock(self.id, LockType.SHARED, current_user.username, LockLevel.SCAN):
        return _get_metadata(self.measures, key, default={})

get_metadata

get_metadata(key=None, default={}, current_user=None, **kwargs)

Get the metadata associated with a scan.

Parameters:

Name	Type	Description	Default
key	str	A key that should exist in the scan's metadata.	None
default	Any	The default value to return if the key does not exist in the metadata. Default is an empty dictionary{}.	{}

Returns:

Type	Description
any	If key is None, returns a dictionary. Else, returns the value attached to this key.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database()
>>> db.connect()
>>> token = db.login('admin', 'admin')
>>> scan = db.get_scan('real_plant_analyzed')
>>> scan.get_metadata('owner')
'guest'
>>> db.logout()
>>> scan.get_metadata('owner')  # Can still access it as 'guest' user

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@use_guest_as_default
@requires_permission(Permission.READ, check_scan_access=False)
def get_metadata(self, key=None, default={}, current_user=None, **kwargs) -> Any:
    """Get the metadata associated with a scan.

    Parameters
    ----------
    key : str
        A key that should exist in the scan's metadata.
    default : Any, optional
        The default value to return if the key does not exist in the metadata.
        Default is an empty dictionary``{}``.

    Returns
    -------
    any
        If `key` is ``None``, returns a dictionary.
        Else, returns the value attached to this key.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database()
    >>> db.connect()
    >>> token = db.login('admin', 'admin')
    >>> scan = db.get_scan('real_plant_analyzed')
    >>> scan.get_metadata('owner')
    'guest'
    >>> db.logout()
    >>> scan.get_metadata('owner')  # Can still access it as 'guest' user
    """
    # Use shared lock for read operations
    with self.db.lock_manager.acquire_lock(self.id, LockType.SHARED, current_user.username, LockLevel.SCAN):
        return _get_metadata(self.metadata, key, default)

get_timelapse

get_timelapse()

Get the parent TimeLapse instance if this scan belongs to one, else None.

Returns:

Type	Description
TimeLapse | None	The parent TimeLapse instance, or None if standalone.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> scan = tl.create_scan('mytl_001_01')
>>> scan.get_timelapse().id
'mytl_001'
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

def get_timelapse(self) -> Optional["TimeLapse"]:
    """Get the parent `TimeLapse` instance if this scan belongs to one, else `None`.

    Returns
    -------
    plantdb.commons.fsdb.core.TimeLapse | None
        The parent `TimeLapse` instance, or `None` if standalone.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> tl = db.create_timelapse('mytl_001')
    >>> scan = tl.create_scan('mytl_001_01')
    >>> scan.get_timelapse().id
    'mytl_001'
    >>> db.disconnect()
    """
    if isinstance(self.metadata, dict):
        tl_meta = self.metadata.get("timelapse")
        if isinstance(tl_meta, dict) and tl_meta.get("id"):
            return self.db.get_timelapse(tl_meta["id"])
    return None

group_share

group_share(groups, current_user=None, **kwargs)

Change the group sharing of the scan.

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db()
>>> new_scan = db.create_scan('007', metadata={'project': 'GoldenEye'})  # create a new scan dataset
>>> print(new_scan.get_metadata('owner'))
admin
>>> print(new_scan.get_metadata('sharing', default=None))
None
>>> new_group = db.create_group('dummy_group', ['admin', 'guest'], 'A test group')
>>> new_scan.group_share('dummy_group')
>>> print(new_scan.get_metadata('sharing'))
['dummy_group']

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
@requires_permission(Permission.DELETE, check_scan_access=True)
def group_share(self, groups, current_user=None, **kwargs) -> None:
    """Change the group sharing of the scan.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db()
    >>> new_scan = db.create_scan('007', metadata={'project': 'GoldenEye'})  # create a new scan dataset
    >>> print(new_scan.get_metadata('owner'))
    admin
    >>> print(new_scan.get_metadata('sharing', default=None))
    None
    >>> new_group = db.create_group('dummy_group', ['admin', 'guest'], 'A test group')
    >>> new_scan.group_share('dummy_group')
    >>> print(new_scan.get_metadata('sharing'))
    ['dummy_group']
    """
    if isinstance(groups, str):
        groups = [groups]
    # Verify that the new group(s) exists:
    valid_groups = self.db.rbac_manager.valid_sharing_groups(groups)

    if len(valid_groups) == 0:
        self.logger.error(f"No valid group(s) given, can not share to non-existent group!")
        return

    # Validate sharing groups if present
    if 'sharing' in self.metadata:
        sharing_groups = self.metadata['sharing']
        if not self.db.rbac_manager.validate_sharing_groups(sharing_groups):
            raise ValueError("One or more sharing groups do not exist")

    # Use exclusive lock for metadata updates
    valid_groups_str = ", ".join(valid_groups)
    self.logger.debug(f"Updating '{self.id}' scan group sharing to: '{valid_groups_str}'")
    with self.db.lock_manager.acquire_lock(self.id, LockType.EXCLUSIVE, current_user.username, LockLevel.SCAN):
        # Update metadata
        _set_metadata(self.metadata, 'sharing', valid_groups)
        # Ensure modification timestamp
        _set_metadata(self.metadata, 'last_modified', iso_date_now())
        _store_scan_metadata(self)

    self.logger.debug(f"Done updating the scan group sharing.")
    return

is_locked

is_locked()

Check if a scan is locked in the system.

Returns:

Type	Description
bool	True if the scan is locked (having neither an exclusive lock nor any shared locks), False otherwise.

See Also

ScanManager.lock_manager: Component responsible for managing scan locks.

Source code in plantdb/commons/fsdb/core.py

def is_locked(self) -> bool:
    """Check if a scan is locked in the system.

    Returns
    -------
    bool
        ``True`` if the scan is locked (having neither an exclusive lock nor any shared locks), ``False`` otherwise.

    See Also
    --------
    ScanManager.lock_manager: Component responsible for managing scan locks.
    """
    return self.db.is_scan_locked(self.id)

list_filesets

list_filesets(query=None, fuzzy=False)

Get the list of filesets identifiers in the scan dataset.

Parameters:

Name	Type	Description	Default
query	dict	A query to use to filter the returned list of filesets. The metadata must match given key and value from the query dictionary.	None
fuzzy	bool	Whether to use fuzzy matching or not, that is the use of regular expressions.	False

Returns:

Type	Description
list of str	The list of filesets identifiers in the scan dataset.

See Also

plantdb.commons.fsdb._filter_query

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan("myscan_001")
>>> scan.list_filesets()
['fileset_001']
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

def list_filesets(self, query=None, fuzzy=False) -> list[str]:
    """Get the list of filesets identifiers in the scan dataset.

    Parameters
    ----------
    query : dict, optional
        A query to use to filter the returned list of filesets.
        The metadata must match given ``key`` and ``value`` from the `query` dictionary.
    fuzzy : bool
        Whether to use fuzzy matching or not, that is the use of regular expressions.

    Returns
    -------
    list of str
        The list of filesets identifiers in the scan dataset.

    See Also
    --------
    plantdb.commons.fsdb._filter_query

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan("myscan_001")
    >>> scan.list_filesets()
    ['fileset_001']
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    if query == None:
        return list(self.filesets.keys())
    else:
        return [fs.id for fs in _filter_query(list(self.filesets.values()), query, fuzzy)]

path

path()

Get the path to the local scan dataset.

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan("myscan_001")
>>> scan.path()  # should be '/tmp/romidb_.../myscan_001'
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

def path(self) -> pathlib.Path:
    """Get the path to the local scan dataset.

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan("myscan_001")
    >>> scan.path()  # should be '/tmp/romidb_.../myscan_001'
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    return _scan_path(self)

set_metadata

set_metadata(data, value=None, current_user=None, **kwargs)

Add a new metadata to the scan.

Parameters:

Name	Type	Description	Default
data	str or dict	If a string, a key to address the value. If a dictionary, update the metadata dictionary with data (value is then unused).	required
value	any	The value to assign to data if the latest is not a dictionary.	None

Raises:

Type	Description
PermissionError	If no user is authenticated. If the user lacks permission to modify the metadata.
ValueError	If metadata validation fails

Examples:

>>> from plantdb.commons.test_database import dummy_db
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan("myscan_001")
>>> print(scan.get_metadata('test'))
1
>>> scan.set_metadata("test", "value")
>>> print(scan.get_metadata('test'))
value
>>> # Changing scan ownership through metadata is blocked:
>>> scan.set_metadata("owner", "guest")
WARNING  [FSDB] Excluding 'owner' key from Scan metadata update!
>>> # Read the scan metadata file
>>> import json
>>> from plantdb.commons.fsdb.path_helpers import _scan_metadata_path
>>> p = _scan_metadata_path(scan)
>>> print(p.exists())
True
>>> print(json.load(p.open(mode='r')))
{'test': 'value'}
>>> db.disconnect()  # clean up (delete) the temporary dummy database

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
@requires_permission(Permission.WRITE, check_scan_access=True)
def set_metadata(self, data, value=None, current_user=None, **kwargs) -> None:
    """Add a new metadata to the scan.

    Parameters
    ----------
    data : str or dict
        If a string, a key to address the `value`.
        If a dictionary, update the metadata dictionary with `data` (`value` is then unused).
    value : any, optional
        The value to assign to `data` if the latest is not a dictionary.

    Raises
    ------
    PermissionError
        If no user is authenticated.
        If the user lacks permission to modify the metadata.
    ValueError
        If metadata validation fails

    Examples
    --------
    >>> from plantdb.commons.test_database import dummy_db
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan("myscan_001")
    >>> print(scan.get_metadata('test'))
    1
    >>> scan.set_metadata("test", "value")
    >>> print(scan.get_metadata('test'))
    value
    >>> # Changing scan ownership through metadata is blocked:
    >>> scan.set_metadata("owner", "guest")
    WARNING  [FSDB] Excluding 'owner' key from Scan metadata update!
    >>> # Read the scan metadata file
    >>> import json
    >>> from plantdb.commons.fsdb.path_helpers import _scan_metadata_path
    >>> p = _scan_metadata_path(scan)
    >>> print(p.exists())
    True
    >>> print(json.load(p.open(mode='r')))
    {'test': 'value'}
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    # The locking mechanism is handled by the `MetadataManager._update_metadata` method
    self._update_metadata(data, value, current_user, _store_scan_metadata, cls_name="Scan")

store

store()

Save changes to the scan main JSON FILE (files.json).

Source code in plantdb/commons/fsdb/core.py

def store(self) -> None:
    """Save changes to the scan main JSON FILE (``files.json``)."""
    _store_scan(self)
    return

TimeLapse

TimeLapse(db, tl_id, metadata=None, created_at=None, owner=None, scans=None)

Bases: TimeLapse, MetadataManager

Implement TimeLapse for the local File System DataBase.

Implementation of a timelapse container as a file structure with: * directory ${TimeLapse.db.basedir}/${TimeLapse.id} as timelapse root directory; * JSON marker file timelapse.json containing timelapse metadata; * member scans located under ${TimeLapse.db.basedir}/${TimeLapse.id}/${Scan.id}.

Attributes:

Name	Type	Description
db	FSDB	A local database instance hosting this TimeLapse instance.
id	str	The identifier of this TimeLapse instance in the local database db.
metadata	dict	A metadata dictionary.
created_at	str	ISO-formatted creation timestamp.

See Also

plantdb.commons.db.TimeLapse plantdb.commons.fsdb.core.Scan

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> tl.list_scans()
[]
>>> scan = tl.create_scan('mytl_001_01')
>>> tl.list_scans()
['mytl_001_01']
>>> scan.timelapse.id
'mytl_001'
>>> tl.to_dict()['counts']
{'scans': 1}
>>> db.disconnect()

Returned by:

• Reference API commons commons fsdb core
  •  FSDB
    •  create_timelapse
    •  get_timelapse
  •  Scan  get_timelapse

Source code in plantdb/commons/fsdb/core.py

def __init__(self, db, tl_id, metadata=None, created_at=None, owner=None, scans=None):
    super().__init__(db, tl_id)
    self.metadata = metadata if metadata is not None else {}
    self.created_at = created_at or iso_date_now()
    self.owner = owner
    self.scans = list(scans) if scans else []
    self.session_manager = self.db.session_manager
    self.logger = self.db.logger

__getitem__

__getitem__(item)

Dict-like access for backward compatibility with dictionary return values.

Source code in plantdb/commons/fsdb/core.py

def __getitem__(self, item):
    """Dict-like access for backward compatibility with dictionary return values."""
    if item == "id":
        return self.id
    elif item == "created_at":
        return self.created_at
    elif item == "metadata":
        return self.metadata
    elif item == "owner":
        return self.owner
    elif item == "scans":
        return list(self.scans)
    elif item == "counts":
        member_scans = [
            s for s in self.db.scans.values()
            if isinstance(s.metadata, dict) and s.metadata.get("timelapse", {}).get("id") == self.id
        ]
        return {"scans": len(member_scans)}
    elif item in self.metadata:
        return self.metadata[item]
    raise KeyError(item)

create_scan

create_scan(scan_id, metadata=None, current_user=None, **kwargs)

Create a new member scan in this timelapse.

Parameters:

Name	Type	Description	Default
scan_id	str	The identifier of the scan to create.	required
metadata	dict	Additional metadata for the scan. The timelapse.id will automatically be set.	None

Returns:

Type	Description
Scan	The created Scan instance.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> scan = tl.create_scan('mytl_001_01')
>>> scan.path().exists()
True
>>> scan.timelapse.id
'mytl_001'
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
@requires_permission(Permission.CREATE, check_scan_access=False)
def create_scan(self, scan_id: str, metadata: dict = None, current_user=None, **kwargs) -> "Scan":
    """Create a new member scan in this timelapse.

    Parameters
    ----------
    scan_id : str
        The identifier of the scan to create.
    metadata : dict, optional
        Additional metadata for the scan. The `timelapse.id` will automatically be set.

    Returns
    -------
    plantdb.commons.fsdb.core.Scan
        The created `Scan` instance.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> tl = db.create_timelapse('mytl_001')
    >>> scan = tl.create_scan('mytl_001_01')
    >>> scan.path().exists()
    True
    >>> scan.timelapse.id
    'mytl_001'
    >>> db.disconnect()
    """
    meta = copy.deepcopy(metadata) if metadata else {}
    # Add timelapse ID to the scan metadata
    tl_meta = meta.setdefault("timelapse", {})
    if isinstance(tl_meta, dict):
        tl_meta["id"] = self.id
    return self.db.create_scan(scan_id, metadata=meta, current_user=current_user, **kwargs)

delete

delete(recursive=False, current_user=None, **kwargs)

Delete this timelapse container from the database.

Parameters:

Name	Type	Description	Default
recursive	bool	Whether to delete member scans recursively. Defaults to False.	False

Returns:

Type	Description
bool	True on successful deletion.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> db.list_timelapses()
['mytl_001']
>>> tl.delete()
True
>>> db.list_timelapses()
[]
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

def delete(self, recursive: bool = False, current_user=None, **kwargs) -> bool:
    """Delete this timelapse container from the database.

    Parameters
    ----------
    recursive : bool, optional
        Whether to delete member scans recursively. Defaults to False.

    Returns
    -------
    bool
        True on successful deletion.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> tl = db.create_timelapse('mytl_001')
    >>> db.list_timelapses()
    ['mytl_001']
    >>> tl.delete()
    True
    >>> db.list_timelapses()
    []
    >>> db.disconnect()
    """
    return self.db.delete_timelapse(self.id, recursive=recursive, current_user=current_user, **kwargs)

delete_scan

delete_scan(scan_id, current_user=None, **kwargs)

Delete a member scan from this timelapse.

Parameters:

Name	Type	Description	Default
scan_id	str	The identifier of the scan to delete.	required

Returns:

Type	Description
bool	True on successful deletion.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> tl.create_scan('mytl_001_01')
>>> tl.delete_scan('mytl_001_01')
True
>>> tl.list_scans()
[]
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
@requires_permission(Permission.DELETE, check_scan_access=False)
def delete_scan(self, scan_id: str, current_user=None, **kwargs) -> bool:
    """Delete a member scan from this timelapse.

    Parameters
    ----------
    scan_id : str
        The identifier of the scan to delete.

    Returns
    -------
    bool
        True on successful deletion.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> tl = db.create_timelapse('mytl_001')
    >>> tl.create_scan('mytl_001_01')
    >>> tl.delete_scan('mytl_001_01')
    True
    >>> tl.list_scans()
    []
    >>> db.disconnect()
    """
    if not self.scan_exists(scan_id):
        raise ScanNotFoundError(self.db, scan_id)
    return self.db.delete_scan(scan_id, current_user=current_user, **kwargs)

get_db

get_db()

Get parent database instance.

Returns:

Type	Description
DB	Database instance where to find the timelapse.

Source code in plantdb/commons/db.py

def get_db(self):
    """Get parent database instance.

    Returns
    -------
    plantdb.commons.db.DB
        Database instance where to find the timelapse.
    """
    return self.db

get_id

get_id()

Get the timelapse instance id.

Returns:

Type	Description
str	Id of the timelapse instance.

Source code in plantdb/commons/db.py

def get_id(self):
    """Get the timelapse instance id.

    Returns
    -------
    str
        Id of the timelapse instance.
    """
    return deepcopy(self.id)

get_metadata

get_metadata(key=None, default={}, current_user=None, **kwargs)

Get metadata associated with this timelapse.

Parameters:

Name	Type	Description	Default
key	str	Specific key to retrieve.	None
default	Any	Default value if key is not found.	{}

Returns:

Type	Description
Any	Metadata dictionary or specific value.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> tl.set_metadata('experiment', 'romi')
>>> tl.get_metadata('experiment')
'romi'
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@use_guest_as_default
@requires_permission(Permission.READ, check_scan_access=False)
def get_metadata(self, key=None, default={}, current_user=None, **kwargs) -> Any:
    """Get metadata associated with this timelapse.

    Parameters
    ----------
    key : str, optional
        Specific key to retrieve.
    default : Any, optional
        Default value if key is not found.

    Returns
    -------
    Any
        Metadata dictionary or specific value.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> tl = db.create_timelapse('mytl_001')
    >>> tl.set_metadata('experiment', 'romi')
    >>> tl.get_metadata('experiment')
    'romi'
    >>> db.disconnect()
    """
    with self.db.lock_manager.acquire_lock(self.id, LockType.SHARED, current_user.username, LockLevel.SCAN):
        return _get_metadata(self.metadata, key, default)

get_scan

get_scan(scan_id, current_user=None, **kwargs)

Get a member Scan instance by id in this timelapse.

Parameters:

Name	Type	Description	Default
scan_id	str	The identifier of the member scan.	required

Returns:

Type	Description
Scan	The retrieved Scan instance.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> tl.create_scan('mytl_001_01')
>>> tl.get_scan('mytl_001_01').id
'mytl_001_01'
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
def get_scan(self, scan_id: str, current_user=None, **kwargs) -> "Scan":
    """Get a member `Scan` instance by id in this timelapse.

    Parameters
    ----------
    scan_id : str
        The identifier of the member scan.

    Returns
    -------
    plantdb.commons.fsdb.core.Scan
        The retrieved `Scan` instance.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> tl = db.create_timelapse('mytl_001')
    >>> tl.create_scan('mytl_001_01')
    >>> tl.get_scan('mytl_001_01').id
    'mytl_001_01'
    >>> db.disconnect()
    """
    scan = self.db.get_scan(scan_id, current_user=current_user, **kwargs)
    if isinstance(scan.metadata, dict):
        tl = scan.metadata.get("timelapse")
        if isinstance(tl, dict) and tl.get("id") == self.id:
            return scan
    raise ScanNotFoundError(self.db, scan_id)

get_scans

get_scans(query=None, current_user=None, **kwargs)

Get the list of member Scan instances in this timelapse.

Parameters:

Name	Type	Description	Default
query	dict	A query dictionary to filter member scans.	None

Returns:

Type	Description
list of plantdb.commons.fsdb.core.Scan	Sorted list of member Scan instances.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> tl.create_scan('mytl_001_01')
>>> tl.create_scan('mytl_001_02')
>>> [scan.id for scan in tl.get_scans()]
['mytl_001_01', 'mytl_001_02']
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
def get_scans(self, query=None, current_user=None, **kwargs) -> list["Scan"]:
    """Get the list of member `Scan` instances in this timelapse.

    Parameters
    ----------
    query : dict, optional
        A query dictionary to filter member scans.

    Returns
    -------
    list of plantdb.commons.fsdb.core.Scan
        Sorted list of member `Scan` instances.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> tl = db.create_timelapse('mytl_001')
    >>> tl.create_scan('mytl_001_01')
    >>> tl.create_scan('mytl_001_02')
    >>> [scan.id for scan in tl.get_scans()]
    ['mytl_001_01', 'mytl_001_02']
    >>> db.disconnect()
    """
    tl_query = {"timelapse": {"id": self.id}}
    if query:
        tl_query.update(query)
    kwargs_copy = dict(kwargs)
    kwargs_copy.setdefault("sort", "timelapse.scheduled")
    return self.db.get_scans(query=tl_query, current_user=current_user, **kwargs_copy)

list_scan

list_scan(query=None, fuzzy=False, current_user=None, **kwargs)

Alias for list_scans.

Source code in plantdb/commons/fsdb/core.py

def list_scan(self, query=None, fuzzy=False, current_user=None, **kwargs) -> list[str]:
    """Alias for list_scans."""
    return self.list_scans(query=query, fuzzy=fuzzy, current_user=current_user, **kwargs)

list_scans

list_scans(query=None, fuzzy=False, current_user=None, **kwargs)

Get the list of member scan identifiers in this timelapse.

Parameters:

Name	Type	Description	Default
query	dict	A query dictionary to filter member scans.	None
fuzzy	bool	Whether to use fuzzy regex matching.	False

Returns:

Type	Description
list[str]	List of member scan IDs sorted chronologically.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> tl.create_scan('mytl_001_01')
>>> tl.create_scan('mytl_001_02')
>>> tl.list_scans()
['mytl_001_01', 'mytl_001_02']
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
def list_scans(self, query=None, fuzzy=False, current_user=None, **kwargs) -> list[str]:
    """Get the list of member scan identifiers in this timelapse.

    Parameters
    ----------
    query : dict, optional
        A query dictionary to filter member scans.
    fuzzy : bool, optional
        Whether to use fuzzy regex matching.

    Returns
    -------
    list[str]
        List of member scan IDs sorted chronologically.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> tl = db.create_timelapse('mytl_001')
    >>> tl.create_scan('mytl_001_01')
    >>> tl.create_scan('mytl_001_02')
    >>> tl.list_scans()
    ['mytl_001_01', 'mytl_001_02']
    >>> db.disconnect()
    """
    tl_query = {"timelapse": {"id": self.id}}
    if query:
        tl_query.update(query)
    kwargs_copy = dict(kwargs)
    kwargs_copy.setdefault("sort", "timelapse.scheduled")
    kwargs_copy.setdefault("owner_only", False)
    return self.db.list_scans(query=tl_query, fuzzy=fuzzy, current_user=current_user, **kwargs_copy)

path

path()

Get the path to the local timelapse directory.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> tl.path().name
'mytl_001'
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

def path(self) -> pathlib.Path:
    """Get the path to the local timelapse directory.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> tl = db.create_timelapse('mytl_001')
    >>> tl.path().name
    'mytl_001'
    >>> db.disconnect()
    """
    return _timelapse_path(self.db, self.id)

scan_exists

scan_exists(scan_id)

Check if a scan exists in this timelapse.

Parameters:

Name	Type	Description	Default
scan_id	str	The identifier of the scan to check.	required

Returns:

Type	Description
bool	True if the scan exists and belongs to this timelapse, False otherwise.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> scan = tl.create_scan('mytl_001_01')
>>> tl.scan_exists('mytl_001_01')
True
>>> tl.scan_exists('unrelated_scan')
False
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

def scan_exists(self, scan_id: str) -> bool:
    """Check if a scan exists in this timelapse.

    Parameters
    ----------
    scan_id : str
        The identifier of the scan to check.

    Returns
    -------
    bool
        True if the scan exists and belongs to this timelapse, False otherwise.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> tl = db.create_timelapse('mytl_001')
    >>> scan = tl.create_scan('mytl_001_01')
    >>> tl.scan_exists('mytl_001_01')
    True
    >>> tl.scan_exists('unrelated_scan')
    False
    >>> db.disconnect()
    """
    if not self.db.scan_exists(scan_id):
        return False
    scan = self.db.scans[scan_id]
    if isinstance(scan.metadata, dict):
        tl = scan.metadata.get("timelapse")
        if isinstance(tl, dict) and tl.get("id") == self.id:
            return True
    return False

set_metadata

set_metadata(data, value=None, current_user=None, **kwargs)

Set metadata for this timelapse.

Parameters:

Name	Type	Description	Default
data	str or dict	Key or metadata dictionary.	required
value	Any	Value if data is a key string.	None

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> tl.set_metadata('experiment', 'romi')
>>> tl.get_metadata('experiment')
'romi'
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

@get_authentication
@require_authentication
@requires_permission(Permission.WRITE, check_scan_access=False)
def set_metadata(self, data, value=None, current_user=None, **kwargs) -> None:
    """Set metadata for this timelapse.

    Parameters
    ----------
    data : str or dict
        Key or metadata dictionary.
    value : Any, optional
        Value if data is a key string.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> tl = db.create_timelapse('mytl_001')
    >>> tl.set_metadata('experiment', 'romi')
    >>> tl.get_metadata('experiment')
    'romi'
    >>> db.disconnect()
    """
    self._update_metadata(data, value, current_user, _store_timelapse_metadata, cls_name="TimeLapse")

store

store()

Save changes to the timelapse JSON file (timelapse.json).

Source code in plantdb/commons/fsdb/core.py

def store(self) -> None:
    """Save changes to the timelapse JSON file (timelapse.json)."""
    _store_timelapse_metadata(self)

to_dict

to_dict()

Serialize timelapse descriptor to dictionary.

Returns:

Type	Description
dict	Dictionary containing timelapse id, created_at, metadata, and scan counts.

Examples:

>>> from plantdb.commons.test_database import test_database
>>> db = test_database(no_auth=True)
>>> db.connect()
>>> tl = db.create_timelapse('mytl_001')
>>> tl.create_scan('mytl_001_01')
>>> tl.to_dict()['id']
'mytl_001'
>>> tl.to_dict()['counts']
{'scans': 1}
>>> db.disconnect()

Source code in plantdb/commons/fsdb/core.py

def to_dict(self) -> dict:
    """Serialize timelapse descriptor to dictionary.

    Returns
    -------
    dict
        Dictionary containing timelapse id, created_at, metadata, and scan counts.

    Examples
    --------
    >>> from plantdb.commons.test_database import test_database
    >>> db = test_database(no_auth=True)
    >>> db.connect()
    >>> tl = db.create_timelapse('mytl_001')
    >>> tl.create_scan('mytl_001_01')
    >>> tl.to_dict()['id']
    'mytl_001'
    >>> tl.to_dict()['counts']
    {'scans': 1}
    >>> db.disconnect()
    """
    member_scans = [
        s for s in self.db.scans.values()
        if isinstance(s.metadata, dict) and s.metadata.get("timelapse", {}).get("id") == self.id
    ]
    return {
        "id": self.id,
        "created_at": self.created_at,
        "owner": self.owner,
        "scans": list(self.scans),
        "metadata": self.metadata,
        "counts": {"scans": len(member_scans)},
    }

get_authentication

get_authentication(method)

Get authentication and inject current_user into the wrapped call.

The wrapper expects the following keyword arguments (all optional): * default_user : fallback username (default: None → unauthenticated). * token : JWT supplied by the client.

Any other **kwargs are passed through unchanged.

See Also

get_logged_username : Retrieves the username of the currently logged-in user.

Source code in plantdb/commons/fsdb/core.py

def get_authentication(method: Callable) -> Callable:
    """Get authentication and inject ``current_user`` into the wrapped call.

    The wrapper expects the following keyword arguments (all optional):
        * ``default_user`` : fallback username (default: ``None`` → unauthenticated).
        * ``token`` : JWT supplied by the client.

    Any other ``**kwargs`` are passed through unchanged.

    See Also
    --------
    get_logged_username : Retrieves the username of the currently logged-in user.
    """

    @functools.wraps(method)
    def wrapper(self, *args, **kwargs):
        """
        Other Parameters
        ----------------
        default_user : str, optional
            A fallback username to use if no valid session or token is found. Defaults to ``None``.
        token : str, optional
            The session token or JWT for validating the user's session.
        """
        # Pull and normalize the authentication‑related arguments
        default_user: Optional[str] = kwargs.pop('default_user', None)
        token: Optional[str] = kwargs.get('token', None)

        # Retrieve username based on the type of `self`
        if isinstance(self, (TimeLapse, Scan, Fileset, File)):
            user = get_logged_username(self.db, default_user=default_user, token=token)
        else:
            user = get_logged_username(self, default_user=default_user, token=token)

        # Update the `current_user` only if not already defined...
        # This is to preserve any existing value of `current_user` directly passed to the method (like a guest user)
        if not kwargs.get('current_user'):
            kwargs['current_user'] = user

        # Call the original method, injecting ``current_user``
        return method(self, *args, **kwargs)

    return wrapper

get_logged_username

get_logged_username(fsdb, default_user=None, token=None, **kwargs)

Returns the username of the currently logged user based on the session management system.

This function identifies the username of the logged-in user by inspecting the session manager associated with the given fsdb object. It supports multiple types of session managers, including SingleSessionManager, JWTSessionManager, and generic SessionManager. If no valid session is found or if necessary arguments for session validation are missing, it falls back to the default_user.

Parameters:

Name	Type	Description	Default
fsdb	FSDB	The filesystem database instance containing configuration, logger, and session manager. The session manager is responsible for handling user sessions.	required
default_user	str	A fallback username to use if no valid session or token is found. Defaults to None.	None
token	str	The session token or JWT for validating the user's session.	None

Returns:

Type	Description
User | TokenUser | None	The username of the logged-in user or the fallback default_user.

Notes

• If no valid session manager is attached to the fsdb object, an error will be logged.
• Token validation for both JWTSessionManager and SessionManager assumes that fsdb implements methods like get_username for retrieving usernames based on the provided token.

See Also

plantdb.commons.fsdb.auth.session.SingleSessionManager : Manages single session systems. plantdb.commons.fsdb.auth.session.JWTSessionManager : Handles JSON Web Token-based authentication. plantdb.commons.fsdb.auth.session.SessionManager : Manages multiple generic user sessions.

Examples:

>>> import os
>>> from plantdb.commons.test_database import dummy_db
>>> from plantdb.commons.fsdb.core import get_logged_username
>>> db = dummy_db()  # NoAuthSessionManager with automatic login as 'admin'
>>> get_logged_username(db)
'admin'
>>> db.logout()
>>> get_logged_username(db) is None
True
>>> get_logged_username(db, default_user='guest')

Source code in plantdb/commons/fsdb/core.py

def get_logged_username(fsdb: "FSDB", default_user=None, token=None, **kwargs) -> User | TokenUser | None:
    """Returns the username of the currently logged user based on the session management system.

    This function identifies the username of the logged-in user by inspecting the session manager
    associated with the given `fsdb` object. It supports multiple types of session managers, including
    ``SingleSessionManager``, ``JWTSessionManager``, and generic ``SessionManager``.
    If no valid session is found or if necessary arguments for session validation are missing, it
    falls back to the `default_user`.

    Parameters
    ----------
    fsdb : plantd.commons.fsdb.FSDB
        The filesystem database instance containing configuration, logger, and session manager.
        The session manager is responsible for handling user sessions.
    default_user : str, optional
        A fallback username to use if no valid session or token is found. Defaults to ``None``.
    token : str, optional
        The session token or JWT for validating the user's session.

    Returns
    -------
    User | TokenUser | None
        The username of the logged-in user or the fallback `default_user`.

    Notes
    -----
    - If no valid session manager is attached to the `fsdb` object, an error will be logged.
    - Token validation for both JWTSessionManager and SessionManager assumes that `fsdb` implements methods like
      `get_username` for retrieving usernames based on the provided token.

    See Also
    --------
    plantdb.commons.fsdb.auth.session.SingleSessionManager : Manages single session systems.
    plantdb.commons.fsdb.auth.session.JWTSessionManager : Handles JSON Web Token-based authentication.
    plantdb.commons.fsdb.auth.session.SessionManager : Manages multiple generic user sessions.

    Examples
    --------
    >>> import os
    >>> from plantdb.commons.test_database import dummy_db
    >>> from plantdb.commons.fsdb.core import get_logged_username
    >>> db = dummy_db()  # NoAuthSessionManager with automatic login as 'admin'
    >>> get_logged_username(db)
    'admin'
    >>> db.logout()
    >>> get_logged_username(db) is None
    True
    >>> get_logged_username(db, default_user='guest')
    """
    if isinstance(fsdb.session_manager, (NoAuthSessionManager, SingleSessionManager)):
        # Get the username from the session manager (as only one user can be logged at once)
        try:
            session = list(fsdb.session_manager.sessions.keys())[0]
        except IndexError:
            logged_user = fsdb.get_user_data(username=default_user)
        else:
            logged_user = fsdb.get_user_data(username=fsdb.session_manager.validate_session(session)['username'])
    elif isinstance(fsdb.session_manager, (JWTSessionManager, SessionManager)):
        # If a JSON Web Token Session Manager or a Session Manager, require the token to retrieve the username
        if token:
            logged_user = fsdb.get_user_data(token=token)
        elif default_user:
            logged_user = fsdb.get_user_data(username=default_user)
        else:
            logged_user = None
    else:
        raise ValueError("Can't serve a local PlantDB without a session manager!")

    return logged_user

require_authentication

require_authentication(method)

Enforce authentication.

Source code in plantdb/commons/fsdb/core.py

def require_authentication(method: Callable) -> Callable:
    """Enforce authentication."""

    @functools.wraps(method)
    def wrapper(self, *args, **kwargs):
        # Abort if no authenticated user could be determined
        if not kwargs.get('current_user'):
            raise NoAuthUserError()
        return method(self, *args, **kwargs)

    return wrapper

require_connected_db

require_connected_db(method)

Decorator that ensures the method is only called when the database is connected.

This ensures that operations that require a valid database connection are properly guarded against calls when the connection is inactive.

Raises:

Type	Description
ValueError	If the decorated method is called while the database connection status is not active

See Also

plantdb.commons.fsdb.core.FSDB.connect : Method typically used to establish a database connection.

Source code in plantdb/commons/fsdb/core.py

def require_connected_db(method: Callable) -> Callable:
    """Decorator that ensures the method is only called when the database is connected.

    This ensures that operations that require a valid database connection are properly guarded against calls when the connection is inactive.

    Raises
    ------
    ValueError
        If the decorated method is called while the database connection status is not active

    See Also
    --------
    plantdb.commons.fsdb.core.FSDB.connect : Method typically used to establish a database connection.
    """

    @functools.wraps(method)
    def wrapper(self, *args, **kwargs):
        if not self.is_connected:
            raise ValueError("Database not connected, use the 'connect()' method first!")

        return method(self, *args, **kwargs)

    return wrapper

require_token

require_token(method)

Decorator that passes the token to the decorated method depending on the session manager.

Source code in plantdb/commons/fsdb/core.py

def require_token(method: Callable) -> Callable:
    """Decorator that passes the token to the decorated method depending on the session manager."""

    @functools.wraps(method)
    def wrapper(self, *args, **kwargs):

        if isinstance(self.session_manager, SingleSessionManager):
            # If a Single SessionManager, get the token from the session manager
            try:
                session = list(self.session_manager.sessions.keys())[0]
                kwargs['token'] = session
            except IndexError:
                self.logger.error("No logged user!")

        elif isinstance(self.session_manager, JWTSessionManager):
            # If a JSON Web Token Session Manager, require the token
            if not 'token' in kwargs:
                self.logger.warning("The 'token' argument is required when using JWTSessionManager")

        elif isinstance(self.session_manager, SessionManager):
            # If a regular Session Manager, require the session token
            if not 'token' in kwargs:
                self.logger.warning("The 'token' argument is required when using JWTSessionManager")

        else:
            self.logger.error("Can't serve a local PlantDB without a session manager!")

        return method(self, *args, **kwargs)

    return wrapper

requires_permission

requires_permission(required_permissions, check_scan_access=False)

Decorator that enforces permission checks before the execution of a method.

The decorator inspects the current_user keyword argument supplied to the wrapped method and verifies that the user holds the required permissions. When check_scan_access is True, the check is performed against a specific scan dataset, including additional token‑based dataset permissions. If the user is missing any required permission, a PermissionError is raised.

Parameters:

Name	Type	Description	Default
required_permissions	Union[Permission, Tuple[Permission, ...]]	Permission or tuple of permissions that the user must possess to invoke the wrapped method.	required
check_scan_access	bool	Determines whether the permission check is scoped to a particular scan dataset (True) or performed at the generic user level (False). Default is False.	False

Returns:

Type	Description
function	A decorator that can be applied to instance methods. The decorator wraps the original method with the described permission validation logic.

Raises:

Type	Description
PermissionError	Raised when no current_user is provided or when the user does not have all of the required permissions for the operation.

Notes

• The wrapper extracts the current_user from kwargs; it must be passed explicitly when the decorated method is called.
• When check_scan_access is enabled, the wrapper retrieves the target Scan object via _get_scan and its metadata via _get_metadata to perform dataset‑specific permission checks.
• For TokenUser instances, the wrapper also verifies that the token’s dataset‑level permissions include the required permissions.

See Also

functools.wraps Preserves the original method’s metadata (name, docstring, etc.) when creating the wrapper.

Source code in plantdb/commons/fsdb/core.py

def requires_permission(required_permissions: Union[Permission, Tuple[Permission, ...]],
                        check_scan_access: bool = False):
    """Decorator that enforces permission checks before the execution of a method.

    The decorator inspects the ``current_user`` keyword argument supplied
    to the wrapped method and verifies that the user holds the required permissions.
    When ``check_scan_access`` is ``True``, the check is performed against a specific
    scan dataset, including additional token‑based dataset permissions. If the
    user is missing any required permission, a ``PermissionError`` is raised.

    Parameters
    ----------
    required_permissions : Union[Permission, Tuple[Permission, ...]]
        Permission or tuple of permissions that the user must possess to invoke
        the wrapped method.
    check_scan_access : bool, optional
        Determines whether the permission check is scoped to a particular scan
        dataset (``True``) or performed at the generic user level (``False``).
        Default is ``False``.

    Returns
    -------
    function
        A decorator that can be applied to instance methods. The decorator wraps
        the original method with the described permission validation logic.

    Raises
    ------
    PermissionError
        Raised when no ``current_user`` is provided or when the user does not have
        all of the required permissions for the operation.

    Notes
    -----
    * The wrapper extracts the ``current_user`` from ``kwargs``; it must be passed
      explicitly when the decorated method is called.
    * When ``check_scan_access`` is enabled, the wrapper retrieves the target
      ``Scan`` object via ``_get_scan`` and its metadata via ``_get_metadata`` to
      perform dataset‑specific permission checks.
    * For ``TokenUser`` instances, the wrapper also verifies that the token’s
      dataset‑level permissions include the required permissions.

    See Also
    --------
    functools.wraps
        Preserves the original method’s metadata (name, docstring, etc.) when
        creating the wrapper.
    """

    def decorator(method):
        @functools.wraps(method)
        def wrapper(self, *args, **kwargs):
            if isinstance(required_permissions, Permission):
                required_permissions_ = (required_permissions,)
            else:
                required_permissions_ = required_permissions

            user: User | TokenUser = kwargs.get("current_user", None)
            if user is None:
                raise NoAuthUserError()

            # TokenUser have scan-based permissions, disabling scan access check is done for scan creation only
            if isinstance(user, TokenUser) and not check_scan_access:
                scan_perms = user.get_permissions_for_dataset(args[0])
                has_perms = all(perm in scan_perms for perm in required_permissions_)
                if not has_perms:
                    raise PermissionError(
                        f"Token from User {user.username} does not have required permissions to use {method.__name__}")
                return method(self, *args, **kwargs)

            # check dataset
            if check_scan_access:
                db, scan = _get_fsdb_and_scan(self, *args)
                # Access metadata directly to avoid nested lock acquisition
                metadata = _get_metadata(scan.metadata, None, {})
                has_perms = all(
                    db.rbac_manager.can_access_scan(user, metadata, perm) for perm in required_permissions_
                )
                # Token users have restricted permissions per dataset
                if isinstance(user, TokenUser):
                    token_permissions = user.get_permissions_for_dataset(scan.id)
                    has_perms = has_perms and all(perm in token_permissions for perm in required_permissions_)

            else:
                db = _get_fsdb(self)
                has_perms = all(db.rbac_manager.has_permission(user, perm) for perm in required_permissions_)

            if not has_perms:
                raise PermissionError(
                    f"User {user.username} does not have required permissions to use {method.__name__}")

            return method(self, *args, **kwargs)

        return wrapper

    return decorator

use_guest_as_default

use_guest_as_default(method)

Injects a guest user when no authentication is provided.

This enables methods that require an authenticated user to operate transparently with a default guest context.

Source code in plantdb/commons/fsdb/core.py

def use_guest_as_default(method: Callable) -> Callable:
    """Injects a guest user when no authentication is provided.

    This enables methods that require an authenticated user to operate transparently with a default guest context.
    """

    @functools.wraps(method)
    def wrapper(self, *args, **kwargs):
        if not kwargs.get('current_user'):
            if isinstance(self, (TimeLapse, Scan, Fileset, File)):
                kwargs["current_user"] = self.db.get_guest_user()
            else:
                kwargs["current_user"] = self.get_guest_user()
        return method(self, *args, **kwargs)

    return wrapper