core

plantdb.commons.fsdb.core

This module implement a database as a local file structure.

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

FSDB

FSDB(basedir, required_filesets=None, logger=None, session_manager=None, session_timeout=3600, max_login_attempts=3, lockout_duration=900, max_concurrent_sessions=10)

Bases: DB

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

Implement as a simple local file structure with following directory structure and marker files: * directory ${FSDB.basedir} as database root directory; * marker file MARKER_FILE_NAME at 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.
required_filesets	List[str]	A list of required filesets to consider a scan valid. Set it to None to accept any subdirectory of basedir as a valid scan. Defaults to ['metadata'].
logger	Logger	Logger 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.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:
>>> import os
>>> from plantdb.commons.fsdb.core import FSDB
>>> db = FSDB(os.environ.get('ROMI_DB', "/data/ROMI/DB/"))
>>> 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
required_filesets	list of str	A list of required filesets to consider a scan valid. By default, None, will set it to ['metadata'] to defines as a Scan the subdirectories with a 'metadata' directory. Use [] to accept any subdirectory of basedir as a valid scan	None
logger	Logger	Logger instance to use for logging. Defaults to the module logger.	None
session_manager	(SingleSessionManager, SessionManager, JWTSessionManager)	The session manager to use for session authentication. Defaults to SingleSessionManager.	SingleSessionManager
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

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

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

def __init__(self, basedir: Union[str, Path],
             required_filesets: Optional[List[str]] = None,
             logger: Optional[logging.Logger] = None,
             session_manager: Union[SingleSessionManager, SessionManager, JWTSessionManager] = None,
             session_timeout: int = 3600, max_login_attempts: int = 3,
             lockout_duration: int = 900, max_concurrent_sessions: int = 10):
    """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.
    required_filesets : list of str, optional
        A list of required filesets to consider a scan valid.
        By default, ``None``,  will set it to ``['metadata']`` to defines as a Scan the subdirectories with a 'metadata' directory.
        Use `[]`  to accept any subdirectory of `basedir` as a valid scan
    logger : logging.Logger, optional
        Logger instance to use for logging. Defaults to the module logger.
    session_manager : {SingleSessionManager, SessionManager, JWTSessionManager}, optional
        The session manager to use for session authentication.
        Defaults to ``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.

    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__)

    basedir = Path(basedir)
    # Check the given path to 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.scans = {}
    self.is_connected: bool = False
    self.required_filesets = required_filesets or ['metadata']

    # Initialize scan lock manager
    self.lock_manager = ScanLockManager(basedir)

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

    # Initialize RBAC manager with groups file in basedir
    users_file = os.path.join(basedir, "users_0.15.json")
    groups_file = os.path.join(basedir, "groups.json")
    self.rbac_manager = RBACManager(users_file, groups_file, max_login_attempts, lockout_duration)

add_user_to_group

add_user_to_group(group_name, user, **kwargs)

Add a user to a group.

Parameters:

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

Returns:

Type	Description
bool	True if user was added successfully

Raises:

Type	Description
PermissionError	If user lacks permission to modify the group

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

@require_authentication
def add_user_to_group(self, group_name, user, **kwargs):
    """Add a user to a group.

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

    Returns
    -------
    bool
        True if user was added successfully

    Raises
    ------
    PermissionError
        If user lacks permission to modify the group
    """
    current_user = self.get_user_data(kwargs.get('username', None))
    if not current_user:
        raise PermissionError("No authenticated user")

    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):
    """
    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) -> bool:
    """Connect the database by loading the scans' dataset."""
    try:
        # Initialize scan discovery
        self.scans = _load_scans(self)
        self.is_connected = True
        self.logger.info("Connected to database successfully")
    except Exception as e:
        self.logger.error(f"Failed to connect to database: {e}")
        raise

    return True

create_group

