利用SQLAlchemy和反射實現DAO的基類

阿新 • • 發佈：2020-11-04

在使用SQLAlchemy時，要根據每一個類的情況寫對應增刪改查，習慣了JAVA的虛類、反射，抽象出daobase類，如果python也有就會少很多重複程式碼，於是有了下面的簡單實現。

注：我的Entity都是最簡單的沒有其他方法，只有屬性，如以下例子：

Base = declarative_base()
class StudentBasicInfo(Base):
    """
    學生基本資訊類
    表名： studentbasic
    用於記錄學生最基本的資訊，在匯入時清除以前資料，用新的資料替換。
    """
    __tablename__ = 'studentbasic 
'
    """表名"""

    id = Column(Integer, primary_key=True)
    studentName = Column(String(10))
    """學生姓名，限10個字串"""
    cardId = Column(String(18))
    """身份證號，限18個字串"""
    sYear = Column(Integer)
    """入學年度，純數字"""
    sClass = Column(Integer)
    """班級，純數字"""
    sex = Column(String(2))
    """性別，字串型別 
"""
    studyNo = Column(String(25))
    """學號"""

StudentEntity

下面是DbFactory用於連線資料庫（本例為Sqlite3資料庫），生成Session：

class DbFactory(object):
    """資料庫連線工廠類"""

    def __init__(self):
        """ 初始化，類私有變數session和engine初始化為NONE，後呼叫init_db方法。 """
        self.session = None
        self.engine = None
        self.init_db()

     
def init_db(self, path="data.db"):
        """設定engine
        :param path 字串，設定資料檔案位置，預設tools目錄下的data.db
        :return 無
        """
        self.engine = create_engine('sqlite:///' + path, echo=True)

    def get_session(self):
        """獲得session，如果self.engine沒有初始化，呼叫init_db方法，如果已經初始化檢查self.session是否有值，沒有獲取，有則返回
        :return: Session()例項，每個DbFactory的引用保持一個Session()例項。
        """
        if self.engine is None:
            self.init_db()
        Session = sessionmaker(bind=self.engine)
        if self.session is None:
            self.session = Session()
        return self.session

    def create_db(self):
        """ 初始化資料庫表，用於正式使用軟體以前初始化資料庫。
        :return 無
        """
        Base.metadata.create_all(self.engine)

DbFactory

最後是DaoBase實現：

class DaoBase(object, metaclass=abc.ABCMeta):
    """MVC中M層基類，在類中描述公共抽象方法"""

    def __init__(self):
        """初始化，引入DbFactory例項，並初始備用
        :return 無
        """
        self.dbfactory = DbFactory()

    @abc.abstractmethod
    def get_entityclass(self):
        """抽象方法，返回DAO維護的實體類，用於查詢語句
        :return 類.class
        """
        pass

    def find_by_id(self, ids):
        """用ID查詢
        :param ids 整數型
        :return 類例項
        """
        return self.dbfactory.get_session().query(self.get_entityclass()).filter(self.get_entityclass().id == ids).one()

    def create(self, obj):
        """存貯類例項至資料庫
        :param obj 具體類例項
        :return 無
        """
        self.dbfactory.get_session().add(obj)
        self.dbfactory.get_session().commit()

    def update(self, obj):
        """
        更新資料，根據obj的ID查詢資料庫中的資料
        如果obj是利用session查詢出來的，且和update使用同一個session，則沒有變化中間的判斷不會執行，這時已經修改過session中的資料直接commit
        如果obj只是保留id，自行定義的一個例項而非在同一個session中查詢出來，則遍歷例項中的所有屬性（屬性非以_開頭）迴圈中會產生不一致的資料，
        利用setattr進行修改並更新，同樣呼叫commit最終提交，所以下面這個commit必須執行。
        :param obj 修改後的例項
        :raise EntryNotFoundException 查詢不到例項
        :raise ObjectNotMatchException obj與資料庫查詢的類不是同一種類
        :return 無
        """
        if isinstance(obj, self.get_entityclass()):
            try:
                obj_saved = self.find_by_id(obj.id)
                if obj_saved is None:
                    raise EntryNotFoundException()
                for attribute_name in dir(obj_saved):
                    if attribute_name[0:1] != "_" and \
                            getattr(obj, attribute_name) != getattr(obj_saved, attribute_name):
                        setattr(obj_saved, attribute_name, getattr(obj, attribute_name))
                self.__commit()
            except Exception as e:
                raise e
        else:
            raise ObjectNotMatchException()

    def __commit(self):
        self.dbfactory.get_session().commit()

    @abc.abstractmethod
    def delete(self, obj):
        pass