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