DB

Interfaces with a tinyDB database for converting search results to dict.

Classes

DBTable

Bases: Table

Extends TinyDB table to cast Document type to dict

Functions

all 
all(*args, **kwargs)
Source code in fedbiomed/common/db.py 
@cast_
def all(self, *args, **kwargs):
    return super().all(*args, **kwargs)
delete 
delete(*args, **kwargs)
Source code in fedbiomed/common/db.py 
@cast_
def delete(self, *args, **kwargs):
    return super().remove(*args, **kwargs)
get 
get(*args, **kwargs)
Source code in fedbiomed/common/db.py 
@cast_
def get(self, *args, **kwargs):
    return super().get(*args, **kwargs)
insert 
insert(*args, **kwargs)
Source code in fedbiomed/common/db.py 
@cast_
def insert(self, *args, **kwargs):
    return super().insert(*args, **kwargs)
search 
search(*args, **kwargs)
Source code in fedbiomed/common/db.py 
@cast_
def search(self, *args, **kwargs):
    return super().search(*args, **kwargs)
update 
update(*args, **kwargs)
Source code in fedbiomed/common/db.py 
@cast_
def update(self, *args, **kwargs):
    return super().update(*args, **kwargs)

TinyDBConnector

Singleton TinyDB connector

Attributes

db property 
db

Return the shared TinyDB instance

Functions

table 
table(name)

Return a table with the given name, ensuring it is a DBTable instance.

Source code in fedbiomed/common/db.py 
def table(self, name: str) -> DBTable:
    """Return a table with the given name, ensuring it is a DBTable instance."""
    # Get the table from the underlying DB instance, forcing cache_size=0
    table = self._db.table(name, cache_size=0)
    return DBTable(table.storage, table.name)

TinyTableConnector 

TinyTableConnector(path)

Base class for tables in TinyDB with common operations

Source code in fedbiomed/common/db.py 
def __init__(self, path: str):
    if type(self) is TinyTableConnector:
        raise TypeError("`TinyTableConnector` cannot be instantiated directly")

    self._query = Query()
    self._table = TinyDBConnector(db_path=path).table(self._table_name)

Functions

all 
all()

Get all entries (or empty list if none found).

Returns:

Type Description
List[dict]

The list of entries as dicts, or empty list if none found.
Source code in fedbiomed/common/db.py 
def all(self) -> List[dict]:
    """Get all entries (or empty list if none found).

    Returns:
        The list of entries as dicts, or empty list if none found.
    """
    return self._table.all()
delete_by_id 
delete_by_id(id_value)

Delete a document by its ID.

Parameters:

Name Type Description Default
id_value Union[str, list[str]]

The ID value or list of ID values to delete.

 required

Returns:

Type Description
List[int]

The list of removed doc IDs.
Source code in fedbiomed/common/db.py 
def delete_by_id(self, id_value: Union[str, list[str]]) -> List[int]:
    """Delete a document by its ID.

    Args:
        id_value: The ID value or list of ID values to delete.

    Returns:
        The list of removed doc IDs.
    """
    id_value = [id_value] if isinstance(id_value, str) else id_value

    for value in id_value:
        if not isinstance(value, str):
            raise TypeError(f"Expected id to be string, got {type(value)} instead.")
        if not id_value:
            raise ValueError("Expected id not to be an empty str.")

    return self._table.delete(self._query[self._id_name].one_of(id_value))
get_all_by_condition 
get_all_by_condition(by, cond)

Search entries by a test condition on a field.

Parameters:

Name Type Description Default
by str

field name to search by.

 required
cond Callable

A function that takes the field value and returns a boolean.

 required

Returns:

Type Description
List[dict]

The list of entries as dicts, or empty list if none found.
Source code in fedbiomed/common/db.py 
def get_all_by_condition(self, by: str, cond: Callable) -> List[dict]:
    """Search entries by a test condition on a field.

    Args:
        by: field name to search by.
        cond: A function that takes the field value and returns a boolean.

    Returns:
        The list of entries as dicts, or empty list if none found.
    """
    return self._table.search(self._query[by].test(cond))
