Functions
	setup ()

StorageItem\|None	fetch_item (int item_id)

None	delete_item (int item_id)

None	create_item (int pos, StorageItemType obj_type, str name, str json_data)

List[StorageItem]	search (str search_term)

None	update_item (int item_id, int pos, StorageItemType obj_type, str name, str json_data)

Variables
	base_dir = os.path.dirname(__file__)

	db_dir = os.path.join(base_dir, "data")

	db_path = os.path.join(db_dir, "storage.db")

	engine = create_engine(f"sqlite:///{db_path}", connect_args={"check_same_thread": False})

Detailed Description

StorageConnector Module

This module provides functions to interact with a SQLite database for managing storage items.
It includes functionalities to set up the database, create, fetch, delete, and search items,
as well as export the data to a CSV file.

Function Documentation

◆ create_item()

None Storage_connector.create_item	(	int	pos,
		StorageItemType	obj_type,
		str	name,
		str	json_data )

Creates a new item in the storage database with the provided position, type, name, and JSON data.

Args:
    pos: LED position of the item.
    obj_type: type of the Item
    name: name of item
    json_data: additional json ifno

Returns:

Definition at line 60 of file Storage_connector.py.

def create_item(pos: int, obj_type: StorageItemType, name: str, json_data: str) -> None:
    """
    Creates a new item in the storage database with the provided position, type, name, and JSON data.
 
    Args:
        pos: LED position of the item.
        obj_type: type of the Item
        name: name of item
        json_data: additional json ifno
 
    Returns:
 
    """
    info_value = None if json_data in ["{}", ""] else json_data
 
    new_item = StorageItem(
        position=pos,
        type=obj_type,
        name=name,
        info=info_value
    )
 
    with Session(engine) as session:
        session.add(new_item)
        session.commit()
        logger.info(f"Item created: {name}")
 

◆ delete_item()

None Storage_connector.delete_item ( int item_id )

Deletes an item from the storage database based on the provided item id.
Args:
    item_id: The id of the item to be deleted from the storage.
Returns:
    None

Definition at line 42 of file Storage_connector.py.

def delete_item(item_id: int) -> None:
    """
    Deletes an item from the storage database based on the provided item id.
    Args:
        item_id: The id of the item to be deleted from the storage.
    Returns:
        None
    """
    with Session(engine) as session:
        item = session.get(StorageItem, item_id)
        if item:
            session.delete(item)
            session.commit()
            logger.info(f"Item with ID {item_id} deleted.")
        else:
            logger.warning(f"Deletion failed: ID {item_id} not found.")
 
 

◆ fetch_item()

StorageItem | None Storage_connector.fetch_item ( int item_id )

Fetch an item from the storage database by its id.
Args:
    item_id: The id of the item to fetch.
Returns:
    The item object if found, otherwise None.

Definition at line 30 of file Storage_connector.py.

def fetch_item(item_id: int) -> StorageItem | None:
    """
    Fetch an item from the storage database by its id.
    Args:
        item_id: The id of the item to fetch.
    Returns:
        The item object if found, otherwise None.
    """
    with Session(engine) as session:
        return session.get(StorageItem, item_id)
 
 

◆ search()

List[StorageItem] Storage_connector.search ( str search_term )

Searches the storage database for entries that match the given search term.
Args:
    search_term: The term to search for in the database. The term will be split
                      into individual words,
                      and each word will be used to search the 'type', 'name', and
                      'info' columns.
Returns:
    A list of tuples containing the rows from the database that match the search criteria.
          Returns None if there is an SQLite programming error.
Raises:
    sqlite3.ProgrammingError: If there is an error executing the query.

Definition at line 87 of file Storage_connector.py.

def search(search_term: str) -> List[StorageItem]:
    """
    Searches the storage database for entries that match the given search term.
    Args:
        search_term: The term to search for in the database. The term will be split
                          into individual words,
                          and each word will be used to search the 'type', 'name', and
                          'info' columns.
    Returns:
        A list of tuples containing the rows from the database that match the search criteria.
              Returns None if there is an SQLite programming error.
    Raises:
        sqlite3.ProgrammingError: If there is an error executing the query.
    """
    with Session(engine) as session:
        statement = select(StorageItem)
        search_words = search_term.split()
 
        for word in search_words:
            pattern = f"%{word}%"
            statement = statement.where(
                or_(
                    StorageItem.type.like(pattern),
                    StorageItem.name.like(pattern),
                    StorageItem.info.like(pattern)
                )
            )
 
        results = session.exec(statement).all()
        return list(results)
 
 

◆ setup()

Storage_connector.setup ( )

Sets up the database by creating the necessary tables if they do not already exist.

Definition at line 26 of file Storage_connector.py.

def setup():
    """Sets up the database by creating the necessary tables if they do not already exist."""
    SQLModel.metadata.create_all(engine)
 

◆ update_item()

None Storage_connector.update_item	(	int	item_id,
		int	pos,
		StorageItemType	obj_type,
		str	name,
		str	json_data )

Updates an item in the storage database.
Args:
    item_id: The id of the item to update.
    pos: The new position of the item.
    obj_type: The new type of the item.
    name: The new name of the item.
    json_data: The new JSON data associated with the item.

Definition at line 119 of file Storage_connector.py.

def update_item(item_id: int, pos: int, obj_type: StorageItemType, name: str, json_data: str) -> None:
    """
    Updates an item in the storage database.
    Args:
        item_id: The id of the item to update.
        pos: The new position of the item.
        obj_type: The new type of the item.
        name: The new name of the item.
        json_data: The new JSON data associated with the item.
    """
    with Session(engine) as session:
        db_item = session.get(StorageItem, item_id)
        if not db_item:
            logger.error(f"Update fehlgeschlagen: Item {item_id} nicht gefunden.")
            return
 
        db_item.position = pos
        db_item.type = obj_type
        db_item.name = name
        db_item.info = json_data if json_data != "{}" else None
 
        session.add(db_item)
        session.commit()
        session.refresh(db_item)
        logger.info(f"Item {item_id} erfolgreich aktualisiert.")

Variable Documentation

◆ base_dir

Storage_connector.base_dir = os.path.dirname(__file__)

Definition at line 15 of file Storage_connector.py.

◆ db_dir

Storage_connector.db_dir = os.path.join(base_dir, "data")

Definition at line 16 of file Storage_connector.py.

◆ db_path

Storage_connector.db_path = os.path.join(db_dir, "storage.db")

Definition at line 17 of file Storage_connector.py.

◆ engine

Storage_connector.engine = create_engine(f"sqlite:///{db_path}", connect_args={"check_same_thread": False})

Definition at line 24 of file Storage_connector.py.

Functions

Variables