ds_sql_read_only.py

"""
Database Utility Module
=======================

This module provides a lightweight database configuration and query execution setup 
using SQLAlchemy and Pandas. It facilitates database interactions with SQLite 
and offers utilities for managing SQL queries in a structured manner.

Features:
---------
- `DatabaseConfig` class:
    Handles database connection setup. Provides a decorator to manage database
    connections for query functions. Stores and formats SQL queries for readability.
    Converts query results into Pandas DataFrames.
- `select` function:
    A decorator-based SQL SELECT function to execute read-only queries.
    Supports dynamic column selection and query clauses.

Dependencies:
-------------
- pandas
- sqlalchemy

Author: Eugen Hoppe (CC0)
Date: 2025-02-20
"""
from functools import wraps

import pandas as pd

from sqlalchemy import create_engine, text
from sqlalchemy.engine import Result
from sqlalchemy.engine.base import Connection


DB_PATH = "path_to.db"


# Database Configuration
# ======================
class DatabaseConfig:
    def __init__(self, path: str):
        """src: https://gist.github.com/eugen-hoppe/485201df52b69397bf44ecee2d0effc8"""
        self.path = path
        self.cache_sql: list = None

    def __create_engine__(self):
        return create_engine(f"sqlite:///{self.path}", echo=False)

    def db_connection(self, func):
        engine = self.__create_engine__()
        @wraps(func)
        def wrapper(*args, **kwargs):
            with engine.connect() as connection:
                try:
                    return func(connection, *args, **kwargs)
                except Exception as e:
                    raise e
        return wrapper
    
    def print_statement(self) -> None:
        print("\n" + " ".join(self.cache_sql) + ";" + "\n")

    def print_statement_pretty(self, *br, sep: str = "\n    ", **shift) -> None:
        sql_ = sep.join(self.cache_sql).replace(f"    FROM{sep}", "FROM ")
        for br_id, (keyword, shift) in enumerate(shift.items()):
            kw_ = keyword.upper().replace("_", " ")
            sql_ = sql_.replace(kw_, br[br_id] + shift * " " + kw_)
        print("\n" + sql_ + ";" + "\n")

    @staticmethod
    def get_df(io: Result) -> pd.DataFrame:
        return pd.DataFrame([tuple(r) for r in io.fetchall()], columns=list(io.keys()))


# Init Database Configuration
# ===========================
sql = DatabaseConfig(path=DB_PATH)


# Decorate Your Read Only Main Query Function for Your Data Science Project
# =========================================================================
@sql.db_connection
def select(conn: Connection, *columns, from_: str = "", **kw) -> pd.DataFrame:
    """src: https://gist.github.com/eugen-hoppe/485201df52b69397bf44ecee2d0effc8"""
    c_ = [col + ", " for col in columns][:-1] + [columns[-1]] if len(columns) else ["*"]
    kw_fx = lambda x: (
        (str(x).upper().replace("_", " ") + " ").replace("  ", " ").replace("  ", " ")
    )
    sql.cache_sql = ["SELECT", *c_, "FROM", from_, *[kw_fx(k) + str(kw[k]) for k in kw]]
    return DatabaseConfig.get_df(conn.execute(text(" ".join(sql.cache_sql) + ";")))


if __name__ == "__main__":
    # Get Data Frame from DB Query
    # ============================
    df = select(from_="users")

    # Print Last SQL Statement
    # ========================
    sql.print_statement()