| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394 | # 导入:import uuidfrom datetime import datetimefrom urllib import parseimport yamlfrom sqlalchemy import Column, Integer, String, DateTime, Text, and_from sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy.orm import sessionmakerfrom config.url_and_db import ai_word_engine# 创建对象的基类:Base = declarative_base()passowrd = parse.quote_plus('hcst@2020')# 定义User对象:class Task(Base):    # 表的名字:    __tablename__ = 'tb_asr_result'    # 表的结构:    signature = Column(String(100))    task_id = Column(Integer, primary_key=True)    task_status = Column(Integer)    task_result = Column(Text)    word_text = Column(Text)    word_split = Column(Text)    word_text_md5 = Column(Text)    create_time = Column(DateTime, nullable=True, default=datetime.now)    update_time = Column(DateTime)# 初始化数据库连接:with open('/data/pythonProject/video_to_word/config/config.yaml', mode='r', encoding='utf-8') as f:    config = yaml.load(f.read(), Loader=yaml.FullLoader)# 创建DBSession类型:DBSession = sessionmaker(bind=ai_word_engine)def query(signature, task_status, task_id):    condition = (1 == 1)    if signature != None:        condition = and_(condition, Task.signature == signature)    if task_status != None:        condition = and_(condition, Task.task_status == task_status)    if task_id != None:        condition = and_(condition, Task.task_id == task_id)    # 创建Session:    session = DBSession()    # 创建Query查询,filter是where条件,最后调用one()返回唯一行,如果调用all()则返回所有行:    task = session.query(Task).filter(condition).order_by(Task.create_time.desc()).all()    # 关闭Session:    session.close()    return taskdef insert(task: Task):    # 创建session对象:    session = DBSession()    # 添加到session:    session.add(task)    session.flush()    # 提交即保存到数据库:    session.commit()    # 关闭session:    session.close()    return taskdef update(task):    # 创建session对象:    session = DBSession()    taskdb = session.query(Task).filter(Task.task_id == task.task_id).one()    taskdb.task_status = task.task_status    taskdb.word_text = task.word_text    taskdb.word_split = task.word_split    taskdb.word_text_md5 = task.word_text_md5    taskdb.task_result = task.task_result    # 提交即保存到数据库:    session.commit()    # 关闭session:    session.close()if __name__ == '__main__':    print('test query')    # task = Task(task_id = '3',task_status=1,signature='dx')    # task = insert(task)    task = query(None, None, 1)[0]    print(task.signature)    # update(task)
 |