python?-?sqlachemy另類用法思路詳解

更新時間：2024年12月06日 08:54:49 作者：瘋狂的妞妞

這篇文章主要介紹了python?-?sqlachemy另類用法,本文結(jié)合實例代碼給大家介紹的非常詳細,感興趣的朋友跟隨小編一起看看吧

這里只是給出一個思路，或許對于未來解決問題有一些參考意義。

仿 JAP 的寫法

這種寫法很像 java 環(huán)境中的 JPA，如果引入模版引擎，則可以大幅增強實用性。

但是，在 python 環(huán)境中，這不符合主流的 ORM 框架。

潛在風險：代碼檢測的時候，可能會被誤判，因為我們定義了一大堆空的函數(shù)。

# 注解式事務(wù) start ---------------------------------------------
@update(sql='UPDATE `t_temp` SET `desc`= :desc WHERE (`id`= :id) LIMIT 1')
def modify(params: dict = None) -> int:
    pass
@query(sql='SELECT * FROM `t_temp` WHERE (`id`= :id) LIMIT 1', result_type=dict)
def queryById(params: dict = None) -> list:
    pass
@query(sql='SELECT * FROM `t_temp` WHERE (`id`= :id) LIMIT 1', result_type=dict)
def queryById2(id: int) -> list:
    pass
@transactional()
def test_annotation():
    ret = modify({'id': 18, 'desc': 'OR 1=1'})
    print(ret)
    result = queryById2(18)
    print(result)

代碼封裝

import inspect
import logger_factory
import typing
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.engine import Result, CursorResult
logger = logger_factory.get_logger()
# 定義數(shù)據(jù)庫連接字符串
DATABASE_URI = 'mysql+pymysql://{username}:{password}@{host}:{port}/{dbname}?charset=utf8mb4'
# 替換為你的數(shù)據(jù)庫用戶名、密碼、主機、端口和數(shù)據(jù)庫名
USERNAME = 'root'
PASSWORD = 'root'
HOST = 'localhost'
PORT = '3306'
DBNAME = 'med'
# 創(chuàng)建數(shù)據(jù)庫引擎，使用連接池
engine = create_engine(
    DATABASE_URI.format(
        username=USERNAME,
        password=PASSWORD,
        host=HOST,
        port=PORT,
        dbname=DBNAME
    ),
    echo=False,  # 如果設(shè)置為True，SQLAlchemy將打印所有執(zhí)行的SQL語句，通常用于調(diào)試
    pool_size=10,  # 連接池大小
    max_overflow=20,  # 超過連接池大小外最多創(chuàng)建的連接數(shù)
    pool_timeout=30,  # 連接池中沒有線程可用時，在拋出異常前等待的時間
    pool_recycle=3600  # 多少秒之后對連接進行一次回收（重置）
)
# do a test
with engine.connect() as con:
    rs = con.execute(text('SELECT 1'))
    rs.fetchone()
    logger.debug('create engine succeed!')
# session-maker
Session = sessionmaker(bind=engine)
# thread safe session-maker
DBSession = scoped_session(Session)
# with Session() as session:
#     # 獲取數(shù)據(jù)庫連接
#     connection = session.connection()
#     savepoint = connection.begin_nested()
#     print(savepoint)
def getEffectRows(result: Result) -> int:
    r"""
    獲取受影響行數(shù)
    這里有點問題：源碼部分 rowcount 是一個 callable，但實際應(yīng)該是 int；
    這里繞一點，確保不會出問題，如果返回 -1，說明出現(xiàn)了意料之外的情況
    :param result: 結(jié)果集
    :return: 受影響行數(shù)
    """
    if isinstance(result, CursorResult):
        effect_row = result.rowcount
        if isinstance(effect_row, int):
            return effect_row
        if callable(effect_row):
            return effect_row()
    return -1
def resultAsDict(result: Result) -> list:
    r"""
    將查詢結(jié)果轉(zhuǎn)換為 dict-list
    :param result: 結(jié)果集
    :return: dict 列表
    """
    keys = result.keys()
    ret = list()
    for item in result.fetchall():
        ret.append(dict(zip(keys, item)))
    return ret
def execute(sql: str, params: dict = None) -> Result:
    r"""
    執(zhí)行一條查詢語句
    :param sql: 查詢語句
    :param params: 參數(shù)
    :return: 結(jié)果集
    """
    if sql is None:
        raise ValueError('sql cannot be None')
    logger.debug('execute sql: ' + sql)
    logger.debug('parameter  : ' + str(params))
    return DBSession().execute(text(sql), params)
