Functions
None	setup ()

None	fetch_csv ()

Annotated[tuple, "tuple containing the item's data if found, otherwise None"]	fetch_item (item_id)

None	delete_item (item_id)

None	create_item (pos, obj_type, name, json_data)

Annotated[ list, "list of tuples containing the rows from the database that match the search criteria",]	search (search_term)

None	update_item (item_id, pos, obj_type, name, json_data)

Variables
Annotated	db_path

Annotated	mydb

Annotated	cursor = mydb.cursor()

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.

Functions:
- setup(): Sets up the database by creating the 'storage' table and
  a trigger for automatic updating of the 'modification_time' column.
- fetch_csv(): Fetches all data from the 'storage' table and
  writes it to a CSV file named 'out.csv'.
- fetch_item(item_id): Fetches an item from the storage database by its id.
- delete_item(item_id): Deletes an item from the storage database based on the provided item id.
- create_item(pos, typ, name, jsonData): Creates an item in the storage database.
- search(search_term): Searches the storage database for entries that match the given search term.

Function Documentation

◆ create_item()

None Storage_connector.create_item	(	pos,
		obj_type,
		name,
		json_data )

Creates an item in the storage database.
Args:
    pos (int): The position of the item.
    typ (str): The type of the item.
    name (str): The name of the item.
    jsonData (str): The JSON data associated with the item. If empty JSON object ("{}"),
    no additional info is stored.
Returns:
    None

Definition at line 128 of file Storage_connector.py.

def create_item(pos, obj_type, name, json_data) -> None:
    """
    Creates an item in the storage database.
    Args:
        pos (int): The position of the item.
        typ (str): The type of the item.
        name (str): The name of the item.
        jsonData (str): The JSON data associated with the item. If empty JSON object ("{}"),
        no additional info is stored.
    Returns:
        None
    """
 
    logger.info(
        "Creating item with position: {}, type: {}, name: {}, json_data: {}",
        pos,
        obj_type,
        name,
        json_data,
    )
    if json_data == "{}":
        query = "INSERT INTO storage (position, type, name) VALUES (?, ?, ?);"
        params = (pos, obj_type, name)
    else:
        query = "INSERT INTO storage (position, type, name, info) VALUES (?, ?, ?, ?);"
        params = (pos, obj_type, name, json_data)
    logger.debug(f"Executing query: {query} with params: {params}")
    cursor.execute(query, params)
    mydb.commit()
 
 

◆ delete_item()

None Storage_connector.delete_item ( item_id )

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

Definition at line 115 of file Storage_connector.py.

def delete_item(item_id) -> None:
    """
    Deletes an item from the storage database based on the provided item id.
    Args:
        itemID (int): The id of the item to be deleted from the storage.
    Returns:
        None
    """
 
    cursor.execute("DELETE FROM storage WHERE ID=?;", (item_id,))
    mydb.commit()
 
 

◆ fetch_csv()

None Storage_connector.fetch_csv ( )

Fetches all data from the 'storage' table in the database and
writes it to a CSV file named 'out.csv'.
The function executes a SQL query to select all rows from the
'storage' table, writes the column headers
and all rows to the CSV file, and saves the file with UTF-8 encoding.
Raises:
    Any exceptions raised by the database cursor execution or file operations.

Definition at line 82 of file Storage_connector.py.

def fetch_csv() -> None:
    """
    Fetches all data from the 'storage' table in the database and
    writes it to a CSV file named 'out.csv'.
    The function executes a SQL query to select all rows from the
    'storage' table, writes the column headers
    and all rows to the CSV file, and saves the file with UTF-8 encoding.
    Raises:
        Any exceptions raised by the database cursor execution or file operations.
    """
 
    cursor.execute("SELECT * FROM storage;")
    with open("static/database.csv", "w", newline="", encoding="utf-8") as csv_file:
        csv_writer = csv.writer(csv_file)
        csv_writer.writerow([i[0] for i in cursor.description])
        csv_writer.writerows(cursor.fetchall())
 
 

◆ fetch_item()

Annotated[tuple, "tuple containing the item's data if found, otherwise None"] Storage_connector.fetch_item ( item_id )

Fetch an item from the storage database by its id.
Args:
    itemID (int): The id of the item to fetch.
Returns:
    tuple: A tuple containing the item's data if found, otherwise None.

Definition at line 100 of file Storage_connector.py.

) -> Annotated[tuple, "tuple containing the item's data if found, otherwise None"]:
    """
    Fetch an item from the storage database by its id.
    Args:
        itemID (int): The id of the item to fetch.
    Returns:
        tuple: A tuple containing the item's data if found, otherwise None.
    """
 
    cursor.execute("select * From storage WHERE id=?;", (item_id,))
    return cursor.fetchone()
 
 

◆ search()

Annotated[ list, "list of tuples containing the rows from the database that match the search criteria", ] Storage_connector.search ( search_term )

Searches the storage database for entries that match the given search term.
Args:
    searchTerm (str): 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:
    list: 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 159 of file Storage_connector.py.

]:
    """
    Searches the storage database for entries that match the given search term.
    Args:
        searchTerm (str): 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:
        list: 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.
    """
    try:
        # Split searchTerm into single words
        search_terms = search_term.split()
 
        # create statement with placeholders for each search term
        conditions = " AND ".join(
            ["(type LIKE ? OR name LIKE ? OR info LIKE ?)" for _ in search_terms]
        )
        query = f"""
        SELECT *
        FROM storage
        WHERE {conditions};
        """
 
        # create parameters for each placeholder
        parameters = tuple(f"%{term}%" for term in search_terms for _ in range(3))
 
        # Debugging: print query and parameters
        logger.debug(f"Executing search query: {query} with parameters: {parameters}")
 
        cursor.execute(query, parameters)
        results = cursor.fetchall()
 
        return results
 
    except sqlite3.ProgrammingError as e:
        logger.error(f"SQLite-Error: {str(e)}")
        return None
 
 

◆ setup()

None Storage_connector.setup ( )

Sets up the database by creating the 'storage' table and a trigger for automatic
updating of the 'modification_time' column.
The 'storage' table contains the following columns:
- id: INTEGER, primary key, autoincrement
- position: INTEGER
- type: sensor | screw | display | nail | display | cable | miscellaneous | Motor Driver
- name: TEXT
- info: TEXT
- modification_time: TIMESTAMP, defaults to the current timestamp
The trigger 'update_modification_time' ensures that the 'modification_time' column
is automatically updated to the current timestamp whenever a row in the 'storage'
table is updated.
Commits the changes to the database and prints a confirmation message.

Definition at line 36 of file Storage_connector.py.

def setup() -> None:
    """
    Sets up the database by creating the 'storage' table and a trigger for automatic
    updating of the 'modification_time' column.
    The 'storage' table contains the following columns:
    - id: INTEGER, primary key, autoincrement
    - position: INTEGER
    - type: sensor | screw | display | nail | display | cable | miscellaneous | Motor Driver
    - name: TEXT
    - info: TEXT
    - modification_time: TIMESTAMP, defaults to the current timestamp
    The trigger 'update_modification_time' ensures that the 'modification_time' column
    is automatically updated to the current timestamp whenever a row in the 'storage'
    table is updated.
    Commits the changes to the database and prints a confirmation message.
    """
    cursor.execute(
        """
    CREATE TABLE IF NOT EXISTS storage (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        position INTEGER,
        type TEXT,
        name TEXT,
        info TEXT,
        modification_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    );
    """
    )
 
    # trigger for automatic update of modification_time
    cursor.execute(
        """
    CREATE TRIGGER IF NOT EXISTS update_modification_time
    AFTER UPDATE ON storage
    FOR EACH ROW
    BEGIN
        UPDATE storage
        SET modification_time = CURRENT_TIMESTAMP
        WHERE id = OLD.id;
    END;
    """
    )
    mydb.commit()
    logger.info("Database setup complete and ready for service")
 
 

◆ update_item()

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

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

Definition at line 208 of file Storage_connector.py.

def update_item(item_id, pos, obj_type, name, json_data) -> None:
    """
    Updates an item in the storage database.
    Args:
        item_id (int): The id of the item to update.
        pos (int): The new position of the item.
        obj_type (str): The new type of the item.
        name (str): The new name of the item.
        json_data (str): The new JSON data associated with the item.
    Returns:
        None
    """
    logger.info(
        "Updating item with id: {}, position: {}, type: {}, name: {}, json_data: {}",
        item_id,
        pos,
        obj_type,
        name,
        json_data,
    )
    query = """
        UPDATE storage
        SET position = ?, type = ?, name = ?, info = ?
        WHERE id = ?;
    """
    params = (pos, obj_type, name, json_data, item_id)
    logger.debug(f"Executing query: {query} with params: {params}")
    cursor.execute(query, params)
    mydb.commit()

Variable Documentation

◆ cursor

Annotated Storage_connector.cursor = mydb.cursor()

Definition at line 33 of file Storage_connector.py.

◆ db_path

Annotated Storage_connector.db_path

Initial value:

=  os.path.join(
    os.path.dirname(__file__), "data/storage.db"
)

Definition at line 26 of file Storage_connector.py.

◆ mydb

Annotated Storage_connector.mydb