create_group(name, users=None, description=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

Returns:

Type	Description
Group	The created group object

Raises:

Type	Description
PermissionError	If user lacks permission to create groups
ValueError	If group already exists

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

@require_authentication
def create_group(self, name, users=None, description=None, **kwargs):
    """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

    Returns
    -------
    Group
        The created group object

    Raises
    ------
    PermissionError
        If user lacks permission to create groups
    ValueError
        If group already exists
    """
    current_user = self.get_user_data(kwargs.get('username', None))
    if not current_user:
        raise PermissionError("No authenticated user")

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

create_scan

create_scan(scan_id, metadata=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
OSError	If the scan_id is not valid or already exists in the local database.

See Also

plantdb.commons.fsdb._is_valid_id plantdb.commons.fsdb._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 'anonymous' for dummy database
anonymous
>>> print(new_scan.get_metadata('project'))
GoldenEye
>>> scan = db.create_scan('007')  # attempt to create an existing scan dataset
OSError: Given scan identifier '007' already exists!
>>> scan = db.create_scan('0/07')  # attempt to create a scan dataset using invalid characters
OSError: 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
@require_authentication
def create_scan(self, scan_id, metadata=None, **kwargs):
    """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
    ------
    OSError
        If the `scan_id` is not valid or already exists in the local database.

    See Also
    --------
    plantdb.commons.fsdb._is_valid_id
    plantdb.commons.fsdb._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 'anonymous' for dummy database
    anonymous
    >>> print(new_scan.get_metadata('project'))
    GoldenEye
    >>> scan = db.create_scan('007')  # attempt to create an existing scan dataset
    OSError: Given scan identifier '007' already exists!
    >>> scan = db.create_scan('0/07')  # attempt to create a scan dataset using invalid characters
    OSError: Invalid scan identifier '0/07'!
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    current_user = self.get_user_data(**kwargs)
    if not current_user:
        raise PermissionError("No authenticated user!")

    # Check CREATE permission
    if not self.rbac_manager.has_permission(current_user, Permission.CREATE):
        raise PermissionError(f"Insufficient permissions to create scan with user '{current_user.username}'")

    if self.scan_exists(scan_id):
        raise ScanExistsError(self, scan_id)

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

    # Set current user as owner if not specified
    if 'owner' not in 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 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")

    # Use exclusive lock for scan creation
    with self.lock_manager.acquire_lock(scan_id, LockType.EXCLUSIVE, current_user.username):
        # Verify if the given `scan_id` already exists in the local database
        if self.scan_exists(scan_id):
            raise IOError(f"Given scan identifier '{scan_id}' already exists!")
        try:
            # Initialize scan object
            scan = Scan(self, scan_id)  # Initialize a new Scan instance
            scan_path = _make_scan(scan)  # Create directory structure

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

            scan.store()  # store the new scan in the local database
            self.scans[scan_id] = scan  # Update scans dictionary with newly created

            self.logger.info(f"Created scan '{scan_id}' for user '{current_user.username}'")
            return scan

        except Exception as e:
            self.logger.error(f"Failed to create scan {scan_id}: {e}")
            # Cleanup on failure
            try:
                if os.path.exists(scan_path):
                    import shutil
                    shutil.rmtree(scan_path)
            except:
                pass
            return None

create_user

create_user(username, fullname, password, roles=None)

Create a new user with the specified details.

Parameters:

Name	Type	Description	Default
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

See Also

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

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

@require_authentication
def create_user(self, username, fullname, password, roles=None) -> None:
    """
    Create a new user with the specified details.

    Parameters
    ----------
    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.

    See Also
    --------
    RBACManager.users.create : Method used to actually create the user.
    """
    return self.rbac_manager.users.create(username, fullname, password, roles)

delete_group

delete_group(group_name, **kwargs)

Delete a group.

Parameters:

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

Returns:

Type	Description
bool	True if group was deleted successfully

Raises:

Type	Description
PermissionError	If user lacks permission to delete groups

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

@require_authentication
def delete_group(self, group_name, **kwargs):
    """Delete a group.

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

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

    Raises
    ------
    PermissionError
        If user lacks permission to delete groups
    """
    current_user = self.get_user_data(kwargs.get('username', None))
    if not current_user:
        raise PermissionError("No authenticated user")

    if not self.rbac_manager.delete_group(current_user, group_name):
        raise PermissionError("Insufficient permissions or group not found")
    return True

delete_scan

delete_scan(scan_id, **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

Raises:

Type	Description
IOError	If the id do not exist in the local database.

See Also

plantdb.commons.fsdb._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
@require_authentication
def delete_scan(self, scan_id, **kwargs):
    """Delete an existing `Scan` from the local database.

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

    Raises
    ------
    IOError
        If the `id` do not exist in the local database.

    See Also
    --------
    plantdb.commons.fsdb._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
    """
    current_user = self.get_user_data(**kwargs)
    if not current_user:
        raise PermissionError("No authenticated user")

    if not self.scan_exists(scan_id):
        raise ValueError(f"Scan {scan_id} does not exist")

    # Check DELETE permission for this specific scan
    scan = self.scans[scan_id]
    metadata = self.rbac_manager.ensure_scan_owner(scan.get_metadata())

    if not self.rbac_manager.can_access_scan(current_user, metadata, Permission.DELETE):
        raise PermissionError("Insufficient permissions to delete scan")

    # Check if scan is locked
    if self.is_scan_locked(scan_id):
        self.logger.error(f"Scan {scan_id} is locked by another user")

    # Use exclusive lock for scan deletion
    with self.lock_manager.acquire_lock(scan_id, LockType.EXCLUSIVE, current_user.username):
        try:
            # Get the Scan instance from database
            scan = self.scans[scan_id]
            _delete_scan(scan)  # delete the scan directory
            self.scans.pop(scan_id)  # remove the scan from the scan list
            self.logger.info(f"Deleted scan '{scan_id}' by user '{current_user.username}'")
        except Exception as e:
            self.logger.error(f"Failed to delete scan {scan_id}: {e}")
            raise
    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
>>> db.disconnect()  # clean up (delete) the temporary dummy database
>>> print(db.is_connected)
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
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    >>> print(db.is_connected)
    False
    """
    for s_id, scan in self.scans.items():
        scan._erase()
    self.scans = {}
    self.is_connected = False
    return

get_guest_user

get_guest_user()

Retrieve the guest user information from the RBAC manager.

Returns the guest user object containing all relevant data.

Parameters:

Name	Type	Description	Default
None			required

Returns:

Type	Description
dict	A dictionary representing the guest user with all attributes. For example, it might contain keys like 'id', 'name', etc.

Examples:

>>> rbac_manager = RBACManager()
>>> user_info = rbac_manager.get_guest_user()
>>> print(user_info)
{'id': 12345, 'name': 'Guest User', 'role': 'Guest'}

Notes

This method interacts with the underlying RBAC manager to fetch guest user information. Ensure that the RBAC manager is correctly configured.

See Also

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

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

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

    Returns the guest user object containing all relevant data.

    Parameters
    ----------
    None

    Returns
    -------
    dict
        A dictionary representing the guest user with all attributes.
        For example, it might contain keys like 'id', 'name', etc.

    Examples
    --------
    >>> rbac_manager = RBACManager()
    >>> user_info = rbac_manager.get_guest_user()
    >>> print(user_info)
    {'id': 12345, 'name': 'Guest User', 'role': 'Guest'}

    Notes
    -----
    This method interacts with the underlying RBAC manager to fetch guest
    user information. Ensure that the RBAC manager is correctly configured.

    See Also
    --------
    rbac_manager.get_guest_user : The underlying method used by this function.
    """
    return self.rbac_manager.get_guest_user()

get_scan

get_scan(scan_id, **kwargs)

Get 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 do not exist in the local database and create is False.

Returns:

Type	Description
Scan	The Scan identified by the scan_id.

Examples:

>>> 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

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

@require_connected_db
@require_authentication
def get_scan(self, scan_id, **kwargs):
    """Get `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` do not exist in the local database and `create` is ``False``.

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

    Examples
    --------
    >>> 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
    """
    if not self.scan_exists(scan_id):
        raise ScanNotFoundError(self, scan_id)

    current_user = self.get_user_data(**kwargs)
    if not current_user:
        raise Exception("No valid user!")

    scan = self.scans[scan_id]
    metadata = self.rbac_manager.ensure_scan_owner(scan.get_metadata())
    if self.rbac_manager.can_access_scan(current_user, metadata, Permission.READ):
        # Use shared lock for read operations
        with self.lock_manager.acquire_lock(scan_id, LockType.SHARED, current_user.username):
            return scan
    else:
        self.logger.warning(f"User '{current_user.username}' cannot access scan {scan_id}!")

    return None

get_scan_access_summary

get_scan_access_summary(scan_id, **kwargs)

Get access summary for current user on a scan.

Parameters:

Name	Type	Description	Default
scan_id	str	The scan identifier	required

Returns:

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

Raises:

Type	Description
PermissionError	If no authenticated user

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

@require_authentication
def get_scan_access_summary(self, scan_id, **kwargs):
    """Get access summary for current user on a scan.

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

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

    Raises
    ------
    PermissionError
        If no authenticated user
    """
    current_user = self.get_user_data(kwargs.get('username', None))
    if not current_user:
        raise PermissionError("No authenticated user")

    if scan_id not in self.scans:
        return None

    scan = self.scans[scan_id]
    try:
        metadata = self.rbac_manager.ensure_scan_owner(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)

get_scans

get_scans(query=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 *x************>]
>>> db.disconnect()  # clean up (delete) the temporary dummy database

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

@require_connected_db
@require_authentication
def get_scans(self, query=None, **kwargs) -> List:
    """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 *x************>]
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    current_user = self.get_user_data(**kwargs)
    if not current_user:
        self.logger.warning("No authenticated user, can not proceed.")
        return []

    # Get all scans and filter by access permissions
    accessible_scans = {}

    for scan_id, scan in self.scans.items():
        try:
            metadata = self.rbac_manager.ensure_scan_owner(scan.get_metadata())
            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())

    return accessible_scans

