SQL Interactions & Functions

Database Interactions

We have implemented the following functionality to interact with the database:

kitab.db.sql_interactions

This module contains the functions for interacting with the SQL database.

SqlHandler

Source code in kitab\db\sql_interactions.py

class SqlHandler:

    def __init__(self, dbname: str, user: str, password: str, host: str, port: str) -> None:
        # Check credentials
        if any(not cred for cred in [dbname, user, password, host, port]):
            raise Exception("Some database credentials were not passed. Please fill in the database credentials (DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME) in db_credentials.py.")

        self.connection = psycopg2.connect(dbname=dbname, user=user, password=password, host=host, port=port)
        self.cursor = self.connection.cursor()
        self.cursor.execute("CREATE EXTENSION IF NOT EXISTS vector;")
        register_vector(self.connection)

    def close_cnxn(self, verbose: bool = False) -> None:
        """
        Close the connection to the database.

        Examples:
            >>> from kitab.db.sql_interactions import SqlHandler
            >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
            >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
            >>> db.close_cnxn()

        Parameters:
            verbose (bool): Whether to print verbose output. Defaults to False.

        Returns:
            None
        """
        if verbose:
            logger.info('Committing the changes.')

        self.connection.commit()
        self.cursor.close()
        self.connection.close()

        if verbose:
            logger.info('The connection has been closed.')


    def get_table_columns(self, table_name: str, verbose: bool = False) -> list:
        """
        Retrieves the columns of a table in the database.

        Examples:
            >>> from kitab.db.sql_interactions import SqlHandler
            >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
            >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
            >>> db.get_table_columns("Book")

        Parameters:
            table_name (str): The name of the table whose columns are to be retrieved.
            verbose (bool): Whether to print verbose output. Defaults to False.

        Returns:
            list: A list of column names in the table.
        """
        try:
            self.cursor.execute(f"SELECT column_name FROM information_schema.columns WHERE table_name = '{table_name}';")
            columns = self.cursor.fetchall()
            column_names = [col[0] for col in columns]
            if verbose:
                logger.info(f'Retrieved columns for table {table_name}: {column_names}')
            return column_names
        except Exception as e:
            if verbose:
                logger.error(f'Error occurred while retrieving columns for table {table_name}: {e}')
            return []


    def execute_commands(self, commands: list, verbose: bool = False) -> None:
        """
        Executes a list of commands in the database.

        Examples:
            >>> from kitab.db.sql_interactions import SqlHandler
            >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
            >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
            >>> commands = [...]
            >>> db.execute_commands(commands)

        Parameters:
            commands (list): A list of SQL commands to be executed.
            verbose (bool): Whether to print verbose output. Defaults to False.

        Returns:
            None
        """
        for command in commands:
            self.cursor.execute(command)

        self.connection.commit()

        if verbose:
            logger.info('Commands executed successfully.')


    def insert_many(self, df: pd.DataFrame, table_name: str, verbose: bool = False) -> None:
        """
        Inserts data from a DataFrame into a table in the database.

        Examples:
            >>> from kitab.db.sql_interactions import SqlHandler
            >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
            >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
            >>> df = pd.DataFrame(...)
            >>> db.insert_many(df, "Book")

        Parameters:
            df (pd.DataFrame): The DataFrame containing the data to be inserted.
            table_name (str): The name of the table to be dropped.
            verbose (bool): Whether to print verbose output. Defaults to False.

        Returns:
            None
        """
        try:
            df = df.replace(np.nan, None)  # for handling NULLS
            df.rename(columns=lambda x: x.lower(), inplace=True)
            columns = list(df.columns)

            if verbose:
                logger.info(f'Columns before intersection: {columns}')

            sql_column_names = [i.lower() for i in self.get_table_columns(table_name)]
            columns = list(set(columns) & set(sql_column_names))

            if verbose:
                logger.info(f'Columns after intersection: {columns}')

            data_to_insert = df.loc[:, columns]
            values = []

            for row in data_to_insert.itertuples(index=False):
                values.append(tuple(row))

            if verbose:
                logger.info(f'Shape of the table to be imported: {data_to_insert.shape}')

            ncolumns = len(columns)
            params = ','.join(['%s'] * ncolumns)

            if len(columns) > 1:
                cols = ', '.join(columns)
            else:
                cols = columns[0]

            if verbose:
                logger.info(f'Insert structure: Columns: {cols}, Parameters: {params}')

            query = f"INSERT INTO {table_name} ({cols}) VALUES ({params});"

            if verbose:
                logger.info(f'Query: {query}')

            self.cursor.executemany(query, values)
            self.connection.commit()

            if verbose:
                logger.info('Data loaded successfully.')
        except Exception as e:
            logger.error(f'Error occurred while inserting data into {table_name}: {e}')


    def truncate_table(self, table_name: str, verbose: bool = False) -> None:
        """
        Truncates a table from the database.

        Examples:
            >>> from kitab.db.sql_interactions import SqlHandler
            >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
            >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
            >>> db.truncate_table("Book")

        Parameters:
            table_name (str): The name of the table to be truncated.
            verbose (bool): Whether to print verbose output. Defaults to False.

        Returns:
            None
        """
        query = f""" TRUNCATE TABLE {table_name} CASCADE; """   #if exists
        self.cursor.execute(query)

        if verbose:
            logger.info(f'the {table_name} is truncated')

        # self.cursor.close()


    def drop_table(self, table_name: str, verbose: bool = False) -> None:
        """
        Drops a table from the database if it exists.

        Examples:
            >>> from kitab.db.sql_interactions import SqlHandler
            >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
            >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
            >>> db.drop_table("Book")

        Parameters:
            table_name (str): The name of the table to be dropped.
            verbose (bool): Whether to print verbose output. Defaults to False.

        Returns:
            None
        """
        query = f"DROP TABLE IF EXISTS {table_name};"
        if verbose:
                logger.info(query)

        self.cursor.execute(query)

        self.close_cnxn.commit()

        if verbose:
            logger.info(f"Table '{table_name}' deleted.")


    def insert_records(self, table_name: str, values_list: list[dict], verbose: bool = False) -> None:
        """
        Insert one or more records into the database table.

        Examples:
            >>> from kitab.db.sql_interactions import SqlHandler
            >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
            >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
            >>> list_of_books = [...]
            >>> db.insert_records("Book", list_of_books)

        Parameters:
            values_list (List[Dict]): A list of dictionaries containing column names as keys and their values as values.
            verbose (bool): Whether to print verbose output. Defaults to False.

        Returns:
            None
        """
        if not values_list:
            logger.warning("No records to insert.")
            return            

        columns = ', '.join(values_list[0].keys())
        placeholders = '(' + ', '.join(['%s'] * len(values_list[0])) + ')'
        query = f"INSERT INTO {table_name} ({columns}) VALUES {placeholders};"        
        values = [tuple(value.values()) for value in values_list]

        self.cursor.executemany(query, values)
        self.connection.commit()

        if verbose:
            logger.info(f"{len(values_list)} records inserted successfully.")


    def update_records(self, table_name: str, updated_values: dict, condition: dict, verbose: bool = False) -> None:
        """
        Update records in the database table based on a given condition.

        Examples:
            >>> from kitab.db.sql_interactions import SqlHandler
            >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
            >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
            >>> updated_values = {...}
            >>> conditions = {...}
            >>> db.update_records("Book", updated_values, conditions)        

        Parameters:
            table_name (str): The name of the table to update records in.
            condition (dict): A dictionary representing the condition for selecting records to update.
            updated_values (dict): A dictionary containing column names as keys and their updated values as values.
            verbose (bool): Whether to print verbose output. Defaults to False.

        Returns:
            None
        """
        if not condition:
            logger.warning("No condition provided for updating records.")
            return

        if not updated_values:
            logger.warning("No values provided for update.")
            return

        set_clause = ', '.join([f"{column} = %s" for column in updated_values.keys()])
        condition_clause = ' AND '.join([f"{column} = %s" for column in condition.keys()])

        query = f"UPDATE {table_name} SET {set_clause} WHERE {condition_clause};"
        values = list(updated_values.values()) + list(condition.values())

        self.cursor.execute(query, tuple(values))
        self.connection.commit()

        if verbose:
            logger.info("Records updated successfully.")

    def remove_records(self, table_name: str, conditions_list: list[dict], verbose: bool = False) -> None:
        """
        Remove records from the database table based on multiple conditions.

        Examples:
            >>> from kitab.db.sql_interactions import SqlHandler
            >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
            >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
            >>> list_of_conditions = [...]
            >>> db.remove_records("Book", list_of_conditions)          

        Parameters:
            table_name (str): The name of the table to remove records from.
            conditions_list (list[dict]): A list of dictionaries representing conditions for selecting records to remove.
                The conditions inside each dictionary are concatenated using AND, and the dictionaries inside the list are concatenated using OR.
            verbose (bool): Whether to print verbose output. Defaults to False.

        Returns:
            None
        """
        if not conditions_list:
            logger.warning("No conditions provided for removing records.")
            return

        condition_clauses = []
        values = []

        for condition in conditions_list:
            condition_clause = ' AND '.join([f"{column} = %s" for column in condition.keys()])
            condition_clauses.append(f"({condition_clause})")
            values.extend(list(condition.values()))

        where_clause = ' OR '.join(condition_clauses)
        query = f"DELETE FROM {table_name} WHERE {where_clause};"

        self.cursor.execute(query, tuple(values))
        self.connection.commit()

        if verbose:
            logger.info("Records removed successfully.")

    def get_table(self, table_name: str, conditions: dict = None, verbose: bool = False) -> pd.DataFrame:
        """
        Retrieve data from the database table.

        Examples:
            >>> from kitab.db.sql_interactions import SqlHandler
            >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
            >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
            >>> list_of_conditions = [...]
            >>> db.get_table("Book", list_of_conditions)          

        Parameters:
            table_name (str): The name of the table to retrieve data from.
            conditions (dict, optional): A dictionary representing the conditions to filter records. Defaults to None.
            verbose (bool): Whether to print verbose output. Defaults to False.

        Returns:
            pd.DataFrame: A DataFrame containing the retrieved data.
        """
        if conditions:
            condition_clauses = []
            values = []
            for column, value in conditions.items():
                if isinstance(value, list):
                    placeholders = ', '.join(['%s'] * len(value))
                    condition_clauses.append(f"{column} IN ({placeholders})")
                    values.extend(value)
                else:
                    condition_clauses.append(f"{column} = %s")
                    values.append(value)

            condition_clause = ' AND '.join(condition_clauses)
            query = f"SELECT * FROM {table_name} WHERE {condition_clause};"
            data = pd.read_sql(query, self.connection, params=values)
        else:
            query = f"SELECT * FROM {table_name};"
            data = pd.read_sql(query, self.connection)

        if verbose:
            logger.info("Table retrieved successfully.")

        return data

