• pymysql
  • 安装
  • 创建表
  • 增删改查
  • 更多查询
• SQLAlchmy
  • 安装
  • 执行原生SQL
  • 执行ORM

参考来源

cnblogs

pymysql和SQLAchemy

python使用MySQL主要有两个模块，pymysql（MySQLdb与pymysql类似，但只支持python2）和SQLAchemy。

• pymysql（MySQLdb）为原生模块，直接执行sql语句
• SQLAchemy为一个ORM框架，将数据对象转换成SQL，然后使用数据API执行SQL并获取执行结果

pymysql

安装

pip3 install pymysql --user

创建表

 
# 连接数据库
conn = pymysql.connect(host="localhost",user="root",passwd="root",db="learningsql",charset="utf8",port=3306)

# 创建游标
cursor = conn.cursor()

# 执行sql语句
cursor.execute("""create table if not exists t_sales(
	id int primary key auto_increment not null,
	nickName varchar(128) not null,
	color varchar(128) not null,
	size varchar(128) not null,
	comment text not null,
	saledata varchr(128) not null)engine=InnoDB defalut charset=utf8;""")

#提交
conn.commit()

#关闭游标
cursor.close()

#关闭连接
conn.close()

增删改查

 
# 连接数据库
conn = pymysql.connect(host="localhost",user="root",passwd="root",db="learningsql",charset="utf8",port=3306)  

# 创建游标
cursor = conn.cursor()

# 新增一条数据
insert_sql = "insert into t_sales(nickName,color,size,comment,saledate) values(%s,%s,%s,%s,%s);"  
## 返回受影响的行数
row1 = cursor.execute(insert_sql,("Bob", "黑色", "XL", "便宜实惠", "2019-04-20"))  

# 新增多条数据
insert_sql = "insert into t_sales(nickName,color,size,comment,saledate) values(%s,%s,%s,%s,%s);"  
data = [("Bob", "黑色", "XL", "便宜实惠", "2019-04-20"),("Ted", "黄色", "M", "便宜实惠", "2019-04-20"),("Gary", "黑色", "M", "穿着舒服", "2019-04-20")]
row1 = cursor.executemany(insert_sql, data)

update_sql = "update t_sales set color='白色' where id=%s;"
#返回受影响的行数
row2 = cursor.execute(update_sql,(1,))

select_sql = "select * from t_sales where id>%s;"
#返回受影响的行数
row3 = cursor.execute(select_sql,(1,))

delete_sql = "delete from t_sales where id=%s;"
#返回受影响的行数
row4 = cursor.execute(delete_sql,(4,))

#提交，不然无法保存新建或者修改的数据(增删改得提交)
conn.commit()
cursor.close()
conn.close()

更多查询

 
conn = pymysql.connect(host="localhost", user="root", passwd="", db='learningsql', charset='utf8', port=3306)  #和mysql服务端设置格式一样（还可设置为gbk, gb2312）
#创建游标
cursor = conn.cursor()

select_sql = "select id,nickname,size from t_sales where id>%s;"
cursor.execute(select_sql, (3,))

row1 = cursor.fetchone()      #获取第一条数据，获取后游标会向下移动一行
row_n = cursor.fetchmany(3)  #获取前n条数据，获取后游标会向下移动n行
row_all = cursor.fetchall()  #获取所有数据，获取后游标会向下移动到末尾
print(row1)
print(row_n)
print(row_all)
#conn.commit()
cursor.close()
conn.close()

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

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

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

 
conn = pymysql.connect(host="localhost", user="root", passwd="", db='learningsql', charset='utf8', port=3306)  #和mysql服务端设置格式一样（还可设置为gbk, gb2312）
#创建游标
cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)

select_sql = "select id,nickname,size from t_sales where id>%s;"
cursor.execute(select_sql, (3,))

row1 = cursor.fetchall()      

print(row1)

conn.commit()
cursor.close()
conn.close()

SQLAlchmy

安装

pip3 install sqlalchemy --user

执行原生SQL

 
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, String, Integer
import threading

