2022年 11月 7日

Python 操作 MySQL 数据库的三个模块

python使用MySQL主要有两个模块，pymysql（MySQLdb）和SQLAchemy。

  • pymysql（MySQLdb）为原生模块，直接执行sql语句，其中pymysql模块支持python 2和python3，MySQLdb只支持python2，两者使用起来几乎一样。

  • SQLAchemy为一个ORM框架，将数据对象转换成SQL，然后使用数据API执行SQL并获取执行结果

  • 另外DBUtils模块提供了一个数据库连接池，方便多线程场景中python操作数据库。

1.pymysql模块

安装：pip install pymysql

创建表格操作(注意中文格式设置)

  1. #coding:utf-8
  2. import pymysql
  3.  
  4. #关于中文问题
  5. #1. mysql命令行创建数据库，设置编码为gbk：create databse demo2 character set utf8; 
  6. #2. python代码中连接时设置charset="gbk"
  7. #3. 创建表格时设置default charset=utf8
  8.  
  9. #连接数据库
  10. conn = pymysql.connect(host="localhost", user="root", passwd="", db='learningsql', charset='utf8', port=3306)  #和mysql服务端设置格式一样（还可设置为gbk, gb2312）
  11. #创建游标
  12. cursor = conn.cursor()
  13. #执行sql语句
  14. cursor.execute("""create table if not exists t_sales(
  15.                 id int primary key auto_increment not null,
  16.                  nickName varchar(128) not null,
  17.                  color varchar(128) not null,
  18.                   size varchar(128) not null, 
  19.                   comment text not null,
  20.                   saledate varchar(128) not null)engine=InnoDB default charset=utf8;""")
  21.                   
  22. # cursor.execute("""insert into t_sales(nickName,color,size,comment,saledate) 
  23.                 # values('%s','%s','%s','%s','%s');""" % ("zack", "黑色", "L", "大小合适", "2019-04-20"))
  24.                 
  25. cursor.execute("""insert into t_sales(nickName,color,size,comment,saledate) 
  26.                 values(%s,%s,%s,%s,%s);""" , ("zack", "黑色", "L", "大小合适", "2019-04-20"))
  27. #提交
  28. conn.commit()
  29. #关闭游标
  30. cursor.close()
  31. #关闭连接
  32. conn.close()

增删改查：

注意execute执行sql语句参数的两种情况：

  • execute("insert into t_sales(nickName, size) values('%s','%s');" % ("zack","L") )  #此时的%s为字符窜拼接占位符，需要引号加’%s’  （有sql注入风险）

  • execute("insert into t_sales(nickName, size) values(%s,%s);" , ("zack","L") ) #此时的%s为sql语句占位符，不需要引号%s

  1. #***************************增删改查******************************************************
  2. conn = pymysql.connect(host="localhost", user="root", passwd="", db='learningsql', charset='utf8', port=3306)  #和mysql服务端设置格式一样（还可设置为gbk, gb2312）
  3. #创建游标
  4. cursor = conn.cursor()
  5.  
  6. insert_sql = "insert into t_sales(nickName,color,size,comment,saledate) values(%s,%s,%s,%s,%s);"
  7. #返回受影响的行数
  8. row1 = cursor.execute(insert_sql,("Bob", "黑色", "XL", "便宜实惠", "2019-04-20"))
  9.  
  10. update_sql = "update t_sales set color='白色' where id=%s;"
  11. #返回受影响的行数
  12. row2 = cursor.execute(update_sql,(1,))
  13.  
  14. select_sql = "select * from t_sales where id>%s;"
  15. #返回受影响的行数
  16. row3 = cursor.execute(select_sql,(1,))
  17.  
  18. delete_sql = "delete from t_sales where id=%s;"
  19. #返回受影响的行数
  20. row4 = cursor.execute(delete_sql,(4,))
  21.  
  22. #提交，不然无法保存新建或者修改的数据(增删改得提交)
  23. conn.commit()
  24. cursor.close()
  25. conn.close()

