SQLAlchemy

時間  2019-11-10
標籤 sqlalchemy 简体版
原文   原文鏈接

 介紹

SQLAlchemy是Python編程語言下的一款ORM框架，該框架創建在數據庫API之上，使用關係對象映射進行數據庫操做，簡言之即是：將對象轉換成SQL，而後使用數據API執行SQL並獲取執行結果。html

組成部分：python

  • Engine，框架的引擎
  • Connection Pooling ，數據庫鏈接池
  • Dialect，選擇鏈接數據庫的DB API種類
  • Schema/Types，架構和類型
  • SQL Exprression Language，SQL表達式語言

SQLAlchemy自己沒法操做數據庫，其必須以來pymsql等第三方插件，Dialect用於和數據API進行交流，根據配置文件的不一樣調用不一樣的數據庫API，從而實現對數據庫的操做，如:mysql

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.htmlsql

安裝

pip install SQLAlchemy

使用

執行原生sql語句

create_engine 方法進行數據庫鏈接，返回一個 db 對象。參數echo = True能夠在控制檯打印sql語句
經過這個engine對象能夠直接execute 進行查詢，例如 engine.execute("SELECT * FROM user") 也能夠經過 engine 獲取鏈接在查詢，例如 conn = engine.connect() 經過 conn.execute()方法進行查詢。二者有什麼差異呢？數據庫

  • 直接使用engine的execute執行sql的方式, 叫作connnectionless執行,
  • 藉助 engine.connect()獲取conn, 而後經過conn執行sql, 叫作connection執行

主要差異在因而否使用transaction模式, 若是不涉及transaction, 兩種方法效果是同樣的. 官網推薦使用後者。django

使用engine的execute執行sql:編程

from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8?charset=utf8")
cur = engine.execute('select * from users')
result = cur.fetchall()
print(result)

使用engine.connect()執行sql語句:安全

from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8?charset=utf8")
conn = engine.connect()
cur = conn.execute('select * from users')
result = cur.fetchall()
print(result)

使用鏈接池session

from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine

engine = create_engine(
    "mysql+pymysql://root:123456@127.0.0.1:3306/s8?charset=utf8",
    max_overflow=0,  # 超過鏈接池大小外最多建立的鏈接
    pool_size=5,  # 鏈接池大小
    pool_timeout=30,  # 池中沒有線程最多等待的時間，不然報錯
    pool_recycle=-1  # 多久以後對線程池中的線程進行一次鏈接的回收（重置）
)

conn = engine.raw_connection()
cursor = conn.cursor()
cursor.execute("select * from users")
result = cursor.fetchall()
print(result)
cursor.close()
conn.close()

ORM

建立一個簡單的表

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
# 建立鏈接
engine = create_engine("mysql+pymysql://root:root@192.168.56.11/oldboydb",encoding='utf-8',echo=True)

#生成ORM基類
Base=declarative_base() 

class User(Base):
    __tablename__ = 'user' #表名
    id = Column(Integer,primary_key=True) #字段，整形，主鍵 column是導入的
    name = Column(String(32))
    password = Column(String(64))

Base.metadata.create_all(engine) #在engine鏈接的數據庫裏建立表結構

經常使用字段架構

Integer/BigInteger/SmallInteger
# 整形.
Boolean
# 布爾類型. Python 中表現爲 True/False , 數據庫根據支持狀況, 表現爲 BOOLEAN 或SMALLINT . 實例化時能夠指定是否建立約束(默認建立).
Date/DateTime/Time (timezone=False)
# 日期類型, Time 和 DateTime 實例化時能夠指定是否帶時區信息.
Interval
# 時間誤差類型. 在 Python 中表現爲 datetime.timedelta() , 數據庫不支持此類型則存爲日期.
Enum (*enums, **kw)
# 枚舉類型, 根據數據庫支持狀況, SQLAlchemy 會使用原生支持或者使用 VARCHAR 類型附加約束的方式實現. 原生支持中涉及新類型建立, 細節在實例化時控制.
Float
# 浮點小數.
Numeric (precision=None, scale=None, decimal_return_scale=None, ...)
# 定點小數, Python 中表現爲 Decimal .
LargeBinary (length=None)
# 字節數據. 根據數據庫實現, 在實例化時可能須要指定大小.
PickleType
# Python 對象的序列化類型.
String (length=None, collation=None, ...)
# 字符串類型, Python 中表現爲 Unicode , 數據庫表現爲 VARCHAR , 一般都須要指定長度.
Unicode
# 相似與字符串類型, 在某些數據庫實現下, 會明確表示支持非 ASCII 字符. 同時輸入輸出也強制是 Unicode 類型.
Text
# 長文本類型, Python 表現爲 Unicode , 數據庫表現爲 TEXT .

Column指定的一些字段參數

default # 默認值,時間字段的默認值datetime.datetime.now不能加(),不然會執行生成固定值
primary_key=True # 設置爲主鍵
autoincrement=True # 主鍵的自增
index=True # 做爲索引
nullable=True # 是否能夠爲空

表的參數

# 在類下定義__table_args__
__table_args__ = (
        # UniqueConstraint('id', 'name', name='uix_id_name'),聯合惟一
        # Index('ix_id_name', 'name', 'extra'),聯合索引
        # 'mysql_engine': 'InnoDB',
        #  'mysql_charset': 'utf8'
    )

　　

外鍵設置

# sqlaichemy不像django擁有外鍵字段,設置外鍵須要導入ForeignKey指定哪張表的那個字段,做爲Column的第二個參數
from sqlalchemy import Column, Integer, ForeignKey
# 字段
hobby_id = Column(Integer, ForeignKey("hobby.id"))

　　另外若是是多對多的表,他也沒法建立第三張表,須要手動建立第三張表關聯兩張表

一張簡單的表

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
from sqlalchemy.orm import relationship

Base = declarative_base()

class Users(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True)
    age = Column(Integer, default=18)
    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', 'extra'),
    )

# 生成這張表
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8?charset=utf8")
Base.metadata.create_all(engine) # 將繼承了Base類的表在engine鏈接的數據庫中生成
# 刪除表
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8?charset=utf8")
Base.metadata.drop_all(engine) # 將繼承了Base類的表在engine鏈接的數據庫中刪除

一對多關係的設置

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
from sqlalchemy.orm import relationship

Base = declarative_base()

class Classes(Base):
    __tablename__ = 'classes'

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True)

class Student(Base):
    __tablename__ = 'student'

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True)
    age = Column(Integer, default=18)
    classes = Column(Integer,ForeignKey('classes.id'))

    # 與生成表結構無關，僅用於查詢方便,指定了關聯的表,和反向查詢的名字
    # 他會根據這個類裏與指定表關聯的字段去查找
    hobby = relationship("Classes", backref='stu')

多對多關係的表設計

class Classes(Base):
    __tablename__ = 'classes'

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True)
class Teacher(Base):
    __tablename__ = 'teacher'
    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True)
    age = Column(Integer, default=19)
    # 與生成表結構無關，僅用於查詢方便,指定了關聯的表,中間表的表名和反向查詢的名字
    servers = relationship('Classes', secondary='server2group', backref='classes')
class Teacher2Class(Base):
    __tablename__ = 'server2group'
    id = Column(Integer, primary_key=True)
    classes_id = Column(Integer,ForeignKey('classes.id'))
    teacher_id = Column(Integer,ForeignKey('teacher.id'))
    __table_args__ = (
     UniqueConstraint('classes_id', 'teacher_id', name='tea_cls'),
    # Index('ix_id_name', 'name', 'extra'),
    )

數據操做

ORM經過Session與數據庫創建鏈接的。當應用第一次載入時，咱們定義一個Session類（聲明create_engine()的同時），這個Session類爲新的Session對象提供工廠服務。

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
engine =create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/s8day128db?charset=utf8")
Session = sessionmaker(bind=engine)

建立鏈接的方式

# 方式一
# 這個定製的Session類會建立綁定到數據庫的Session對象。若是須要和數據庫創建鏈接，只須要實例化一個Session：
session = Session() # 此處只可以在視圖函數內執行
# 執行數據庫操做
# 修改操做進行session.commit()
session.close()


# 方式二:支持線程安全，爲每一個線程建立一個session
session = scoped_session(Session)
# 執行數據庫操做
# 修改操做進行session.commit()
session.remove()

方式二的源碼:

class scoped_session(object):
    session_factory = None
    def __init__(self, session_factory, scopefunc=None):
        self.session_factory = session_factory # 原來的Session

        if scopefunc:
            self.registry = ScopedRegistry(session_factory, scopefunc)
        else:
            self.registry = ThreadLocalRegistry(session_factory)

    def __call__(self, **kw):
        if kw:
            if self.registry.has():
                raise sa_exc.InvalidRequestError(
                    "Scoped session is already present; "
                    "no new arguments may be specified.")
            else:
                sess = self.session_factory(**kw)
                self.registry.set(sess)
                return sess
        else:
            return self.registry()
    ...
class ThreadLocalRegistry(ScopedRegistry):
    def __init__(self, createfunc):
        self.createfunc = createfunc # 原來的Session
        self.registry = threading.local()

    def __call__(self):
        try:
            return self.registry.value
        except AttributeError:
            val = self.registry.value = self.createfunc()
            return val

def instrument(name):
    def do(self, *args, **kwargs):
        return getattr(self.registry(), name)(*args, **kwargs)
    return do # 這裏返回的是函數,至關於self.query = do,self.query()至關於do()

for meth in Session.public_methods: # meth就是原Session對象的屬性
    setattr(scoped_session, meth, instrument(meth))

添加數據

添加一條

clsobj = models.Classes(name='全棧1期') # 實例化類
session.add(clsobj) # 經過session將clsobj添加進數據庫中
# 上面並不須要指定庫,由於clsobj是Classes的實例,他們存在着對應關係
session.commit() # 提交

添加多條

clsobj = models.Classes(name='全棧2期')
stuobj = models.Student(name="李淳罡",classes=1)
session.add_all([clsobj,stuobj]) # 也正是由於對應關係的存在,咱們能夠將不一樣類的實例一塊兒提交
session.commit()

簡單的查詢數據

r1 = session.query(models.Classes).all()
print(r1) # [<models.Classes object at 0x0000020B3F849240>, <models.Classes object at 0x0000020B3F8492E8>]
r2 = session.query(models.Classes.name.label('xx'), models.Classes.id).all()
print(r2,type(r2[0])) # [('全棧1期', 1), ('全棧2期', 2)] <class 'sqlalchemy.util._collections.result'>
# 看似元組其實並非,也能夠經過.字段的方式得到值,.label('xx')至關於爲這個字段從新齊了名字,至關於sql中的as
r3 = session.query(models.Classes).filter(models.Classes.name == "全棧1期").all()
print(r3) # filter(表達式) [<models.Classes object at 0x0000020B3F849240>] 
r4 = session.query(models.Classes).filter_by(name='alex').all()
print(r4) # filter_by(字段=值) []
r5 = session.query(models.Classes).filter_by(name='alex').first()
print(r5) # None

刪除

session.query(models.Classes).filter(models.Classes.id>2).delete()
# 此處應注意是什麼調用的.delete(),all()方法返回的是個列表
session.commit()

修改

session.query(Users).filter(Users.id > 0).update({"name" : "099"})
session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False)
session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate")

# updata方法傳遞字典的鍵能夠是字符串類型的字段名,也能夠是表下的字段,
# 若是是對原數據進行修改還要指定synchronize_session

更多

#　條件
ret = session.query(Users).filter_by(name='alex').all()
# 多條件,隔開,and關係
ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all() 
# between在a,b之間
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()
# in_(),接收一個列表
ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
# ~取反
ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
# 子查詢
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
from sqlalchemy import and_, or_
# 查詢條件的關係
# and關係
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
# or關係
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
# 組合關係
ret = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == 'eric', Users.id > 3),
        Users.extra != ""
    )).all()


# 通配符,模糊匹配like()
ret = session.query(Users).filter(Users.name.like('e%')).all()
ret = session.query(Users).filter(~Users.name.like('e%')).all()

# 限制 limit
ret = session.query(Users)[1:2]


# 排序
ret = session.query(Users).order_by(Users.name.desc()).all()
# 多條件排序是按第一種方式排出現相同時,將相同值按第二種方式排
ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()

# 分組,及聚合函數
from sqlalchemy.sql import func

ret = session.query(Users).group_by(Users.extra).all()
ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).all()

ret = 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()

# 連表
ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
# 內鏈接
ret = session.query(Person).join(Favor).all()
# souter=True左鏈接,他們的鏈接不用設置關聯字段,由於他們之間存在外鍵關係
ret = session.query(Person).join(Favor, isouter=True).all()

# 無外鍵關聯則須要本身設置關聯字段,query什麼就能查到什麼,不能.別的屬性,通常用做關聯查詢
obj = session.query(models.Student).join(models.Userinfo,models.Student.id==models.Userinfo.user_id).first()

# 組合,將具備相同字段數量的查詢結果聯合成一個結果
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union(q2).all() # 去重

q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union_all(q2).all() # 不去重
# 原生查詢,查詢結果必須在query中
result = session.query(models.Classes).from_statement(text("SELECT * FROM student where age=:age")).params(age=18).first()
obj = session.query("name","age").from_statement(text('SELECT name,age from teacher')).all()

跨表查詢實例

# 找到全部學生，打印學生信息(包含班級名稱)
# 子查詢
objs = session.query(models.Student).all()
for obj in objs:
    cls_obj = session.query(models.Classes).filter(models.Classes.id==obj.classes).first()
    print(obj.id,obj.name,obj.classes,cls_obj.name)

# 連表,已有外鍵關聯
objs = session.query(models.Student.id,models.Student.name,models.Classes.name).join(models.Classes,isouter=True).all()
print(objs)

# 還記不記得relationship()
objs = session.query(models.Student).all()
for item in objs:
    print(item.id,item.name,item.classes,item.cls.name)

另外relationship()還能夠用來添加數據

# 一對多示例
# 向Student增長一條記錄,順便向Classes增長一條記錄
session.add(models.Student(name='小韓',cls=models.Classes(name='全棧8期')))
# 向學生表增長一條記錄
clsobj = session.query(models.Classes).filter(models.Classes.name == "全棧8期").first()
session.add(models.Student(name='崔絲塔娜',cls=clsobj))

# 反向操做
# 建立班級同時建立學生
# 由於是一對多的關係,因此stu應該是一個集合
session.add(models.Classes(name='全棧3期',stu=[models.Student(name='奧利安娜'),models.Student(name='莫甘娜')]))


# 多對多
# 建立講師關聯班級,建立班級
obj = models.Teacher(name='奧菲娜')
obj.servers = [models.Classes(name='全棧4期'),models.Classes(name='全棧5期')]
session.add(obj)
# 建立講師不建立班級
clsobj = session.query(models.Classes).filter(models.Classes.name == "全棧8期").first()
session.add(models.Teacher(name='露露',servers=[clsobj]))

# 反向添加
session.add(models.Classes(name='全棧9期',classes=[models.Teacher(name='奧瑞利亞'),models.Teacher(name='索拉卡')]))
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程