get_user

get_user(session_token)

Get the username.

Returns:

Type	Description
str or None	Username if token is valid

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

def get_user(self, session_token):
    """Get the username.

    Returns
    -------
    str or None
        Username if token is valid
    """
    return self.session_manager.session_username(session_token)

get_user_data

get_user_data(username=None, session_token=None)

Get the user data.

Returns:

Type	Description
User or None	Current user object if authenticated, None otherwise

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

def get_user_data(self, username=None, session_token=None):
    """Get the user data.

    Returns
    -------
    User or None
        Current user object if authenticated, None otherwise
    """
    if username:
        return self.rbac_manager.users.get_user(username)
    elif session_token:
        return self.rbac_manager.users.get_user(self.get_user(session_token))
    else:
        self.logger.error("No username or session token provided")
        return None

get_user_groups

get_user_groups(username=None, **kwargs)

Get groups for a user.

Parameters:

Name	Type	Description	Default
username	str	Username to query. If None, uses current user.	None

Returns:

Type	Description
list	A list of Group objects the user belongs to

Raises:

Type	Description
PermissionError	If no authenticated user

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

@require_authentication
def get_user_groups(self, username=None, **kwargs):
    """Get groups for a user.

    Parameters
    ----------
    username : str, optional
        Username to query. If None, uses current user.

    Returns
    -------
    list
        A list of Group objects the user belongs to

    Raises
    ------
    PermissionError
        If no authenticated user
    """
    current_user = self.get_user_data(kwargs.get('username', None))
    if not current_user:
        raise PermissionError("No authenticated user")

    if username is None:
        username = current_user.username

    return self.rbac_manager.get_user_groups(username)

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(**kwargs)