批量插入和自增id

  1. #***************************批量插入******************************************************
  2. conn = pymysql.connect(host="localhost", user="root", passwd="", db='learningsql', charset='utf8', port=3306)  #和mysql服务端设置格式一样（还可设置为gbk, gb2312）
  3. #创建游标
  4. cursor = conn.cursor()
  5.  
  6. insert_sql = "insert into t_sales(nickName,color,size,comment,saledate) values(%s,%s,%s,%s,%s);"
  7. data = [("Bob", "黑色", "XL", "便宜实惠", "2019-04-20"),("Ted", "黄色", "M", "便宜实惠", "2019-04-20"),("Gary", "黑色", "M", "穿着舒服", "2019-04-20")]
  8. row1 = cursor.executemany(insert_sql, data)
  9.  
  10. conn.commit()
  11. #为插入的第一条数据的id，即插入的为5,6,7，new_id=5
  12. new_id = cursor.lastrowid
  13. print(new_id)
  14. cursor.close()
  15. conn.close()

获取查询数据

  1. #***************************获取查找sql的查询数据******************************************************
  2. conn = pymysql.connect(host="localhost", user="root", passwd="", db='learningsql', charset='utf8', port=3306)  #和mysql服务端设置格式一样（还可设置为gbk, gb2312）
  3. #创建游标
  4. cursor = conn.cursor()
  5.  
  6. select_sql = "select id,nickname,size from t_sales where id>%s;"
  7. cursor.execute(select_sql, (3,))
  8.  
  9. row1 = cursor.fetchone()      #获取第一条数据，获取后游标会向下移动一行
  10. row_n = cursor.fetchmany(3)  #获取前n条数据，获取后游标会向下移动n行
  11. row_all = cursor.fetchall()  #获取所有数据，获取后游标会向下移动到末尾
  12. print(row1)
  13. print(row_n)
  14. print(row_all)
  15. #conn.commit()
  16. cursor.close()
  17. conn.close()

注：在fetch数据时按照顺序进行，可以使用cursor.scroll(num,mode)来移动游标位置，如：

  • cursor.scroll(1,mode=’relative’)  # 相对当前位置移动

  • cursor.scroll(2,mode=’absolute’) # 相对绝对位置移动

fetch获取数据类型

fetch获取的数据默认为元组格式，还可以获取字典类型的，如下：

  1. #***************************获取字典格式数据******************************************************
  2. conn = pymysql.connect(host="localhost", user="root", passwd="", db='learningsql', charset='utf8', port=3306)  #和mysql服务端设置格式一样（还可设置为gbk, gb2312）
  3. #创建游标
  4. cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
  5.  
  6. select_sql = "select id,nickname,size from t_sales where id>%s;"
  7. cursor.execute(select_sql, (3,))
  8.  
  9. row1 = cursor.fetchall()      
  10.  
  11. print(row1)
  12.  
  13. conn.commit()
  14. cursor.close()
  15. conn.close()

2.SQLAlchmy框架

