SQLAlchemy连接MySQL及记录的查询、更新、删除、多表关联查询_sqlalchemy mysql

SQLAlchemy是Python的ORM库，支持多种数据库。

建立连接

连接MySQL要用到Engine，Engine集成了连接池pool和方言Dialect（支持不通数据库的SQL语法），最后都统一成标准DBAPI。

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://user:pass@localhost/db?charset=utf8mb4',
                       echo=True, pool_recycle=7200, pool_size=5, max_overflow=10, pool_timeout=30)
1
2
3

create_engine第一个参数是database url，用到了pymysql驱动。
参数echo用于打印执行中的关键日志，包括SQL执行语句。
pool_recycle表示connection空闲7200秒就会重新获取。
pool_size指定连接池大小。
max_overflow指定超过连接池大小的最大数。
pool_timeout获取连接的超时阈值。
poolclass设置为NullPool，禁用连接池，session.Close()之后会立即断开数据库连接。

报错：Lost connection to MySQL server during query

2013, 'Lost connection to MySQL server during query 
([Errno 104] Connection reset by peer)
1
2

设置pool_pre_ping=True和pool_recycle可以每次执行都查看连接是否可用。
pool_recycle的值 应该比mysql中设置的 interactive_timeout 和 wait_timeout 值小才有效。
pool_timeout的值(默认10s) 应该比 mysql的 connect_timeout值 小或者等于。

SHOW GLOBAL VARIABLES LIKE '%timeout%';
1

@等特殊字符的处理

用urllib.parse.quote_plus编码@特殊字符。

from urllib import parse
from sqlalchemy import create_engine
 
user = "root"
password = "******@108"
host = "127.0.0.1"
db = "db_name"
 
pwd = parse.quote_plus(password)
 
engine = create_engine(f"mysql+pymysql://{user}:{pwd}@{host}:3306/{db}?charset=utf8")
 
1
2
3
4
5
6
7
8
9
10
11
12

建立映射

sqlalchemy把表table映射成model class，把字段field映射成column，把记录record映射成实例instance。

手动创建映射

from sqlalchemy.sql.schema import Column
from sqlalchemy.types import BigInteger, String, Integer

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(BigInteger, primary_key=True)
    name = Column(String(255))
    age = Column(Integer)
    gender = Column(String(255))
1
2
3
4
5
6
7
8
9
10
11
12

另外可以使用sqlacodegen（pip install sqlacodegen）自动生成model：

sqlacodegen mysql+pymysql://user:password@localhost/test --outfile=models.py --tables users
1

# coding: utf-8
from sqlalchemy import Column, String
from sqlalchemy.dialects.mysql import BIGINT, INTEGER
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
metadata = Base.metadata


class User(Base):
    __tablename__ = 'users'

    id = Column(BIGINT(20), primary_key=True)
    name = Column(String(255, 'utf8_unicode_ci'))
    age = Column(INTEGER(11))
    gender = Column(String(255, 'utf8_unicode_ci'))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

参考类型：

自动创建映射

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql.schema import Table

Base = declarative_base()
metadata = Base.metadata
metadata.bind = engine


# class User(Base):
#     __table__ = Table('users', metadata, autoload=True)
class User(Base):
    __table__ = Table('users', metadata, autoload_with=engine)
1
2
3
4
5
6
7
8
9
10
11
12

设置autoload或autoload_with就可以自动生成映射，另外Base.metadata需要提前绑定engine。

查询

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)
session = Session()
query = session.query(User).filter(User.age >= 19).filter(User.name.like("王%"))
for user in query.all():
    print(user.age)
1
2
3
4
5
6
7

查询用到了Session对象，类似一个容器，容器里是identity map 的结构对象。
用filter过滤数据。

等值过滤器（==）

session.query(Account).filter(Account.user_name=='Mark') #判断字符串类型
session.query(Account).filter(Account.salary==2000) #判断数值类型
1
2

不等过滤器（!=、<、>、<=、>=）

session.query(Account).filter(Account.user_name !="mark" ) #不等于字符串类型
session.query(Account).filter(Account.salary !=2000) #不等于数值类型
session.query(Account).filter(Account.salary >2000) #大于过滤器
session.query(Account).filter(Account.salary <2000) #小于过滤器
session.query(Account).filter(Account.salary <=2000) #小于等于过滤器
session.query(Account).filter(Account.salary >=2000) #大于等于过滤器

