# 导入: import uuid from datetime import datetime from urllib import parse import yaml from sqlalchemy import Column, Integer, String, DateTime, Text, and_ from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from common_func import get_db_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) # 1-1 数据库连接引擎,依据开发环境/生产环境 进行切换 mac = uuid.UUID(int=uuid.getnode()).hex[-12:] if mac in ['5254003fa716', '52540003f5dd']: engine = get_db_engine(config['ai_word_dev_db']) else: engine = get_db_engine(config['ai_word_product_db']) # 创建DBSession类型: DBSession = sessionmaker(bind=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 task def insert(task: Task): # 创建session对象: session = DBSession() # 添加到session: session.add(task) session.flush() # 提交即保存到数据库: session.commit() # 关闭session: session.close() return task def 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)