前言:
字段声明类型中,最右边的是数据库中对应的字段,我们依然可以使用,其左边的的 SQLAchemy 则是其自身封装的自定义类型。
本篇不会讲太多的理论知识,因为这个实用性更强,所以通篇全部都是案例,每个案例都会输出对应的 sql , 这样你也能更清晰的明白自己写出的代码最终都转化成什么样的 sql 了。
本篇的最后一个案例是 upsert “存在则更新,不存在则插入”的高级用法。
案例一: 创建一个自定义类
数据库结构
1 CREATE TABLE `student` ( 2 `id` int(2) NOT NULL AUTO_INCREMENT, 3 `name` char(20) NOT NULL, 4 `code` char(64) NOT NULL, 5 `sex` char(4) NOT NULL, 6 PRIMARY KEY (`id`) USING BTREE 7 ) ENGINE=InnoDB AUTO_INCREMENT=11 DEFAULT CHARSET=utf8;
数据库中的值
Id name code sex ------------------------------------ 1 Bob AU dddd 2 Bob BR girl 3 Hua CA boy 4 Lan CN girl 5 Hua RU girl 6 Smith US boy 7 Bob AU boy 8 Smith BM girl 9 Hub BU boy 10 Hip HK boy
1 映射关系 2 from sqlalchemy import create_engine, Column, INT, VARCHAR 3 from sqlalchemy.ext.declarative import declarative_base 4 from sqlalchemy.orm import sessionmaker 5 6 7 # 创建引擎 , echo=True ,表示需要开启 sql 打印,调试的以后特别好用 8 engine = create_engine("mysql+mysqldb://root:[email protected]:3306/yinguohai", pool_size=2, max_overflow=0, echo=True) 9 10 11 # 创建基类,返回一个定制的metaclass 类 12 Base = declarative_base() 13 14 15 # 自定义类 16 class Student(Base): 17 # 表名 18 __tablename__ = 'student' 19 # 字段映射 20 id = Column('id', INT, primary_key=True) 21 name = Column('name', VARCHAR) 22 code = Column('code', VARCHAR) 23 sex = Column('sex', VARCHAR) 24 25 26 def to_dict(self): 27 """ 28 将查询的结果转化为字典类型 29 Student 对象的内容如下 {'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x10174c898>, 'sex': 'nan', 'name': 'ygh', 'code': 'AU', 'school': 'hua'} 30 获取其值剔除 "_sa_instance_state 即可。但不能在self.__dict__上直接删除”_sa_instance_state” 这个值是公用的。 31 :return: 32 """ 33 return {k: v for k, v in self.__dict__.items() if k != "_sa_instance_state”} 34 35 # 创建会话对象,用于操作数据库 36 Session = sessionmaker(bind=engine) 37 session = Session()
ps: 下面的例子全部是依据这个来
案例二: 全部查询
1 result = session.query(Student).all() 2 for i in result: 3 # i是一个Student对象,所以可以使用其 to_dict() 去格式化其对象的值 4 if isinstance(i, Student): 5 print(i.to_dict())
-------------------------结果------------------------- {'sex': 'dddd', 'name': 'Bob', 'id': 1, 'code': 'AU'} {'sex': 'girl', 'name': 'Bob', 'id': 2, 'code': 'BR'} {'sex': 'boy', 'name': 'Hua', 'id': 3, 'code': 'CA'} {'sex': 'girl', 'name': 'Lan', 'id': 4, 'code': 'CN'} {'sex': 'girl', 'name': 'Hua', 'id': 5, 'code': 'RU'} {'sex': 'boy', 'name': 'Smith', 'id': 6, 'code': 'US'} {'sex': 'boy', 'name': 'Bob', 'id': 7, 'code': 'AU'} {'sex': 'girl', 'name': 'Smith', 'id': 8, 'code': 'BM'} {'sex': 'boy', 'name': 'Hub', 'id': 9, 'code': 'BU'} {'sex': 'boy', 'name': 'Hip', 'id': 10, 'code': 'HK'}
Sql:
1 SELECT 2 student.id AS student_id, 3 student.NAME AS student_name, 4 student.CODE AS student_code, 5 student.sex AS student_sex 6 FROM 7 student
案例三: 部分字段查询
1 result = session.query(Student.id, Student.name).all() 2 for i in result: 3 # 此时返回的是一个tuple ,而不是一个Student对象 4 print(i)
-------------------------结果------------------------- (1, 'Bob') (2, 'Bob') (3, 'Hua') (4, 'Lan') (5, 'Hua') (6, 'Smith') (7, 'Bob') (8, 'Smith') (9, 'Hub') (10, 'Hip')
Sql:
1 SELECT 2 student.id AS student_id, 3 student.NAME AS student_name 4 FROM 5 student
案例四:多条件查询
或, or_
result = session.query(Student).filter(
or_(Student.name == "Bob", Student.sex != "aa")).
first()
print(result.to_dict())
-------------------------结果-------------------------
{'sex': 'dddd', 'name': 'Bob', 'id': 1, 'code': 'AU'}
Sql:
SELECT
student.id AS student_id,
student.NAME AS student_name,
student.CODE AS student_code,
student.sex AS student_sex
FROM
student
WHERE
student.NAME = % s
OR student.sex != % s ( 'Bob', 'aa', 1 )
且,and_
result = session.query(Student).filter(
and_(Student.name == "Bob" , Student.sex != "aa")).
first()
print(result.to_dict())
-------------------------结果-------------------------
{'sex': 'dddd', 'name': 'Bob', 'id': 1, 'code': 'AU'}
Sql:
SELECT
student.id AS student_id,
student.NAME AS student_name,
student.CODE AS student_code,
student.sex AS student_sex
FROM
student
WHERE
student.NAME = % s
AND student.sex != % s
LIMIT % s ( 'Bob', 'aa', 1 )
案例五:like , 模糊查询
result = session.query(Student).filter(
Student.sex.like('%bo%')).
first()
print(result.to_dict())
-------------------------结果-------------------------
{'sex': 'dddd', 'name': 'Bob', 'id': 1, 'code': 'AU'}
Sql:
SELECT
student.id AS student_id,
student.NAME AS student_name,
student.CODE AS student_code,
student.sex AS student_sex
FROM
student
WHERE
student.sex
LIKE % s
LIMIT %s ('%bo%', 1)
案例六:in_ , 范围查询
result = session.query(Student).filter(
Student.name.in_(["Bob", "Smith"])).all()
for i in result:
print(i.to_dict())
-------------------------结果-------------------------
{'code': 'AU', 'id': 1, 'name': 'Bob', 'sex': 'dddd'}
{'code': 'BR', 'id': 2, 'name': 'Bob', 'sex': 'girl'}
{'code': 'US', 'id': 6, 'name': 'Smith', 'sex': 'boy'}
{'code': 'AU', 'id': 7, 'name': 'Bob', 'sex': 'boy'}
{'code': 'BM', 'id': 8, 'name': 'Smith', 'sex': 'girl'}
Sql:
SELECT
student.id AS student_id,
student.NAME AS student_name,
student.CODE AS student_code,
student.sex AS student_sex
FROM
student
WHERE
student.NAME IN (% s, % s ) ( 'Bob', 'Smith' )
案例七:排序,asc() , desc()
#result = session.query(Student).order_by(
Student.id.desc()).all()
result = session.query(Student).order_by(
Student.id.asc()).all()
for i in result:
print(i.to_dict())
-------------------------结果-------------------------
{'sex': 'dddd', 'name': 'Bob', 'id': 1, 'code': 'AU'}
{'sex': 'girl', 'name': 'Bob', 'id': 2, 'code': 'BR'}
{'sex': 'boy', 'name': 'Hua', 'id': 3, 'code': 'CA'}
{'sex': 'girl', 'name': 'Lan', 'id': 4, 'code': 'CN'}
{'sex': 'girl', 'name': 'Hua', 'id': 5, 'code': 'RU'}
{'sex': 'boy', 'name': 'Smith', 'id': 6, 'code': 'US'}
{'sex': 'boy', 'name': 'Bob', 'id': 7, 'code': 'AU'}
{'sex': 'girl', 'name': 'Smith', 'id': 8, 'code': 'BM'}
{'sex': 'boy', 'name': 'Hub', 'id': 9, 'code': 'BU'}
{'sex': 'boy', 'name': 'Hip', 'id': 10, 'code': 'HK'}
Sql:
SELECT
student.id AS student_id,
student.NAME AS student_name,
student.CODE AS student_code,
student.sex AS student_sex
FROM
student
ORDER BY
student.id
ASC
案例八:分页
方式一 ,limit( position )
result = session.query(Student).
limit(2).all()
for i in result:
print(i.to_dict())
-------------------------结果-------------------------
{'sex': 'dddd', 'name': 'Bob', 'id': 1, 'code': 'AU'}
{'sex': 'girl', 'name': 'Bob', 'id': 2, 'code': 'BR'}
Sql:
SELECT
student.id AS student_id,
student.NAME AS student_name,
student.CODE AS student_code,
student.sex AS student_sex
FROM
student
LIMIT % s (2,)
方式二: slice(start, end)
result = session.query(Student).order_by(Student.id.asc()).
slice(2, 3).all()
for i in result:
print(i.to_dict())
-------------------------结果-------------------------
{'sex': 'boy', 'code': 'CA', 'id': 3, 'name': 'Hua'}
Sql:
SELECT
student.id AS student_id,
student.NAME AS student_name,
student.CODE AS student_code,
student.sex AS student_sex
FROM
student
ORDER BY
student.id ASC
LIMIT % s,% s ( 2, 1 )
案例九:统计
result = session.query(Student).
count()
print(result)
-------------------------结果-------------------------
10
Sql:
SELECT
count(*) AS count_1
FROM
( SELECT student.id AS student_id, student.NAME AS student_name, student.CODE AS student_code, student.sex AS student_sex FROM student ) AS anon_1
案例十:去重
result = session.query(Student.name).
distinct(Student.name).all()
-------------------------结果-------------------------
('Bob',)
('Hua',)
('Lan',)
('Smith',)
('Hub',)
('Hip',)
Sql:
SELECT
DISTINCT
student.NAME AS student_name
FROM
student
案例十 一: 联合查询,默认 inner join查询
result = session.query(Student.id, Student.code, Student.name, Country.population).
join(Country, Student.code == Country.code).all()
for i in result:
print(i)
-------------------------结果-------------------------
(1, 'AU', 'Bob', 18886000)
(2, 'BR', 'Bob', 170115000)
(3, 'CA', 'Hua', 1147000)
(4, 'CN', 'Lan', 1277558000)
(5, 'RU', 'Hua', 146934000)
(6, 'US', 'Smith', 278357000)
(7, 'AU', 'Bob', 18886000)
Sql:
SELECT
student.id AS student_id,
student.CODE AS student_code,
student.NAME AS student_name,
a_country.population AS a_country_population
FROM
student
INNER JOIN a_country ON student.CODE = a_country.CODE
案例十二:添加
方式一,单条插入,add( )
result =
session.add(Student(name="Bob", code="AU", sex="boy"))
print(result)
#事务需要提交才能生效,有别与查询
session.commit()
Sql:
BEGIN
INSERT INTO student (name, code, sex) VALUES (%s, %s, %s) ('Bob', 'AU', 'boy’)
COMMIT
-------------------------结果-------------------------
None
方式二,批量插入, add_all( )
result =
session.add_all([
Student(name="Smith", code="BM", sex="girl"),
Student(name="Hub", code="BU", sex="boy"),
Student(name="Hip", code="HK", sex="boy"),
])
session.commit()
print(result)
-------------------------结果-------------------------
None
Sql:
BEGIN
INSERT INTO student (name, code, sex) VALUES (%s, %s, %s) ('Smith', 'BM', 'girl’)
INSERT INTO student (name, code, sex) VALUES (%s, %s, %s) ('Hub', 'BU', 'boy’)
INSERT INTO student (name, code, sex) VALUES (%s, %s, %s) ('Hip', 'HK', 'boy’)
COMMIT
案例十三: 更新
result = session.query(Student).filter(Student.id == 1).
update({Student.sex: "dddd”})
# 如果想回滚,则使用 session.rollback() 回滚即可
session.commit()
# 返回修改的记录函数
print(result)
-------------------------结果-------------------------
1
Sql:
BEGIN
UPDATE student SET sex=%s WHERE student.id = %s ('dddd', 1)
COMMIT
案例十四: 不存在则插入,存在则更新
这个属于一种高级的用法,不过也特别简单,看此案例你基本上就秒懂了。
insert_smt =
insert(Student).values(id=1, name="bb", code="AA", sex="boy").
on_duplicate_key_update(sex="aaaaa",code="uuuuu")
result =
session.execute(insert_smt)
session.commit()
print(result.rowcount)
-------------------------结果-------------------------
1
注意事项:
- 需要引入 一个特别函数 , insert( ) , 它是mysql包下的。from sqlalchemy.dialects.mysql import insert
- 使用 on_duplicate_key_update( ) 这个函数进行异常处理,别用错了
- 使用execute , 执行insert( ) 函数创建的 Sql 语句即可
- 最后一定要记得 commit( ) 一下。
Sql:
BEGIN
INSERT INTO student ( id, NAME, CODE, sex )
VALUES (% s, % s, % s, % s )
ON DUPLICATE KEY UPDATE code = %s, sex = %s
(1, 'bb', 'AA', 'boy', 'uuuuu', 'aaaaa')
COMMIT