Database CRUD Functions

Dropbase provides you with a helper package known as dropbase, which contains an array of useful utilities, including a database connection manager.

To import the database connector, import the connect module from dropbase.database as follows:

from dropbase.database import connect

Subsequently, you can connect to one of the sources you've specified in the .env file by name, as per the following example:

db = connect("mydb")

In the background, the dropbase package manages the connection and session for you. It extracts credentials saved in the .env file, establishes a connection, and returns a connection instance.

A few handy methods are provided, including query and execute:

Query

The query function enables you to query a database by passing a SQL statement directly.

res = db.query("SELECT * FROM my_table")

If successful, the query output is a list of dictionaries containing the results. This result object can be passed directly to a Pandas DataFrame as follows:

df = pd.DataFrame(res)

Here is a full example:

from workspace.app.page1 import State
from dropbase.database import connect
import pandas as pd


def function1(state: State) -> pd.DataFrame:
        try:
                # connect to database
            db = connect("mydb")
                # run query
            res = db.query("SELECT * FROM my_table")
                # pass response to dataframe
            df = pd.DataFrame(res)
            return df
        except SQLAlchemyError as e:
        context.widgets.widget1.message = str(e)
        context.widgets.widget1.message_type = "error"
                return context

If the query fails, the session is rolled back, and an error will be raised.

The query method is implemented as:

    def query(self, sql: str):
        try:
            result_proxy = self.session.execute(text(sql))
            result = [dict(row) for row in result_proxy.fetchall()]
            result_proxy.close()
            return result
        except SQLAlchemyError as e:
            self.session.rollback()  # Rollback the session on error.
            raise e  # Propagate the error.

Execute

Execute is another built-in helper method that permits you to perform updates to your database. Below is an example of its usage:

res = db.execute(f"UPDATE my_table SET status = '{value}' WHERE id = {id}")

The method returns the number of affected rows in case of success and raises an exception in case of errors.

Here is a sample usage of execute:

from workspace.app.page1 import Context, State
from dropbase.database import connect


def function2(state: State, context: Context) -> Context:
        try:
                # get values from state to be used in update query
            value = state.widgets.widget1.input1
            id = state.tables.table1.id
                # connect to database
                db = connect("mydb")
                # run execute
            res = db.execute(f"update my_table set status = '{value}' where id = {id}")
          context.widgets.widget1.message = "Status has been updated"
          context.tables.table1.reload = True
    except SQLAlchemyError as e:
        context.widgets.widget1.message = str(e)
        context.widgets.widget1.message_type = "error"
    return context

The execute method is implemented as:

    def execute(self, sql: str):
        try:
            result = self.session.execute(text(sql))
            self.session.commit()
            return result.rowcount
        except SQLAlchemyError as e:
            self.session.rollback()  # Roll back the session on error.
            raise e  # Propagate the error.