engine = create_engine(
            "mysql+pymysql://root@127.0.0.1:3306/learningsql?charset=utf8",
            max_overflow = 0,  #超过连接池大小外最多创建的连接，为0表示超过5个连接后，其他连接请求会阻塞 （默认为10）
            pool_size = 5,    #连接池大小（默认为5）
            pool_timeout = 30,  #连接线程池中，没有连接时最多等待的时间，不设置无连接时直接报错 （默认为30）
            pool_recycle = -1)  #多久之后对线程池中的线程进行一次连接的回收（重置） （默认为-1）


# 三种方式执行原生sql

# def task():
    # conn= engine.raw_connection() #建立原生连接,和pymysql的连接一样
    # cur = conn.cursor()
    # cur.execute("select * from t_sales where id>%s",(2,))
    # result = cur.fetchone()
    # cur.close()
    # conn.close()
    # print(result)


    
# def task():
    # conn = engine.contextual_connect() #建立上下文管理器连接，自动打开和关闭
    # with conn:
        # cur = conn.execute("select * from t_sales where id>%s",(2,))
        # result = cur.fetchone()
    # print(result)
    
    
def task():
    cur =engine.execute("select * from t_sales where id>%s",(2,))  #engine直接执行
    result = cur.fetchone()
    cur.close()
    print(result)

if __name__=="__main__":
    for i in range(10):
        t = threading.Thread(target=task)
        t.start()

执行ORM

创建和删除表

 
import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, String, Integer, DateTime, Text

Base = declarative_base()

class User(Base):
    __tablename__="users"
    id = Column(Integer,primary_key=True)
    name = Column(String(32),index=True, nullable=False) #创建索引，不为空
    email = Column(String(32),unique=True)
    ctime = Column(DateTime, default = datetime.datetime.now)  #传入方法名datetime.datetime.now
    extra = Column(Text,nullable=True)  
    
    __table_args__ = {
    
        # UniqueConstraint('id', 'name', name='uix_id_name'), #设置联合唯一约束
        # Index('ix_id_name', 'name', 'email'),               # 创建索引
    }

def create_tbs():
    engine = create_engine("mysql+pymysql://root@127.0.0.1:3306/learningsql?charset=utf8",max_overflow=2,pool_size=5)  
    Base.metadata.create_all(engine)   #创建所有定义的表

def drop_dbs():
    engine = create_engine("mysql+pymysql://root@127.0.0.1:3306/learningsql?charset=utf8",max_overflow=2,pool_size=5)  
    Base.metadata.drop_all(engine)   #删除所有创建的表

if __name__=="__main__":
    create_tbs() #创建表
    #drop_dbs()   #删除表
    

定义外键关系

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

 
#coding:utf-8

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String,Text,DateTime,ForeignKey,Float
from sqlalchemy.orm import relationship
import datetime

engine = create_engine("mysql+pymysql://root@127.0.0.1:3306/learningsql?charset=utf8")  #数据库有密码时，//root:12345678@127.0.0.1:3306/  
Base = declarative_base()

class Customer(Base):
    __tablename__="customer"  #数据库中保存的表名字
    
    id = Column(Integer,primary_key=True)
    name = Column(String(64),index=True,nullable=False)
    phone = Column(String(16),nullable=False)
    address = Column(String(256),nullable=False)
    purchase_order_id = Column(Integer,ForeignKey("purchase_order.id"))  #关键关系，关联表的__tablename__="purchase_order"
    
    # 和建立表结构无关，方便外键关系查询，backref反向查询时使用order_obj.customer
    purchase_order = relationship("PurchaseOrder",backref="customer")

    
    
class PurchaseOrder(Base):  
    __tablename__ = "purchase_order"   #mysql数据库中表的名字避免使用order，order为一个关键字，使用时必须用反引号`order` (键盘数字1旁边的符号)  
    id=Column(Integer,primary_key=True)
    cost = Column(Float,nullable=True)
    ctime = Column(DateTime,default =datetime.datetime.now)
    desc = Column(String(528))
    
    #多对多关系时，secondary为中间表
    product = relationship("Product",secondary="order_to_product",backref="purchase_order")
    
class Product(Base):
    __tablename__ = "product"
    id = Column(Integer,primary_key=True)
    name = Column(String(256))
    price = Column(Float,nullable=False)
    