List all groups.

Returns:

Type	Description
list	A list of Group objects

Raises:

Type	Description
PermissionError	If user is not authenticated

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

@require_authentication
def list_groups(self, **kwargs):
    """List all groups.

    Returns
    -------
    list
        A list of Group objects

    Raises
    ------
    PermissionError
        If user is not authenticated
    """
    current_user = self.get_user_data(kwargs.get('username', None))
    if not current_user:
        raise PermissionError("No authenticated user")

    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, **kwargs)

Get the list of scans in identifiers 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 in the local database.

See Also

plantdb.commons.fsdb._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
@require_authentication
def list_scans(self, query=None, fuzzy=False, owner_only=True, **kwargs) -> list:
    """Get the list of scans in identifiers 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 in the local database.

    See Also
    --------
    plantdb.commons.fsdb._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
    """
    current_user = self.get_user_data(**kwargs)
    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})

    if query is None:
        return list(self.scans.keys())
    else:
        return [scan.id for scan in _filter_query(list(self.scans.values()), query, fuzzy)]

login

login(username, password)

Authenticate user and create session.

Parameters:

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

Returns:

Type	Description
Union[str, None]	Returns the user session ID if successful, None otherwise.

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

@require_connected_db
def login(self, username: str, password: str) -> Union[str, None]:
    """Authenticate user and create session.

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

    Returns
    -------
    Union[str, None]
        Returns the user session ID if successful, ``None`` otherwise.
    """
    if self.validate_user(username, password):
        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.info(f"User {username} logged in successfully")
        return session_token
    else:
        self.logger.error(f"Failed to login user {username}")
        return None

logout

logout(**kwargs)

Logout user and by invalidating its session.

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

