yixueServer/DBTools.py

from typing import Any

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

# test.db.cxhy.cn
# localhost
engine = create_engine('mysql+pymysql://yixue:Lai123@test.db.cxhy.cn/yixue', pool_pre_ping=True)

Base = declarative_base()


class DBBase(Base):
    __abstract__ = True

    def __repr__(self):
        result = self.__class__.__name__ + "("
        for attr in self.__dict__:
            if attr.startswith("_") or attr.startswith("__") or attr == "__tablename__":
                continue
            result += attr + "=" + str(getattr(self, attr)) + " "
        result += ")"
        return result

    def to_json_object(self, **kwargs: Any):
        obj = {}
        for attr in self.__dict__:
            if attr.startswith("_") or attr.startswith("__") or attr == "__tablename__":
                continue
            obj[attr] = getattr(self, attr)
        for key, value in kwargs.items():
            obj[key] = value
        return obj

    @classmethod
    def new_session(cls):
        Session = sessionmaker(bind=engine)
        session = Session()
        return session

    @classmethod
    def insert(cls, **kwargs: Any):
        ss = cls.new_session()
        obj = cls()
        for key in kwargs.keys():
            if hasattr(obj, key):
                setattr(obj, key, kwargs[key])
        ss.add(obj)
        ss.commit()
        ss.close()

    @classmethod
    def insert_all(cls, items: []):
        ss = cls.new_session()
        lst = []
        for item in items:
            obj = cls()
            for key in item.keys():
                if hasattr(obj, key):
                    setattr(obj, key, item[key])
            lst.append(obj)
        ss.add_all(lst)
        ss.commit()
        ss.close()

    @classmethod
    def delete_by(cls, **kwargs: Any):
        ss = cls.new_session()
        data = ss.query(cls).filter_by(**kwargs).all()
        for dt in data:
            ss.delete(dt)
        ss.commit()
        ss.close()

    @classmethod
    def query_by(cls, **kwargs: Any):
        ss = cls.new_session()
        data = ss.query(cls).filter_by(**kwargs).all()
        ss.close()
        return data

    @classmethod
    def query_first_by(cls, **kwargs: Any):
        ss = cls.new_session()
        data = ss.query(cls).filter_by(**kwargs).first()
        ss.close()
        return data

    # 也是查询，但是不关闭session，用于后续继续执行其他如update操作
    @classmethod
    def query_keep(cls, **kwargs: Any):
        ss = cls.new_session()
        data = ss.query(cls).filter_by(**kwargs).all()
        return [data, ss]

    @classmethod
    def check_exist(cls, **kwargs: Any):
        return cls.query_first_by(**kwargs) is not None


class DBTaiYangPianYi(DBBase):
    __tablename__ = 'taiyangpianyi'
    id = Column(Integer, primary_key=True)
    month = Column(Integer)
    day = Column(Integer)
    diff = Column(Integer)


class DBUserInfo(DBBase):
    __tablename__ = 'user_info'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    beizhu = Column(String)
    man = Column(Integer)
    leibie = Column(Integer)
    year = Column(Integer)
    month = Column(Integer)
    day = Column(Integer)
    hour = Column(Integer)
    minute = Column(Integer)
    sheng = Column(String)
    shi = Column(String)
    qu = Column(String)
    niangan = Column(String)
    nianzhi = Column(String)
    yuegan = Column(String)
    yuezhi = Column(String)
    rigan = Column(String)
    rizhi = Column(String)
    shigan = Column(String)
    shizhi = Column(String)
    customer = Column(String)
    joinTime = Column(String)
    enabled = Column(Integer)


class DBWanNianLi(DBBase):
    __tablename__ = 'wannianli'
    id = Column(Integer, primary_key=True)
    nian = Column(Integer)
    yue = Column(Integer)
    ri = Column(Integer)
    nian_gan = Column(String)
    nian_zhi = Column(String)
    yue_gan = Column(String)
    yue_zhi = Column(String)
    ri_gan = Column(String)
    ri_zhi = Column(String)


class DBShiShenJieDu(DBBase):
    __tablename__ = 'shishen_jiedu'
    id = Column(Integer, primary_key=True)
    arg1 = Column(String)
    arg2 = Column(String)
    guanxi = Column(String)
    tip1 = Column(String)
    tip2 = Column(String)
    tip3 = Column(String)
    tip4 = Column(String)
    tip5 = Column(String)


class DBLuckDay(DBBase):
    __tablename__ = 'luck_day'
    id = Column(Integer, primary_key=True)
    year = Column(Integer)
    month = Column(Integer)
    day = Column(Integer)
    desc = Column(String)
    jiedu1 = Column(String)
    jiedu2 = Column(String)
    jiedu3 = Column(String)
    jiedu4 = Column(String)
    jiedu5 = Column(String)
    tip = Column(String)


class DBGaoNeng(DBBase):
    __tablename__ = 'gaoneng_mid'
    id = Column(Integer, primary_key=True)
    arg1 = Column(String)
    arg2 = Column(String)
    arg3 = Column(String)
    wuxing1 = Column(String)
    wuxing2 = Column(String)
    wuxing3 = Column(String)
    guanxi = Column(String)
    desc = Column(String)
    father = Column(String)
    leibie = Column(String)
    hehua = Column(String)
    fangxiang = Column(String)
    tip = Column(String)


class DBCustomUser(DBBase):
    __tablename__ = 'custom_user'
    id = Column(Integer, primary_key=True)
    user = Column(String)
    psd = Column(String)
    name = Column(String)
    sexy = Column(Integer)
    online = Column(Integer)