【python】-- SQLAlchemy操作MySQL

ORM、SQLAchemy

orm英文全稱object relational mapping,就是對象映射關系程序，簡單來說就是類似python這種面向對象的程序來說一切皆對象，但是使用的數據庫卻都是關系型的，為了保證一致的使用習慣，通過orm將編程語言的對象模型和數據庫的關系模型建立映射關系，這樣在使用編程語言對數據庫進行操作的時候可以直接使用編程語言的對象模型進行操作就可以了，而不用直接使用sql語言。

SQLAlchemy是Python編程語言下的一款ORM框架，該框架建立在數據庫API之上，使用關系對象映射進行數據庫操作，簡言之便是：將對象轉換成SQL，然后使用數據API執行SQL並獲取執行結果。SQLAlchemy本身無法操作數據庫，其必須以來pymsql等第三方插件，Dialect用於和數據API進行交流，根據配置文件的不同調用不同的數據庫API，從而實現對數據庫的操作：

MySQL-Python
    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
    
pymysql
    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
    
MySQL-Connector
    mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
    
cx_Oracle
    oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]

更多數據庫ORM鏈接方式資料:查看

 

1、安裝

pip3 install SQLAlchemy  

 

2、創建sql表

2.1、原本創建sql表命令

CREATE TABLE user (
    id INTEGER NOT NULL AUTO_INCREMENT,
    name VARCHAR(32),
    password VARCHAR(64),
    PRIMARY KEY (id)
)

2.2、使用sqlalchemy創建表

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


connect = create_engine("mysql+pymysql://root:123456@localhost:3306/mysql",
                        encoding="utf-8",
                        echo=True)  # 連接數據庫，echo=True =>把所有的信息都打印出來


Base = declarative_base()  # 生成ORM基類


class User(Base):
    __tablename__ = "hello_word"  # 表名
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    password = Column(String(64))

Base.metadata.create_all(connect)   # 創建表結構

2.3、sqlaicheym原始創建表的方式（2.2創建表的方式就是基於2.3的再封裝，所以2.3基本這種方式基本不使用）

from sqlalchemy import Table, MetaData, Column, Integer, String, ForeignKey
from sqlalchemy.orm import mapper
 
metadata = MetaData()
 
# 表結構
user = Table('user', metadata,
            Column('id', Integer, primary_key=True),
            Column('name', String(50)),
            Column('fullname', String(50)),
            Column('password', String(12))
        ) 
 
class User(object):
# User類的靜態變量
    def __init__(self, name, fullname, password):
        self.name = name
        self.fullname = fullname
        self.password = password
 
mapper(User, user) # mapper 函數使表結構和User類的靜態變量關聯起來

 

 

最基本的表建立好了之后，再建立與數據的Session會話連接，就可以進行增刪查改等操作：

session_class = sessionmaker(bind=connect)  # 創建與數據庫的會話session class ,這里返回給session的是個class,不是實例
session = session_class()   # 生成session實例

3、增

#新增一條數據
#原生sql：insert  into mysql.hello_word(name,password) values("test2","1234")；
obj = User(name="test", password="1234")  #生成你要創建的數據對象
session.add(obj) #把要創建的數據對象添加到這個session里， 一會統一創建
session.commit() #統一提交，創建數據，在此之前數據庫是不會有新增數據的

#新增多條數據
#原生sql:insert  into mysql.hello_word(name,password) values("test2","1234"),("test3","123");
obj = User(name="test", password="1234")
obj1 = User(name="test", password="1234")
session.add_all([obj,obj1])
session.commit()

#回滾，在session.add()之后，在session.commit()之前，想把添加至session緩存中的數據清除，使用rollback()函數回滾即可
Session.rollback()

4、刪

#原生sql:mysql.hello_word where id > 5;
session.query(User).filter(User.id > 5).delete() #通過session查詢User類，然后過濾出id>5的進行刪除
session.commit() #提交

5、改

#①第一種方式
data = Session.query(User).filter_by(name="test1").first()  #獲取數據
data.name = "test"  #修改數據
Session.commit()   #提交

#②第二種方式，通過查找表，過濾條件，然后更新對應參數
session.query(User).filter(User.id > 15).update({"name": "test"})
session.query(User).filter(User.id == 18).update({User.name: "hello"}, synchronize_session=False)
session.query(User).filter_by(name="test1").update({User.password: User.name}, synchronize_session="evaluate")
session.commit()

#③synchronize_session解釋，用於query在進行delete or update操作時，對session的同步策略：
#1、synchronize_session=False，不對session進行同步，直接進行delete or update操作。
#2、synchronize_session="evaluate"，在delete or update操作之前，用query中的條件直接對session的identity_map中的objects進行eval操作，將符合條件的記錄下來，　在delete or update操作之后，將符合條件的記錄刪除或更新。

6、查

6.1、幾種查詢方式的使用

#原生sql:select * from mysql.hello_word;
ret = session.query(User).all()  #查詢所有
#也可以這樣寫：
ret = Session.query(User.name,User.id).all() 