@require_token
def logout(self, **kwargs):
    """Logout user and by invalidating its session."""
    success, username = self.session_manager.invalidate_session(kwargs.get('token', None))
    if success:
        self.logger.info(f"User {username} logged out successfully")
        return True
    else:
        self.logger.warning(f"Failed to logout!")
        return False

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: Optional[Union[str, Iterable[str]]] = 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, **kwargs)

Remove a user from a group.

Parameters:

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

Returns:

Type	Description
bool	True if user was removed successfully

Raises:

Type	Description
PermissionError	If user lacks permission to modify the group

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

@require_authentication
def remove_user_from_group(self, group_name, user, **kwargs):
    """Remove a user from a group.

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

    Returns
    -------
    bool
        True if user was removed successfully

    Raises
    ------
    PermissionError
        If user lacks permission to modify the group
    """
    current_user = self.get_user_data(kwargs.get('username', None))
    if not current_user:
        raise PermissionError("No authenticated user")

    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

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.

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

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.
    """
    return self.rbac_manager.users.validate(username, password)

File

File(fileset, f_id, **kwargs)

Bases: File

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 writen 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!

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={})

Get the metadata associated to 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 do 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

def get_metadata(self, key=None, default={}):
    """Get the metadata associated to 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 do 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
    """
    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, **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

@require_authentication
def import_file(self, path, **kwargs):
    """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
    """
    current_user = self.db.get_user_data(**kwargs)
    if not current_user:
        raise PermissionError("No authenticated user!")

    # Check ownership
    if self.scan.owner != current_user.username:
        raise PermissionError(f"Only the owner can create filesets in scan '{self.id}'")

    # Check if the path is a file
    if isinstance(path, str):
        path = Path(path)
    if not os.path.isfile(path):
        raise ValueError("The provided path is not a file.")
    # 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
    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):
    """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):
    """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)

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

def set_metadata(self, data, value=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
    """
    _set_metadata(self.metadata, data, value)
    # Ensure modification timestamp
    self.metadata['last_modified'] = iso_date_now()
    _store_file_metadata(self)
    return

store

store()

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

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

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

write

write(data, ext='', **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

@require_authentication
def write(self, data, ext="", **kwargs):
    """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
    """
    current_user = self.db.get_user_data(**kwargs)
    if not current_user:
        raise PermissionError("No authenticated user!")

    # Check ownership
    if self.scan.owner != current_user.username:
        raise PermissionError(f"Only the owner can create filesets in scan '{self.id}'")

    self.filename = _get_filename(self, ext)
    path = _file_path(self)
    with path.open(mode="w") as f:
        f.write(data)
    self.store()
    return

write_raw

write_raw(data, ext='', **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

@require_authentication
def write_raw(self, data, ext="", **kwargs):
    """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
    """
    current_user = self.db.get_user_data(**kwargs)
    if not current_user:
        raise PermissionError("No authenticated user!")

    # Check ownership
    if self.scan.owner != current_user.username:
        raise PermissionError(f"Only the owner can create filesets in scan '{self.id}'")

    self.filename = _get_filename(self, ext)
    path = _file_path(self)
    with path.open(mode="wb") as f:
        f.write(data)
    self.store()
    return

Fileset

Fileset(scan, fs_id)

Bases: Fileset

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

Implementation of a fileset as a simple files structure with: * directory ${FSDB.basedir}/${FSDB.scan.id}/${Fileset.id} containing set of files; * directory ${FSDB.basedir}/${FSDB.scan.id}/metadata containing JSON metadata associated to 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

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, **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.

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 exist 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

@require_authentication
def create_file(self, f_id, metadata=None, **kwargs):
    """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.

    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 exist 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
    """
    current_user = self.db.get_user_data(**kwargs)
    if not current_user:
        raise PermissionError("No authenticated user!")

    # Check ownership
    if self.scan.owner != current_user.username:
        raise PermissionError(f"Only the owner can create filesets in scan '{self.id}'")

    # Verify if the given `fs_id` is valid
    if not _is_valid_id(f_id):
        raise IOError(f"Invalid file identifier '{f_id}'!")

    # Use exclusive lock for file creation
    with self.db.lock_manager.acquire_lock(self.scan.id, LockType.EXCLUSIVE, current_user.username):
        # Verify if the given `fs_id` already exists in the local database
        if self.file_exists(f_id):
            raise FileExistsError(self, f_id)

        # 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.username

        # 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 files dictionary
        self.store()  # Store fileset instance to the JSON

        self.logger.info(
            f"Created new file '{f_id}' in '{self.scan.id}/{self.id}' for user '{current_user.username}'")

    return file

delete_file

delete_file(f_id, **kwargs)

Delete a given file from the current fileset.

Parameters:

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

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

@require_authentication
def delete_file(self, f_id, **kwargs):
    """Delete a given file from the current fileset.

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

    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
    """
    current_user = self.db.get_user_data(**kwargs)
    if not current_user:
        raise PermissionError("No authenticated user!")

    # Check ownership
    if self.scan.owner != current_user.username:
        raise PermissionError(f"Only the owner can create filesets in scan '{self.id}'")

    # Verify if the given `fs_id` exists in the local database
    if not self.file_exists(f_id):
        logging.warning(f"Given file identifier '{f_id}' does NOT exists!")
        return

    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``)
    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/create.	required

Returns:

Type	Description
File	The retrieved or created 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):
    """Get a `File` instance, of given `f_id`, in the current fileset.

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

    Returns
    -------
    plantdb.commons.fsdb.core.File
        The retrieved or created 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
    """
    # TODO: should probably create a `@retrieve_username` decorator to provide it to method that do not require a username but could use it
    # TODO: Or maybe get rid of the lock_manager here?
    # with self.db.lock_manager.acquire_lock(self.scan.id, LockType.SHARED, current_user.username or "guest"):

    # Use shared lock for read operations
    with self.db.lock_manager.acquire_lock(self.scan.id, LockType.SHARED, "guest"):
        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 *x************>,
 <plantdb.commons.fsdb.core.File at *x************>,
 <plantdb.commons.fsdb.core.File at *x************>]
