🚀 DBFlux: Lightweight Database Management Library

🛠️ Version 1.0.2

🌟 Introduction

DBFlux is a lightweight, easy-to-use library built on top of SQLAlchemy to simplify database operations in Python.

It provides a streamlined interface for connecting to databases, managing sessions, and performing CRUD operations with minimal effort.

---

✨ Features

• 🔁 Automatic Transaction Management
• 🛠️ Session Handling
• 🔗 Flexibility – Supports multiple database engines via SQLAlchemy
• ⚡ Lightweight & Efficient
• 🔍 Advanced Filtering
• 📥 Data Insertion
• ✏️ Data Modification
• 📄 Easy Pagination
• 🛡️ Safe Deletion
• 📦 Consistent Output Handling

---

📚 Requirements

• Python 3.8+
• SQLAlchemy >= 2.0

---

🔧 Installation

Install dbflux via pip:

pip install dbflux

Or install from source:

git clone https://github.com/abbas-bachari/dbflux.git
cd dbflux
pip install .

---

💡 Quick Start

from dbflux  import Sqlite,DBModel
from sqlalchemy import Column, Integer, String, Float
from sqlalchemy.orm import declarative_base
from time import time

Base=declarative_base()
db = Sqlite(db_name="example.db")


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))
    age = Column(Integer)
    def __repr__(self):
        return f"User(id={self.id}, name={self.name}, email={self.email}, age={self.age})"

class Order(Base):
    __tablename__ = "orders"
    order_id = Column(Integer, primary_key=True)
    product = Column(String, nullable=False)
    price = Column(Float, nullable=False)
    time = Column(Integer, nullable=False)
    
    def to_dict(self):
        return {
            "order_id": self.order_id,
            "product": self.product,
            "price": self.price,
            "time": self.time
        }
    
    def __str__(self):
        return json.dumps(self.to_dict(), indent=4,ensure_ascii=False)

    def __repr__(self):
        return f"Order(order_id={self.order_id}, product={self.product}, price={self.price}, time={self.time})"


db.create_tables(Base)


users=DBModel(User,db)
orders=DBModel(Order,db)


users_data=[
    {"id": 1, "name": "Alice", "email": "alice@test.com","age":22},
    {"id": 2, "name": "Bob", "email": "bob@test.com","age":21},
    {"id": 3, "name": "Carol", "email": "carol@test.com","age":18}
]

orders_data=[
    {"order_id": 1, "product": "Product A", "price": 100, "time": time()},
    {"order_id": 2, "product": "Product B", "price": 200, "time": time()},
    {"order_id": 3, "product": "Product C", "price": 300, "time": time()}
]

users.insert(users_data)
orders.insert(orders_data)

---

💡 Examples Usage DBFactory

Base = declarative_base()

class Order(Base):
    __tablename__ = "orders"

    order_id = Column(Integer, primary_key=True)
    product = Column(String, nullable=False)
    price = Column(Float, nullable=False)
    time = Column(Integer, nullable=False)
    
    def to_dict(self):
        return {
            "order_id": self.order_id,
            "product": self.product,
            "price": self.price,
            "time": self.time
        }
    
    def __str__(self):
        return json.dumps(self.to_dict(), indent=4,ensure_ascii=False)

    def __repr__(self):
        return f"Order(order_id={self.order_id}, product={self.product}, price={self.price}, time={self.time})"
    


factory = DBFactory(db_name="data.db")

db = factory.create("sqlite")

db.create_tables(Base)

orders_db = DBModel(Order ,db)


order = Order(order_id=1, product="Product A", price=100, time=time())

orders_db.insert( order)

orders:list[Order] = orders_db.get(limit=1)

print(orders)

>>> [Order(order_id=1, product=Product A, price=100.0, time=1755924289.1132557)]

print(orders[0])

>>> {
    "order_id": 1,
    "product": "Product A",
    "price": 100.0,
    "time": 1755924289.1132557
    }

---

🔹 Supported Database Types

Type	Aliases
SQLite	sqlite
MySQL	mysql
PostgreSQL	postgres, postgresql
MariaDB	mariadb
Oracle	oracle
DB2	db2, ibmdb2
Firebird	firebird
MSSQL	mssql, sqlserver

---

🔹 Examples for Different Databases

from dbflux.databases import Sqlite, MySQL, PostgreSQL

# Example 1: SQLite
sqlite_db = Sqlite(db_name="data.db")
sqlite_db.create_tables(Base)
sqlite_db.insert(model_class= Order ,data=Order(order_id=10, product="SQLite Product", price=50, time=time()))

# Example 2: MySQL
mysql_db = MySQL(db_name="test_db",username="root", password="password", host="localhost", )
mysql_db.create_tables(Base)
mysql_db.insert(model_class= Order ,data=Order(order_id=11, product="MySQL Product", price=60, time=time()))

# Example 3: PostgreSQL
postgres_db = PostgreSQL(db_name="test_db",username="postgres", password="secret", host="localhost")
postgres_db.create_tables(Base)
postgres_db.insert(model_class= Order ,data=Order(order_id=12, product="PostgreSQL Product", price=70, time=time()))

---

🎯 Summary of Features

✅ CRUD Operations

✅ Bulk Insert & Bulk Update

✅ Advanced Filtering (OR/AND/Range)

✅ Pagination

✅ JSON Output

✅ Transaction Safety

✅ Direct SQLAlchemy Access via BaseDB

---

📖 Documentation

For more details, visit the official SQLAlchemy documentation.

---

📜 License

This project is licensed under the MIT License.

---

👤 Publisher / ناشر

Abbas Bachari / عباس بچاری

---

💖 Sponsor

Support development by sponsoring on Github Sponsors.