close_cnxn(verbose=False)

Close the connection to the database.

Examples:

>>> from kitab.db.sql_interactions import SqlHandler
>>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
>>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
>>> db.close_cnxn()

Parameters:

Name	Type	Description	Default
verbose	bool	Whether to print verbose output. Defaults to False.	False

Returns:

Type	Description
None	None

Source code in kitab\db\sql_interactions.py

def close_cnxn(self, verbose: bool = False) -> None:
    """
    Close the connection to the database.

    Examples:
        >>> from kitab.db.sql_interactions import SqlHandler
        >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
        >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
        >>> db.close_cnxn()

    Parameters:
        verbose (bool): Whether to print verbose output. Defaults to False.

    Returns:
        None
    """
    if verbose:
        logger.info('Committing the changes.')

    self.connection.commit()
    self.cursor.close()
    self.connection.close()

    if verbose:
        logger.info('The connection has been closed.')

drop_table(table_name, verbose=False)

Drops a table from the database if it exists.

Examples:

>>> from kitab.db.sql_interactions import SqlHandler
>>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
>>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
>>> db.drop_table("Book")

Parameters:

Name	Type	Description	Default
table_name	str	The name of the table to be dropped.	required
verbose	bool	Whether to print verbose output. Defaults to False.	False

Returns:

Type	Description
None	None

Source code in kitab\db\sql_interactions.py

def drop_table(self, table_name: str, verbose: bool = False) -> None:
    """
    Drops a table from the database if it exists.

    Examples:
        >>> from kitab.db.sql_interactions import SqlHandler
        >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
        >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
        >>> db.drop_table("Book")

    Parameters:
        table_name (str): The name of the table to be dropped.
        verbose (bool): Whether to print verbose output. Defaults to False.

    Returns:
        None
    """
    query = f"DROP TABLE IF EXISTS {table_name};"
    if verbose:
            logger.info(query)

    self.cursor.execute(query)

    self.close_cnxn.commit()

    if verbose:
        logger.info(f"Table '{table_name}' deleted.")

execute_commands(commands, verbose=False)

Executes a list of commands in the database.

Examples:

>>> from kitab.db.sql_interactions import SqlHandler
>>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
>>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
>>> commands = [...]
>>> db.execute_commands(commands)

Parameters:

Name	Type	Description	Default
commands	list	A list of SQL commands to be executed.	required
verbose	bool	Whether to print verbose output. Defaults to False.	False

Returns:

Type	Description
None	None

Source code in kitab\db\sql_interactions.py

def execute_commands(self, commands: list, verbose: bool = False) -> None:
    """
    Executes a list of commands in the database.

    Examples:
        >>> from kitab.db.sql_interactions import SqlHandler
        >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
        >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
        >>> commands = [...]
        >>> db.execute_commands(commands)

    Parameters:
        commands (list): A list of SQL commands to be executed.
        verbose (bool): Whether to print verbose output. Defaults to False.

    Returns:
        None
    """
    for command in commands:
        self.cursor.execute(command)

    self.connection.commit()

    if verbose:
        logger.info('Commands executed successfully.')

get_table(table_name, conditions=None, verbose=False)

Retrieve data from the database table.

Examples:

>>> from kitab.db.sql_interactions import SqlHandler
>>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
>>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
>>> list_of_conditions = [...]
>>> db.get_table("Book", list_of_conditions)          

Parameters:

Name	Type	Description	Default
table_name	str	The name of the table to retrieve data from.	required
conditions	dict	A dictionary representing the conditions to filter records. Defaults to None.	None
verbose	bool	Whether to print verbose output. Defaults to False.	False

Returns:

Type	Description
DataFrame	pd.DataFrame: A DataFrame containing the retrieved data.

Source code in kitab\db\sql_interactions.py

def get_table(self, table_name: str, conditions: dict = None, verbose: bool = False) -> pd.DataFrame:
    """
    Retrieve data from the database table.

    Examples:
        >>> from kitab.db.sql_interactions import SqlHandler
        >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
        >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
        >>> list_of_conditions = [...]
        >>> db.get_table("Book", list_of_conditions)          

    Parameters:
        table_name (str): The name of the table to retrieve data from.
        conditions (dict, optional): A dictionary representing the conditions to filter records. Defaults to None.
        verbose (bool): Whether to print verbose output. Defaults to False.

    Returns:
        pd.DataFrame: A DataFrame containing the retrieved data.
    """
    if conditions:
        condition_clauses = []
        values = []
        for column, value in conditions.items():
            if isinstance(value, list):
                placeholders = ', '.join(['%s'] * len(value))
                condition_clauses.append(f"{column} IN ({placeholders})")
                values.extend(value)
            else:
                condition_clauses.append(f"{column} = %s")
                values.append(value)

        condition_clause = ' AND '.join(condition_clauses)
        query = f"SELECT * FROM {table_name} WHERE {condition_clause};"
        data = pd.read_sql(query, self.connection, params=values)
    else:
        query = f"SELECT * FROM {table_name};"
        data = pd.read_sql(query, self.connection)

    if verbose:
        logger.info("Table retrieved successfully.")

    return data

get_table_columns(table_name, verbose=False)

Retrieves the columns of a table in the database.

Examples:

>>> from kitab.db.sql_interactions import SqlHandler
>>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
>>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
>>> db.get_table_columns("Book")

Parameters:

Name	Type	Description	Default
table_name	str	The name of the table whose columns are to be retrieved.	required
verbose	bool	Whether to print verbose output. Defaults to False.	False

Returns:

Name	Type	Description
list	list	A list of column names in the table.

Source code in kitab\db\sql_interactions.py

def get_table_columns(self, table_name: str, verbose: bool = False) -> list:
    """
    Retrieves the columns of a table in the database.

    Examples:
        >>> from kitab.db.sql_interactions import SqlHandler
        >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
        >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
        >>> db.get_table_columns("Book")

    Parameters:
        table_name (str): The name of the table whose columns are to be retrieved.
        verbose (bool): Whether to print verbose output. Defaults to False.

    Returns:
        list: A list of column names in the table.
    """
    try:
        self.cursor.execute(f"SELECT column_name FROM information_schema.columns WHERE table_name = '{table_name}';")
        columns = self.cursor.fetchall()
        column_names = [col[0] for col in columns]
        if verbose:
            logger.info(f'Retrieved columns for table {table_name}: {column_names}')
        return column_names
    except Exception as e:
        if verbose:
            logger.error(f'Error occurred while retrieving columns for table {table_name}: {e}')
        return []

insert_many(df, table_name, verbose=False)

Inserts data from a DataFrame into a table in the database.

Examples:

>>> from kitab.db.sql_interactions import SqlHandler
>>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
>>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
>>> df = pd.DataFrame(...)
>>> db.insert_many(df, "Book")

Parameters:

Name	Type	Description	Default
df	DataFrame	The DataFrame containing the data to be inserted.	required
table_name	str	The name of the table to be dropped.	required
verbose	bool	Whether to print verbose output. Defaults to False.	False

Returns:

Type	Description
None	None

Source code in kitab\db\sql_interactions.py

def insert_many(self, df: pd.DataFrame, table_name: str, verbose: bool = False) -> None:
    """
    Inserts data from a DataFrame into a table in the database.

    Examples:
        >>> from kitab.db.sql_interactions import SqlHandler
        >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
        >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
        >>> df = pd.DataFrame(...)
        >>> db.insert_many(df, "Book")

    Parameters:
        df (pd.DataFrame): The DataFrame containing the data to be inserted.
        table_name (str): The name of the table to be dropped.
        verbose (bool): Whether to print verbose output. Defaults to False.

    Returns:
        None
    """
    try:
        df = df.replace(np.nan, None)  # for handling NULLS
        df.rename(columns=lambda x: x.lower(), inplace=True)
        columns = list(df.columns)

        if verbose:
            logger.info(f'Columns before intersection: {columns}')

        sql_column_names = [i.lower() for i in self.get_table_columns(table_name)]
        columns = list(set(columns) & set(sql_column_names))

        if verbose:
            logger.info(f'Columns after intersection: {columns}')

        data_to_insert = df.loc[:, columns]
        values = []

        for row in data_to_insert.itertuples(index=False):
            values.append(tuple(row))

        if verbose:
            logger.info(f'Shape of the table to be imported: {data_to_insert.shape}')

        ncolumns = len(columns)
        params = ','.join(['%s'] * ncolumns)

        if len(columns) > 1:
            cols = ', '.join(columns)
        else:
            cols = columns[0]

        if verbose:
            logger.info(f'Insert structure: Columns: {cols}, Parameters: {params}')

        query = f"INSERT INTO {table_name} ({cols}) VALUES ({params});"

        if verbose:
            logger.info(f'Query: {query}')

        self.cursor.executemany(query, values)
        self.connection.commit()

        if verbose:
            logger.info('Data loaded successfully.')
    except Exception as e:
        logger.error(f'Error occurred while inserting data into {table_name}: {e}')

insert_records(table_name, values_list, verbose=False)

Insert one or more records into the database table.

Examples:

>>> from kitab.db.sql_interactions import SqlHandler
>>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
>>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
>>> list_of_books = [...]
>>> db.insert_records("Book", list_of_books)

Parameters:

Name	Type	Description	Default
values_list	List[Dict]	A list of dictionaries containing column names as keys and their values as values.	required
verbose	bool	Whether to print verbose output. Defaults to False.	False

Returns:

Type	Description
None	None

Source code in kitab\db\sql_interactions.py

def insert_records(self, table_name: str, values_list: list[dict], verbose: bool = False) -> None:
    """
    Insert one or more records into the database table.

    Examples:
        >>> from kitab.db.sql_interactions import SqlHandler
        >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
        >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
        >>> list_of_books = [...]
        >>> db.insert_records("Book", list_of_books)

    Parameters:
        values_list (List[Dict]): A list of dictionaries containing column names as keys and their values as values.
        verbose (bool): Whether to print verbose output. Defaults to False.

    Returns:
        None
    """
    if not values_list:
        logger.warning("No records to insert.")
        return            

    columns = ', '.join(values_list[0].keys())
    placeholders = '(' + ', '.join(['%s'] * len(values_list[0])) + ')'
    query = f"INSERT INTO {table_name} ({columns}) VALUES {placeholders};"        
    values = [tuple(value.values()) for value in values_list]

    self.cursor.executemany(query, values)
    self.connection.commit()

    if verbose:
        logger.info(f"{len(values_list)} records inserted successfully.")

remove_records(table_name, conditions_list, verbose=False)

Remove records from the database table based on multiple conditions.

Examples:

>>> from kitab.db.sql_interactions import SqlHandler
>>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
>>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
>>> list_of_conditions = [...]
>>> db.remove_records("Book", list_of_conditions)          

Parameters:

Name	Type	Description	Default
table_name	str	The name of the table to remove records from.	required
conditions_list	list[dict]	A list of dictionaries representing conditions for selecting records to remove. The conditions inside each dictionary are concatenated using AND, and the dictionaries inside the list are concatenated using OR.	required
verbose	bool	Whether to print verbose output. Defaults to False.	False

Returns:

Type	Description
None	None

Source code in kitab\db\sql_interactions.py

def remove_records(self, table_name: str, conditions_list: list[dict], verbose: bool = False) -> None:
    """
    Remove records from the database table based on multiple conditions.

    Examples:
        >>> from kitab.db.sql_interactions import SqlHandler
        >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
        >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
        >>> list_of_conditions = [...]
        >>> db.remove_records("Book", list_of_conditions)          

    Parameters:
        table_name (str): The name of the table to remove records from.
        conditions_list (list[dict]): A list of dictionaries representing conditions for selecting records to remove.
            The conditions inside each dictionary are concatenated using AND, and the dictionaries inside the list are concatenated using OR.
        verbose (bool): Whether to print verbose output. Defaults to False.

    Returns:
        None
    """
    if not conditions_list:
        logger.warning("No conditions provided for removing records.")
        return

    condition_clauses = []
    values = []

    for condition in conditions_list:
        condition_clause = ' AND '.join([f"{column} = %s" for column in condition.keys()])
        condition_clauses.append(f"({condition_clause})")
        values.extend(list(condition.values()))

    where_clause = ' OR '.join(condition_clauses)
    query = f"DELETE FROM {table_name} WHERE {where_clause};"

    self.cursor.execute(query, tuple(values))
    self.connection.commit()

    if verbose:
        logger.info("Records removed successfully.")

truncate_table(table_name, verbose=False)

Truncates a table from the database.

Examples:

>>> from kitab.db.sql_interactions import SqlHandler
>>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
>>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
>>> db.truncate_table("Book")

Parameters:

Name	Type	Description	Default
table_name	str	The name of the table to be truncated.	required
verbose	bool	Whether to print verbose output. Defaults to False.	False

Returns:

Type	Description
None	None

Source code in kitab\db\sql_interactions.py

def truncate_table(self, table_name: str, verbose: bool = False) -> None:
    """
    Truncates a table from the database.

    Examples:
        >>> from kitab.db.sql_interactions import SqlHandler
        >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
        >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
        >>> db.truncate_table("Book")

    Parameters:
        table_name (str): The name of the table to be truncated.
        verbose (bool): Whether to print verbose output. Defaults to False.

    Returns:
        None
    """
    query = f""" TRUNCATE TABLE {table_name} CASCADE; """   #if exists
    self.cursor.execute(query)

    if verbose:
        logger.info(f'the {table_name} is truncated')

update_records(table_name, updated_values, condition, verbose=False)

Update records in the database table based on a given condition.

Examples:

>>> from kitab.db.sql_interactions import SqlHandler
>>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
>>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
>>> updated_values = {...}
>>> conditions = {...}
>>> db.update_records("Book", updated_values, conditions)        

Parameters:

Name	Type	Description	Default
table_name	str	The name of the table to update records in.	required
condition	dict	A dictionary representing the condition for selecting records to update.	required
updated_values	dict	A dictionary containing column names as keys and their updated values as values.	required
verbose	bool	Whether to print verbose output. Defaults to False.	False

Returns:

Type	Description
None	None

Source code in kitab\db\sql_interactions.py

def update_records(self, table_name: str, updated_values: dict, condition: dict, verbose: bool = False) -> None:
    """
    Update records in the database table based on a given condition.

    Examples:
        >>> from kitab.db.sql_interactions import SqlHandler
        >>> from kitab.db.db_credentials import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
        >>> db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
        >>> updated_values = {...}
        >>> conditions = {...}
        >>> db.update_records("Book", updated_values, conditions)        

    Parameters:
        table_name (str): The name of the table to update records in.
        condition (dict): A dictionary representing the condition for selecting records to update.
        updated_values (dict): A dictionary containing column names as keys and their updated values as values.
        verbose (bool): Whether to print verbose output. Defaults to False.

    Returns:
        None
    """
    if not condition:
        logger.warning("No condition provided for updating records.")
        return

    if not updated_values:
        logger.warning("No values provided for update.")
        return

    set_clause = ', '.join([f"{column} = %s" for column in updated_values.keys()])
    condition_clause = ' AND '.join([f"{column} = %s" for column in condition.keys()])

    query = f"UPDATE {table_name} SET {set_clause} WHERE {condition_clause};"
    values = list(updated_values.values()) + list(condition.values())

    self.cursor.execute(query, tuple(values))
    self.connection.commit()

    if verbose:
        logger.info("Records updated successfully.")

Functions

kitab.db.functions

This module contains tailored functions for interacting with the database. These are used by the API and the recommendation model.

add_book_db(book, verbose=False)

Adds a book to the database.

Examples:

>>> from kitab.db.functions import add_book_db
>>> add_book_db({
        "isbn": "1442942355",
        "title": "The Ghostly Rental",
        "description": "Employing the subtle methods of presenting mysterious ghost stories in the backdrop of psychological troubles, the novel presents the life of James. The troubles that he faces, combined with the baffling events around him give an aura to the novel that is almost unsurpassable",
        "available": False,
        "authors": [
            "Henry James"
        ],
        "genres": [
            "Horror",
            "Short Stories",
            "The United States Of America"
        ]
    })

Parameters:

Name	Type	Description	Default
book	dict	A dictionary containing the book information.	required

Returns:

Name	Type	Description
bool	bool	True if the book was successfully added, False otherwise.

Source code in kitab\db\functions.py

def add_book_db(book: dict, verbose: bool = False) -> bool:
    """
    Adds a book to the database.

    Examples:
        >>> from kitab.db.functions import add_book_db
        >>> add_book_db({
                "isbn": "1442942355",
                "title": "The Ghostly Rental",
                "description": "Employing the subtle methods of presenting mysterious ghost stories in the backdrop of psychological troubles, the novel presents the life of James. The troubles that he faces, combined with the baffling events around him give an aura to the novel that is almost unsurpassable",
                "available": False,
                "authors": [
                    "Henry James"
                ],
                "genres": [
                    "Horror",
                    "Short Stories",
                    "The United States Of America"
                ]
            })

    Parameters:
        book (dict): A dictionary containing the book information.

    Returns:
        bool: True if the book was successfully added, False otherwise.
    """
    try:
        # Open connection to the database
        db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
        if verbose:
            logger.info("Database connection opened.")

        # Extract information from the book dictionary
        ISBN = book["isbn"]
        title = book["title"]
        description = book["description"]
        available = book["available"]
        embedding = get_embedding(book["description"]).tolist()

        db.insert_records("book", [{"isbn": ISBN, "title": title, "description": description, "embedding": embedding, "available": available}])
        if verbose:
            logger.info("Book table populated.")

        # Add author(s) to the author table if doesn't exist
        authors = book["authors"]
        if len(authors) > 0:
            if verbose:
                logger.info("Authors to be added.")
            author_ids = _get_or_add_authors(db, authors, verbose=verbose)

            db.insert_records("bookauthor", [{"isbn": ISBN, "author_id": int(author_id)} for author_id in author_ids], verbose=verbose)
            if verbose:
                logger.info("Author table populated.")
        else:
            if verbose:
                logger.info("No authors to be added.")

        # Add genres to the genres table if doesn't exist
        genres = book["genres"]    
        if len(genres) > 0:
            if verbose:
                logger.info("Genres to be added.")
            genre_ids = _get_or_add_genres(db, genres, verbose=verbose)

            db.insert_records("bookgenre", [{"isbn": ISBN, "genre_id": int(genre_id)} for genre_id in genre_ids], verbose=verbose)
            if verbose:
                logger.info("Genre table populated.")
        else:
            if verbose:
                logger.info("No genres to be added.")

        return True
    except:
        return False    

add_recommendation_log(description, recommendation_isbn, successful, verbose=False)

Adds a recommendation log to the history table.

Examples:

>>> from kitab.db.functions import add_recommendation_log
>>> add_recommendation_log(description="In a masterful blend of psychological intrigue and spectral disturbances, this novel unfurls the complex life of Clara. Her internal struggles are mirrored by eerie, inexplicable occurrences, weaving a tale that is both deeply personal and chillingly atmospheric, offering an unparalleled exploration of the human psyche shadowed by the paranormal.", recommendation_isbn="1442942355", successful=True)

Parameters:

Name	Type	Description	Default
description	str	The description of the recommendation.	required
recommendation_isbn	str	The ISBN of the recommended book.	required
successful	bool	Whether the recommendation was successful or not.	required
verbose	bool	Whether to print verbose output. Defaults to False.	False

Returns:

Name	Type	Description
bool	bool	True if the recommendation log was successfully added, False otherwise.

Source code in kitab\db\functions.py

def add_recommendation_log(description: str, recommendation_isbn: str, successful: bool, verbose: bool = False) -> bool:
    """
    Adds a recommendation log to the history table.

    Examples:
        >>> from kitab.db.functions import add_recommendation_log
        >>> add_recommendation_log(description="In a masterful blend of psychological intrigue and spectral disturbances, this novel unfurls the complex life of Clara. Her internal struggles are mirrored by eerie, inexplicable occurrences, weaving a tale that is both deeply personal and chillingly atmospheric, offering an unparalleled exploration of the human psyche shadowed by the paranormal.", recommendation_isbn="1442942355", successful=True)

    Parameters:
        description (str): The description of the recommendation.
        recommendation_isbn (str): The ISBN of the recommended book.
        successful (bool): Whether the recommendation was successful or not.
        verbose (bool): Whether to print verbose output. Defaults to False.

    Returns:
        bool: True if the recommendation log was successfully added, False otherwise.
    """
    try:
        # Open connection to the database
        db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
        if verbose:
            logger.info("Database connection opened.")

        # Insert the recommendation log into the history table
        db.insert_records("history", [{"description": description, "recommendation_isbn": recommendation_isbn, "successful": successful}], verbose=verbose)

        return True
    except Exception as e:
        logger.error("Error adding recommendation log.")
        return False

get_authors(ISBNs, verbose=False)

Get the authors for the given list of ISBNs.

Examples:

>>> from kitab.db.functions import get_authors
>>> get_authors(ISBNs=["1442942355", "1613720211"])

Parameters:

Name	Type	Description	Default
ISBNs	list[str]	A list of ISBNs.	required
verbose	bool	Whether to print verbose output. Defaults to False.	False

Returns:

Type	Description
dict[str:list]	dict[str:list]: A dictionary containing the authors for each ISBN.

Source code in kitab\db\functions.py

def get_authors(ISBNs: list[str], verbose: bool = False) -> dict[str:list]:
    """
    Get the authors for the given list of ISBNs.

    Examples:
        >>> from kitab.db.functions import get_authors
        >>> get_authors(ISBNs=["1442942355", "1613720211"])

    Parameters:
        ISBNs (list[str]): A list of ISBNs.
        verbose (bool): Whether to print verbose output. Defaults to False.

    Returns:
        dict[str:list]: A dictionary containing the authors for each ISBN.
    """
    # Open connection to the database
    db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
    if verbose:
        logger.info("Database connection opened.")

    # Retrieve the authors of the books with the given ISBNs
    authors = db.get_table("bookauthor", conditions={"isbn": ISBNs}, verbose=verbose)
    author_ids = authors["author_id"].tolist()

    author_table = db.get_table("author", conditions={"author_id": author_ids}, verbose=verbose)

    # Initialize dictionary to store authors for each ISBN
    isbn_authors = {isbn: [] for isbn in ISBNs}

    # Populate dictionary with authors
    for _, row in authors.iterrows():
        isbn = row["isbn"]
        author_id = row["author_id"]
        author_name = author_table.loc[author_table['author_id'] == author_id, 'full_name'].iloc[0]
        isbn_authors[isbn].append(author_name)

    # Return the dictionary of lists
    return isbn_authors

