Python?ORM數(shù)據(jù)庫(kù)框架Sqlalchemy的使用教程詳解

更新時(shí)間：2022年10月19日 14:31:41 作者：胡安民-獨(dú)行者

對(duì)象關(guān)系映射（Object?Relational?Mapping，簡(jiǎn)稱ORM）模式是一種為了解決面向?qū)ο笈c關(guān)系數(shù)據(jù)庫(kù)存在的互不匹配的現(xiàn)象的技術(shù)。本文主要介紹了其使用的相關(guān)資料，感興趣的小伙伴可以學(xué)習(xí)一下

對(duì)象關(guān)系映射（Object Relational Mapping，簡(jiǎn)稱ORM）模式是一種為了解決面向?qū)ο笈c關(guān)系數(shù)據(jù)庫(kù)存在的互不匹配的現(xiàn)象的技術(shù)。面向?qū)ο蟮拈_發(fā)方法是當(dāng)今企業(yè)級(jí)應(yīng)用開發(fā)環(huán)境中的主流開發(fā)方法，關(guān)系數(shù)據(jù)庫(kù)是企業(yè)級(jí)應(yīng)用環(huán)境中永久存放數(shù)據(jù)的主流數(shù)據(jù)存儲(chǔ)系統(tǒng)。對(duì)象和關(guān)系數(shù)據(jù)是業(yè)務(wù)實(shí)體的兩種表現(xiàn)形式，業(yè)務(wù)實(shí)體在內(nèi)存中表現(xiàn)為對(duì)象，在數(shù)據(jù)庫(kù)中表現(xiàn)為關(guān)系數(shù)據(jù)。內(nèi)存中的對(duì)象之間存在關(guān)聯(lián)和繼承關(guān)系，而在數(shù)據(jù)庫(kù)中，關(guān)系數(shù)據(jù)無法直接表達(dá)多對(duì)多關(guān)聯(lián)和繼承關(guān)系。因此，對(duì)象-關(guān)系映射(ORM)系統(tǒng)一般以中間件的形式存在，主要實(shí)現(xiàn)程序?qū)ο蟮疥P(guān)系數(shù)據(jù)庫(kù)數(shù)據(jù)的映射。

學(xué)過java的hibernate框架的那么這個(gè)很好上手,非常簡(jiǎn)單 ,他有兩種模式一種純orm另一種模式是支持原生sql這兩種可以混合使用

優(yōu)點(diǎn)：

簡(jiǎn)潔易讀：將數(shù)據(jù)表抽象為對(duì)象（數(shù)據(jù)模型），更直觀易讀
可移植：封裝了多種數(shù)據(jù)庫(kù)引擎，面對(duì)多個(gè)數(shù)據(jù)庫(kù)，操作基本一致，代碼易維護(hù)
更安全：有效避免SQL注入

缺點(diǎn): 雖然性能稍稍不及原生SQL，但是操作數(shù)據(jù)庫(kù)真的很方便！

官網(wǎng): https://www.sqlalchemy.org/

概念和數(shù)據(jù)類型

概念

常見數(shù)據(jù)類型

安裝

pip install SQLAlchemy

???????pip install mysqlclient   # 安裝自己的數(shù)據(jù)庫(kù)客戶端(可以是mysql 可以是oracle)

連接

from sqlalchemy import create_engine
engine = create_engine("mysql://user:password@hostname/dbname?charset=uft8",
            echo=True,
            pool_size=8,
            pool_recycle=60*30
            )

創(chuàng)建好了Engine的同時(shí)，Pool和Dialect也已經(jīng)創(chuàng)建好了，但是此時(shí)并沒有真正與數(shù)據(jù)庫(kù)連接，等到執(zhí)行具體的語句.connect()等時(shí)才會(huì)連接到數(shù)據(jù)庫(kù)。

