globalbiodata.version

Version dataclass

Class to represent Version information

Attributes:
  • id (int) –

    Database ID for Version

  • name (str) –

    Name of version/pipeline/type

  • date (date) –

    Date of run

  • user (str) –

    User who ran pipeline

  • additional_metadata (dict) –

    Additional data in JSON format

Source code in globalbiodata/version.py
@dataclass
class Version:
    """
    Class to represent Version information

    Attributes:
        id (int): Database ID for Version
        name (str): Name of version/pipeline/type
        date (date): Date of run
        user (str): User who ran pipeline
        additional_metadata (dict): Additional data in JSON format
    """
    id:int
    name:str
    date:date
    user:str
    additional_metadata:dict

    def __init__(self, p):
        self.id = p.get('id')
        self.name = p.get('version_name') or p.get('name')
        self.date = p.get('version_date') or p.get('date')
        self.user = p.get('version_user') or p.get('user')
        self.additional_metadata = p.get('additional_version_metadata') or p.get('additional_metadata')

        self.date = datetime.strptime(self.date, "%Y-%m-%d").date() if self.date and type(self.date) is str else self.date

    def __str__(self):
        version_str = f"Version(id={self.id}, name={self.name}, date={self.date}, user={self.user}, additional_metadata={self.additional_metadata})"
        return version_str

    def write(self, conn: Optional[Connection] = None, engine: Optional[Engine] = None, debug: bool = False) -> int:
        """Write Version to database.

        Args:
            conn (Optional[Connection], optional): SQLAlchemy Connection object.
            engine (Optional[Engine], optional): SQLAlchemy Engine object.
            debug (bool, optional): If `True`, print debug information.

        Returns:
            The ID of the Version written to the database.
        """
        new_version_id = insert_into_table('version', self.__dict__, conn=conn, engine=engine, debug=debug)
        self.id = new_version_id
        return self.id

    def delete(self, conn: Optional[Connection] = None, engine: Optional[Engine] = None, debug: bool = False) -> int:
        """Delete Version from database.

        Args:
            conn (Optional[Connection], optional): SQLAlchemy Connection object.
            engine (Optional[Engine], optional): SQLAlchemy Engine object.
            debug (bool, optional): If `True`, print debug information.

        Returns:
            Number of rows deleted.
        """
        if conn is None:
            conn = engine.connect()

        if not self.id:
            raise ValueError("Version object must have an ID to delete.")

        del_result = delete_from_table('version', {'id':self.id}, conn=conn, engine=engine, debug=debug)
        return del_result

    def fetch_by_id(version_id: int, conn: Optional[Connection] = None, engine: Optional[Engine] = None, debug: bool = False) -> Version:
        """Fetch Version from database by ID.

        Args:
            version_id (int): ID of the Version to fetch.
            conn (Optional[Connection], optional): SQLAlchemy Connection object.
            engine (Optional[Engine], optional): SQLAlchemy Engine object.
            debug (bool, optional): If `True`, print debug information.

        Returns:
            The fetched Version object.
        """
        return fetch_version({'id': version_id}, conn=conn, engine=engine, debug=debug)

delete

delete(conn=None, engine=None, debug=False)

Delete Version from database.

Parameters:
  • conn (Optional[Connection], default: None ) –

    SQLAlchemy Connection object.

  • engine (Optional[Engine], default: None ) –

    SQLAlchemy Engine object.

  • debug (bool, default: False ) –

    If True, print debug information.

Returns:
  • int

    Number of rows deleted.

Source code in globalbiodata/version.py
def delete(self, conn: Optional[Connection] = None, engine: Optional[Engine] = None, debug: bool = False) -> int:
    """Delete Version from database.

    Args:
        conn (Optional[Connection], optional): SQLAlchemy Connection object.
        engine (Optional[Engine], optional): SQLAlchemy Engine object.
        debug (bool, optional): If `True`, print debug information.

    Returns:
        Number of rows deleted.
    """
    if conn is None:
        conn = engine.connect()

    if not self.id:
        raise ValueError("Version object must have an ID to delete.")

    del_result = delete_from_table('version', {'id':self.id}, conn=conn, engine=engine, debug=debug)
    return del_result

fetch_by_id

fetch_by_id(
    version_id, conn=None, engine=None, debug=False
)

Fetch Version from database by ID.

Parameters:
  • version_id (int) –

    ID of the Version to fetch.

  • conn (Optional[Connection], default: None ) –

    SQLAlchemy Connection object.

  • engine (Optional[Engine], default: None ) –

    SQLAlchemy Engine object.

  • debug (bool, default: False ) –

    If True, print debug information.

Returns:
  • Version

    The fetched Version object.

Source code in globalbiodata/version.py
def fetch_by_id(version_id: int, conn: Optional[Connection] = None, engine: Optional[Engine] = None, debug: bool = False) -> Version:
    """Fetch Version from database by ID.

    Args:
        version_id (int): ID of the Version to fetch.
        conn (Optional[Connection], optional): SQLAlchemy Connection object.
        engine (Optional[Engine], optional): SQLAlchemy Engine object.
        debug (bool, optional): If `True`, print debug information.

    Returns:
        The fetched Version object.
    """
    return fetch_version({'id': version_id}, conn=conn, engine=engine, debug=debug)

write

write(conn=None, engine=None, debug=False)

Write Version to database.

Parameters:
  • conn (Optional[Connection], default: None ) –

    SQLAlchemy Connection object.

  • engine (Optional[Engine], default: None ) –

    SQLAlchemy Engine object.

  • debug (bool, default: False ) –

    If True, print debug information.

Returns:
  • int

    The ID of the Version written to the database.

Source code in globalbiodata/version.py
def write(self, conn: Optional[Connection] = None, engine: Optional[Engine] = None, debug: bool = False) -> int:
    """Write Version to database.

    Args:
        conn (Optional[Connection], optional): SQLAlchemy Connection object.
        engine (Optional[Engine], optional): SQLAlchemy Engine object.
        debug (bool, optional): If `True`, print debug information.

    Returns:
        The ID of the Version written to the database.
    """
    new_version_id = insert_into_table('version', self.__dict__, conn=conn, engine=engine, debug=debug)
    self.id = new_version_id
    return self.id