Hamana

One of the features of hamana library is the possibility to initiate a connection to a locally (or in memory) SQLite database that can be used to cetralize the data storage from various sources (by using the different connectors available in the library), and perform the analysis using the SQL language or pandas library.

To ensure consistency, it is possible to have only one single instance of the database that can be shared among different connectors. The database is created in memory by default, but it is possible to store it in a file by providing the path parameter when it is initiated.

Even if the internal database is of SQLite type, it was implemented a dedicated connector HamanaConnector to handle the single instance behavior, that inherits from the SQLiteConnector class. To work with the database, it is necessary first to create an instance of the HamanaConnector class or use the hamana.connect() method as a shortcut. Once the database is not needed anymore, it is possible to close the connection by calling the HamanaConnector.close() method or using the hamana.disconnect() shortcut.

import hamana as hm

# connect to internal database
hm.connect()

# perform operations with the database
# ...

# disconnect from the database
hm.disconnect()

The above example shows how to work with the internal database by using the corresponding shortcuts. The hamana.connect() method creates a new instance of the HamanaConnector class and stores it in the hamana module, while the hamana.disconnect() method closes the connection and removes the instance from the hamana module. The next example, is equivalent to the previous one, but it shows how to work with the HamanaConnector class directly.

from hamana.connector.db.hamana import HamanaConnector

# create a new instance of the HamanaConnector class
hamana_db = HamanaConnector()

# perform operations with the database
# ...

# disconnect from the database
hamana_db.close()

HamanaConnector

hamana.connector.db.hamana.HamanaConnector

HamanaConnector(path: str | Path = ':memory:')

Bases: SQLiteConnector

This class is responsible for handling the database connection of the library.
For each execution, only one instance of this class is allowed to manage a single database connection.

Parameters:

Name	Type	Description	Default
path	str | Path	path like string that define the SQLite database to load/create. By default the database is created in memory.	':memory:'

Example

from hamana.connector.db.hamana import HamanaConnector

# init internal database
hamana_db = HamanaConnector()

# doing operations..
query.to_sqlite(table_name = "t_oracle_output")

# close connection
hamana_db.close()

Source code in src/hamana/connector/db/hamana.py

def __init__(self, path: str | Path = ":memory:") -> None:
    path_str = str(path)
    super().__init__(path_str)
    self._connection = sqlite_connect(database = path_str)

get_instance classmethod

get_instance() -> HamanaConnector

Get the instance of the internal database connector.

Raises:

Type	Description
HamanaConnectorNotInitialised	If the database is not initialized.

Source code in src/hamana/connector/db/hamana.py

@classmethod
def get_instance(cls) -> "HamanaConnector":
    """
        Get the instance of the internal database connector.

        Raises:
            HamanaConnectorNotInitialised: If the database is not initialized.
    """
    logger.debug("start")
    _instance = cls._instances.get(cls)

    if _instance is None:
        logger.error("HamanaConnector is not initialized.")
        raise HamanaConnectorNotInitialised()

    logger.debug("end")
    return _instance

get_connection

get_connection() -> Connection

Get the database connection.

Raises:

Type	Description
HamanaConnectorNotInitialised	If the database is not initialized.

Source code in src/hamana/connector/db/hamana.py

def get_connection(self) -> Connection:
    """
        Get the database connection.

        Raises:
            HamanaConnectorNotInitialised: If the database is not initialized.
    """
    logger.debug("start")
    if self._connection is None:
        logger.error("Database connection is not initialized.")
        raise HamanaConnectorNotInitialised()
    logger.debug("end")
    return self._connection

close

close() -> None

Close the database connection.
Use this method at the end of the process to ensure that the database connection is properly closed. If the connection is already closed, this method will log a warning message.

Source code in src/hamana/connector/db/hamana.py

def close(self) -> None:
    """
        Close the database connection.  
        Use this method at the end of the process to ensure that the database 
        connection is properly closed. If the connection is already closed, 
        this method will log a warning message.
    """
    logger.debug("start")
    if self._connection is None:
        logger.warning("Database connection is already closed.")
    else:
        self._connection.close()
        self._connection = None

        # remove instance from singleton
        HamanaConnector._instances.pop(HamanaConnector)
    logger.debug("end")
    return

Shortcut Methods

hamana.connector.db.hamana.connect

connect(path: str | Path = ':memory:') -> None

Connect to the database using the path provided.
This function is a helper function to connect to the database without creating an instance of the HamanaConnector class.

Parameters:

Name	Type	Description	Default
path	str | Path	path like string that define the SQLite database to load/create. By default the database is created in memory.	':memory:'

Source code in src/hamana/connector/db/hamana.py

def connect(path: str | Path = ":memory:") -> None:
    """
        Connect to the database using the path provided.  
        This function is a helper function to connect to the database 
        without creating an instance of the `HamanaConnector` class.

        Parameters:
            path: path like string that define the SQLite database to load/create.  
                By default the database is created in memory.
    """
    logger.debug("start")

    # create connection
    HamanaConnector(path)
    logger.info(f"Connected to the database ({path}).")

    logger.debug("end")
    return

hamana.connector.db.hamana.execute

execute(query: str) -> Query

execute(query: Query) -> None

execute(query: Query | str) -> None | Query

Execute the query on the internal database.
This function is a helper function to execute a query on the hamana SQLite database.

Parameters:

Name	Type	Description	Default
query	Query | str	query to execute on database. The query could be a string or a Query object. If the query is a string, then the function automatically creates a Query object.	required

Returns:

Type	Description
None | Query	The result depends on the input provided. If query is a string, then the function automatically creates a Query object, executes the extraction and returns the Query object with the result. If query is a Query object, then the function performs the extraction and return None because the result is stored in the object itself.

Source code in src/hamana/connector/db/hamana.py

def execute(query: Query | str) -> None | Query:
    """
        Execute the query on the internal database.  
        This function is a helper function to execute a query on the `hamana` SQLite database.

        Parameters:
            query: query to execute on database. The query could be 
                a string or a `Query` object. If the query is a string, 
                then the function automatically creates a `Query` object.

        Returns:
            The result depends on the input provided. 
                If query is a string, then  the function automatically 
                creates a `Query` object, executes the extraction and 
                returns the `Query` object with the result. 
                If query is a `Query` object, then the function performs 
                the extraction and return None because the result is stored 
                in the object itself.
    """
    logger.debug("start")

    # execute query
    result = HamanaConnector.get_instance().execute(query)
    logger.info(f"Query executed successfully: {query}")

    logger.debug("end")
    return result

hamana.connector.db.hamana.disconnect

disconnect() -> None

Disconnect from the database.
This function is a helper function to disconnect from the database without using an instance of the HamanaConnector class.

Source code in src/hamana/connector/db/hamana.py

def disconnect() -> None:
    """
        Disconnect from the database.  
        This function is a helper function to disconnect from the database 
        without using an instance of the `HamanaConnector` class.
    """
    logger.debug("start")

    # close connection
    HamanaConnector.get_instance().close()
    logger.info("Disconnected from the database.")

    logger.debug("end")
    return