class OrdertoProduct(Base):
    __tablename__ = "order_to_product"
    id = Column(Integer,primary_key=True)
    product_id = Column(Integer,ForeignKey("product.id"))
    purchase_order_id  = Column(Integer,ForeignKey("purchase_order.id"))
    


if __name__ == "__main__":
    Base.metadata.create_all(engine)
    #Base.metadata.drop_all(engine)

增删改查

 
#coding:utf-8

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String,Text,DateTime,ForeignKey,Float
from sqlalchemy.orm import relationship,sessionmaker
from sqlalchemy.sql import text
import datetime

engine = create_engine("mysql+pymysql://root@127.0.0.1:3306/learningsql?charset=utf8")  #数据库有密码时，//root:12345678@127.0.0.1:3306/, 设置utf8防止中文乱码
Base = declarative_base()

class Customer(Base):
    __tablename__="customer"  #数据库中保存的表名字
    
    id = Column(Integer,primary_key=True)
    name = Column(String(64),index=True,nullable=False)
    phone = Column(String(16),nullable=False)
    address = Column(String(256),nullable=False)
    purchase_order_id = Column(Integer,ForeignKey("purchase_order.id"))  #关键关系，关联表的__tablename__="purchase_order"
    
    # 和建立表结构无关，方便外键关系查询，backref反向查询时使用order_obj.customer
    purchase_order = relationship("PurchaseOrder",backref="customer")
    
    
class PurchaseOrder(Base):  
    __tablename__ = "purchase_order"   #mysql数据库中表的名字避免使用order，order为一个关键字，使用时必须用反引号`order` (键盘数字1旁边的符号)
    id=Column(Integer,primary_key=True)
    cost = Column(Float,nullable=True)
    ctime = Column(DateTime,default =datetime.datetime.now)
    desc = Column(String(528))
    
    #多对多关系时，secondary为中间表
    product = relationship("Product",secondary="order_to_product",backref="purchase_order")
    
class Product(Base):
    __tablename__ = "product"
    id = Column(Integer,primary_key=True)
    name = Column(String(256))
    price = Column(Float,nullable=False)
    
class OrdertoProduct(Base):
    __tablename__ = "order_to_product"
    id = Column(Integer,primary_key=True)
    product_id = Column(Integer,ForeignKey("product.id"))
    purchase_order_id  = Column(Integer,ForeignKey("purchase_order.id"))


if __name__ == "__main__":
    #Base.metadata.create_all(engine)
    #Base.metadata.drop_all(engine)
    
    Session = sessionmaker(bind=engine)

    #每次进行数据库操作时都要创建session
    session = Session()

    #*****************增加数据********************
    # pur_order = PurchaseOrder(cost=19.7,desc="python编程之路")

    # session.add(pur_order)
    # session.add_all(
                # [PurchaseOrder(cost=39.7,desc="linux操作系统"),
                # PurchaseOrder(cost=59.6,desc="python cookbook")])
    # session.commit()
    
    #*****************修改数据********************
    
    #session.query(PurchaseOrder).filter(PurchaseOrder.id>2).update({"cost":29.7})
    #session.query(PurchaseOrder).filter(PurchaseOrder.id==2).update({"cost":PurchaseOrder.cost+40.1},synchronize_session=False)  #synchronize_session用于query在进行delete or update操作时，对session的同步策略。
    #session.commit()
    
    #*****************删除数据********************
    #session.query(PurchaseOrder).filter(PurchaseOrder.id==1).delete()
    #session.commit()
    
    #*****************查询数据********************
    #ret = session.query(PurchaseOrder).all()
    # ret = session.query(PurchaseOrder).filter(PurchaseOrder.id==2).all()  #包含对象的列表
    # ret = session.query(PurchaseOrder).filter(PurchaseOrder.id==2).first() #单个对象
    # ret = session.query(PurchaseOrder).filter_by(id=2).all()  #通过列名字的表达式
    # ret = session.query(PurchaseOrder).filter_by(id=2).first()
    #ret = session.query(PurchaseOrder).filter(text("id<:value and cost>:price")).params(value=6,price=15).order_by(PurchaseOrder.cost).all()
    #ret = session.query(PurchaseOrder).from_statement(text("SELECT * FROM purchase_order WHERE cost>:price")).params(price=40).all()
    # print ret
    # for i in ret:
        # print i.id, i.cost, i.ctime,i.desc
        
    #ret2 = session.query(PurchaseOrder.id,PurchaseOrder.cost.label('totalcost')).all()  #只查询两列，ret2为列表
    #print ret2
    
    #关闭session
    session.close()

