- 偿试修复可执行文件打包插件数据表缺失问题

app/db/__init__.py

@@ -1,7 +1,10 @@
+from typing import Any, Self, List
 from typing import Tuple, Optional, Generator
 
 from sqlalchemy import create_engine, QueuePool
-from sqlalchemy.orm import sessionmaker, Session, scoped_session
+from sqlalchemy import inspect
+from sqlalchemy.orm import declared_attr
+from sqlalchemy.orm import sessionmaker, Session, scoped_session, as_declarative
 
 from app.core.config import settings
 
@@ -135,6 +138,52 @@ def db_query(func):
     return wrapper
 
 
+@as_declarative()
+class Base:
+    id: Any
+    __name__: str
+
+    @db_update
+    def create(self, db: Session):
+        db.add(self)
+
+    @classmethod
+    @db_query
+    def get(cls, db: Session, rid: int) -> Self:
+        return db.query(cls).filter(cls.id == rid).first()
+
+    @db_update
+    def update(self, db: Session, payload: dict):
+        payload = {k: v for k, v in payload.items() if v is not None}
+        for key, value in payload.items():
+            setattr(self, key, value)
+        if inspect(self).detached:
+            db.add(self)
+
+    @classmethod
+    @db_update
+    def delete(cls, db: Session, rid):
+        db.query(cls).filter(cls.id == rid).delete()
+
+    @classmethod
+    @db_update
+    def truncate(cls, db: Session):
+        db.query(cls).delete()
+
+    @classmethod
+    @db_query
+    def list(cls, db: Session) -> List[Self]:
+        result = db.query(cls).all()
+        return list(result)
+
+    def to_dict(self):
+        return {c.name: getattr(self, c.name, None) for c in self.__table__.columns}
+
+    @declared_attr
+    def __tablename__(self) -> str:
+        return self.__name__.lower()
+
+
 class DbOper:
     """
     数据库操作基类