一SQLAlchemy簡介
SQLAlchemy是一個基于Python實作的ORM架構。該架構建立在 DB API之上,使用關系對象映射進行資料庫操作,簡言之便是:将類和對象轉換成SQL,然後使用資料API執行SQL并擷取執行結果。
1.與Django中models的差別
很多小夥伴說SQLAlchemy不如Django的models好用,這裡我們需要知道。
Models其實隻是配置和使用比較簡單,畢竟是Django自帶的ORM架構,但是相容性遠不如SQLAchemy,真正算得上全面的ORM架構必然是SQLAlchemy。
無論使用什麼ORM架構,其實都是為了友善不熟練資料庫使用的同學,最推薦的還是使用原生的SQL語句,也建議大家攻克SQL難關。
2.SQLAlchemy組成
![](https://img.laitimes.com/img/9ZDMuAjOiMmIsIjOiQnIsIyZuBnL5kTM0MTO4QDMx0iNzIzMyAjM4ETMzcDM5EDMy0SMzcDNxYTMvw1NwkTMwIzLcFzM3QTM2EzLcd2bsJ2Lc12bj5ycn9Gbi52YugTMwIzZtl2Lc9CX6MHc0RHaiojIsJye.png)
組成部分:
- Engine,架構的引擎
- Connection Pooling ,資料庫連接配接池
- Dialect,選擇連接配接資料庫的DB API種類
- Schema/Types,架構和類型
- SQL Exprression Language,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...]
更多:http://docs.sqlalchemy.org/en/latest/dialects/index.html
二、SQLAlchemy使用
1.執行原生sql語句
通過SQLAlchemy執行源生的sql語句
方式一:
from sqlalchemy import create_engine
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/sqlalchemy01?charset=utf8",
max_overflow=0, # 超過連接配接池大小外最多建立的連接配接
pool_size=5, # 連接配接池大小
pool_timeout=30, # 池中沒有線程最多等待的時間,否則報錯
pool_recycle=-1 # 多久之後對線程池中的線程進行一次連接配接的回收(重置)
)
def task():
conn = engine.raw_connection()
cursor = conn.cursor()
cursor.execute(
"select * from t1"
)
result = cursor.fetchall()
print(">>>",result)
cursor.close()
conn.close()
task()
方式二:
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:[email protected]:3306/sqlalchemy01", max_overflow=0, pool_size=5)
def task():
conn = engine.connect()
with conn:
cur = conn.execute(
"select * from t1"
)
result = cur.fetchall()
print(result)
task()
方式三
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:[email protected]:3306/sqlalchemy01", max_overflow=0, pool_size=5)
def task():
cur = engine.execute("select * from t1")
result = cur.fetchall()
cur.close()
print(result)
task()
2.資料表的操作
通過sqlalchemy來建立表和删除表
import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
# 建立基礎類 R關系 M映射 類
Base = declarative_base()
class Users(Base):
__tablename__ = 'users' # 指定建立的表名
# 寫字段
id = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=False)
email = Column(String(32), unique=True)
# ctime = Column(DateTime, default=datetime.datetime.now)
# extra = Column(Text, nullable=True)
__table_args__ = (
UniqueConstraint('id', 'name', name='uix_id_name'), # 設定位移限制
Index('ix_id_name', 'name', 'email'), # 設定索引
)
# 建立資料庫的引擎
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/sqlalchemy01?charset=utf8",
max_overflow=0, # 超過連接配接池大小外最多建立的連接配接
pool_size=5, # 連接配接池大小
pool_timeout=30, # 池中沒有線程最多等待的時間,否則報錯
pool_recycle=-1 # 多久之後對線程池中的線程進行一次連接配接的回收(重置)
)
# 檢索所有繼承Base的Object并在 engine 指向的資料庫中建立所有的表
Base.metadata.create_all(engine)
# 删除所有的資料庫表
Base.metadata.drop_all(engine)
1.單表建立示例
import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
# 建立基礎類 R關系 M映射 類
Base = declarative_base()
class Users(Base):
__tablename__ = 'users' # 指定建立的表名
# 寫字段
id = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=False)
email = Column(String(32), unique=True)
# ctime = Column(DateTime, default=datetime.datetime.now)
# extra = Column(Text, nullable=True)
__table_args__ = (
UniqueConstraint('id', 'name', name='uix_id_name'), # 設定位移限制
Index('ix_id_name', 'name', 'email'), # 設定索引
)
# 建立資料庫的引擎
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/sqlalchemy01?charset=utf8",
max_overflow=0, # 超過連接配接池大小外最多建立的連接配接
pool_size=5, # 連接配接池大小
pool_timeout=30, # 池中沒有線程最多等待的時間,否則報錯
pool_recycle=-1 # 多久之後對線程池中的線程進行一次連接配接的回收(重置)
)
# 檢索所有繼承Base的Object并在 engine 指向的資料庫中建立所有的表
Base.metadata.create_all(engine)
# 删除所有的資料庫表
Base.metadata.drop_all(engine)
2.一對多示例
# ########## 一對多示例 ##########
class School(Base):
__tablename__ = "school"
id = Column(Integer,primary_key=True)
name = Column(String(32),nullable=False)
class Student(Base):
__tablename__ = "student"
id = Column(Integer,primary_key=True)
name = Column(String(32),nullable=False)
school_id = Column(Integer,ForeignKey("school.id")) # 多對一關系存儲列
# 與生成表結構無關,僅用于查詢友善
school = relationship("School", backref='student')
engine = create_engine("mysql+pymysql://root:[email protected]:3306/sqlalchemy01?charset=utf8")
# 檢索所有繼承 Model 的Object 并在 engine 指向的資料庫中建立 所有的表
Model.metadata.create_all(engine)
3.多對多表結建構立
from sqlalchemy import Column,Integer,String,ForeignKey
from sqlalchemy.orm import relationship
class Girls(Model):
__tablename__ = "girl"
id = Column(Integer,primary_key=True)
name = Column(String(32),nullable=False)
# relationship
g2b = relationship("Boys",backref="b2g",secondary="hotel")
class Boys(Model):
__tablename__ = "boy"
id = Column(Integer,primary_key=True)
name = Column(String(32),nullable=False)
class Hotels(Model):
__tablename__ = "hotel"
id = Column(Integer,primary_key=True)
boy_id = Column(Integer,ForeignKey("boy.id"))
girl_id = Column(Integer,ForeignKey("girl.id"))
engine = create_engine("mysql+pymysql://root:[email protected]:3306/sqlalchemy01?charset=utf8")
# 檢索所有繼承 Model 的Object 并在 engine 指向的資料庫中建立 所有的表
Model.metadata.create_all(engine)
4.定義函數來建立和删除表
def init_db():
"""
根據類建立資料庫表
:return:
"""
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/s6?charset=utf8",
max_overflow=0, # 超過連接配接池大小外最多建立的連接配接
pool_size=5, # 連接配接池大小
pool_timeout=30, # 池中沒有線程最多等待的時間,否則報錯
pool_recycle=-1 # 多久之後對線程池中的線程進行一次連接配接的回收(重置)
)
Base.metadata.create_all(engine)
def drop_db():
"""
根據類删除資料庫表
:return:
"""
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/s6?charset=utf8",
max_overflow=0, # 超過連接配接池大小外最多建立的連接配接
pool_size=5, # 連接配接池大小
pool_timeout=30, # 池中沒有線程最多等待的時間,否則報錯
pool_recycle=-1 # 多久之後對線程池中的線程進行一次連接配接的回收(重置)
)
Base.metadata.drop_all(engine)
if __name__ == '__main__':
drop_db()
init_db()
3.記錄的增删改查
資料庫記錄操作的兩種方式
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Users
engine = create_engine("mysql+pymysql://root:[email protected]:3306/s6", max_overflow=0, pool_size=5)
############方式一#############
Session = sessionmaker(bind=engine)
# 每次執行資料庫操作時,都需要建立一個session
session = Session()
# ############# 執行ORM操作 #############
obj1 = Users(name="alex1")
session.add(obj1)
# 送出事務
session.commit()
# 關閉session
session.close()
###########方式二###########
# 方式二:支援線程安全,為每個線程建立一個session
# - threading.Local
# - 唯一辨別
# ScopedSession對象
# self.registry(), 加括号 建立session
# self.registry(), 加括号 建立session
# self.registry(), 加括号 建立session
from greenlet import getcurrent as get_ident
Session = sessionmaker(bind=engine)
session = scoped_session(Session,get_ident)
# session.add
# 操作
session.remove()
1.單表的增删改查
from day101_sqlAlchemy.SQLAlchemy02_create_table_single import engine,Users
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(engine) # 建立資料庫的查詢視窗
db_session = Session() # 打開查詢視窗
# 增加單條資料
# u = Users(name="ryxiong") # 建立insert語句 insert into
# db_session.add(u) # 将insert語句移動到 db_session 查詢視窗
# db_session.commit() # 執行查詢視窗中的所有語句
# db_session.close() # 關閉查詢視窗
# 增加多條資料
# u_list = [Users(name="egon"),Users(name="alex")]
# db_session.add_all(u_list) # 添加多條資料
# db_session.commit()
# db_session.close()
# 查詢資料
# res = db_session.query(Users).all() # 查詢所有資料
# for user in res:
# print(user.id,user.name)
# res = db_session.query(Users).first() # 查詢符合條件的第一條資料
# print(res.id,res.name) # 3 alex
# 并列條件查詢
# res = db_session.query(Users).filter(Users.id<3,Users.name=="ryxiong").all()
# for user in res:
# print(user.id,user.name) # 1 ryxiong
# res = db_session.query(Users).filter(Users.id<3,Users.name=="ryxiong").first()
# print(res.id,res.name) # 1 ryxiong
# 修改資料
# db_session.query(Users).filter(Users.id==2).update({"name":"Egon"})
# db_session.commit()
# 删除資料
db_session.query(Users).filter(Users.id==3).delete()
db_session.commit()
2.一對多的增删改查
from day101_sqlAlchemy.SQLAlchemy03_create_table_foreignKey import engine,Student,School
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(engine) # 建立資料庫的查詢視窗
db_session = Session() # 打開查詢視窗
# 增加一條資料
# school = School(name="新東方")
# db_session.add(school)
# db_session.commit()
# 在添加學生
# school_fir = db_session.query(School).filter(School.name=="新東方").first()
#
# student = Student(name="ryxiong",school_id=school_fir.id)
# db_session.add(student)
# db_session.commit()
# 1.添加資料 relationship 正向添加資料
# stu = Student(name="alex",school=School(name="藍翔"))
# db_session.add(stu)
# db_session.commit()
# 2.添加資料relationship 反向添加資料
# sch = School(name="藍翔")
# sch.student = [
# Student(name="egon"),
# Student(name="wusir")
# ]
# db_session.add(sch)
# db_session.commit()
# 查詢
# 1.relationship正向查詢
res = db_session.query(Student).all()
for stu in res:
print(stu.id,stu.name,stu.school.name)
# 2.relationship反向查詢
res = db_session.query(School).all()
for sch in res:
for stu in sch.student:
print(sch.name,stu.id,stu.name)
3.多對多查詢
from day101_sqlAlchemy.SQLAlchemy04_create_table_M2M import engine,Boys,Girls
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(engine) # 建立資料庫的查詢視窗
db_session = Session() # 打開查詢視窗
# 添加資料
# 1.relationship正向添加
# girl = Girls(name="Nancy",boy=[Boys(name="ryxiong"),Boys(name="alex")])
# db_session.add(girl)
# db_session.commit()
# 2.relationship反向添加
# boy = Boys(name="egon")
# boy.girl = [
# Girls(name="羅玉鳳"),
# Girls(name="朱利安"),
# ]
#
# db_session.add(boy)
# db_session.commit()
# 查詢資料
# 1.relationship 正向查詢
res = db_session.query(Girls).all()
for girl in res:
for boy in girl.boy:
print(girl.name,boy.name)
# 2.relationship 反向查詢
res = db_session.query(Boys).all()
for boy in res:
for girl in boy.girl:
print(boy.name,girl.name)
4.記錄的進階查詢
from day101_sqlAlchemy.SQLAlchemy02_create_table_single import engine,Users
from sqlalchemy.sql import text
from sqlalchemy.orm import sessionmaker
from sqlalchemy import and_,or_
Session = sessionmaker(engine) # 建立資料庫的查詢視窗
db_session = Session() # 打開查詢視窗
# 邏輯條件查詢 and/or
# ret1 = db_session.query(Users).filter(and_(Users.id<3,Users.name=="ryxiong")).all()
# print(ret1)
# ret2 = db_session.query(Users).filter(or_(Users.id<2,Users.name=="egon")).all()
# print(ret2)
#
# ret3 = db_session.query(Users).filter(
# or_(
# and_(Users.id==1,Users.name=="ryxiong"),
# and_(Users.id==2,Users.name=="egon")
# )
# ).all()
# print(ret3)
# 查詢所有資料排序
# ret = db_session.query(Users).order_by(Users.id.asc()).all() # 按照id升序排列
#
# print(ret)
# 查詢資料,指定查詢資料列,加入别名
# ret = db_session.query(Users.name.label("username"),Users.id).first()
#
# print(ret) # ('alex', 3)
# print(ret.id,ret.username) # 3 alex
# 表達式篩選條件
# user_list = db_session.query(Users).filter(Users.name=="ryxiong").all()
# user_list1 = db_session.query(Users).filter_by(name="ryxiong").all()
# for user in user_list:
# print(user.name)
# 複雜查詢
# user_list2 = db_session.query(Users).filter(text("id<:value and name=:name")).params(value=3,name="ryxiong")
# print(user_list2)
# 查詢語句
# user_list3 = db_session.query(Users).filter(text("select * from user id<:value and name=:name")).params(value=3,name="ryxiong")
# print(user_list3)
# 其他查詢條件
# ret = db_session.query(Users).filter(Users.id.between(1,3)).all() # 查詢id值在1-3之間,不包含3的
# print(ret)
#
# ret1 = db_session.query(Users).filter(Users.id.in_([1,2])).all() # 查詢id在清單[1,2]中的使用者
# print(ret1)
#
# ret2 = db_session.query(Users).filter(~Users.id.in_([1,2])).all() # 查詢使用者id不在清單[1,2]中的。
# print(ret2)
# 子查詢
# ret3 = db_session.query(Users).filter(Users.id.in_(db_session.query(Users.id).filter_by(name="ryxiong"))).all()
# print(ret3)
# 通配符
# ret4 = db_session.query(Users).filter(Users.name.like("%ong")).all()
# print(ret4)
# ret5 = db_session.query(Users).filter(~Users.name.like("%ong")).all()
# print(ret5)
# 切片
# ret6 = db_session.query(Users)[1:2]
# print(ret6)
# 分組 group_by
from sqlalchemy.sql import func
# ret7 = db_session.query(Users).group_by(Users.name).all()
# print(ret7)
# 聚合函數
ret8 = db_session.query(
func.max(Users.id),
func.sum(Users.id),
func.min(Users.id),
).group_by(Users.name).all()
print(ret8) # [(3, Decimal('3'), 3), (2, Decimal('2'), 2), (1, Decimal('1'), 1)]
ret9 = db_session.query(
func.max(Users.id),
func.sum(Users.id),
func.min(Users.id),
).group_by(Users.name).having(func.min(Users.id)>2).all()
print(ret9) # [(3, Decimal('3'), 3)]
轉載于:https://www.cnblogs.com/ryxiong-blog/p/11278235.html