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 ~