get_book_by_ISBN(ISBN, verbose=False)

Retrieves a book from the database based on its ISBN.

Examples:

>>> from kitab.db.functions import get_book_by_ISBN
>>> get_book_by_ISBN("1442942355")

Parameters:

Name	Type	Description	Default
ISBN	str	The ISBN of the book to retrieve.	required
verbose	bool	Whether to print verbose output. Defaults to False.	False

Returns:

Type	Description
tuple[dict]	tuple[dict]: A tuple containing the book information, authors, and genres if found, or None if no book is found.

Source code in kitab\db\functions.py

def get_book_by_ISBN(ISBN: str, verbose: bool = False) -> tuple[dict]:
    """
    Retrieves a book from the database based on its ISBN.

    Examples:
        >>> from kitab.db.functions import get_book_by_ISBN
        >>> get_book_by_ISBN("1442942355")

    Parameters:
        ISBN (str): The ISBN of the book to retrieve.
        verbose (bool): Whether to print verbose output. Defaults to False.

    Returns:
        tuple[dict]: A tuple containing the book information, authors, and genres if found, or None if no book is found.
    """

    # Open connection to the database
    db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
    if verbose:
        logger.info("Database connection opened.")

    # Retrieve the book with the given ISBN
    book = db.get_table("book", conditions={"isbn": ISBN})

    if len(book) == 0:
        logger.info("Book not found.")
        return None

    if verbose:
        logger.info("Book retrieved.")

    book_author = db.get_table("bookauthor", conditions={"isbn": ISBN})
    book_genre = db.get_table("bookgenre", conditions={"isbn": ISBN})

    # If no book found, return None
    if len(book) == 0:
        return None, None, None

    book.drop(columns=["embedding"], inplace=True)
    book = book.to_dict(orient='records')[0]

    author_ids = book_author["author_id"].tolist()
    authors = []
    if len(author_ids) > 0:
        author = db.get_table("author", conditions={"author_id": author_ids})
        authors = author["full_name"].tolist()
        if verbose:
            logger.info(f"Authors retrieved.")
    else:
        if verbose:
            logger.info("No authors found.")

    genre_ids = book_genre["genre_id"].tolist()
    genres = []
    if len(genre_ids) > 0:
        genre = db.get_table("genre", conditions={"genre_id": genre_ids})
        genres = genre["genre"].tolist()
        if verbose:
            logger.info(f"Genres retrieved.")
    else:
        if verbose:
            logger.info("No genres found.")

    book["authors"] = authors
    book["genres"] = genres

    # Return the book
    return book

get_book_by_title(title, verbose=False)

Retrieves a book from the database based on its title.

Examples:

>>> from kitab.db.functions import get_book_by_title
>>> get_book_by_title("The Ghostly Rental")

Parameters:

Name	Type	Description	Default
title	str	The title of the book to retrieve.	required
verbose	bool	Whether to print verbose output. Defaults to False.	False

Returns:

Type	Description
tuple[dict]	tuple[dict]: A tuple containing the book information, authors, and genres if found, or None if no book is found.

Source code in kitab\db\functions.py

def get_book_by_title(title: str, verbose: bool = False) -> tuple[dict]:
    """
    Retrieves a book from the database based on its title.

    Examples:
        >>> from kitab.db.functions import get_book_by_title
        >>> get_book_by_title("The Ghostly Rental")

    Parameters:
        title (str): The title of the book to retrieve.
        verbose (bool): Whether to print verbose output. Defaults to False.

    Returns:
        tuple[dict]: A tuple containing the book information, authors, and genres if found, or None if no book is found.
    """
    # Open connection to the database
    db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
    if verbose:
        logger.info("Database connection opened.")

    # Retrieve the book with the given title
    books = db.get_table("book", conditions={"title": title}, verbose=verbose)

    if len(books) == 0:
        if verbose:
            logger.info("Book not found.")
        return None
    else:
        if verbose:
            logger.info("Book ISBN retrieved.")
        ISBN = books["isbn"].values[0]

    # Return the book
    return get_book_by_ISBN(ISBN)

get_genres(ISBNs, verbose=False)

Get the genres for the given list of ISBNs.

Examples:

>>> from kitab.db.functions import get_genres
>>> get_genres(ISBNs=["1442942355", "1613720211"])

Parameters:

Name	Type	Description	Default
ISBNs	list[str]	A list of ISBNs.	required
verbose	bool	Whether to print verbose output. Defaults to False.	False

Returns:

Type	Description
dict[str:list]	dict[str:list]: A dictionary containing the genres for each ISBN.

Source code in kitab\db\functions.py

def get_genres(ISBNs: list[str], verbose: bool = False) -> dict[str:list]:
    """
    Get the genres for the given list of ISBNs.

    Examples:
        >>> from kitab.db.functions import get_genres
        >>> get_genres(ISBNs=["1442942355", "1613720211"])

    Parameters:
        ISBNs (list[str]): A list of ISBNs.
        verbose (bool): Whether to print verbose output. Defaults to False.

    Returns:
        dict[str:list]: A dictionary containing the genres for each ISBN.
    """
    # Open connection to the database
    db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
    if verbose:
        logger.info("Database connection opened.")

    # Retrieve the genres of the books with the given ISBNs
    genres = db.get_table("bookgenre", conditions={"isbn": ISBNs}, verbose=verbose)
    genre_ids = genres["genre_id"].tolist()

    genre_table = db.get_table("genre", conditions={"genre_id": genre_ids}, verbose=verbose)

    # Initialize dictionary to store genres for each ISBN
    isbn_genres = {isbn: [] for isbn in ISBNs}

    # Populate dictionary with genres
    for _, row in genres.iterrows():
        isbn = row["isbn"]
        genre_id = row["genre_id"]
        genre_name = genre_table.loc[genre_table['genre_id'] == genre_id, 'genre'].iloc[0]
        isbn_genres[isbn].append(genre_name)

    # Return the dictionary of lists
    return isbn_genres

get_history_by_recommendation_isbn(recommendation_isbn, verbose=False)

Get the history of recommendations for a book with the given ISBN.

Examples:

>>> from kitab.db.functions import get_history_by_recommendation_isbn
>>> get_history_by_recommendation_isbn(recommendation_isbn="1442942355")

Parameters:

Name	Type	Description	Default
recommendation_isbn	str	The ISBN of the recommended book.	required

Returns:

Name	Type	Description
dict	dict	A dictionary containing the history of recommendations for the book.

Source code in kitab\db\functions.py

def get_history_by_recommendation_isbn(recommendation_isbn: str, verbose: bool = False) -> dict:
    """
    Get the history of recommendations for a book with the given ISBN.

    Examples:
        >>> from kitab.db.functions import get_history_by_recommendation_isbn
        >>> get_history_by_recommendation_isbn(recommendation_isbn="1442942355")

    Parameters:
        recommendation_isbn (str): The ISBN of the recommended book.

    Returns:
        dict: A dictionary containing the history of recommendations for the book.
    """
    # Open connection to the database
    db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
    if verbose:
        logger.info("Database connection opened.")

    # Retrieve the history of books that have been recommended
    history = db.get_table("history", conditions={"recommendation_ISBN": recommendation_isbn}, verbose=verbose)

    # Return the history
    return history.drop(columns="log_id").to_dict(orient='records')

get_table_from_db(table_name, conditions=None, verbose=False)

Retrieves a table from the database.

Examples:

>>> from kitab.db.functions import get_table_from_db
>>> get_table_from_db("book", conditions={"available": True})

Parameters:

Name	Type	Description	Default
table_name	str	The name of the table to retrieve.	required
conditions	dict	A dictionary of conditions to filter the table.	None
verbose	bool	Whether to print verbose output. Defaults to False.	False

Returns:

Type	Description
DataFrame	pd.DataFrame: A DataFrame containing the table information.

Source code in kitab\db\functions.py

def get_table_from_db(table_name: str, conditions: dict = None, verbose: bool = False) -> pd.DataFrame:
    """
    Retrieves a table from the database.

    Examples:
        >>> from kitab.db.functions import get_table_from_db
        >>> get_table_from_db("book", conditions={"available": True})

    Parameters:
        table_name (str): The name of the table to retrieve.
        conditions (dict): A dictionary of conditions to filter the table.
        verbose (bool): Whether to print verbose output. Defaults to False.

    Returns:
        pd.DataFrame: A DataFrame containing the table information.
    """
    # Open connection to the database
    db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
    if verbose:
        logger.info("Database connection opened.")

    # Retrieve the table from the database
    if conditions:
        table = db.get_table(table_name, conditions=conditions, verbose=verbose)
    else:
        table = db.get_table(table_name, verbose=verbose)

    if verbose:
        logger.info(f"Table {table_name} retrieved.")

    # Return the table
    return table

update_book_db(ISBN, new_book, verbose=False)

Updates a book in the database.

Examples:

>>> from kitab.db.functions import update_book_db
>>> update_book_db("1442942355", {
        "available": True,
        "genres": [
            "Horror",
            "Short Stories",
            "Mystery"
        ]
    })

Parameters:

Name	Type	Description	Default
ISBN	str	The ISBN of the book to update.	required
new_book	dict	A dictionary containing the updated book information.	required
verbose	bool	Whether to print verbose output. Defaults to False.	False

Returns:

Name	Type	Description
bool	bool	True if the book was successfully updated, False otherwise.

Source code in kitab\db\functions.py

def update_book_db(ISBN: str, new_book: dict, verbose: bool = False) -> bool:
    """
    Updates a book in the database.

    Examples:
        >>> from kitab.db.functions import update_book_db
        >>> update_book_db("1442942355", {
                "available": True,
                "genres": [
                    "Horror",
                    "Short Stories",
                    "Mystery"
                ]
            })

    Parameters:
        ISBN (str): The ISBN of the book to update.
        new_book (dict): A dictionary containing the updated book information.
        verbose (bool): Whether to print verbose output. Defaults to False.

    Returns:
        bool: True if the book was successfully updated, False otherwise.
    """
    try:
        # Open connection to the database
        db = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)
        if verbose:
            logger.info("Database connection opened.")

        condition = {"ISBN": ISBN}
        new_values = {}

        latest_ISBN = ISBN

        for key in new_book.keys():
            if key in ["ISBN", "title", "description", "available"]:
                new_values[key] = new_book[key]
            if key == "ISBN":
                latest_ISBN = new_book["ISBN"]
            if key == "description":
                new_values["embedding"] = get_embedding(new_book["description"]).tolist()

        db.update_records("book", new_values, condition, verbose=verbose)
        if verbose:
            logger.info("Book table updated.")

        ISBN = latest_ISBN

        # Get the tables
        book_author = db.get_table("bookauthor", conditions={"isbn": ISBN}, verbose=verbose)
        book_genre = db.get_table("bookgenre", conditions={"isbn": ISBN}, verbose=verbose)

        # Add author(s) to the author table if doesn't exist
        if "authors" in new_book:
            if verbose:
                logger.info("Authors to be updated.")

            authors = new_book["authors"]
            new_author_ids = set(_get_or_add_authors(db, authors, verbose=verbose))
            current_author_ids = set(book_author[book_author["isbn"] == ISBN]["author_id"].tolist())

            removed_authors = current_author_ids - new_author_ids
            added_authors = new_author_ids - current_author_ids

            db.remove_records("bookauthor", [{"isbn": ISBN, "author_id": int(removed_author)} for removed_author in removed_authors], verbose=verbose)
            db.insert_records("bookauthor", [{"isbn": ISBN, "author_id": int(added_author)} for added_author in added_authors], verbose=verbose)

            if verbose:
                logger.info("Author table updated.")
        else:
            if verbose:
                logger.info("No authors to be updated.")

        # Add genres to the genres table if doesn't exist
        if "genres" in new_book:
            if verbose:
                logger.info("Genres to be updated.")

            genres = new_book["genres"]    
            new_genre_ids = set(_get_or_add_genres(db, genres, verbose=verbose))
            current_genre_ids = set(book_genre[book_genre["isbn"] == ISBN]["genre_id"].tolist())

            removed_genres = current_genre_ids - new_genre_ids
            added_genres = new_genre_ids - current_genre_ids

            db.remove_records("bookgenre", [{"isbn": ISBN, "genre_id": int(removed_genre)} for removed_genre in removed_genres], verbose=verbose)
            db.insert_records("bookgenre", [{"isbn": ISBN, "genre_id": int(added_genre)} for added_genre in added_genres], verbose=verbose)

            if verbose:
                logger.info("Genre table updated.")
        else:
            if verbose:
                logger.info("No genres to be updated.")

        return True
    except:
        return False

Loading Data Into the Database

kitab.db.get_data

get_full_data(folder_path='data', verbose=False)