>>> 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):
    """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 *x************>,
     <plantdb.commons.fsdb.core.File at *x************>,
     <plantdb.commons.fsdb.core.File at *x************>]
    >>> 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={})

Get the metadata associated to 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 do 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.dummy=False  # to avoid cleaning up the temporary dummy database
>>> db.disconnect()
>>> db.connect('anonymous')
>>> scan = db.get_scan("myscan_001")
>>> fs = scan.get_fileset('fileset_001')
>>> print(fs.get_metadata("test"))
'value'
>>> db.dummy=True
>>> db.disconnect()  # clean up (delete) the temporary dummy database

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

def get_metadata(self, key=None, default={}):
    """Get the metadata associated to 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 do 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.dummy=False  # to avoid cleaning up the temporary dummy database
    >>> db.disconnect()
    >>> db.connect('anonymous')
    >>> scan = db.get_scan("myscan_001")
    >>> fs = scan.get_fileset('fileset_001')
    >>> print(fs.get_metadata("test"))
    'value'
    >>> db.dummy=True
    >>> db.disconnect()  # clean up (delete) the temporary dummy database
    """
    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, **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

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

@require_authentication
def set_metadata(self, data, value=None, **kwargs):
    """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.

    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
    """
    current_user = self.db.get_user_data(**kwargs)
    if not current_user:
        raise PermissionError("No authenticated user!")

    # Check ownership
    if self.scan.owner != current_user.username:
        raise PermissionError(f"Only the owner can create filesets in scan '{self.id}'")

    _set_metadata(self.metadata, data, value)
    # Ensure modification timestamp
    self.metadata['last_modified'] = iso_date_now()
    _store_fileset_metadata(self)
    return

store

store()

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

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

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

Scan

Scan(db, scan_id)

Bases: Scan

Implement Scan for the local File System DataBase from 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.
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: 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.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.dummy = True  # to clean up the temporary dummy database
>>> db.disconnect()  # clean up (delete) the temporary dummy database

>>> # Example #3: 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

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

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

    Parameters
    ----------
    db : plantdb.commons.fsdb.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.measures = None

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

create_fileset

create_fileset(fs_id, metadata=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
IOError	If the id already exists in the current Scan instance. If the id is not valid.

See Also

plantdb.commons.fsdb._is_valid_id plantdb.commons.fsdb._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

@require_authentication
def create_fileset(self, fs_id, metadata=None, **kwargs):
    """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
    ------
    IOError
        If the `id` already exists in the current `Scan` instance.
        If the `id` is not valid.

    See Also
    --------
    plantdb.commons.fsdb._is_valid_id
    plantdb.commons.fsdb._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
    """
    current_user = self.db.get_user_data(**kwargs)
    if not current_user:
        raise PermissionError("No authenticated user!")

    # Check ownership
    if self.owner != current_user.username:
        raise PermissionError(f"Only the owner can create filesets in scan '{self.id}'")
    # Verify if the given `fs_id` is valid
    if not _is_valid_id(fs_id):
        raise IOError(f"Invalid fileset identifier '{fs_id}'!")

    # Use exclusive lock for fileset creation
    self.logger.info(f"Creating fileset '{fs_id}' from scan '{self.id}'")
    with self.db.lock_manager.acquire_lock(self.id, LockType.EXCLUSIVE, current_user.username):
        # Verify if the given `fs_id` already exists in the local database
        if self.fileset_exists(fs_id):
            raise FilesetExistsError(self, fs_id)

        # 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.username

        # 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.info(f"Created new fileset '{fs_id}' in scan '{self.id}' for user '{current_user.username}'")

    return fileset

delete_fileset

delete_fileset(fs_id, **kwargs)

Delete a given fileset from the scan dataset.

Parameters:

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

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

@require_authentication
def delete_fileset(self, fs_id, **kwargs):
    """Delete a given fileset from the scan dataset.

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

    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
    """
    current_user = self.db.get_user_data(**kwargs)
    if not current_user:
        raise PermissionError("No authenticated user!")

    # Check ownership
    if self.owner != current_user.username:
        raise PermissionError(f"Only the owner can delete filesets from scan '{self.id}'")

    # Use exclusive lock for fileset deletion
    with self.db.lock_manager.acquire_lock(self.id, LockType.EXCLUSIVE, current_user.username):
        # 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}'")

        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.info(f"Deleted fileset '{fs_id}' from scan '{self.id}' by user '{current_user.username}'")
    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_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.

Notes

If the id do not exist in the local database and create is False, None is returned.

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):
    """Get a `Fileset` instance, of given `id`, in the current scan dataset.

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

    Returns
    -------
    Fileset
        The retrieved or created fileset.

    Notes
    -----
    If the `id` do not exist in the local database and `create` is `False`, `None` is returned.

    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
    """
    # TODO: should probably create a `@retrieve_username` decorator to provide it to method that do not require a username but could use it
    # TODO: Or maybe get rid of the lock_manager here?
    # with self.db.lock_manager.acquire_lock(self.id, LockType.SHARED, current_user.username or "guest"):

    # Use shared lock for read operations
    with self.db.lock_manager.acquire_lock(self.id, LockType.SHARED, "guest"):
        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 *x************>]
>>> 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):
    """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 *x************>]
    >>> 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)