SQLAlchemy的整体架构如下，建立在第三方的DB API上，将类和对象操作转换为数据库sql，然后利用DB API执sql语句得到结果。其适用于多种数据库。另外其内部实现了数据库连接池，方便进行多线程操作。

  • Engine，框架的引擎

  • Connection Pooling ，数据库连接池

  • Dialect(http://docs.sqlalchemy.org/en/latest/dialects/index.html)，选择连接数据库的DB API种类，（pymysql，mysqldb等）“

  • Schema/Types，架构和类型

  • SQL Exprression Language，SQL表达式语言

  • DB API(https://www.python.org/dev/peps/pep-0249/)：Python Database API Specification

2.1 执行原生sql

安装：pip install sqlalchemy

SQLAlchmy也可以不利用ORM，使用数据库连接池，类似pymysql模块执行原生sql

  1. #coding:utf-8
  2.  
  3. from sqlalchemy import create_engine
  4. from sqlalchemy.ext.declarative import declarative_base
  5. from sqlalchemy import Column, String, Integer
  6. import threading
  7.  
  8. engine = create_engine(
  9.             "mysql+pymysql://root@127.0.0.1:3306/learningsql?charset=utf8",
  10.             max_overflow = 0,  #超过连接池大小外最多创建的连接，为0表示超过5个连接后，其他连接请求会阻塞 （默认为10）
  11.             pool_size = 5,    #连接池大小（默认为5）
  12.             pool_timeout = 30,  #连接线程池中，没有连接时最多等待的时间，不设置无连接时直接报错 （默认为30）
  13.             pool_recycle = -1)  #多久之后对线程池中的线程进行一次连接的回收（重置） （默认为-1）
  14.             
  15. # def task():
  16.     # conn= engine.raw_connection() #建立原生连接,和pymysql的连接一样
  17.     # cur = conn.cursor()
  18.     # cur.execute("select * from t_sales where id>%s",(2,))
  19.     # result = cur.fetchone()
  20.     # cur.close()
  21.     # conn.close()
  22.     # print(result)
  23.  
  24.  
  25.     
  26. # def task():
  27.     # conn = engine.contextual_connect() #建立上下文管理器连接，自动打开和关闭
  28.     # with conn:
  29.         # cur = conn.execute("select * from t_sales where id>%s",(2,))
  30.         # result = cur.fetchone()
  31.     # print(result)
  32.     
  33.     
  34. def task():
  35.     cur =engine.execute("select * from t_sales where id>%s",(2,))  #engine直接执行
  36.     result = cur.fetchone()
  37.     cur.close()
  38.     print(result)
  39.  
  40. if __name__=="__main__":
  41.     for i in range(10):
  42.         t = threading.Thread(target=task)
  43.         t.start()

2.2 执行ORM语句

A. 创建和删除表

  1. #coding:utf-8
  2.  
  3. import datetime
  4. from sqlalchemy import create_engine
  5. from sqlalchemy.ext.declarative import declarative_base
  6. from sqlalchemy import Column, String, Integer, DateTime, Text
  7.  
  8. Base = declarative_base()
  9.  
  10. class User(Base):
  11.     __tablename__="users"
  12.     id = Column(Integer,primary_key=True)
  13.     name = Column(String(32),index=True, nullable=False) #创建索引，不为空
  14.     email = Column(String(32),unique=True)
  15.     ctime = Column(DateTime, default = datetime.datetime.now)  #传入方法名datetime.datetime.now
  16.     extra = Column(Text,nullable=True)  
  17.     
  18.     __table_args__ = {
  19.     
  20.         # UniqueConstraint('id', 'name', name='uix_id_name'), #设置联合唯一约束
  21.         # Index('ix_id_name', 'name', 'email'),               # 创建索引
  22.     }
  23.  
  24. def create_tbs():
  25.     engine = create_engine("mysql+pymysql://root@127.0.0.1:3306/learningsql?charset=utf8",max_overflow=2,pool_size=5)
  26.     Base.metadata.create_all(engine)   #创建所有定义的表
  27.  
  28. def drop_dbs():
  29.     engine = create_engine("mysql+pymysql://root@127.0.0.1:3306/learningsql?charset=utf8",max_overflow=2,pool_size=5)
  30.     Base.metadata.drop_all(engine)   #删除所有创建的表
  31.  
  32. if __name__=="__main__":
  33.     create_tbs() #创建表
  34.     #drop_dbs()   #删除表
  35.     

B.表中定义外键关系（一对多，多对多）

思考：下面代码中的一对多关系，relationship 定义在了 customer 表中，应该定义在 PurchaseOrder 更合理？

注意：mysql 数据库中避免使用 order做为表的名字，order 为一个 mysql 关键字，做为表名字时必须用反引号order (键盘数字1旁边的符号)

  1. #coding:utf-8
  2.  
  3. from sqlalchemy import create_engine
  4. from sqlalchemy.ext.declarative import declarative_base
  5. from sqlalchemy import Column,Integer,String,Text,DateTime,ForeignKey,Float
  6. from sqlalchemy.orm import relationship
  7. import datetime
  8.  
  9. engine = create_engine("mysql+pymysql://root@127.0.0.1:3306/learningsql?charset=utf8")  #数据库有密码时，//root:12345678@127.0.0.1:3306/
  10. Base = declarative_base()
  11.  
  12. class Customer(Base):
  13.     __tablename__="customer"  #数据库中保存的表名字
  14.     
  15.     id = Column(Integer,primary_key=True)
  16.     name = Column(String(64),index=True,nullable=False)
  17.     phone = Column(String(16),nullable=False)
  18.     address = Column(String(256),nullable=False)
  19.     purchase_order_id = Column(Integer,ForeignKey("purchase_order.id"))  #关键关系，关联表的__tablename__="purchase_order"
  20.     
  21.     # 和建立表结构无关，方便外键关系查询，backref反向查询时使用order_obj.customer
  22.     purchase_order = relationship("PurchaseOrder",backref="customer")
  23.  
  24.     
  25.     
  26. class PurchaseOrder(Base):  
  27.     __tablename__ = "purchase_order"   #mysql数据库中表的名字避免使用order，order为一个关键字，使用时必须用反引号`order` (键盘数字1旁边的符号)
  28.     id=Column(Integer,primary_key=True)
  29.     cost = Column(Float,nullable=True)
  30.     ctime = Column(DateTime,default =datetime.datetime.now)
  31.     desc = Column(String(528))
  32.     
  33.     #多对多关系时，secondary为中间表
  34.     product = relationship("Product",secondary="order_to_product",backref="purchase_order")
  35.     
  36. class Product(Base):
  37.     __tablename__ = "product"
  38.     id = Column(Integer,primary_key=True)
  39.     name = Column(String(256))
  40.     price = Column(Float,nullable=False)
  41.     
  42. class OrdertoProduct(Base):
  43.     __tablename__ = "order_to_product"
  44.     id = Column(Integer,primary_key=True)
  45.     product_id = Column(Integer,ForeignKey("product.id"))
  46.     purchase_order_id  = Column(Integer,ForeignKey("purchase_order.id"))
  47.     
  48.  
  49.  
  50. if __name__ == "__main__":
  51.     Base.metadata.create_all(engine)
  52.     #Base.metadata.drop_all(engine)

C.增删改查操作

增删改查

  1. #coding:utf-8
  2.  
  3. from sqlalchemy import create_engine
  4. from sqlalchemy.ext.declarative import declarative_base
  5. from sqlalchemy import Column,Integer,String,Text,DateTime,ForeignKey,Float
  6. from sqlalchemy.orm import relationship,sessionmaker
  7. from sqlalchemy.sql import text
  8. import datetime
  9.  
  10. engine = create_engine("mysql+pymysql://root@127.0.0.1:3306/learningsql?charset=utf8")  #数据库有密码时，//root:12345678@127.0.0.1:3306/, 设置utf8防止中文乱码
  11. Base = declarative_base()
  12.  
  13. class Customer(Base):
  14.     __tablename__="customer"  #数据库中保存的表名字
  15.     
  16.     id = Column(Integer,primary_key=True)
  17.     name = Column(String(64),index=True,nullable=False)
  18.     phone = Column(String(16),nullable=False)
  19.     address = Column(String(256),nullable=False)
  20.     purchase_order_id = Column(Integer,ForeignKey("purchase_order.id"))  #关键关系，关联表的__tablename__="purchase_order"
  21.     
  22.     # 和建立表结构无关，方便外键关系查询，backref反向查询时使用order_obj.customer
  23.     purchase_order = relationship("PurchaseOrder",backref="customer")
  24.     
  25.     
  26. class PurchaseOrder(Base):  
  27.     __tablename__ = "purchase_order"   #mysql数据库中表的名字避免使用order，order为一个关键字，使用时必须用反引号`order` (键盘数字1旁边的符号)
  28.     id=Column(Integer,primary_key=True)
  29.     cost = Column(Float,nullable=True)
  30.     ctime = Column(DateTime,default =datetime.datetime.now)
  31.     desc = Column(String(528))
  32.     
  33.     #多对多关系时，secondary为中间表
  34.     product = relationship("Product",secondary="order_to_product",backref="purchase_order")
  35.     
  36. class Product(Base):
  37.     __tablename__ = "product"
  38.     id = Column(Integer,primary_key=True)
  39.     name = Column(String(256))
  40.     price = Column(Float,nullable=False)
  41.     
  42. class OrdertoProduct(Base):
  43.     __tablename__ = "order_to_product"
  44.     id = Column(Integer,primary_key=True)
  45.     product_id = Column(Integer,ForeignKey("product.id"))
  46.     purchase_order_id  = Column(Integer,ForeignKey("purchase_order.id"))
  47.  
  48.  
  49. if __name__ == "__main__":
  50.     #Base.metadata.create_all(engine)
  51.     #Base.metadata.drop_all(engine)
  52.     
  53.     Session = sessionmaker(bind=engine)
  54.  
  55.     #每次进行数据库操作时都要创建session
  56.     session = Session()
  57.  
  58.     #*****************增加数据********************
  59.     # pur_order = PurchaseOrder(cost=19.7,desc="python编程之路")
  60.  
  61.     # session.add(pur_order)
  62.     # session.add_all(
  63.                 # [PurchaseOrder(cost=39.7,desc="linux操作系统"),
  64.                 # PurchaseOrder(cost=59.6,desc="python cookbook")])
  65.     # session.commit()
  66.     
  67.     #*****************修改数据********************
  68.     
  69.     #session.query(PurchaseOrder).filter(PurchaseOrder.id>2).update({"cost":29.7})
  70.     #session.query(PurchaseOrder).filter(PurchaseOrder.id==2).update({"cost":PurchaseOrder.cost+40.1},synchronize_session=False)  #synchronize_session用于query在进行delete or update操作时，对session的同步策略。
  71.     #session.commit()
  72.     
  73.     #*****************删除数据********************
  74.     #session.query(PurchaseOrder).filter(PurchaseOrder.id==1).delete()
  75.     #session.commit()
  76.     
  77.     #*****************查询数据********************
  78.     #ret = session.query(PurchaseOrder).all()
  79.     # ret = session.query(PurchaseOrder).filter(PurchaseOrder.id==2).all()  #包含对象的列表
  80.     # ret = session.query(PurchaseOrder).filter(PurchaseOrder.id==2).first() #单个对象
  81.     # ret = session.query(PurchaseOrder).filter_by(id=2).all()  #通过列名字的表达式
  82.     # ret = session.query(PurchaseOrder).filter_by(id=2).first()
  83.     #ret = session.query(PurchaseOrder).filter(text("id<:value and cost>:price")).params(value=6,price=15).order_by(PurchaseOrder.cost).all()
  84.     #ret = session.query(PurchaseOrder).from_statement(text("SELECT * FROM purchase_order WHERE cost>:price")).params(price=40).all()
  85.     # print ret
  86.     # for i in ret:
  87.         # print i.id, i.cost, i.ctime,i.desc
  88.         
  89.     #ret2 = session.query(PurchaseOrder.id,PurchaseOrder.cost.label('totalcost')).all()  #只查询两列，ret2为列表
  90.     #print ret2
  91.     
  92.     #关闭session
  93.     session.close()

查询语句

  1. # 条件
  2. ret = session.query(Users).filter_by(name='alex').all()
  3. ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()
  4. ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()
  5. ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
  6. ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
  7. ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
  8. from sqlalchemy import and_, or_
  9. ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
  10. ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
  11. ret = session.query(Users).filter(
  12.     or_(
  13.         Users.id < 2,
  14.         and_(Users.name == 'eric', Users.id > 3),
  15.         Users.extra != ""
  16.     )).all()
  17.  
  18.  
  19. # 通配符
  20. ret = session.query(Users).filter(Users.name.like('e%')).all()
  21. ret = session.query(Users).filter(~Users.name.like('e%')).all()
  22.  
  23. # 限制
  24. ret = session.query(Users)[1:2]
  25.  
  26. # 排序
  27. ret = session.query(Users).order_by(Users.name.desc()).all()
  28. ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()
  29.  
  30. # 分组
  31. from sqlalchemy.sql import func
  32.  
  33. ret = session.query(Users).group_by(Users.extra).all()
  34. ret = session.query(
  35.     func.max(Users.id),
  36.     func.sum(Users.id),
  37.     func.min(Users.id)).group_by(Users.name).all()
  38.  
  39. ret = session.query(
  40.     func.max(Users.id),
  41.     func.sum(Users.id),
  42.     func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()
  43.  
  44. # 连表
  45.  
  46. ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
  47.  
  48. ret = session.query(Person).join(Favor).all()
  49.  
  50. ret = session.query(Person).join(Favor, isouter=True).all()
  51.  
  52.  
  53. # 组合
  54. q1 = session.query(Users.name).filter(Users.id > 2)
  55. q2 = session.query(Favor.caption).filter(Favor.nid < 2)
  56. ret = q1.union(q2).all()
  57.  
  58. q1 = session.query(Users.name).filter(Users.id > 2)
  59. q2 = session.query(Favor.caption).filter(Favor.nid < 2)
  60. ret = q1.union_all(q2).all()

补充

  1. #coding:utf-8
  2.  
  3. from sqlalchemy import create_engine
  4. from sqlalchemy.orm import sessionmaker
  5. from sqlalchemy.sql import text, func
  6. from sqlalchemy_orm2 import PurchaseOrder  #导入定义的PurchaseOrder表格类
  7.  
  8. engine = create_engine("mysql+pymysql://root@127.0.0.1:3306/learningsql?charset=utf8")
  9. Session = sessionmaker(bind=engine)
  10. session = Session()
  11. #查询
  12. ret = session.execute("select * from purchase_order where id=:value",params={"value":3})
  13. print ret
  14. for i in ret:
  15.     print i.id, i.cost, i.ctime,i.desc
  16.  
  17. #插入
  18. purchase_order  = PurchaseOrder.__table__  #拿到PurchaseOrder表格对象
  19. ret=session.execute(purchase_order.insert(),
  20.                 [{"cost":46.3,"desc":'python2'},
  21.                 {"cost":43.3,"desc":'python3'}])
  22. session.commit()
  23. print(ret.lastrowid)
  24. session.close()
  25.  
  26. # 关联子查询
  27. subqry = session.query(func.count(Server.id).label("sid")).filter(Server.id == Group.id).correlate(Group).as_scalar()
  28. result = session.query(Group.name, subqry)
  29. """
  30. SELECT `group`.name AS group_name, (SELECT count(server.id) AS sid 
  31. FROM server 
  32. WHERE server.id = `group`.id) AS anon_1 
  33. FROM `group`
  34. """

D.多线程操作

  1. #coding:utf-8
  2.  
  3. from sqlalchemy import create_engine
  4. from sqlalchemy.orm import sessionmaker
  5. from sqlalchemy_orm2 import Product
  6. from threading import Thread
  7.  
  8. engine = create_engine("mysql+pymysql://root@127.0.0.1:3306/learningsql?charset=utf8",max_overflow=0,pool_size=5)
  9. Session = sessionmaker(bind=engine)
  10.  
  11. def task(name,price):
  12.     session = Session()
  13.     pro = Product(name=name,price=price)
  14.     session.add(pro)
  15.     session.commit()
  16.     session.close()
  17.  
  18. if __name__=="__main__":
  19.     for i in range(6):
  20.         t = Thread(target=task,args=("pro"+str(i),i*5))
  21.         t.start()
  22.     

E. 通过relationship操纵一对多和多对多关系

一对多

  1. #coding:utf-8
  2.  
  3. from sqlalchemy import create_engine
  4. from sqlalchemy.orm import sessionmaker
  5. from sqlalchemy.sql import text, func
  6. from sqlalchemy_orm2 import PurchaseOrder,Product,OrdertoProduct,Customer  #导入定义的表格类
  7.  
  8.  
  9. engine = create_engine("mysql+pymysql://root@127.0.0.1:3306/learningsql?charset=utf8")
  10. Session = sessionmaker(bind=engine)
  11. session = Session()
  12.  
  13. # #通过定义的关键关系添加(id值)
  14. # cus1 = Customer(name="zack",phone="13567682333",address="Nanjing",purchase_order_id=3)
  15. # session.add(cus1)
  16.  
  17.  
  18. # #通过relationship正向添加
  19. # cus2 = Customer(name="zack2",phone="13567682333",address="Nanjing",purchase_order=PurchaseOrder(cost=53,desc="java"))
  20. # session.add(cus2)
  21. # session.commit()
  22.  
  23. #通过relationship反向添加
  24. # purchase_order=PurchaseOrder(cost=53,desc="php")
  25. # cus3 = Customer(name="zack3",phone="13567682333",address="Nanjing")
  26. # cus4 = Customer(name="zack4",phone="13567682333",address="Nanjing")
  27. # purchase_order.customer=[cus3,cus4]  #cus3,cus4的purchase_order_id都是purchase_order.id值,即同时添加了两组外键关系
  28. # session.add(purchase_order)
  29. # session.commit()
  30.  
  31. ##通过relationship正向查询
  32. cus = session.query(Customer).first()
  33. print(cus.purchase_order_id)
  34. print(cus.purchase_order.desc)
  35.  
  36. #通过relationship反向查询
  37. pur = session.query(PurchaseOrder).filter(PurchaseOrder.id==3).first()
  38. print(pur.desc)
  39. print(pur.customer) #返回一个list

多对多

  1. #coding:utf-8
  2.  
  3.  
  4. from sqlalchemy import create_engine
  5. from sqlalchemy.orm import sessionmaker
  6. from sqlalchemy.sql import text, func
  7. from sqlalchemy_orm2 import PurchaseOrder,Product,OrdertoProduct,Customer  #导入定义的表格类
  8.  
  9. engine = create_engine("mysql+pymysql://root@127.0.0.1:3306/learningsql?charset=utf8")
  10. Session = sessionmaker(bind=engine)
  11. session = Session()
  12.  
  13. # session.add_all([Product(name="java",price=24),
  14.                 # Product(name="python",price=34),
  15.                 # Product(name="php",price=27)])
  16. # session.commit()
  17.  
  18. # #通过定义的关键关系添加(id值)
  19. # op = OrdertoProduct(product_id=1,purchase_order_id=16)
  20. # session.add(op)
  21. # session.commit()
  22.  
  23. # #通过relationship添加
  24. # pur = PurchaseOrder(cost=27,desc="xxxx")
  25. # pur.product = [Product(name="C++",price=60),]  #正向
  26. # session.add(pur)
  27.  
  28. # pro = Product(name="C",price=40)
  29. # pro.purchase_order=[PurchaseOrder(cost=27,desc="xxxx"),]  #反向
  30. # session.add(pro)
  31. # session.commit()
  32.  
  33.  
  34. #通过relationship正向查询
  35. pur = session.query(PurchaseOrder).filter(PurchaseOrder.id==19).first()
  36. print(pur.desc)
  37. print(pur.product) #结果为列表
  38.  
  39. #通过relationship反向查询
  40. pro = session.query(Product).filter(Product.id==5).first()
  41. print(pro.name)
  42. print(pro.purchase_order) #结果为列表
  43.  
  44. session.close()

3.数据库连接池

对于ORM框架，其内部维护了链接池，可以直接通过多线程操控数据库。对于pymysql模块，通过多线程操控数据库容易出错，得加锁串行执行。进行并发时，可以利用DBUtils模块来维护数据库连接池。

3.1 多线程操控pymysql

不采用DBUtils连接池时, pymysql多线程代码如下：

每个线程创建链接

  1. import pymysql
  2. import threadind
  3.  
  4. #**************************无连接池*******************************    
  5. #每个线程都要创立一次连接，线程并发操作间可能有问题？
  6. def func():
  7.     conn = pymysql.connect(host="127.0.0.1",port=3306,db="learningsql",user="root",passwd="",charset="utf8")
  8.     cursor = conn.cursor()
  9.     cursor.execute("select * from user where nid>1;")
  10.     result = cursor.fetchone()
  11.     print(result)
  12.     cursor.close()
  13.     conn.close()
  14.     
  15. if __name__=="__main__":
  16.     for i in range(5):
  17.         t = threading.Thread(target=func,name="thread-%s"%i)
  18.         t.start()

一个连接串行执行

  1. #**************************无连接池*******************************
  2. #创建一个连接，加锁串行执行
  3. from threading import Lock
  4. import pymysql
  5. import threading
  6. conn = pymysql.connect(host="127.0.0.1",port=3306,db="learningsql",user="root",passwd="",charset="utf8")    
  7.  
  8.  
  9. lock = Lock()    
  10. def func():
  11.     with lock:
  12.         cursor = conn.cursor()
  13.         cursor.execute("select * from user where nid>1;")
  14.         result = cursor.fetchone()
  15.         print(result)
  16.         cursor.close()
  17.         
  18.         #conn.close()不能在线程中关闭连接，否则其他线程不可用了
  19.         
  20. if __name__=="__main__":
  21.     threads = []
  22.     for i in range(5):
  23.         t = threading.Thread(target=func,name="thread-%s"%i)
  24.         threads.append(t)
  25.         t.start()
  26.         
  27.     for t in threads:
  28.         t.join()
  29.     
  30.     conn.close()

3.2 DBUtils连接池

DBUtils连接池有两种连接模式：PersistentDB和PooledDB

官网文档：https://cito.github.io/DBUtils/UsersGuide.html

模式一(DBUtils.PersistentDB):

为每个线程创建一个连接，线程即使调用了close方法，也不会关闭，只是把连接重新放到连接池，供自己线程再次使用。当线程终止时，连接自动关闭。

PersistentDB使用代码如下：

  1. #coding:utf-8
  2.  
  3. from DBUtils.PersistentDB import PersistentDB
  4. import pymysql
  5. import threading
  6.  
  7. pool = PersistentDB(
  8.     creator = pymysql,  # 使用链接数据库的模块
  9.     maxusage = None,    # 一个链接最多被重复使用的次数，None表示无限制
  10.     setsession=[],     # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]
  11.     ping = 0,           # ping MySQL服务端，检查是否服务可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always
  12.     closeable = False,    # 如果为False时， conn.close() 实际上被忽略，供下次使用，再线程关闭时，才会自动关闭链接。如果为True时， conn.close()则关闭链接，那么再次调用pool.connection时就会报错，因为已经真的关闭了连接（pool.steady_connection()可以获取一个新的链接）
  13.     threadlocal = None,    # 本线程独享值得对象，用于保存链接对象，如果链接对象被重置
  14.     host="127.0.0.1",
  15.     port = 3306,
  16.     user = "root",
  17.     password="",
  18.     database="learningsql",
  19.     charset = "utf8"
  20. )
  21.  
  22. def func():
  23.     conn = pool.connection()
  24.     cursor = conn.cursor()
  25.     cursor.execute("select * from user where nid>1;")
  26.     result = cursor.fetchone()
  27.     print(result)
  28.     cursor.close()
  29.     conn.close()
  30.     
  31. if __name__ == "__main__":
  32.     for i in range(5):
  33.         t = threading.Thread(target=func,name="thread-%s"%i)
  34.         t.start()

模式二(DBUtils.PooledDB)：

创建一批连接到连接池，供所有线程共享使用。

(由于pymysql、MySQLdb等threadsafety值为1，所以该模式连接池中的线程会被所有线程共享。)

PooledDB使用代码如下：

  1. from DBUtils.PooledDB import PooledDB
  2. import pymysql
  3. import threading
  4. import time
  5.  
  6. pool = PooledDB(
  7.     creator = pymysql,  # 使用链接数据库的模块
  8.     maxconnections = 6,  # 连接池允许的最大连接数，0和None表示不限制连接数
  9.     mincached = 2,   # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
  10.     maxcached = 5,   # 链接池中最多闲置的链接，0和None不限制
  11.     maxshared = 3,   # 链接池中最多共享的链接数量，0和None表示全部共享。PS: 无用，因为pymysql和MySQLdb等模块的 threadsafety都为1，所有值无论设置为多少，_maxcached永远为0，所以永远是所有链接都共享。
  12.     blocking = True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
  13.     maxusage = None,   # 一个链接最多被重复使用的次数，None表示无限制
  14.     setsession = [],   # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]
  15.     ping = 0,           # ping MySQL服务端，检查是否服务可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always
  16.     host="127.0.0.1",
  17.     port = 3306,
  18.     user="root",
  19.     password="",
  20.     database = "learningsql",
  21.     charset = "utf8"
  22. )
  23.  
  24. def func():
  25.     conn = pool.connection()
  26.     cursor = conn.cursor()
  27.     cursor.execute("select * from user where nid>1;")
  28.     result = cursor.fetchone()
  29.     print(result)
  30.     time.sleep(5)  #为了查看mysql端的线程数量
  31.     cursor.close()
  32.     conn.close()
  33.     
  34. if __name__=="__main__":
  35.     for i in range(5):
  36.         t = threading.Thread(target=func,name="thread-%s"%i)
  37.         t.start()

上述代码中加入了sleep(5)使线程连接数据库时间延长，方便查看mysql数据库连接线程情况，下图分别为代码执行中和执行后的线程连接情况，可以发现，代码执行时，同时有6个线程连接上了数据库（有一个为mysql命令客户端），代码执行后，只有一个线程连接数据库，但仍有5个线程等待连接。

（show status like “Threads%” 查看线程连接情况）