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
            )

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)

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

from sqlalchemy.orm import sessionmaker

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

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

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

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

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

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

url=mysql://root:root@106.12.174.220:3306/demo?charset=utf8
echo=True # 是否打印sql語(yǔ)句
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:  過(guò)濾條件
        :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ù)無(wú)法轉(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語(yǔ)句 ,function(conn)是一個(gè)函數(shù),里面包含多條sql語(yǔ)句,需要使用原生的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語(yǔ)句(包括增刪改查,和存儲(chǔ)過(guò)程...只要是sql語(yǔ)句都可以執(zhí)行)
    def executeSql(self, sql, params=None):
        """
        :param sql:  sql語(yǔ)句  如: "select * from User where id = :id and name = :name "
        :param params:  參數(shù)  例如: {"id":1,"name":"張三"} 注意:參數(shù)名必須和sql語(yǔ)句中的參數(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語(yǔ)句
    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 為注釋但是不會(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)

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ǔ)過(guò)程,函數(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ǔ)過(guò)程
        # 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ǔ)過(guò)程
        # 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)