前言：
- 字段聲明類型中，最右邊的是數據庫中對應的字段，我們依然可以使用，其左邊的的 SQLAchemy 則是其自身封裝的自定義類型。
- 本篇不會講太多的理論知識，因為這個實用性更強，所以通篇全部都是案例，每個案例都會輸出對應的 sql , 這樣你也能更清晰的明白自己寫出的代碼最終都轉化成什么樣的 sql 了。
- 本篇的最后一個案例是 upsert “存在則更新，不存在則插入”的高級用法。
- 本篇中的所有案例都親測可用
案例一： 創建一個自定義類
數據庫結構
CREATE TABLE `student` ( `id` int(2) NOT NULL AUTO_INCREMENT, `name` char(20) NOT NULL, `code` char(64) NOT NULL, `sex` char(4) NOT NULL, PRIMARY KEY (`id`) USING BTREE ) 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
ps: 下面的例子全部是依據這
創建自定義類，后面的所有操作的 session，都根據此案例的 session 來操作的。
from sqlalchemy import create_engine, Column, INT, VARCHAR
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 創建基類，返回一個定制的metaclass 類
Base = declarative_base()
# 自定義類
class Student(Base):
# 表名
__tablename__ = 'student'
# 字段映射
id = Column('id', INT, primary_key=True)
name = Column('name', VARCHAR)
code = Column('code', VARCHAR)
sex = Column('sex', VARCHAR)
def to_dict(self):
"""
將查詢的結果轉化為字典類型
Student 對象的內容如下 {'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x10174c898>, 'sex': 'nan', 'name': 'ygh', 'code': 'AU', 'school': 'hua'}
獲取其值剔除 "_sa_instance_state 即可。但不能在self.__dict__上直接刪除”_sa_instance_state” 這個值是公用的。
:return:
"""
return {k: v for k, v in self.__dict__.items() if k != "_sa_instance_state”}
# 創建引擎 , echo=True ,表示需要開啟 sql 打印，調試的以后特別好用
engine =create_engine("mysql+mysqldb://root:123qwe@192.168.1.254:3306/yinguohai", pool_size=2, max_overflow=0, echo=True
# 創建會話對象，用於操作數據庫 Session = sessionmaker(bind=engine) session = Session()
案例二： 全部查詢
result = session.query(Student).all()
for i in result:
# i是一個Student對象，所以可以使用其 to_dict() 去格式化其對象的值
if isinstance(i, Student):
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
案例三： 部分字段查詢
result = session.query(Student.id, Student.name).all()
for i in result:
# 此時返回的是一個tuple ,而不是一個Student對象
print(i)
-------------------------結果------------------------- (1, 'Bob') (2, 'Bob') (3, 'Hua') (4, 'Lan') (5, 'Hua') (6, 'Smith') (7, 'Bob') (8, 'Smith') (9, 'Hub') (10, 'Hip')
對應Sql:
SELECT
student.id AS student_id,
student.NAME AS student_name
FROM
student
案例四：多條件查詢, or_ , and_
或， 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'}
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'}
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'}
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 ， slice
方式一 ,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'}
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 )
案例九：統計，count()
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
案例十：去重,distinct()
result = session.query(Student.name).distinct(Student.name).all()
-------------------------結果------------------------- ('Bob',) ('Hua',) ('Lan',) ('Smith',) ('Hub',) ('Hip',)
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)
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() , add_all()
方式一，單條插入,add( )
result = session.add(Student(name="Bob", code="AU", sex="boy")) print(result) #事務需要提交才能生效，有別與查詢 session.commit()
-------------------------結果-------------------------
None
Sql:
BEGIN INSERT INTO student (name, code, sex) VALUES (%s, %s, %s) ('Bob', 'AU', 'boy') COMMIT
方式二，批量插入, 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
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
案例十三： 更新,update()
result = session.query(Student).filter(Student.id == 1).update({Student.sex: "dddd”})
# 如果想回滾，則使用 session.rollback() 回滾即可
session.commit()
# 返回修改的記錄函數
print(result)
-------------------------結果------------------------- 1
BEGIN
UPDATE student SET sex=%s WHERE student.id = %s ('dddd', 1)
COMMIT
案例十四： 不存在則插入，存在則更新,on_duplicate_key_update()
這個屬於一種高級的用法，不過也特別簡單，看此案例你基本上就秒懂了。
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