#原生slq:select name,password from mysql.hello_word;
ret = session.query(User.name, User.extra).all()  #只查詢name和extra字段所以所有數據

#原生sql:select * from mysql.hello_word where name="test1";
ret = session.query(User).filter_by(name='test1').all() #查詢name='alex'的所有數據
ret = session.query(User).filter_by(name='test1').first()#查詢name='alex'的第一條數據

#查詢id>5的name字段內容，且以id大小排序
#原生sql；select name from mysql.hello_word where id >5 order by id；
ret = session.query(User).filter(text("id>:value and name=:name")).params(value=5, name='test2').order_by(User.id).all() 

#根據原生sql查詢數據
ret = session.query(User).from_statement(text("SELECT * FROM hello_word where name=:name")).params(name='test1').all()

6.2、filter和filter_by使用的區別

#filter用於sql表達式查詢過濾，如>，<, ==,等表達式
session.query(MyClass).filter(MyClass.name == 'some name')
#filter_by用於關鍵字查詢過濾，如id=value,name=value
session.query(MyClass).filter_by(name = 'some name')

6.3、重構__repr__方法，將5.1 中ret內存對象按__repr__方法中定義的格式進行打印顯示

class User(Base):
    __tablename__ = "hello_word"  # 表名
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    password = Column(String(64))

    def __repr__(self):    # 使返回的內存對象變的可讀
        return "<id:{0} name:{1} password:{2}>".format(self.id, self.name, self.password)

#Base.metadata.create_all(connect)  # 創建標結構

session_class = sessionmaker(bind=connect)  # 創建與數據庫的會話session class ,這里返回給session的是個class,不是實例
session = session_class()   # 生成session實例

user = session.query(User).all()  #查詢全部
print(user) 

#輸出
[<id:1 name:test1 password:1234>, <id:2 name:test1 password:1234>, <id:8 name:test2 password:1234>, <id:9 name:test3 password:123>, <id:10 name:test4 password:123>, <id:11 name:test5 password:123>, <id:12 name:test2 password:1234>, <id:13 name:test3 password:123>, <id:14 name:test4 password:123>, <id:15 name:test5 password:123>, <id:16 name:test2 password:1234>, <id:17 name:test3 password:123>, <id:18 name:test4 password:123>, <id:19 name:test5 password:123>]

7、其他操作

#多條件查詢
#原生sql:select * from mysql.hello_word where id >2 and id < 19
data = session.query(User).filter(Use.id>2).filter(Use.id<19).all()


#通配符
#原生sql:select * from mysql.hello_word where name like "test%" #"test_"、%test%
data = session.query(User).filter(User.name.like('test%')).all() #匹配以test開頭，而后跟多個字符
data = session.query(User).filter(User.name.like('test_')).all() #匹配以test開頭，而后跟一個字符
data = session.query(User).filter(~User.name.like('e%')).all() #加~后，忽略like(),直接匹配所有
#原生sql select count(name) from mysql.hello_word where name like "%test%"
data = session.query(User).filter(User.name.like("%qigao%")).count() # 模糊匹配並計數


#分組
from sqlalchemy import func #導入func 進行函數操作
#原生sql:select count(name),name from mysql.hello_word group by name
data =session.query(func.count(User.name),User.name).group_by(User.name).all()  #根據User.name分組
#原生sql:select max(id),sum(id),min(id) from mysql.hello_word group by name  #根據name 分組
data =session.query(func.max(User.id),func.sum(User.id),func.min(User.id)).group_by(User.name).all()
#原生sql:select max(id),sum(id),min(id) from mysql.hello_word group by name having min(id > 2) # 根據name分組且id>2
data = session.query(func.max(User.id),func.sum(User.id),func.min(User.id)).group_by(User.name).having(func.min(User.id) >2).all()


#排序
#原生sql:select * from mysql.hello_word  order by id asc
data = session.query(User).order_by(User.id.asc()).all() #將所有數據根據 “列” 從小到大排列
#原生sql:select * from mysql.hello_word  order by id desc, id asc 
data = session.query(User).order_by(User.id.desc(), User.id.asc()).all()#將所有數據根據 “列1” 從大到小排列,如果相同則按照“列2”由小到大排列


#條件表達式 in、between、 and 、or
data = session.query(User).filter_by(name='test').all()
data = session.query(User).filter(User.id > 1, Users.name == 'test').all()
data = session.query(User).filter(User.id.between(1, 3), Users.name == 'test').all()
data = session.query(User).filter(User.id.in_([1,3,4])).all()
data = session.query(User).filter(~User.id.in_([1,3,4])).all()
data = session.query(User).filter(Users.id.in_(session.query(User.id).filter_by(name='test'))).all()

from sqlalchemy import and_, or_
data = session.query(User).filter(and_(User.id > 3, Users.name == 'test')).all()
data = session.query(User).filter(or_(User.id < 2, Users.name == 'test')).all()
data = session.query(User).filter(or_(User.id < 2,and_(User.name == 'test',User.id > 3),User.password != "")).all()