python orm 框架中sqlalchemy用法实例详解

 """ Created on 19-10-22 @author: apple @description:建表 """ import pymysql server = '127.0.0.1' user = 'root' # dev password = '123456' conn = pymysql.connect(server, user, password, database='DataSave') # 获取连接 cursor = conn.cursor() # 获取游标 # "**ENGINE=InnoDB DEFAULT CHARSET=utf8**"-创建表的过程中增加这条，中文就不是乱码 # 创建表 cursor.execute (""" CREATE TABLE if not exists lamp_result( result_id INT NOT NULL auto_increment primary key, product_number VARCHAR(100), record_time VARCHAR(100), lamp_color INT NOT NULL, detect_result VARCHAR(100), old_pic_path VARCHAR(100), result_pic_path VARCHAR(100) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) # 查询数据 cursor.execute('SELECT * FROM lamp_result') row = cursor.fetchone() print(row) # cursor.execute("INSERT INTO user VALUES('%d', '%s','%s','%s','%s')" % ('xiaoming','qwe','ming','@163.com')) # 提交数据，才会写入表格 conn.commit() # 关闭游标关闭数据库 cursor.close() conn.close() 

 """ Created on 19-10-22 @author: apple @requirement:Anaconda 4.3.0 (64-bit) Python3.6 @description:数据存储 """ from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, String, Integer, create_engine from sqlalchemy.orm import sessionmaker Base = declarative_base() # 连接数据库 # alter table students convert to character set utf8; conn = "mysql+pymysql://root:password@0.0.0.0:3306/DataSave" engine = create_engine(conn, encoding='UTF8', echo=False) # echo=True 打印日志 # 创建session对象 Session = sessionmaker(bind=engine) session = Session() # 数据库表模型ORM class DataSaveSystem(Base): """ 员工自助信息采集系统 """ __tablename__ = 'lamp_result' # 定义表名 # 定义列名 result_id = Column(Integer, primary_key=True, autoincrement=True, nullable=False) product_number = Column(String(50), nullable=True) record_time = Column(String(50), nullable=False) lamp_color = Column(Integer, nullable=False) detect_result = Column(String(100), nullable=False) old_pic_path = Column(String(100), nullable=False) result_pic_path = Column(String(100), nullable=False) def __repr__(self): """ 引用该类别，输出结果 :return: """ return str(self.__dict__) # return ''.format(self.detect_result) # 插入数据 def insert_to_db(product_number=None, record_time=None, lamp_color=None, detect_result=None, old_pic_path=None, result_pic_path=None): ''' :param product_number: 产品编号 :param record_time: 取原图时间 :param lamp_color: 灯的颜色:1 2 3 4 :param detect_result: 检测结果 :param old_pic_path: 原图路径 :param result_pic_path: 结果图路径 :return: 数据是否写入成功 ''' information_system_instance = DataSaveSystem( product_number=product_number, record_time=record_time, lamp_color=lamp_color, detect_result=detect_result, old_pic_path=old_pic_path, result_pic_path=result_pic_path) # session.add_all([ #   lamp_result(id=2, name="张2", age=19), #   lamp_result(id=3, name="张3", age=20) # ]) session.add(information_system_instance) try: session.commit() # 尝试提交数据库事务 # print('数据库数据提交成功') return { "code": 200, "status": True, "message": "写入数据库成功", } except SQLAlchemyError as e: session.rollback() print(e) return { "code": 500, "status": False, "message": str(e) } # url = "mysql+pymysql://root:password@0.0.0.1:3306/DataSave" # # echo为True时，打印sql,可用于调试 # engine = create_engine(url, echo=False, encoding='utf-8', pool_size=5) # sessionClass = sessionmaker(bind=engine) # # 创建会话 # session = sessionClass() # # 查所有,并排序 # stuList = session.query(DataSaveSystem).order_by(DataSaveSystem.result_id).all() # print(stuList) # stu = DataSaveSystem(product_number='id1', record_time='20191022170400', lamp_color='1', detect_result='ok', old_pic_path='picture/', result_pic_path='d') # session.add(stu) stuList = [DataSaveSystem(product_number='id1', record_time='20191022170400', lamp_color='1', detect_result='ok', old_pic_path='picture/', result_pic_path='d'), DataSaveSystem(product_number='id1', record_time='20191022170400', lamp_color='1', detect_result='ok', old_pic_path='picture/', result_pic_path='d')] # session.add_all(stuList) # session.commit() # print('数据成功') if __name__ == '__main__': result = insert_to_db(stu) print(result) 

 """ Created on 19-10-31 @author: apple @requirement:Anaconda 4.3.0 (64-bit) Python3.6 @description:调取函数基类 """ from data_sql.airconditioning_lamp_datasave.datasave import DataSaveSystem from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, String, Integer, create_engine from sqlalchemy.orm import sessionmaker Base = declarative_base() # 连接数据库 # alter table students convert to character set utf8; conn = "mysql+pymysql://root:password@0.0.0.1:3306/DataSave" engine = create_engine(conn, encoding='UTF8', echo=False) # echo=True 打印日志 # 创建session对象 Session = sessionmaker(bind=engine) session = Session() stuList = [DataSaveSystem(product_number='id1', record_time='20191022170400', lamp_color='1', detect_result='ok', old_pic_path='picture/', result_pic_path='F'), DataSaveSystem(product_number='id1', record_time='20191022170400', lamp_color='1', detect_result='ok', old_pic_path='picture/', result_pic_path='F'),DataSaveSystem(product_number='id1', record_time='20191022170400', lamp_color='1', detect_result='ok', old_pic_path='picture/', result_pic_path='F'),DataSaveSystem(product_number='id1', record_time='20191022170400', lamp_color='1', detect_result='ok', old_pic_path='picture/', result_pic_path='F')] session.add_all(stuList) session.commit() print('数据成功') # # 根据主建查询数据 # result = session.query(DataSaveSystem).get(3) # print(result.old_pic_path) # # 查询第一条 # result = session.query(DataSaveSystem).first() # print(result) #打印对象属性 # 查询表关键字的数据 result = session.query(DataSaveSystem).filter_by(result_pic_path='a/').first() print(result) #修改 session.query(DataSaveSystem).filter(DataSaveSystem.result_pic_path=='a/').update({"detect_result":"不合格"}) session.commit()