Retrieves and combines data from multiple CSV files and corresponding pickle files.

Examples:

>>> from kitab.db.get_data import get_full_data
>>> get_full_data(folder_path="data")

Parameters:

Name	Type	Description	Default
folder_path	str	The path to the directory containing the CSV and pickle files.	'data'
verbose	bool	Whether to display logs. Default is False.	False

Returns:

Type	Description
DataFrame	pd.DataFrame: A DataFrame containing the combined data with an additional 'embedding' column.

Source code in kitab\db\get_data.py

def get_full_data(folder_path: str = "data", verbose: bool = False) -> pd.DataFrame:
    """
    Retrieves and combines data from multiple CSV files and corresponding pickle files.

    Examples:
        >>> from kitab.db.get_data import get_full_data
        >>> get_full_data(folder_path="data")

    Parameters:
        folder_path (str): The path to the directory containing the CSV and pickle files.
        verbose (bool): Whether to display logs. Default is False.

    Returns:
        pd.DataFrame: A DataFrame containing the combined data with an additional 'embedding' column.
    """
    data_paths = sorted(glob(f"{folder_path}/*.csv"))
    emb_paths = sorted(glob(f"{folder_path}/*.pkl"))

    if len(data_paths) == 0:
        raise Exception("No CSV files found in the specified folder.")
    elif len(emb_paths) == 0:
        raise Exception("No PKL files found in the specified folder.")
    elif len(data_paths) != len(emb_paths):
        raise Exception("The number of CSV and PKL files do not match.")

    if verbose:
        logger.info("Data and embeddings found successfully.")

    datas = [pd.read_csv(data_path) for data_path in data_paths]
    embs = []

    for emb_path in emb_paths:
        with open(emb_path, "rb") as f:
            emb = pickle.load(f)
        embs.append(emb)

    df = pd.concat(datas).reset_index(drop=True)
    df["embedding"] = np.concatenate(embs).tolist()

    if verbose:
        logger.info("Data and embeddings combined successfully.")

    return df

load_data(folder_path='data', verbose=False)

Load data from a specified folder path and insert it into the database.

Examples:

>>> from kitab.db.get_data import load_data
>>> load_data(folder_path="data")

Parameters:

Name	Type	Description	Default
folder_path	str	The path to the folder containing the data files. Default is "data".	'data'
verbose	bool	Whether to display logs. Default is False.	False

Returns:

Type	Description
None	None

Source code in kitab\db\get_data.py

def load_data(folder_path: str = "data", verbose: bool = False) -> None:
    """
    Load data from a specified folder path and insert it into the database.

    Examples:
        >>> from kitab.db.get_data import load_data
        >>> load_data(folder_path="data")

    Parameters:
        folder_path (str): The path to the folder containing the data files. Default is "data".
        verbose (bool): Whether to display logs. Default is False.

    Returns:
        None
    """
    try:
        # Getting the full data
        data = get_full_data(folder_path, verbose=verbose)

        data.fillna({"genre": ""}, inplace=True)
        data = data[REQUIRED_COLUMNS + ["embedding", "available"]]

        data.dropna(subset = ["isbn", "description"], inplace=True)

        book_table = data[["isbn", "title", "description", "embedding", "available"]]

        if verbose:
            logger.info("Data loaded successfully.")

        def split_and_filter(cell):
            if cell:
                genres = cell.split(",")
                return [g.strip() for g in genres if g]
            else:
                return []

        authors = data["author"].apply(lambda x: split_and_filter(x))
        data["author"] = authors
        unique_authors = authors.explode().dropna().unique()
        author_table = pd.DataFrame({"author_id": range(1, len(unique_authors)+1), "full_name": unique_authors})

        if verbose:
            logger.info("Authors extracted successfully.")

        books_with_authors = data[data['author'].map(lambda d: len(d)) > 0]
        book_author = books_with_authors.explode("author")[["author", "isbn"]]
        book_author = pd.merge(book_author, author_table, how='left', left_on='author', right_on='full_name')[["isbn", "author_id"]]
        # book_author.rename(columns={"isbn":"ISBN"}, inplace=True)
        book_author.drop_duplicates(inplace=True)
        book_author.reset_index(drop=True, inplace=True)
        book_author["author_id"] = book_author["author_id"].astype(int)

        if verbose:
            logger.info("Book-Author mapping created successfully.")

        genres = data["genre"].apply(lambda x: split_and_filter(x))
        data["genre"] = genres
        unique_genres = genres.explode().dropna().unique()
        genre_table = pd.DataFrame({"genre_id": range(1, len(unique_genres)+1), "genre": unique_genres})

        if verbose:
            logger.info("Genres extracted successfully.")

        books_with_genres = data[data['genre'].map(lambda d: len(d)) > 0]
        book_genre = books_with_genres.explode("genre")[["genre", "isbn"]]
        book_genre["genre"] = book_genre["genre"].str.strip()
        book_genre = pd.merge(book_genre, genre_table, how='left', left_on='genre', right_on='genre')[["isbn", "genre_id"]]
        # book_genre.rename(columns={"isbn":"ISBN"}, inplace=True)
        book_genre.drop_duplicates(inplace=True)
        book_genre.reset_index(drop=True, inplace=True)
        book_genre["genre_id"] = book_genre["genre_id"].astype(int)

        if verbose:
            logger.info("Book-Genre mapping created successfully.")

        # Establish connection with the database
        sql_handler = SqlHandler(DB_NAME, user=DB_USER, password=DB_PASSWORD, host=DB_HOST, port=DB_PORT)

        # Create tables in the DB
        for command in COMMANDS:
            sql_handler.cursor.execute(command)

        if verbose:
            logger.info("Tables created successfully.")

        # Inserting data
        # Book table
        sql_handler.insert_many(book_table, "book", verbose=verbose)

        if verbose:
            logger.info("Book data inserted successfully.")

        # Author table
        sql_handler.insert_many(author_table, "author", verbose=verbose)

        if verbose:
            logger.info("Author data inserted successfully.")

        # Genre table
        sql_handler.insert_many(genre_table, "genre", verbose=verbose)

        if verbose:
            logger.info("Genre data inserted successfully.")

        # BookAuthor table
        sql_handler.insert_many(book_author, "bookauthor", verbose=verbose)

        if verbose:
            logger.info("Book-Author mapping inserted successfully.")

        # BookGenre table
        sql_handler.insert_many(book_genre, "bookgenre", verbose=verbose)

        if verbose:
            logger.info("Book-Genre mapping inserted successfully.")

        # Close the connection
        sql_handler.close_cnxn(verbose=verbose)

    except psycopg2.Error as e:
        logger.error("Unable to connect to the PostgreSQL server.")