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

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

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

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

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

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

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

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

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

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

概念

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

安裝

pip install SQLAlchemy

???????pip install mysqlclient   # 安裝自己的數(shù)據(jù)庫客戶端(可以是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的同時，Pool和Dialect也已經(jīng)創(chuàng)建好了，但是此時并沒有真正與數(shù)據(jù)庫連接，等到執(zhí)行具體的語句.connect()等時才會連接到數(shù)據(jù)庫。

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

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

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

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

class UserDB(SQLalchemyFast.Base):
    __tablename__ = "User"   # __tablename__ 聲明表名
    # primary_key為主鍵,autoincrement為自增, doc 為注釋但是不會在數(shù)據(jù)庫中生成
    id = Column(Integer, primary_key=True,autoincrement=True,doc="主鍵")
    name = Column(String(64), unique=True, doc="用戶名") # unique=True 為唯一約束會在數(shù)據(jù)庫中生成索引
    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ù)庫表關(guān)聯(lián)映射起來。數(shù)據(jù)庫表模型類通過__tablename__和表關(guān)聯(lián)起來，Column表示數(shù)據(jù)表的列

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

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

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

會話

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

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

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

from sqlalchemy.orm import sessionmaker

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

session的常見操作方法包括：

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

增刪改查

增

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

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

查

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)閉會話

        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ù)庫配置文件database.properties

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

工具

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)建(必須對象繼承Base)

    # 創(chuàng)建會話
    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ù),不支持分組查詢(因為聚合后的數(shù)據(jù)無法轉(zhuǎn)換成對象)
    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)是一個函數(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語句(包括增刪改查,和存儲過程...只要是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ā)送多個參數(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

測試實體

from sqlalchemy import Column, Integer, String

from src.database.SQLalchemyFast import SQLalchemyFast


class UserDB(SQLalchemyFast.Base):
    __tablename__ = "User"   # __tablename__ 聲明表名
    # primary_key為主鍵,autoincrement為自增, doc 為注釋但是不會在數(shù)據(jù)庫中生成
    id = Column(Integer, primary_key=True,autoincrement=True,doc="主鍵")
    name = Column(String(64), unique=True, doc="用戶名") # unique=True 為唯一約束會在數(shù)據(jù)庫中生成索引
    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 為注釋但是不會在數(shù)據(jù)庫中生成
    id = Column(Integer, primary_key=True,autoincrement=True,doc="主鍵")
    name = Column(String(64), unique=True, doc="用戶名") # unique=True 為唯一約束會在數(shù)據(jù)庫中生成索引
    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)

驗證代碼

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):
    # 測試添加數(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"))

    # 測試添加多條數(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")])
    # 測試刪除數(shù)據(jù)
    def test_deleteDataById(self):
        path = FileTool.getProjectPath(subpath="src/database/database.properties")
        db = SQLalchemyFast(path)
        db.deleteDataById(UserDB, 1)

    # 測試條件刪除數(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")

    # 測試更新數(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"})

    # 測試條件更新數(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")

    # 測試查詢數(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)

    # 測試查詢指定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)

    # 測試條件查詢數(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)

    # 測試創(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)

    # 測試執(zhí)行sql(執(zhí)行失敗會回滾的(存儲過程,函數(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)建存儲過程
        # 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)用存儲過程
        # 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"})



    # 測試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ù)庫框架Sqlalchemy的使用教程詳解的詳細(xì)內(nèi)容，更多關(guān)于Python Sqlalchemy的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: