1. 程式人生 > >chapter13.2、SQLAlchemy

chapter13.2、SQLAlchemy

 

ORM

ORM,物件關係對映,物件和關係之間的對映,使用面向物件的方式來操作資料庫。 關係模型和Python物件之間的對映
  • table => class ,表對映為類
  • row => object ,行對映為例項
  • column => property ,欄位對映為屬性
  SQLAlchemy 是一個ORM框架   安裝 pip install sqlalchemy 文件  https://docs.sqlalchemy.org/en/latest/   檢視版本 import sqlalchemy print(sqlalchemy.__version__)   開發 內部使用了 連線池
  建立連線 交給引擎來做 格式為: dialect+driver://username:[email protected]:port/database   mysqldb的連線 格式 : mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname> 程式碼 : engine = sqlalchemy.create_engine("mysql+mysqldb://wayne:[email protected]
:3306/magedu")   pymysql的連線 格式 : mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>] engine = sqlalchemy.create_engine("mysql+pymysql://wayne:[email protected]:3306/magedu") engine = sqlalchemy.create_engine("mysql+pymysql://wayne:[email protected]
:3306/magedu",echo=True)   echo=True : 引擎是否列印執行的語句,除錯的時候開啟很方便。 懶連線: 建立引擎並不會馬上連線資料庫,直到讓資料庫執行任務時才連線。從連線池分配一個連線。   Declare a Mapping建立對映 建立基類
from sqlalchemy.ext.declarative import declarative_base
# 建立基類,便於實體類繼承。SQLAlchemy大量使用了超程式設計
Base = declarative_base()

 

建立實體類
# 建立實體類
class Student(Base):
    # 指定表名
    __tablename__ = 'student'
    # 定義類屬性對應欄位
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(64), nullable=False)
    age = Column(Integer)
    # 第一引數是欄位名,如果和屬性名不一致,一定要指定
    # age = Column('age', Integer)
    def __repr__(self):
        return "{} id={} name={} age={}".format(
            self.__class__.__name__, self.id, self.name, self.age)

 

__tablename__ 指定表名 Column類指定對應的欄位,必須指定   例項化
s = Student(name='tom')
print(s.name)
s.age = 20
print(s.age)

 

建立表 可以使用SQLAlchemy來建立、刪除表
# 刪除繼承自Base的所有表
Base.metadata.drop_all(engine)
# 建立繼承自Base的所有表
Base.metadata.create_all(engine)

metadata 記錄的是繼承自Base的表,

生產環境很少這樣建立表,都是系統上線的時候由指令碼生成。 生成環境很少刪除表, 寧可廢棄都不能刪除。  

建立會話session

在一個會話中操作資料庫,會話建立在連線上,連線被引擎管理。 當第一次使用資料庫時,從引擎維護的連線池中獲取一個連線使用。
# 建立session
Session = sessionmaker(bind=engine) # 返回類
session = Session() # 例項化
# 依然在第一次使用時連線資料庫

session物件執行緒不安全。所以不同執行緒應該使用不用的session物件。

Session類和engine有一個就行了。    
import sqlalchemy
from sqlalchemy import create_engine,Column,String,Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

USER = "root"
PWD = "root"
HOST = "172.16.34.34"
PORT = "3306"
DB = "test"

conn_str = 'mysql+pymysql://{}:{}@{}:{}/{}'.format(
    USER,PWD,HOST,PORT,DB
)
engine = create_engine(conn_str, echo=True)

Base = declarative_base()

# 建立實體類
class Student(Base):
    # 指定表名,必須指定
    __tablename__ = 'student'
        # 定義類屬性對應欄位
    id = Column(Integer,autoincrement=True,primary_key=True)
    name = Column(String(64),nullable=False)
    age = Column("age",Integer,nullable=False)
    gen = Column("gender",String(20))
        # 第一引數是欄位名,如果和屬性名不一致,一定要指定
        # age = Column('age', Integer)
    def __repr__(self):
        return "{} id={} name={} age={} gender={}".format(
            self.__class__.__name__, self.id, self.name, self.age, self.gen)

print(Student)
print(repr(Student.__table__))

s = Student(name='tom',age=20)
print(s.name)
s.gender = "M"
print(s.gender)

# Base.metadata.drop_all(bind=engine)
# Base.metadata.create_all(bind=engine) #  metadata 記錄的是繼承自Base的表,刪除也是

Session = sessionmaker(bind=engine) # 不做操作不會處理連線
session =Session() # 建立session物件,也不會在此時連線資料庫

 

 

CRUD操作

add():增加一個物件
add_all():可迭代物件,元素是物件

session.add(s) #新增一次物件s,未提交
print(s)
session.commit() #提交一次物件s
print(s) 

try: 
    session.add_all([s])  
    print(s) 
    session.commit() #在同一個執行緒內,再次提交同一個未修改過的物件不會執行
except: 
    session.rollback() # 使用要加日誌記錄
    raise

add_all()方法不會提交成功的,不是因為它不對,而是s,s成功提交後,s的主鍵就有了值,所以,只要s沒有修改過,就認為沒有改動。如下,s變化了,就可以提交修改了。

s.name = 'jerry' # 修改
session.add_all([s])

s主鍵沒有值,就是新增;主鍵有值,就是找到主鍵對應的記錄修改。

簡單查詢
使用query()方法,返回一個Query物件

 
students = session.query(Student) # 無條件,相當於select * from student
for student in students:
    print(student)
print('~~~~~~~~~~~~~')
student = session.query(Student).get(3) # 通過主鍵查詢,相當於select * from student where id=3
print(student)

query方法將實體類傳入,返回類的物件可迭代物件,這時候並不查詢。迭代它就執行SQL來查詢資料庫,封裝資料到指定類的例項。
get方法使用主鍵查詢,返回一條傳入類的一個例項。物件不存在,返回None

 改

先查後改

student = session.query(Student).get(3)
print(student)
student.name = 'sam'
student.age = 30
print(student)
session.add(student)
session.commit()

先查後刪  嘗試執行以下程式碼,會發現產生一個異常,未持久的異常
try:
    student = Student(id=2, name="serry", age=10)
    session.delete(student)
    session.commit()
except Exception as e:
    session.rollback()
    print('~~~~~~~~')
    print(e)
Instance '<Student at 0x26edf10b438>' is not persisted

 資料庫查詢資料慢且忙,最好在容器中就記錄下來,不要重複查詢相同的資料

 狀態

每一個實體,都有一個狀態屬性_sa_instance_state,其型別是sqlalchemy.orm.state.InstanceState,可以使用sqlalchemy.inspect(entity)函式檢視狀態。
常見的狀態值有transient、pending、persistent、deleted、detached。

 
狀態 說明
transient 實體類尚未加入到session中,同時並沒有儲存到資料庫中
pending transient的實體被add()到session中,狀態切換到pending,但它還沒有flush到資料庫中
persistent

session中的實體物件對應著資料庫中的真實記錄。pending狀態在提交成功後可以變成persistent狀態,或者查詢成功返回的實體也是persistent狀態

deleted

實體被刪除且已經flush但未commit完成。事務提交成功了,實體變成detached,事務失敗,
返回persistent狀態

detached 刪除成功的實體進入這個狀態
           

 

 

新建一個實體,狀態是transient臨時的。

一旦add()後從transient變成pending狀態。

成功commit()後從pending變成persistent狀態。

成功查詢返回的實體物件,也是persistent狀態。

persistent狀態的實體,修改依然是persistent狀態。

persistent狀態的實體,刪除後,flush後但沒有commit,就變成deteled狀態,成功提交,變為detached狀態,提交失敗,還原到persistent狀態。flush方法,主動把改變應用到資料庫中去。