1
2
3
4
5
6
7

模糊查询（like）

#查询所有名字包含字母i的用户，结果包括id为1、2、3、4的4条记录
session.query(Account).filter(Account.user_name.like('%i%'))

#查询所有title中以Manager结尾的用户，结果包括id为1、5的两条记录
session.query(Account).filter(Account.title.like('%Manager'))

#查询所有名字中以Da开头的用户，结果包括id为1、3的两条记录
session.query(Account).filter(Account.user_name.like('Da%'))

1
2
3
4
5
6
7
8
9

包括过滤器（in_）

#查询id不为1，3，5的记录，结果包含id为2，4的两条记录
session.query(Account).filter(~Account.id.in_([1,3,5]))
#查询工资不为2000、3000、4000的记录，结果包含id为5的1条记录
session.query(Account).filter(~Account.id.in_([2000,3000,4000]))
#查询所有title不为Engineer和Accountant记录，结果包括id为1、5的两条记录
session.query(Account).filter(~Account.title.in_(['Accountant','Engineer']))

1
2
3
4
5
6
7

判断是否为空（is NULL、is not NULL）

#查询salary为空值的记录，结果包含id为5的记录
#下面两方式效果相同
session.query(Account).filter(Account.salary==None)
session.query(Account).filter(Account,salary.is_(None))

#查询salary不为空值的记录，结果包含id为1、2、3、4的记录
#下面两方式效果相同
session.query(Account).filter(Account.salary!=None)
session.query(Account).filter(Account.salary.isnot(None))

1
2
3
4
5
6
7
8
9
10

非逻辑（~）

#查询id不为1、3、5的记录，结果包含id为2、4的两条记录
session.query(Account).filter(~Account.id.in_([1,3,5]))

#查询工资不为2000、3000、4000的记录，结果包含id为5的1条记录
session.query(Account).filter(~Account.id.in_([2000,3000,4000]))

#查询所有title不为Engineer和Accountant的记录，结果包括id为1、5的2条记录。
session.query(Account).filter(~Account.title.in(['Accountant','Engineer']))

1
2
3
4
5
6
7
8
9

与逻辑（and_）

#直接在filter中添加多个条件即表示与逻辑
session.query(Account).filter(Account.title=='Engineer',Account.salary=3000)

#用关机子and_进行逻辑查询
from sqlalchemy import and_
session.query(Account).filter(and_(Account.title=='Engineer',Account.salary=3000))

#通过多个filter的链接表示与逻辑
session.query(Account).filter(Account.title=='Engineer').filter(Account.salary=3000)

1
2
3
4
5
6
7
8
9
10

或逻辑（or_）

#引入或逻辑关键字or_
from sqlalchemy import or_

#查询title是Engineer或者salary为3000的记录，返回结果为id为1、2、3、4的记录
session.query(Account).filter(or_(Account.title=='Engineer',Account.salary=3000))

1
2
3
4
5
6

新增

user1 = User()
user1.name = "赵六"
user1.age = 21
user1.gender = "女"
session.add(user1)
session.commit()
1
2
3
4
5
6

先建立一个User实例，然后调用session的add方法，最后commit。

批量插入新增

bulk_save_objects方法

session = Session()
objects = [
    User(name="u1"),
    User(name="u2"),
    User(name="u3")]
session.bulk_save_objects(objects)

1
2
3
4
5
6
7

add_all方法

session = Session()
objects = [User(name="u1"), User(name="u2"), User(name="u3")]
session.add_all(objects)
session.commit()

1
2
3
4
5

更新

user2 = session.query(User).filter_by(name='赵六').first()
user2.gender = "男"
session.commit()
1
2
3

先查找到User实例，然后修改实例属性，最后调用session的commit修改数据库。

删除

user3 = session.query(User).filter_by(name='赵六').first()
session.delete(user3)
session.commit()
1
2
3

多表查询

Session = sessionmaker(bind=engine)
session = Session()
query = session.query(User, Address).join(Address, User.id == Address.user_id)
for user in query.all():
    print(user[0].name)
    print(user[1].add)
1
2
3
4
5
6

这里用了join方法。

参考

http://sunnyingit.github.io/book/section_python/SQLalchemy-session.html