123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101 |
- # 导入:
- 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)
|