get_all_by_value 
get_all_by_value(by, value)

Get all entries by a field value (or empty list if none found).

Parameters:

Name Type Description Default
by str

field name to search by.

 required
value Any

field value to search for.

 required

Returns:

Type Description
List[dict]

The list of entries as dicts, or empty list if none found.
Source code in fedbiomed/common/db.py 
def get_all_by_value(self, by: str, value: Any) -> List[dict]:
    """Get all entries by a field value (or empty list if none found).

    Args:
        by: field name to search by.
        value: field value to search for.

    Returns:
        The list of entries as dicts, or empty list if none found.
    """
    return self._table.search(
        self._query[by].one_of(value)
        if isinstance(value, (list, tuple))
        else self._query[by] == value
    )
get_by_id 
get_by_id(id_value)

Get a document by its ID.

Parameters:

Name Type Description Default
id_value str

The ID value to search for.

 required

Returns:

Type Description
Optional[dict]

The document if found, otherwise None.
Source code in fedbiomed/common/db.py 
def get_by_id(self, id_value: str) -> Optional[dict]:
    """Get a document by its ID.

    Args:
        id_value: The ID value to search for.

    Returns:
        The document if found, otherwise None.
    """
    response = self._table.search(self._query[self._id_name] == id_value)
    assert len(response) < 2, (
        f"Multiple entries found for {self._id_name}={id_value}, "
        "which should be unique."
    )
    return response[0] if response else None
insert 
insert(entry)

Insert a new document.

Parameters:

Name Type Description Default
entry dict

The document to insert.

 required

Returns:

Type Description
int

The document ID of the inserted entry.
Source code in fedbiomed/common/db.py 
def insert(self, entry: dict) -> int:
    """Insert a new document.

    Args:
        entry: The document to insert.

    Returns:
        The document ID of the inserted entry.
    """
    if not isinstance(entry, dict):
        raise TypeError(f"Expected entry to be dict, got {type(entry)}.")
    if self._id_name not in entry:
        raise ValueError(f"Entry must contain '{self._id_name}' field.")
    if self.get_by_id(entry[self._id_name]) is not None:
        raise KeyError(
            f"Entry with {self._id_name}={entry[self._id_name]} already exists."
        )
    _ = self._table.insert(entry)
    return entry[self._id_name]
update_by_id 
update_by_id(id_value, update)

Update a document by its ID.

Parameters:

Name Type Description Default
id_value str

The ID value to update.

 required
update Dict[str, Any]

A dictionary of fields to update.

 required

Returns:

Type Description
dict

Updated document.
Source code in fedbiomed/common/db.py 
def update_by_id(self, id_value: str, update: Dict[str, Any]) -> dict:
    """Update a document by its ID.

    Args:
        id_value: The ID value to update.
        update: A dictionary of fields to update.

    Returns:
        Updated document.
    """
    if self.get_by_id(id_value) is None:
        raise KeyError(f"No entry found with {self._id_name}={id_value}")

    _ = self._table.update(update, self._query[self._id_name] == id_value)

    return self.get_by_id(id_value)

Functions

cast_ 

cast_(func)

Decorator function for typing casting

Source code in fedbiomed/common/db.py 
def cast_(func):
    """Decorator function for typing casting"""

    # Wraps TinyDb get, all, search and upsert methods
    def wrapped(*args, **kwargs):
        add_docs = kwargs.get("add_docs")
        if add_docs is not None:
            kwargs.pop("add_docs")

        document = func(*args, **kwargs)
        if isinstance(document, list):
            if document and isinstance(document[0], Document):
                casted = [dict(r) for r in document]
            else:
                casted = document
        elif isinstance(document, Document):
            casted = dict(document)
        else:
            # Plain python type
            casted = document

        if add_docs:
            return casted, document
        else:
            return casted

    return wrapped