更多查询

 
#　条件
ret = session.query(Users).filter_by(name='alex').all()
ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
from sqlalchemy import and_, or_
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
ret = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == 'eric', Users.id > 3),
        Users.extra != ""
    )).all()


# 通配符
ret = session.query(Users).filter(Users.name.like('e%')).all()
ret = session.query(Users).filter(~Users.name.like('e%')).all()

# 限制
ret = session.query(Users)[1:2]

# 排序
ret = session.query(Users).order_by(Users.name.desc()).all()
ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()

# 分组
from sqlalchemy.sql import func

ret = session.query(Users).group_by(Users.extra).all()
ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).all()

ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()

# 连表

ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()

ret = session.query(Person).join(Favor).all()

ret = session.query(Person).join(Favor, isouter=True).all()


# 组合
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union(q2).all()

q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union_all(q2).all()

relationship操纵一对多和多对多关系

一对多

 
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.sql import text, func
from sqlalchemy_orm2 import PurchaseOrder,Product,OrdertoProduct,Customer  #导入定义的表格类


engine = create_engine("mysql+pymysql://root@127.0.0.1:3306/learningsql?charset=utf8")
Session = sessionmaker(bind=engine)
session = Session()

# #通过定义的关键关系添加(id值)
# cus1 = Customer(name="zack",phone="13567682333",address="Nanjing",purchase_order_id=3)
# session.add(cus1)


# #通过relationship正向添加
# cus2 = Customer(name="zack2",phone="13567682333",address="Nanjing",purchase_order=PurchaseOrder(cost=53,desc="java"))
# session.add(cus2)
# session.commit()

#通过relationship反向添加
# purchase_order=PurchaseOrder(cost=53,desc="php")
# cus3 = Customer(name="zack3",phone="13567682333",address="Nanjing")
# cus4 = Customer(name="zack4",phone="13567682333",address="Nanjing")
# purchase_order.customer=[cus3,cus4]  #cus3,cus4的purchase_order_id都是purchase_order.id值,即同时添加了两组外键关系
# session.add(purchase_order)
# session.commit()

##通过relationship正向查询
cus = session.query(Customer).first()
print(cus.purchase_order_id)
print(cus.purchase_order.desc)

#通过relationship反向查询
pur = session.query(PurchaseOrder).filter(PurchaseOrder.id==3).first()
print(pur.desc)
print(pur.customer) #返回一个list

多对多

 
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.sql import text, func
from sqlalchemy_orm2 import PurchaseOrder,Product,OrdertoProduct,Customer  #导入定义的表格类

engine = create_engine("mysql+pymysql://root@127.0.0.1:3306/learningsql?charset=utf8")
Session = sessionmaker(bind=engine)
session = Session()

# session.add_all([Product(name="java",price=24),
                # Product(name="python",price=34),
                # Product(name="php",price=27)])
# session.commit()

# #通过定义的关键关系添加(id值)
# op = OrdertoProduct(product_id=1,purchase_order_id=16)
# session.add(op)
# session.commit()

# #通过relationship添加
# pur = PurchaseOrder(cost=27,desc="xxxx")
# pur.product = [Product(name="C++",price=60),]  #正向
# session.add(pur)

# pro = Product(name="C",price=40)
# pro.purchase_order=[PurchaseOrder(cost=27,desc="xxxx"),]  #反向
# session.add(pro)
# session.commit()


#通过relationship正向查询
pur = session.query(PurchaseOrder).filter(PurchaseOrder.id==19).first()
print(pur.desc)
print(pur.product) #结果为列表

#通过relationship反向查询
pro = session.query(Product).filter(Product.id==5).first()
print(pro.name)
print(pro.purchase_order) #结果为列表

session.close()