Get the manual measurements associated to 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.

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

def get_measures(self, key=None):
    """Get the manual measurements associated to 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.
    """
    return _get_metadata(self.measures, key, default={})

get_metadata

get_metadata(key=None, default={})

Get the metadata associated to 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 do 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.

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

def get_metadata(self, key=None, default={}):
    """Get the metadata associated to 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 do 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.
    """
    # TODO: should probably create a `@retrieve_username` decorator to provide it to method that do not require a username but could use it
    # TODO: Or maybe get rid of the lock_manager here?
    # with self.db.lock_manager.acquire_lock(self.id, LockType.SHARED, current_user.username or "guest"):
    #    return _get_metadata(self.metadata, key, default)

    # Use shared lock for read operations
    with self.db.lock_manager.acquire_lock(self.id, LockType.SHARED, "guest"):
        return _get_metadata(self.metadata, key, default)

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):
    """
    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[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:
    """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[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, **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 user lacks permission to modify metadata
ValueError	If metadata validation fails

Examples:

>>> import json
>>> from plantdb.commons.test_database import dummy_db
>>> from plantdb.commons.fsdb.path_helpers import _scan_metadata_path
>>> db = dummy_db(with_file=True)
>>> scan = db.get_scan("myscan_001")
>>> scan.set_metadata("test", "value")
>>> 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

@require_authentication
def set_metadata(self, data, value=None, **kwargs):
    """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 user lacks permission to modify metadata
    ValueError
        If metadata validation fails

    Examples
    --------
    >>> import json
    >>> from plantdb.commons.test_database import dummy_db
    >>> from plantdb.commons.fsdb.path_helpers import _scan_metadata_path
    >>> db = dummy_db(with_file=True)
    >>> scan = db.get_scan("myscan_001")
    >>> scan.set_metadata("test", "value")
    >>> 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
    """
    current_user = self.db.get_user_data(**kwargs)
    if not current_user:
        raise PermissionError("No authenticated user!")

    # Get current metadata for validation
    old_metadata = self.db.rbac_manager.ensure_scan_owner(self.get_metadata())

    if isinstance(data, str):
        if value is None:
            self.logger.warning(f"No value given for key '{data}'!")
        new_metadata = {data: value}
    else:
        try:
            assert isinstance(data, dict)
        except AssertionError:
            raise PermissionError(f"Invalid metadata type '{type(data)}'")
        else:
            new_metadata = data

    # Validate metadata changes
    if not self.db.rbac_manager.validate_scan_metadata_access(current_user, old_metadata, new_metadata):
        raise PermissionError("Insufficient permissions to modify scan metadata")

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

    # Check WRITE permission for this scan
    if not self.db.rbac_manager.can_access_scan(current_user, old_metadata, Permission.WRITE):
        raise PermissionError("Insufficient permissions to modify scan")

    # Use exclusive lock for metadata updates
    with self.db.lock_manager.acquire_lock(self.id, LockType.EXCLUSIVE, current_user.username):
        # Update metadata
        _set_metadata(self.metadata, new_metadata, None)
        # Ensure modification timestamp
        _set_metadata(self.metadata, 'last_modified', iso_date_now())
        _store_scan_metadata(self)

        self.logger.info(f"Updated metadata for scan '{self.id}' by user '{current_user.username}'")

    return

store

store()

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

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

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

require_authentication

require_authentication(method)

Decorator that extracts the username using the session manager and passes it to the decorated method.

The object of the decorated method is expected to have the following attributes: - session_manager: a class managing user session(s) - logger: a logger instance to log messages

The object of the decorated method is expected to have the following methods: - get_user: a method that returns the username to use for authentication credentials

Notes

• The token should be passed as a 'token' kwarg to the decorated method.
• The username will default to 'guest'.
• The username should be passed as a 'username' kwarg to the decorated method.
• The user data will be passed as a 'username' kwarg to the decorated method.

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

def require_authentication(method):
    """
    Decorator that extracts the username using the session manager and passes it to the decorated method.

    The object of the decorated method is expected to have the following attributes:
    - session_manager: a class managing user session(s)
    - logger: a logger instance to log messages

    The object of the decorated method is expected to have the following methods:
    - get_user: a method that returns the username to use for authentication credentials

    Notes
    -----
    - The token should be passed as a 'token' kwarg to the decorated method.
    - The username will default to 'guest'.
    - The username should be passed as a 'username' kwarg to the decorated method.
    - The user data will be passed as a 'username' kwarg to the decorated method.
    """

    def wrapper(self, *args, **kwargs):

        if isinstance(self.session_manager, SingleSessionManager):
            # If a Single SessionManager, get the username from the session manager or use 'guest' user
            try:
                session = list(self.session_manager.sessions.keys())[0]
            except IndexError:
                kwargs['username'] = "guest"
            else:
                kwargs['username'] = self.session_manager.validate_session(session)['username']

        elif isinstance(self.session_manager, JWTSessionManager):
            # If a JSON Web Token Session Manager, require the token or default to 'guest' user
            if 'token' in kwargs:
                jwt_token = kwargs.pop('token', None)
                # Get username from JWT
                if isinstance(self, (Scan, Fileset, File)):
                    username = self.db.get_user(jwt_token)
                else:
                    username = self.get_user(jwt_token)
                kwargs['username'] = username
            else:
                kwargs['username'] = 'guest'

        elif isinstance(self.session_manager, SessionManager):
            # If a regular Session Manager, require the session token or default to 'guest' user
            if 'token' in kwargs:
                token = kwargs.pop('token', None)
                # Get username from session token
                if isinstance(self, (Scan, Fileset, File)):
                    username = self.db.get_user(token)
                else:
                    username = self.get_user(token)
                kwargs['username'] = username
            else:
                kwargs['username'] = 'guest'

        else:
            self.logger.error("Can't serve a local PlantDB without a session manager!")

        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):
    """
    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.
    """

    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):
    """Decorator that passes the token to the decorated method depending on the session manager."""

    def wrapper(self, *args, **kwargs):

        if isinstance(self.session_manager, SingleSessionManager):
            # If a Single SessionManager, get the token from the session manager
            session = list(self.session_manager.sessions.keys())[0]
            kwargs['token'] = session

        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