Connect to SQLite and cursor for different getting data functions-CodePudding

I have a SQLite database and I have multiple functions for multiple queries like below. My question is about connecting to the database and cursor for each function. Is my code true and efficient? Is there another way to do it?

def get_all_data(database_addr: str):
    conn = sqlite3.connect(database_addr)
    c = conn.cursor()
    c.execute(
        """
        SELECT
        a.data
        FROM data_table a
        """
    )
    df = pd.DataFrame(c.fetchall(), columns=["data"])
    return df


def get_last_record(database_addr: str):
    conn = sqlite3.connect(database_addr)
    c = conn.cursor()
    c.execute(
        """
        SELECT * 
        FROM data_table 
        ORDER BY data_table.id DESC 
        LIMIT 1
        """
    )
    df = pd.DataFrame(c.fetchall(), columns=["id"])
    return df


def clear_local_data(last_record: int, database_addr: str):
    conn = sqlite3.connect(database_addr)
    c = conn.cursor()
    c.execute(
        """
        DELETE 
        FROM data_table a 
        WHERE a.id<=?
        """,
        (last_record,),
    )
    conn.commit()

CodePudding user response：

You should not be connecting to the database for every time you query. If your program is a one off short running program, you can use a long lived connection, but if the connection dies then your program will stop working. If your program is a long running service, use pooling which will handle disconnects and other issues for you.

conn = sqlite3.connect(database_addr)

def get_all_data(database_addr: str):
    c = conn.cursor()
    c.execute(
        """
        SELECT
        a.data
        FROM data_table a
        """
    )
    df = pd.DataFrame(c.fetchall(), columns=["data"])
    return df


def get_last_record(database_addr: str):
    c = conn.cursor()
    c.execute(
        """
        SELECT * 
        FROM data_table 
        ORDER BY data_table.id DESC 
        LIMIT 1
        """
    )
    df = pd.DataFrame(c.fetchall(), columns=["id"])
    return df

import sqlalchemy.pool as pool
import sqlite3

mypool = pool.QueuePool(
    lambda: sqlite3.connect(database_addr),
    max_overflow=10,
    pool_size=5,
)

def get_all_data(database_addr: str):
    c = mypool.connect().cursor()
    c.execute(
        """
        SELECT
        a.data
        FROM data_table a
        """
    )
    df = pd.DataFrame(c.fetchall(), columns=["data"])
    return df


def get_last_record(database_addr: str):
    c = mypool.connect().cursor()
    c.execute(
        """
        SELECT * 
        FROM data_table 
        ORDER BY data_table.id DESC 
        LIMIT 1
        """
    )
    df = pd.DataFrame(c.fetchall(), columns=["id"])
    return df