database.py 2.7 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586
  1. # 导入:
  2. from sqlalchemy.orm import sessionmaker
  3. from sqlalchemy.ext.declarative import declarative_base
  4. from sqlalchemy import Column, Integer, String,DateTime,Text,create_engine,and_
  5. from datetime import datetime
  6. from urllib import parse
  7. # 创建对象的基类:
  8. Base = declarative_base()
  9. passowrd = parse.quote_plus('hcst@2021')
  10. # 定义User对象:
  11. class Task(Base):
  12. # 表的名字:
  13. __tablename__ = 'tb_swap_face_task'
  14. # 表的结构:
  15. id = Column(Integer, primary_key=True, autoincrement=True)
  16. input_video_url = Column(Text)
  17. input_image_url = Column(Text)
  18. output_video_url = Column(Text)
  19. start_time = Column(DateTime)
  20. finish_time = Column(DateTime)
  21. input_video_md5 = Column(String(100))
  22. input_image_md5 = Column(String(100))
  23. status = Column(String(100))
  24. create_by = Column(String(100))
  25. create_time = Column(DateTime, nullable=True, default=datetime.now)
  26. update_time = Column(DateTime)
  27. # 初始化数据库连接:
  28. engine = create_engine('mysql+mysqlconnector://data:'+passowrd+'@139.186.27.96:4000/db_ai_swap')
  29. # 创建DBSession类型:
  30. DBSession = sessionmaker(bind=engine)
  31. def query(id,video_md5,image_md5,create_by):
  32. condition = (1==1)
  33. if id != None:
  34. condition = and_(condition, Task.id == id)
  35. if video_md5 != None:
  36. condition = and_(condition, Task.input_video_md5 == video_md5)
  37. if image_md5 != None:
  38. condition = and_(condition, Task.input_image_md5 == image_md5)
  39. if create_by != None:
  40. condition = and_(condition, Task.create_by == create_by)
  41. # 创建Session:
  42. session = DBSession()
  43. # 创建Query查询,filter是where条件,最后调用one()返回唯一行,如果调用all()则返回所有行:
  44. task = session.query(Task).filter(condition).order_by(Task.create_time.desc()).all()
  45. # 关闭Session:
  46. session.close()
  47. return task
  48. def insert(task:Task):
  49. # 创建session对象:
  50. session = DBSession()
  51. # 添加到session:
  52. session.add(task)
  53. session.flush()
  54. id = task.id
  55. # 提交即保存到数据库:
  56. session.commit()
  57. task.id = id
  58. # 关闭session:
  59. session.close()
  60. return task
  61. def update(task):
  62. # 创建session对象:
  63. session = DBSession()
  64. taskdb = session.query(Task).filter(Task.id==task.id).one()
  65. taskdb.status = task.status
  66. taskdb.output_video_url = task.output_video_url
  67. taskdb.start_time = task.start_time
  68. taskdb.finish_time = task.finish_time
  69. # 提交即保存到数据库:
  70. session.commit()
  71. # 关闭session:
  72. session.close()
  73. if __name__ == '__main__':
  74. print('test query')
  75. task = Task(input_video_url = 'a',input_image_url='a',start_time=datetime.now(),status='test')
  76. task = insert(task)
  77. print(task.id)
  78. #query(1)
  79. #update(task)