刪除、修改操作,需要對應一個真實的記錄,所以要求實體物件是persistent狀態。

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


USER = "root"
PWD = "root"
HOST = "172.16.34.34"
PORT = "3306"
DB = "test"

conn_str = 'mysql+pymysql://{}:{}@{}:{}/{}'.format(
    USER,PWD,HOST,PORT,DB
)
engine = create_engine(conn_str, echo=True)

Base = declarative_base()

# 建立實體類
class Student(Base):
    # 指定表名,必須指定
    __tablename__ = 'student'
        # 定義類屬性對應欄位
    id = Column(Integer,autoincrement=True,primary_key=True)
    name = Column(String(64),nullable=False)
    age = Column("age",Integer,nullable=False)
    gen = Column("gender",String(20))
        # 第一引數是欄位名,如果和屬性名不一致,一定要指定
        # age = Column('age', Integer)
    def __repr__(self):
        return "{} id={} name={} age={} gender={}".format(
            self.__class__.__name__, self.id, self.name, self.age, self.gen)

print(Student)
print(repr(Student.__table__))

Session = sessionmaker(bind=engine)
session =Session()

from sqlalchemy.orm.state import InstanceState

def getstate(entity,i):
    insp = sqlalchemy.inspect(entity)
    state = "session={}, attached={},\ntransient={},persistent={},\npending={},deleted={}.detached={}".format(
        insp.session_id,
        insp._attached,
        insp.transient,
        insp.persistent,
        insp.pending,
        insp.deleted,
        insp.detached
    )
    print(i,state)
    print(insp.key)
    print("_"*30)

# student = session.query(Student).get(3)
# getstate(student,1)

try:
    student = Student(name="Tony", age=30)
    getstate(student, 2) # transit
    student = Student(name="sammy", age=30)
    getstate(student, 3) # transit
    session.add(student) # add後變成pending
    getstate(student, 4) # pending
    # session.delete(student) # 刪除的前提是persistent,否則拋異常
    # getstate(student, 5)
    session.commit()
    getstate(student, 6) # persistent
    session.delete(student) # 刪除的前提是persistent,否則拋異常
    getstate(student, 7)
    session.flush()
    getstate(student,8)
    session.commit()
    getstate(student, 9)
except Exception as e:
    session.rollback()
    print('~~~~~~~~')
    print(e)

返回結果

