多进程，多线程，使用sqlalchemy 创建引擎（单例模式），闭包装饰器承载数据库会话，装饰模型类的类方法

python 多进程，多线程，使用 sqlalchemy 对数据库进行操作

创建引擎 & 获取数据库会话：

使用类的方式，然后在对象方法中去创建数据库引擎（使用单例，确保只创建一个对象，方法里对引擎做判断，确保只创建一个数据库引擎）

# mysql全局基类方法
class MysqlGlobal(object):
    __instance = None
    __engine = None

    def __new__(cls, *args, **kwargs):
        if not cls.__instance:
            cls.__instance = object.__new__(cls)
        return cls.__instance

    def gen_engine(self):
        if not MysqlGlobal.__engine:
            engine = create_engine("mysql+{driver}://{username}:{password}@{server}/{database}?charset={charset}"
                                   .format(driver=MYSQL_DRIVER,
                                           username=MYSQL_USERNAME,
                                           password=MYSQL_PASSWORD,
                                           server=MYSQL_SERVER,
                                           database=DB_NAME,
                                           charset=DB_CHARSET),
                                   pool_size=100,
                                   max_overflow=100,
                                   # pool_recycle=7200,
                                   pool_recycle=2,
                                   echo=False)
            engine.execute("SET NAMES {charset};".format(charset=DB_CHARSET))
            MysqlGlobal.__engine = engine
        return MysqlGlobal.__engine

    @property
    def mysql_session(self):
        self.gen_engine()
        mysql_db = sessionmaker(bind=MysqlGlobal.__engine)
        return mysql_db()

数据表模型类

# 数据库orm映射绑定引擎
MapBase = declarative_base(bind=MysqlGlobal().gen_engine())


class WfCarInfo(MapBase):
    __tablename__ = "wf_carinfo"
    untreated_status = 0
    treated_status = 1

    id = Column(Integer, primary_key=True, nullable=True, autoincrement=True)
    status = Column(Integer, nullable=True, default=0)  # 接入状态，0：待处理，1：已处理
    # create_time = Column(DateTime, nullable=True, default=func.now())  # 记录的创建时间
    # update_time = Column(DateTime, nullable=True, default=func.now(), onupdate=func.now())  # 记录的更新时间
    。。。省略

数据库会话闭包装饰器

包装了数据库会话 session

def mysql_session(method):
    @functools.wraps(method)
    def wrapper(*args, **kwargs):
        session = MysqlGlobal().mysql_session
        return method(*args, session, **kwargs)
    return wrapper

　　

装饰器装饰模型类的类方法

这样在外面调用类方法进行数据库操作的时候，就不需要传数据库会话过来（session 参数）

@classmethod
@mysql_session
def query_all(cls, session):
    carinfo_obj_list = session.query(cls).all()
    info_list = [carinfo_obj.to_dict() for carinfo_obj in carinfo_obj_list]
    return info_list

@classmethod
@mysql_session
def query_all_by_status(cls, status, session):
    carinfo_obj_list = session.query(cls).filter(cls.status == status).all()
    info_list = [carinfo_obj.to_dict() for carinfo_obj in carinfo_obj_list]
    return info_list

@classmethod
@mysql_session
def query_all_by_index(cls, current_index, next_index, session):
    carinfo_obj_list = session.query(cls).filter(and_(cls.id > current_index, cls.id <= next_index)).all()
    info_list = [carinfo_obj.to_dict() for carinfo_obj in carinfo_obj_list]
    return info_list

@classmethod
@mysql_session
def insert_one(cls, row, session):
    if not row:
        return
    new_record = cls(**row)
    session.add(new_record)
    session.commit()
    return new_record

@classmethod
@mysql_session
def update_status(cls, record_id, status, session):
    session.query(cls).filter(cls.recordId == record_id).update({"status": status})
    session.commit()
    return 1

@classmethod
@mysql_session
def delete_treated_data(cls, session):
    # session.query(cls).filter(cls.status == cls.treated_status).delete()
    treated_obj_list = session.query(cls).filter(cls.status == cls.treated_status).all()
    count = len(treated_obj_list)
    [session.delete(treated_obj) for treated_obj in treated_obj_list]
    session.commit()
    return count

　　

附加（数据库初始化）

# 数据库初始化
def init_db_data():
    MapBase.metadata.create_all()
    logging.info("init mysql_db success")
    print("init mysql_db success")

　　

end ~