Python SQLiteDB Class with SQLAlchemy Integration: Easy Data Management for Your Projects with SQLite

SQLiteDB.py

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()


class SQLiteDB:
    def __init__(self, db_file):
        self.engine = create_engine(f'sqlite:///{db_file}')
        Base.metadata.create_all(self.engine)
        self.Session = sessionmaker(bind=self.engine)
        self.session_instance = self.Session()

    def add_item(self, item):
        with self.session_instance as s:
            s.add(item)
            s.commit()
            s.refresh(item)
            return item

    def update_item(self, item, update_dict):
        with self.session_instance as s:
            for key, value in update_dict.items():
                setattr(item, key, value)
            s.commit()
            s.refresh(item)
            return item

    def save_items(self, items):
        with self.session_instance as s:
            s.bulk_save_objects(items)
            s.flush()
            s.commit()
            return items

    def get_items(self, model):
        items = self.session_instance.query(model).all()
        self.session_instance.close()
        return items

    def get_items_by_field(self, model, relationship_id_dict: dict):
        items = self.session_instance.query(model).filter_by(**relationship_id_dict).all()
        return items

    def get_item_by_id(self, model, id):
        item = self.session_instance.get(model, id)
        self.session_instance.close()
        return item

    def get_item_by_field(self, model, field_dict):
        item = self.session_instance.query(model).filter_by(**field_dict).first()
        return item

    def delete_item_by_id(self, model, item_id):
        item = self.session_instance.get(model, item_id)
        if item is not None:
            self.session_instance.delete(item)
            self.session_instance.commit()
            self.session_instance.close()

    def get_uniques_by_field(self, field):
        return self.session_instance.query(field).distinct().all()


class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

    def __repr__(self):
        return f"<User(id='{self.id}', name='{self.name}', email='{self.email}')>"


if __name__ == '__main__':
    db = SQLiteDB('test.db')
    user1 = User(name='Alice', email='alice@example.com')
    user2 = User(name='Bob', email='bob@example.com')
    db.add_item(user1)
    db.add_item(user2)
    users = db.get_items(User)
    print("user list: ")
    for user in users:
        print(user)
    print("user with name Alice:")
    user = db.get_item_by_field(User, {"name": 'Alice'})
    print(user)
    db.update_item(user, {"name": 'Alice 2'})
    print("Updated user: ", user)
    print("Deleted user: ", user)
    db.delete_item_by_id(User, user.id)
    users = db.get_items(User)
    print("user list: ")
    for user in users:
        print(user)