2 session=None, attached=False,
transient=True,persistent=False,
pending=False,deleted=False.detached=False
None
______________________________
3 session=None, attached=False,
transient=True,persistent=False,
pending=False,deleted=False.detached=False
None
______________________________
4 session=1, attached=True,
transient=False,persistent=False,
pending=True,deleted=False.detached=False
None
______________________________
2018-11-12 18:04:27,682 INFO sqlalchemy.engine.base.Engine SHOW VARIABLES LIKE 'sql_mode'
2018-11-12 18:04:27,682 INFO sqlalchemy.engine.base.Engine {}
2018-11-12 18:04:27,683 INFO sqlalchemy.engine.base.Engine SHOW VARIABLES LIKE 'lower_case_table_names'
2018-11-12 18:04:27,683 INFO sqlalchemy.engine.base.Engine {}
2018-11-12 18:04:27,685 INFO sqlalchemy.engine.base.Engine SELECT DATABASE()
2018-11-12 18:04:27,685 INFO sqlalchemy.engine.base.Engine {}
2018-11-12 18:04:27,686 INFO sqlalchemy.engine.base.Engine show collation where `Charset` = 'utf8mb4' and `Collation` = 'utf8mb4_bin'
2018-11-12 18:04:27,686 INFO sqlalchemy.engine.base.Engine {}
2018-11-12 18:04:27,687 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS CHAR(60)) AS anon_1
2018-11-12 18:04:27,687 INFO sqlalchemy.engine.base.Engine {}
2018-11-12 18:04:27,688 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS CHAR(60)) AS anon_1
2018-11-12 18:04:27,688 INFO sqlalchemy.engine.base.Engine {}
2018-11-12 18:04:27,689 INFO sqlalchemy.engine.base.Engine SELECT CAST('test collated returns' AS CHAR CHARACTER SET utf8mb4) COLLATE utf8mb4_bin AS anon_1
2018-11-12 18:04:27,689 INFO sqlalchemy.engine.base.Engine {}
2018-11-12 18:04:27,690 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2018-11-12 18:04:27,691 INFO sqlalchemy.engine.base.Engine INSERT INTO student (name, age, gender) VALUES (%(name)s, %(age)s, %(gender)s)
2018-11-12 18:04:27,691 INFO sqlalchemy.engine.base.Engine {'name': 'sammy', 'age': 30, 'gender': None}
2018-11-12 18:04:27,692 INFO sqlalchemy.engine.base.Engine COMMIT
6 session=1, attached=True,
transient=False,persistent=True,
pending=False,deleted=False.detached=False
(<class '__main__.Student'>, (11,), None)
______________________________
7 session=1, attached=True,
transient=False,persistent=True,
pending=False,deleted=False.detached=False
(<class '__main__.Student'>, (11,), None)
______________________________
2018-11-12 18:04:27,693 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2018-11-12 18:04:27,694 INFO sqlalchemy.engine.base.Engine SELECT student.gender AS student_gender, student.id AS student_id, student.name AS student_name, student.age AS student_age 
FROM student 
WHERE student.id = %(param_1)s
2018-11-12 18:04:27,694 INFO sqlalchemy.engine.base.Engine {'param_1': 11}
2018-11-12 18:04:27,695 INFO sqlalchemy.engine.base.Engine DELETE FROM student WHERE student.id = %(id)s
2018-11-12 18:04:27,695 INFO sqlalchemy.engine.base.Engine {'id': 11}
8 session=1, attached=True,
transient=False,persistent=False,
pending=False,deleted=True.detached=False
(<class '__main__.Student'>, (11,), None)
______________________________
2018-11-12 18:04:27,696 INFO sqlalchemy.engine.base.Engine COMMIT
9 session=None, attached=False,
transient=False,persistent=False,
pending=False,deleted=False.detached=True
(<class '__main__.Student'>, (11,), None)
______________________________

 

複雜查詢

實體類

import enum
class GenderEnum(enum.Enum):
    M = "M"
    F = "F"

class Employee(Base):
    __tablename__ = "employees"
    emp_no = Column(Integer, primary_key=True)
    birth_date = Column(Date, nullable=False)
    first_name = Column(String(14), nullable=False)
    last_name = Column(String(16), nullable=False)
    gender = Column(Enum(GenderEnum), nullable=False)
    hire_date = Column(Date, nullable=False)

    def __repr__(self):
        return "{} no={} name={} {} gender={}".format(
            self.__class__.__name__, self.emp_no, self.first_name, self.last_name,
            self.gender.value
        )

# 返回的迭代器,檢視內容
def show(emps): for x in emps: print(x)

以下語句為條件

最簡單的查詢:

emps = session.query(Employee).filter(Employee.emp_no > 10015)
show(emps)

與,或,非

and條件可以使用兩個filter實現,也可以使用and_,也可使用運算子過載 &

emps = session.query(Employee).filter(Employee.emp_no > 10015).filter(Employee.gender ==GenderEnum.F)
show(emps)

emps = session.query(Employee).filter(and_(Employee.emp_no > 10015, Employee.gender ==GenderEnum.M))
show(emps)
#運算子過載注意表示式要加括號
emps = session.query(Employee).filter((Employee.emp_no > 10015) & (Employee.gender == GenderEnum.M))
show(emps)

or 條件可以使用or_ 或者運算子 | 

emps = session.query(Employee).filter(or_(Employee.emp_no > 10018, Employee.emp_no < 10003))
show(emps)
#加括號
emps = session.query(Employee).filter((Employee.emp_no > 10018) | (Employee.emp_no < 10003))
show(emps)

not 條件使用not_ 或者運算子 ~

emps = session.query(Employee).filter(not_(Employee.emp_no < 10018))
show(emps)
#加括號
emps = session.query(Employee).filter(~(Employee.emp_no < 10018))
show(emps)

總之,與或非的運算子&、|、~,一定要在表示式上加上括號

in

emplist = [10010, 10015, 10018]
emps = session.query(Employee).filter(Employee.emp_no.in_(emplist))
show(emps)

not in 

emplist = [10010, 10015, 10018]
emps = session.query(Employee).filter(~Employee.emp_no.in_(emplist))
show(emps)

emps = session.query(Employee).filter(Employee.emp_no.notin_(emplist))
show(emps)

like  ,少用

emps = session.query(Employee).filter(Employee.last_name.like('P%'))
show(emps)

not like ,少用

emps = session.query(Employee).filter(Employee.last_name.notlike('P%'))

ilike 忽略大小寫

emps = session.query(Employee).filter(Employee.last_name.ilike('P%'))
show(emps)

 

排序

升序

emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.emp_no)
emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.emp_no.asc())
show(emps)

降序

emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.emp_no.desc())
show(emps)

多列排序

emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.last_name).order_by(Employee.emp_no.desc())
show(emps)
emps = session.query(Employee).filter(Employee.emp_no > 10010).order_by(Employee.last_name.desc(), Employee.emp_no.asc())
show(emps)

 

分頁

emps = session.query(Employee).limit(4)
show(emps)
emps
= session.query(Employee).limit(4).offset(18) show(emps)

 

消費者方法

總共的行數

emps = session.query(Employee)
print(len(list(emps))) # 返回大量的結果集,然後轉換list
print(emps.count()) # 聚合函式count(*)的查詢

取所有資料

emps = session.query(Employee)
print(emps.all()) # 返回列表,查不到返回空列表

取首行  :  first方法本質上就是limit語句

emps = session.query(Employee)
print(emps.limit(1).one()) #返回一行
print(emps.one()) #如果查詢結果是多行拋異常

刪除

# 刪除 delete by query
session.query(Employee).filter(Employee.emp_no > 10018).delete()
#session.commit() # 提交則刪除

 

聚合,分組

聚合

from sqlalchemy import func
query = session.query(func.count(Employee.emp_no))
print(query.one()) # 只能有一行結果
print(query.scalar()) # 取one()返回元組的第一個元素

max,min,avg

print(session.query(func.max(Employee.emp_no)).scalar())
print(session.query(func.min(Employee.emp_no)).scalar())
print(session.query(func.avg(Employee.emp_no)).scalar())

分組

print(session.query(Employee.gender,func.count(Employee.emp_no)).group_by(Employee.gender).all())

 

關聯查詢

有兩張表,其中的物件多對多關聯,就要建立第三張表。

如果是一對多關聯,就在多的一端建立外來鍵。

有一個員工,即屬於A部門,又屬於B部門,同時每個部門都有許多員工,這就是多對多

先把這些表的Model類和欄位屬性建立起來。

class Employee(Base):
# 指定表名
    __tablename__ = 'employees'
    # 定義屬性對應欄位
    emp_no = Column(Integer, primary_key=True)
    birth_date = Column(Date, nullable=False)
    first_name = Column(String(14), nullable=False)
    last_name = Column(String(16), nullable=False)
    gender = Column(Enum(GenderEnum), nullable=False)
    hire_date = Column(Date, nullable=False)
    # 第一引數是欄位名,如果和屬性名不一致,一定要指定
    # age = Column('age', Integer)

    def __repr__(self):
        return "{} no={} name={} {} gender={}".format(
        self.__class__.__name__, self.emp_no, self.first_name, self.last_name,
        self.gender.value
        )

class Department(Base):
    __tablename__ = 'departments'
    dept_no = Column(String(4), primary_key=True)
    dept_name = Column(String(40), nullable=False, unique=True)
    def __repr__(self):
        return "{} no={} name={}".format(type(self).__name__, self.dept_no, self.dept_name)

class Dept_emp(Base):
    __tablename__ = "dept_emp"
    emp_no = Column(Integer, ForeignKey('employees.emp_no',    ondelete='CASCADE'), primary_key=True)
    dept_no = Column(String(4), ForeignKey('departments.dept_no', ondelete='CASCADE'), primary_key=True)
    from_date = Column(Date, nullable=False)
    to_date = Column(Date, nullable=False)
    def __repr__(self):
        return "{} empno={} deptno={}".format(type(self).__name__, self.emp_no, self.dept_no)

 

查詢10010員工的所在部門和標號資訊

results = session.query(Employee,Dept_emp).filter((Employee.emp_no == Dept_emp.emp_no) & (Employee.emp_no == 10010)).all()
show(results)

這種方式產生隱式連線的語句

使用join

results = session.query(Employee).join(Dept_emp).filter(Employee.emp_no == 10010).all()
results = session.query(Employee).join(Dept_emp, Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010).all() print(results)

返回的都只有一行資料。它們生成的SQL語句是一樣的,執行該SQL語句返回確實是2行記錄,可以Python中的返回值列表中只有一個元素。

原因在於 query(Employee) 這個只能返回一個實體物件中去,為了解決這個問題,需要修改實體類Employee,增加屬性用來存放部門資訊

 

sqlalchemy.orm.relationship(實體類名字串)

from sqlachemy import relationship
class Employee(Base):
# 指定表名
    __tablename__ = 'employees'
    # 定義屬性對應欄位
    emp_no = Column(Integer, primary_key=True)
    birth_date = Column(Date, nullable=False)
    first_name = Column(String(14), nullable=False)
    last_name = Column(String(16), nullable=False)
    gender = Column(Enum(GenderEnum), nullable=False)
    hire_date = Column(Date, nullable=False)
    # 第一引數是欄位名,如果和屬性名不一致,一定要指定
    # age = Column('age', Integer)

    departments = relationship("Dept_emp") 

    def __repr__(self):
        return "{} no={} name={} {} gender={} depts={}".format(
        self.__class__.__name__, self.emp_no, self.first_name, self.last_name,
        self.gender.value,self.departments
        )

查詢

# 第一種
# results = session.query(Employee).join(Dept_emp).filter(Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010)
# 第二種
# results = session.query(Employee).join(Dept_emp, Employee.emp_no == Dept_emp.emp_no).filter(Employee.emp_no == 10010)
# 第三種
results = session.query(Employee).join(Dept_emp, (Employee.emp_no == Dept_emp.emp_no) & (Employee.emp_no == 10010))
show(results.all()) 

第一種方法join(Dept_emp)中沒有等值條件,會自動生成一個等值條件,如果後面有filter,哪怕是filter(Employee.emp_no == Dept_emp.emp_no),這個條件會在where中出現。第一種這種自動增加join的等值條件的方式不好,不要這麼寫
第二種方法在join中增加等值條件,阻止了自動的等值條件的生成。這種方式推薦
第三種方法就是第二種,這種方式也可以

只要不訪問departments屬性,就不會查dept_emp這張表。

總結
在開發中,一般都會採用ORM框架,這樣就可以使用物件操作表了。

定義表對映的類,使用Column的描述器定義類屬性,使用ForeignKey來定義外來鍵約束。

如果在一個物件中,想檢視其它表對應的物件的內容,就要使用relationship來定義關係。

是否使用外來鍵約束?
1、力挺派
  能使資料保證完整性一致性
2、嫌棄派
  開發難度增加,大資料的時候影響插入、修改、刪除的效率。
  在業務層保證資料的一致性。