# 导入: from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String,DateTime,Text,create_engine,and_ from datetime import datetime from urllib import parse # 创建对象的基类: Base = declarative_base() passowrd = parse.quote_plus('hcst@2021') # 定义User对象: class Task(Base): # 表的名字: __tablename__ = 'tb_swap_face_task' # 表的结构: id = Column(Integer, primary_key=True, autoincrement=True) input_video_url = Column(Text) input_image_url = Column(Text) output_video_url = Column(Text) start_time = Column(DateTime) finish_time = Column(DateTime) input_video_md5 = Column(String(100)) input_image_md5 = Column(String(100)) output_video_md5 = Column(String(100)) status = Column(String(100)) create_by = Column(String(100)) create_time = Column(DateTime, nullable=True, default=datetime.now) update_time = Column(DateTime) # 初始化数据库连接: engine = create_engine('mysql+mysqlconnector://data:'+passowrd+'@139.186.27.96:3390/db_ai_swap') # 创建DBSession类型: DBSession = sessionmaker(bind=engine) def query(id,video_md5,image_md5,create_by): condition = (1==1) if id != None: condition = and_(condition, Task.id == id) if video_md5 != None: condition = and_(condition, Task.input_video_md5 == video_md5) if image_md5 != None: condition = and_(condition, Task.input_image_md5 == image_md5) if create_by != None: condition = and_(condition, Task.create_by == create_by) # 创建Session: session = DBSession() # 创建Query查询,filter是where条件,最后调用one()返回唯一行,如果调用all()则返回所有行: task = session.query(Task).filter(condition).order_by(Task.create_time.desc()).all() # 关闭Session: print(task) session.close() return task def insert(task:Task): # 创建session对象: session = DBSession() # 添加到session: session.add(task) session.flush() id = task.id # 提交即保存到数据库: session.commit() task.id = id # 关闭session: session.close() return task def update(task): # 创建session对象: session = DBSession() taskdb = session.query(Task).filter(Task.id==task.id).one() taskdb.status = task.status taskdb.output_video_url = task.output_video_url taskdb.start_time = task.start_time taskdb.finish_time = task.finish_time taskdb.output_video_md5 = task.output_video_md5 # 提交即保存到数据库: session.commit() # 关闭session: session.close() if __name__ == '__main__': print('test query') task = Task(input_video_url = 'a',input_image_url='a',start_time=datetime.now(),status='test') task = insert(task) print(task.id) #query(1) #update(task)