def executeQuery(sql: str, params: dict = None, result_type: type = tuple) -> typing.Sequence:
    r"""
    執(zhí)行一個查詢
    :param sql: sql
    :param params: dict
    :param result_type: 結(jié)果集類型，可選：tuple、dict
    :return: 序列
    """
    result = execute(sql, params)
    if result_type == dict:
        return resultAsDict(result)
        pass
    # default return_type tuple-list
    return result.fetchall()
def executeUpdate(sql: str, params: dict = None) -> int:
    r"""
    執(zhí)行一個查詢
    :param sql: sql 執(zhí)行語句
    :param params: dict 查詢參數(shù)
    :return: 受影響行數(shù)
    """
    result = execute(sql, params)
    return getEffectRows(result)
def transactional(rollback: type = Exception):
    r"""
    注解式事務(wù)
    用法類似于 spring 環(huán)境下的 @Transactional 注解
    注意: 事務(wù)控制在 session 級別，不能兼容事務(wù)嵌套的場景（理想狀態(tài)下，應(yīng)當通過 save-point 實現(xiàn)）
    推薦: 如果遇到很復雜的事務(wù)嵌套，顯式調(diào)用 session，手動控制事務(wù)
    :param rollback: 指定觸發(fā)回滾的異常類型
    :return: 裝飾器函數(shù)
    """
    def decorator(func):
        def call(*args, **kwargs):
            session = None
            try:
                session = DBSession()
                ret = func(*args, **kwargs)
                session.commit()
                return ret
            except rollback as e:
                if session:
                    session.rollback()
                logger.exception(f'transaction exception, rollback: {str(e)}')
                raise
            finally:
                if session:
                    session.close()
        return call
    return decorator
    pass
def update(sql: str = None):
    r"""
    注解式查詢，E.G.::
        @update(sql='UPDATE `t_temp` SET `desc`= :desc WHERE (`id`= :id) LIMIT 1')
        def modify(params: dict = None) -> int:
            pass
    :param sql: 要執(zhí)行的 sql
    :return: decorator
    """
    def decorator(func):
        def call(*args, **kwargs):
            result = execute(sql, args[0])
            return getEffectRows(result)
        return call
    return decorator
    pass
def query(sql: str = None, result_type: type = tuple):
    r"""
    注解式查詢，E.G.::
    E.G.::
        @query(sql='SELECT * FROM `t_temp` WHERE (`id`= :id) LIMIT 1', result_type=dict)
        def queryById2(id: int) -> list:
            pass
    :param sql: 要執(zhí)行的 sql
    :param result_type: 結(jié)果集類型，可選：tuple、dict
    :return:  decorator
    """
    def decorator(func):
        def call(*args, **kwargs):
            if sql is None:
                raise ValueError('sql cannot be None')
            first = args[0]
            if isinstance(first, dict):
                result = DBSession().execute(text(sql), args)
            else:
                names = inspect.signature(func).parameters.values()
                params = dict()
                for idx, name in enumerate(names):
                    params[name.name] = args[idx]
                print(params)
                result = DBSession().execute(text(sql), params)
            if result_type == dict:
                keys = result.keys()
                ret = list()
                for item in result.fetchall():
                    ret.append(dict(zip(keys, item)))
                return ret
                # default return_type tuple
                pass
            return result.fetchall()
        return call
    return decorator
    pass
@transactional()
def test_transaction():
    r"""
    測試注解式事務(wù)
    :return: None
    """
    session = DBSession()
    session.execute(text("UPDATE `t_temp` SET `desc`= :desc WHERE (`id`= :id) LIMIT 1"), {'id': 18, 'desc': 'OR 1=3'})
    session.execute(text("UPDATE `t_temp` SET `desc`= :desc WHERE (`id`= :id) LIMIT 1"), {'id': 18, 'desc': 'OR 1=4'})
    # raise exception
    raise SyntaxError('Syntax error')
@transactional()
def test_api():
    r"""
    測試封裝過的函數(shù)
    :return: None
    """
    execute("UPDATE `t_temp` SET `desc`= :desc WHERE (`id`= :id) LIMIT 1", {'id': 18, 'desc': 'OR 1=1'})
    execute("UPDATE `t_temp` SET `desc`= :desc WHERE (`id`= :id) LIMIT 1", {'id': 18, 'desc': 'OR 1=2'})
    # raise exception
    raise SyntaxError('Syntax error')

到此這篇關(guān)于python - sqlachemy另類用法的文章就介紹到這了,更多相關(guān)python sqlachemy另類用法內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！