echo: 當(dāng)設(shè)置為True時(shí)會(huì)將orm語句轉(zhuǎn)化為sql語句打印，一般debug的時(shí)候可用
pool_size: 連接池的大小，默認(rèn)為5個(gè)，設(shè)置為0時(shí)表示連接無限制
pool_recycle: 設(shè)置時(shí)間以限制數(shù)據(jù)庫(kù)多久沒連接自動(dòng)斷開

創(chuàng)建數(shù)據(jù)庫(kù)表類（模型）

ORM的重要特點(diǎn)就是操作類來操作數(shù)據(jù)庫(kù)，現(xiàn)在我們來創(chuàng)建一個(gè)類，以常見的用戶表舉例：

from sqlalchemy import Column, Integer, String
from src.database.SQLalchemyFast import SQLalchemyFast

class UserDB(SQLalchemyFast.Base):
    __tablename__ = "User"   # __tablename__ 聲明表名
    # primary_key為主鍵,autoincrement為自增, doc 為注釋但是不會(huì)在數(shù)據(jù)庫(kù)中生成
    id = Column(Integer, primary_key=True,autoincrement=True,doc="主鍵")
    name = Column(String(64), unique=True, doc="用戶名") # unique=True 為唯一約束會(huì)在數(shù)據(jù)庫(kù)中生成索引
    email = Column(String(64), doc="郵箱")
    def __init__(self, name=None, email=None):
        self.name = name
        self.email = email

    def __str__(self):
           return "UserDB(id=%s,name=%s,email=%s)" % (self.id, self.name, self.email)

上面的SQLalchemyFast.Base是我自己封裝的Base ,用于統(tǒng)一管理所有模型類,可以將Python類和數(shù)據(jù)庫(kù)表關(guān)聯(lián)映射起來。數(shù)據(jù)庫(kù)表模型類通過__tablename__和表關(guān)聯(lián)起來，Column表示數(shù)據(jù)表的列

生成數(shù)據(jù)庫(kù)表

Base = declarative_base()
Base.metadata.create_all(engine)

會(huì)自動(dòng)創(chuàng)建表，如果存在則忽略，執(zhí)行以上代碼，就會(huì)發(fā)現(xiàn)在db中創(chuàng)建了users表。前提必須有模型類繼承了Base

會(huì)話

會(huì)話就和打電話一樣,打一次電話就是一個(gè)會(huì)話,就相當(dāng)于和數(shù)據(jù)庫(kù)交互一次就是一個(gè)會(huì)話,一個(gè)會(huì)話可以運(yùn)行多個(gè)或單個(gè)語句,會(huì)話結(jié)束必須關(guān)閉

sqlalchemy中使用session用于創(chuàng)建程序和數(shù)據(jù)庫(kù)之間的會(huì)話，所有對(duì)象的載入和保存都需要通過session對(duì)象。

通過sessionmaker調(diào)用創(chuàng)建一個(gè)工廠，并關(guān)聯(lián)Engine以確保每個(gè)session都可以使用該Engine連接資源：

from sqlalchemy.orm import sessionmaker

# 創(chuàng)建session
DbSession = sessionmaker(bind=engine)
session = DbSession()

session的常見操作方法包括：

flush：預(yù)提交，提交到數(shù)據(jù)庫(kù)文件，還未寫入數(shù)據(jù)庫(kù)文件中 (沒事用)
commit：提交了一個(gè)事務(wù)
rollback：回滾
close：關(guān)閉

增刪改查

增

add_user = UserDB("test", "test123@qq.com")
session.add(add_user)
session.commit()

session.add()將會(huì)把Model加入當(dāng)前session維護(hù)的持久空間(可以從session.dirty看到)中，直到commit時(shí)提交到數(shù)據(jù)庫(kù)。

查

users = session.query(UserDB).filter(UserDB.id=1).all()
for item in users:
  print(item.name)

改

session.query(Users).filter(UserDB.id=1).update({'name': "Jack"})

刪

session.query(UserDB).filter(UserDB.name == "test").delete()
session.commit()

執(zhí)行裸sql

session.execute(text(sql), params)
session.commit()

sql: select * from User where id = :id and name = :name

params: {"id":1,"name":"張三"}`

參數(shù)名必須和sql語句中的參數(shù)名一致

with關(guān)閉會(huì)話

        DbSession = sessionmaker(bind=engine)
        with DbSession() as conn:
            # 代碼
            conn.commit()

sql建造者模式

需要導(dǎo)入的包

from sqlalchemy import  delete, update, text, select, join, desc, asc

        sql = select(UserDB.id,UserDB.name).select_from(UserDB).\
            where(text("id = :id and name = :name")).\
            group_by(UserDB.id,UserDB.name).\
            having(text("id = :id and name = :name")).\
            order_by(desc("id"),asc("name")).\
            offset(1).limit(10).\
            params(id=1, name="張三")
        print(sql)

以上sql放入到execute里直接就能跑了

多表聯(lián)查(只支持內(nèi)查詢和左查詢和全查詢)

        sql = select(UserDB.id,UserDB.name).select_from(UserDB).\
            join(BookDB,UserDB.id == BookDB.id).\
            join(alias(BookDB,"b"),text("b.id == b.id"),isouter=True).\
            join(alias(BookDB,"e"),text("e.id == e.id"),full=True). \
            where(text("id = :id and name = :name"))
        print(sql)

封裝的工具

數(shù)據(jù)庫(kù)配置文件database.properties

url=mysql://root:root@106.12.174.220:3306/demo?charset=utf8
echo=True # 是否打印sql語句
pool_size=10 # 連接池大小
pool_recycle=1800 # 連接池回收時(shí)間
pool_timeout=30 # 連接池超時(shí)時(shí)間
isolation_level=READ_COMMITTED # 事務(wù)隔離級(jí)別

工具

from sqlalchemy import create_engine, delete, update, text, alias
from sqlalchemy.future import select
from sqlalchemy.orm import declarative_base, sessionmaker

from src.file.FileReadAndWrite import FileReadAndWrite
from src.log.Log import Log


class SQLalchemyFast(object):
    Base = declarative_base()
    """
    功能: SQLalchemy工具
    """

    def __init__(self, dbFile):
        file = FileReadAndWrite.readPropertiesFile(dbFile)
        self.engine = create_engine(
            url=file['url'],
            echo=bool(file['echo']),
            pool_size=int(file['pool_size']),
            pool_recycle=int(file['pool_recycle']),
            pool_timeout=int(file['pool_timeout']),
            isolation_level=file['isolation_level'],
        )

        SQLalchemyFast.Base.metadata.create_all(self.engine)  # 創(chuàng)建表,如果表存在則不創(chuàng)建(必須對(duì)象繼承Base)

    # 創(chuàng)建會(huì)話
    def createSession(self):
        Session = sessionmaker(bind=self.engine)
        return Session()

    # 添加一條數(shù)據(jù)
    def addData(self, object):
        with self.createSession() as conn:
            conn.add(object)
            conn.commit()

    # 添加多條數(shù)據(jù)
    def addDataList(self, objectList):
        with self.createSession() as conn:
            conn.add_all(objectList)
            conn.commit()

    # 刪除主鍵id的數(shù)據(jù)
    def deleteDataById(self, cla, id):
        with self.createSession() as conn:
            conn.query(cla).filter(cla.id == id).delete()
            conn.commit()

    # 刪除指定數(shù)據(jù)(where是并且的關(guān)系,不支持or和其他復(fù)雜查詢)
    def deleteDataWhere(self, cla, *where):
        with self.createSession() as conn:
            stmt = delete(cla).where(*where)
            conn.execute(stmt)
            conn.commit()

    # 清空表
    def truncateTable(self, cla):
        with self.createSession() as conn:
            conn.query(cla).delete()
            conn.commit()

    # 更新指定主鍵id的數(shù)據(jù)
    def updateDataById(self, cla, id, data):
        """
        :param cla:  類(表)
        :param id:  主鍵id
        :param data:  {'key': "value",...}  key為表中的字段名,value為要修改的值
        :return:
        """
        with self.createSession() as conn:
            stmt = update(cla).where(cla.id == id).values(data)
            result = conn.execute(stmt)
            conn.commit()
            return result

    # 更新指定條件的數(shù)據(jù) (where是并且的關(guān)系,不支持or和其他復(fù)雜查詢)
    def updateDataWhere(self, cla, data, *where):
        """
        :param cla:  類(表)
        :param data:  {'key': "value",...}  key為表中的字段名,value為要修改的值
        :param where:  過濾條件
        :return:
        """
        with self.createSession() as conn:
            stmt = update(cla).where(*where).values(data)
            conn.execute(stmt)
            conn.commit()

    # 查詢?nèi)繑?shù)據(jù)
    def queryDataAll(self, cla):
        with self.createSession() as conn:
            result = conn.query(cla).all()
        return result

    # 查詢主鍵id的數(shù)據(jù)
    def queryDataById(self, cla, id):
        with self.createSession() as conn:
            result = conn.query(cla).filter(cla.id == id).first()
        return result

    # 查詢指定數(shù)據(jù),不支持分組查詢(因?yàn)榫酆虾蟮臄?shù)據(jù)無法轉(zhuǎn)換成對(duì)象)
    def queryDataWhere(self, cla,aliasName=None, column=None, where=None,
                       join=None, on=None, left=None, full=None,
                       order="", limit="", offset="", distinct=None, params=None):
        with self.createSession() as conn:
            stmt = select(cla)
            if aliasName:
                stmt = select(alias(cla,aliasName))
            if column:
                stmt = stmt.with_only_columns(text(column)) .select_from(cla)
            if join is not None and on is not None:
                if left:
                    stmt = stmt.join(join, text(on), isouter=True)
                elif full:
                    stmt = stmt.join(join, text(on), full=True)
                else:
                    stmt = stmt.join(join, text(on))
            if where:
                stmt = stmt.where(text(where))
            if order:
                stmt = stmt.order_by(text(order))
            if limit:
                stmt = stmt.limit(limit)
            if offset:
                stmt = stmt.offset(offset)
            if distinct:
                stmt = stmt.distinct()
            result = conn.execute(stmt,params).all()
            result= [row[0] for row in result]
        return result

    # 創(chuàng)建事物(運(yùn)行多條sql語句 ,function(conn)是一個(gè)函數(shù),里面包含多條sql語句,需要使用原生的sqlalchemy)
    def createTransaction(self, function):
        with  self.createSession() as conn:
            conn.begin()
            try:
                function(conn)
                conn.commit()
            except Exception as e:
                Log.logError(e)
                conn.rollback()

    # 執(zhí)行sql語句(包括增刪改查,和存儲(chǔ)過程...只要是sql語句都可以執(zhí)行)
    def executeSql(self, sql, params=None):
        """
        :param sql:  sql語句  如: "select * from User where id = :id and name = :name "
        :param params:  參數(shù)  例如: {"id":1,"name":"張三"} 注意:參數(shù)名必須和sql語句中的參數(shù)名一致
        發(fā)送多個(gè)參數(shù)時(shí),參數(shù)名必須以列表的形式傳入,例如: {"id":["1","2"],"name":"張三"}
        "INSERT INTO some_table (x, y) VALUES (:x, :y)" 參數(shù)可以是 [{"x": 11, "y": 12}, {"x": 13, "y": 14}]
        :return:
        """
        with self.createSession() as conn:
            result = conn.execute(text(sql), params)
            conn.commit()
        return result

    # 執(zhí)行構(gòu)建sql語句
    def executeSqlBuild(self, sql):
        with self.createSession() as conn:
            result = conn.execute(sql)
            conn.commit()
        return result

測(cè)試實(shí)體

from sqlalchemy import Column, Integer, String

from src.database.SQLalchemyFast import SQLalchemyFast


class UserDB(SQLalchemyFast.Base):
    __tablename__ = "User"   # __tablename__ 聲明表名
    # primary_key為主鍵,autoincrement為自增, doc 為注釋但是不會(huì)在數(shù)據(jù)庫(kù)中生成
    id = Column(Integer, primary_key=True,autoincrement=True,doc="主鍵")
    name = Column(String(64), unique=True, doc="用戶名") # unique=True 為唯一約束會(huì)在數(shù)據(jù)庫(kù)中生成索引
    email = Column(String(64), doc="郵箱")
    def __init__(self, name=None, email=None):
        self.name = name
        self.email = email

    def __str__(self):
           return "UserDB(id=%s,name=%s,email=%s)" % (self.id, self.name, self.email)

from sqlalchemy import Column, Integer, String

from src.database.SQLalchemyFast import SQLalchemyFast


class BookDB(SQLalchemyFast.Base):
    __tablename__ = "Book"   # __tablename__ 聲明表名
    # primary_key為主鍵,autoincrement為自增, doc 為注釋但是不會(huì)在數(shù)據(jù)庫(kù)中生成
    id = Column(Integer, primary_key=True,autoincrement=True,doc="主鍵")
    name = Column(String(64), unique=True, doc="用戶名") # unique=True 為唯一約束會(huì)在數(shù)據(jù)庫(kù)中生成索引
    email = Column(String(64), doc="郵箱")
    def __init__(self, name=None, email=None):
        self.name = name
        self.email = email

    def __str__(self):
           return "UserDB(id=%s,name=%s,email=%s)" % (self.id, self.name, self.email)

驗(yàn)證代碼

import unittest
from sqlalchemy import delete, update, text, select, join, desc, asc, alias

from src.database.BookDB import BookDB
from src.database.SQLalchemyFast import SQLalchemyFast
from src.database.UserDB import UserDB
from src.file.FileTool import FileTool


class SQLalchemyFastTest(unittest.TestCase):
    # 測(cè)試添加數(shù)據(jù)
    def test_add(self):
        path = FileTool.getProjectPath(subpath="src/database/database.properties")
        db = SQLalchemyFast(path)
        db.addData(UserDB("name1", "123456789"))
        db.addData(UserDB("name2", "123456789"))
        db.addData(UserDB("name3", "123456789"))
        db.addData(UserDB("name4", "123456789"))

    # 測(cè)試添加多條數(shù)據(jù)
    def test_addAll(self):
        path = FileTool.getProjectPath(subpath="src/database/database.properties")
        db = SQLalchemyFast(path)
        db.addDataList([UserDB("name111", "123456789"), UserDB("name211", "123456789")])
    # 測(cè)試刪除數(shù)據(jù)
    def test_deleteDataById(self):
        path = FileTool.getProjectPath(subpath="src/database/database.properties")
        db = SQLalchemyFast(path)
        db.deleteDataById(UserDB, 1)

    # 測(cè)試條件刪除數(shù)據(jù)
    def test_deleteWhere(self):
        path = FileTool.getProjectPath(subpath="src/database/database.properties")
        db = SQLalchemyFast(path)
        db.deleteDataWhere(UserDB, UserDB.name == "name1", UserDB.email == "123456789")

    # 測(cè)試更新數(shù)據(jù)
    def test_update(self):
        path = FileTool.getProjectPath(subpath="src/database/database.properties")
        db = SQLalchemyFast(path)
        db.updateDataById(UserDB, 10, {"name": "name31", "email": "123456789"})

    # 測(cè)試條件更新數(shù)據(jù)
    def test_updateFilter(self):
        path = FileTool.getProjectPath(subpath="src/database/database.properties")
        db = SQLalchemyFast(path)
        db.updateDataWhere(UserDB, {"name": "name33", "email": "123456789"}, UserDB.name == "name2",
                           UserDB.email == "1231")

    # 測(cè)試查詢數(shù)據(jù)
    def test_queryDataAll(self):
        path = FileTool.getProjectPath(subpath="src/database/database.properties")
        db = SQLalchemyFast(path)
        data_all = db.queryDataAll(UserDB)
        for data in data_all:
            print(data)

    # 測(cè)試查詢指定id數(shù)據(jù)
    def test_queryDataById(self):
        path = FileTool.getProjectPath(subpath="src/database/database.properties")
        db = SQLalchemyFast(path)
        data = db.queryDataById(UserDB, 10)
        print(data)

    # 測(cè)試條件查詢數(shù)據(jù)(不支持分組查詢和鏈表查詢)
    def test_queryDataWhere(self):
        path = FileTool.getProjectPath(subpath="src/database/database.properties")
        db = SQLalchemyFast(path)
        data_all = db.queryDataWhere(UserDB,
                                     where="name like CONCAT(:name,'%')",
                                     order="id desc",
                                     offset=1,
                                     limit=3,
                                     params={"name": "name"})

        # db.queryDataWhere(UserDB,
        #                              where="name like CONCAT(:name,'%')",
        #                              order="id desc",
        #                              offset=1,
        #                              limit=3,
        #                              params={"name": "name"})

        # db.queryDataWhere(UserDB,aliasName="a",
        #                   join=alias(BookDB,"b"),on="a.id == b.id",
        #                   where="a.name like CONCAT(:name,'%')",
        #                   params={"name": "name"})
        for data in data_all:
            print(data)

    # 測(cè)試創(chuàng)建事物
    def test_createTransaction(self):
        path = FileTool.getProjectPath(subpath="src/database/database.properties")
        db = SQLalchemyFast(path)

        def test1(conn):
            conn.add(UserDB("name111", "123456789"))
            conn.add(UserDB("name211", "123456789"))
            # raise Exception("test122")
            # conn.add(UserDB("name333", "123456789"))
            # conn.add(UserDB("name444", "123456789"))

        db.createTransaction(test1)

    # 測(cè)試執(zhí)行sql(執(zhí)行失敗會(huì)回滾的(存儲(chǔ)過程,函數(shù)))
    def test_executeSql(self):
        path = FileTool.getProjectPath(subpath="src/database/database.properties")
        db = SQLalchemyFast(path)
        # data_all = db.executeSql("select * from User")
        # data_all = db.executeSql("select * from User where name like CONCAT(:name,'%')", params={"name":"name"})
        # for data in data_all:
        #     print(data)

        # 創(chuàng)建存儲(chǔ)過程
        # db.executeSql("CREATE PROCEDURE `test_procedure` \
        # (IN `in_name` VARCHAR(255), IN `in_email` VARCHAR(255)) \
        # BEGIN \
        # INSERT INTO `User` (`name`, email) VALUES (in_name, in_email); \
        # END")

        # 調(diào)用存儲(chǔ)過程
        # db.executeSql("call test_procedure(:name, :email)", params={"name": "name555", "email": "email12131"})

        # 創(chuàng)建函數(shù)
        # db.executeSql("CREATE FUNCTION `test_function`( `in_name` VARCHAR(255),  `in_email` VARCHAR(255)) \
        #  RETURNS INT(11) \
        #  BEGIN \
        #  DELETE FROM `User`; \
        #  INSERT INTO `User` (`name`, email) VALUES (in_name, in_email); \
        #  INSERT INTO `User` (`name`, email) VALUES (in_name, in_email); \
        #  RETURN 1; \
        #  END")
        # 調(diào)用函數(shù)
        # data_all = db.executeSql("select test_function(:name, :email)", params={"name": "name5551", "email": "email12131"})



    # 測(cè)試sql構(gòu)造
    def test_executeSqlBuild(self):
        path = FileTool.getProjectPath(subpath="src/database/database.properties")
        db = SQLalchemyFast(path)
        # sql = select(UserDB.id,UserDB.name).select_from(UserDB).\
        #     join(BookDB,UserDB.id == BookDB.id)
        # print(sql)
        # db.executeSqlBuild(sql)

以上就是Python ORM數(shù)據(jù)庫(kù)框架Sqlalchemy的使用教程詳解的詳細(xì)內(nèi)容，更多關(guān)于Python Sqlalchemy的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: