一個(gè)超方便使用SQL的Python神器

作者：言淦，一枚喜歡淦代碼的碼農(nóng)，"言淦說(shuō)"主理人

其實(shí)一開(kāi)始用的是pymysql，但是發(fā)現(xiàn)維護(hù)比較麻煩，還存在代碼注入的風(fēng)險(xiǎn)，所以就干脆直接用ORM框架。

• sqlalchemy版本: 1.3.15
• pymysql版本: 0.9.3
• mysql版本: 5.7

初始化工作

一般使用ORM框架，都會(huì)有一些初始化工作，比如數(shù)據(jù)庫(kù)連接，定義基礎(chǔ)映射等。

以MySQL為例，創(chuàng)建數(shù)據(jù)庫(kù)連接只需要傳入DSN字符串即可。其中echo表示是否輸出對(duì)應(yīng)的sql語(yǔ)句，對(duì)調(diào)試比較有幫助。

from sqlalchemy import create_engine

engine = create_engine('mysql+pymysql://$user:$password@$host:$port/$db?charset=utf8mb4', echo=True)

個(gè)人設(shè)計(jì)

對(duì)于我個(gè)人而言，引進(jìn)ORM框架時(shí)，我的項(xiàng)目會(huì)參考MVC模式做以下設(shè)計(jì)。其中model存儲(chǔ)的是一些數(shù)據(jù)庫(kù)模型，即數(shù)據(jù)庫(kù)表映射的Python類(lèi)；model_op存儲(chǔ)的是每個(gè)模型對(duì)應(yīng)的操作，即增刪查改；調(diào)用方（如main.py）執(zhí)行數(shù)據(jù)庫(kù)操作時(shí)，只需要調(diào)用model_op層，并不用關(guān)心model層，從而實(shí)現(xiàn)解耦。

├── main.py
├── model
│   ├── __init__.py
│   ├── base_model.py
│   ├── ddl.sql
│   └── py_orm_model.py
└── model_op
    ├── __init__.py
    └── py_orm_model_op.py

映射聲明(Model介紹)

舉個(gè)栗子，如果我們有這樣一張測(cè)試表

create table py_orm (
    `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '唯一id',
    `name` varchar(255) NOT NULL DEFAULT '' COMMENT '名稱(chēng)',
    `attr` JSON NOT NULL COMMENT '屬性',
    `ct` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '創(chuàng)建時(shí)間',
    `ut` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON update CURRENT_TIMESTAMP COMMENT '更新時(shí)間',
    PRIMARY KEY(`id`)
)ENGINE=InnoDB COMMENT '測(cè)試表';

在ORM框架中，映射的結(jié)果就是下文這個(gè)Python類(lèi)

# py_orm_model.py
from .base_model import Base
from sqlalchemy import Column, Integer, String, TIMESTAMP, text, JSON


class PyOrmModel(Base):
    __tablename__ = 'py_orm'

    id = Column(Integer, autoincrement=True, 
                primary_key=True, comment='唯一id')
    name = Column(String(255), nullable=False, 
                  default='', comment='名稱(chēng)')
    attr = Column(JSON, nullable=False, comment='屬性')
    ct = Column(TIMESTAMP, nullable=False, server_default=text('CURRENT_TIMESTAMP'), comment='創(chuàng)建時(shí)間')
    ut = Column(TIMESTAMP, nullable=False, 
                server_default=text('CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP'),
                comment='更新時(shí)間')

首先

我們可以看到PyOrmModel繼承了Base類(lèi)，該類(lèi)是sqlalchemy提供的一個(gè)基類(lèi)，會(huì)對(duì)我們聲明的Python類(lèi)做一些檢查，我將其放在base_model中。

# base_model.py
# 一般base_model做的都是一些初始化的工作

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

engine = create_engine("mysql+pymysql://root:[email protected]:33306/orm_test?charset=utf8mb4", echo=False)

其次

每個(gè)Python類(lèi)都必須包含__tablename__屬性，不然無(wú)法找到對(duì)應(yīng)的表。

第三

關(guān)于數(shù)據(jù)表的創(chuàng)建有兩種方式，第一種當(dāng)然是手動(dòng)在MySQL中創(chuàng)建，只要你的Python類(lèi)定義沒(méi)有問(wèn)題，就可以正常操作；第二種是通過(guò)orm框架創(chuàng)建，比如下面

# main.py
# 注意這里的導(dǎo)入路徑，Base創(chuàng)建表時(shí)會(huì)尋找繼承它的子類(lèi)，如果路徑不對(duì)，則無(wú)法創(chuàng)建成功

from sqlachlemy_lab import Base, engine

if __name__ == '__main__':
    Base.metadata.create_all(engine)

創(chuàng)建效果：

...
2020-04-04 10:12:53,974 INFO sqlalchemy.engine.base.Engine 
CREATE TABLE py_orm (
    id INTEGER NOT NULL AUTO_INCREMENT, 
    name VARCHAR(255) NOT NULL DEFAULT '' COMMENT '名稱(chēng)', 
    attr JSON NOT NULL COMMENT '屬性', 
    ct TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, 
    ut TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, 
    PRIMARY KEY (id)
)

第四

關(guān)于字段屬性

• 1.primary_key和autoincrement比較好理解，就是MySQL的主鍵和遞增屬性。
• 2.如果是int類(lèi)型，不需要指定長(zhǎng)度，而如果是varchar類(lèi)型，則必須指定。
• 3.nullable對(duì)應(yīng)的就是MySQL中的NULL 和 NOT NULL
• 4.關(guān)于default和server_default: default代表的是ORM框架層面的默認(rèn)值，即插入的時(shí)候如果該字段未賦值，則會(huì)使用我們定義的默認(rèn)值；server_default代表的是數(shù)據(jù)庫(kù)層面的默認(rèn)值，即DDL語(yǔ)句中的default關(guān)鍵字。

Session介紹

在SQLAlchemy的文檔中提到，數(shù)據(jù)庫(kù)的增刪查改是通過(guò)session來(lái)執(zhí)行的。

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)

session = Session()
orm = PyOrmModel(id=1, name='test', attr={})
session.add(orm)

session.commit()
session.close()

如上，我們可以看到，對(duì)于每一次操作，我們都需要對(duì)session進(jìn)行獲取，提交和釋放。這樣未免過(guò)于冗余和麻煩，所以我們一般會(huì)進(jìn)行一層封裝。

1.采用上下文管理器的方式

處理session的異常回滾和關(guān)閉，這部分與所參考的文章是幾乎一致的。

# base_model.py
from contextlib import contextmanager
from sqlalchemy.orm import sessionmaker, scoped_session

def _get_session():
    """獲取session"""
    return scoped_session(sessionmaker(bind=engine, expire_on_commit=False))()

# 在這里對(duì)session進(jìn)行統(tǒng)一管理，包括獲取，提交，回滾和關(guān)閉
@contextmanager
def db_session(commit=True):
    session = _get_session()
    try:
        yield session
        if commit:
            session.commit()
    except Exception as e:
        session.rollback()
        raise e
    finally:
        if session:
            session.close()

2.model和dict轉(zhuǎn)換

在PyOrmModel中增加兩個(gè)方法，用于model和dict之間的轉(zhuǎn)換

class PyOrmModel(Base):
    ...

    @staticmethod
    def fields():
        return ['id', 'name', 'attr']

    @staticmethod
    def to_json(model):
        fields = PyOrmModel.fields()
        json_data = {}
        for field in fields:
            json_data[field] = model.__getattribute__(field)
        return json_data

    @staticmethod
    def from_json(data: dict):
        fields = PyOrmModel.fields()

        model = PyOrmModel()
        for field in fields:
            if field in data:
                model.__setattr__(field, data[field])
        return model

3.數(shù)據(jù)庫(kù)操作的封裝

與參考的文章不同，我是直接調(diào)用了session，從而使調(diào)用方不需要關(guān)注model層，減少耦合。

# py_orm_model_op.py
from sqlachlemy_lab.model import db_session
from sqlachlemy_lab.model import PyOrmModel


class PyOrmModelOp:
    def __init__(self):
        pass

    @staticmethod
    def save_data(data: dict):
        with db_session() as session:
            model = PyOrmModel.from_json(data)
            session.add(model)

    # 查詢(xún)操作，不需要commit
    @staticmethod
    def query_data(pid: int):
        data_list = []
        with db_session(commit=False) as session:
            data = session.query(PyOrmModel).filter(PyOrmModel.id == pid)
            for d in data:
                data_list.append(PyOrmModel.to_json(d))

            return data_list

4.調(diào)用方

# main.py
from sqlachlemy_lab.model_op import PyOrmModelOp


if __name__ == '__main__':
    PyOrmModelOp.save_data({'id': 1, 'name': 'test', 'attr': {}})

完整代碼請(qǐng)參見(jiàn)：  
https://github.com/yangancode/python_lab/tree/master/sqlachlemy_lab

推薦閱讀:
入門(mén): 最全的零基礎(chǔ)學(xué)Python的問(wèn)題  | 零基礎(chǔ)學(xué)了8個(gè)月的Python  | 實(shí)戰(zhàn)項(xiàng)目 |學(xué)Python就是這條捷徑

干貨:爬取豆瓣短評(píng)，電影《后來(lái)的我們》 | 38年NBA最佳球員分析 |   從萬(wàn)眾期待到口碑撲街！唐探3令人失望  | 笑看新倚天屠龍記 | 燈謎答題王 ｜用Python做個(gè)海量小姐姐素描圖 ｜碟中諜這么火，我用機(jī)器學(xué)習(xí)做個(gè)迷你推薦系統(tǒng)電影

趣味:彈球游戲  | 九宮格  | 漂亮的花 | 兩百行Python《天天酷跑》游戲!

AI: 會(huì)做詩(shī)的機(jī)器人 | 給圖片上色 | 預(yù)測(cè)收入 | 碟中諜這么火，我用機(jī)器學(xué)習(xí)做個(gè)迷你推薦系統(tǒng)電影

小工具: Pdf轉(zhuǎn)Word，輕松搞定表格和水印！ | 一鍵把html網(wǎng)頁(yè)保存為pdf！|  再見(jiàn)PDF提取收費(fèi)！ | 用90行代碼打造最強(qiáng)PDF轉(zhuǎn)換器，word、PPT、excel、markdown、html一鍵轉(zhuǎn)換 | 制作一款釘釘?shù)蛢r(jià)機(jī)票提示器！ ｜60行代碼做了一個(gè)語(yǔ)音壁紙切換器天天看小姐姐！｜


年度爆款文案
1).臥槽！Pdf轉(zhuǎn)Word用Python輕松搞定！
2).學(xué)Python真香！我用100行代碼做了個(gè)網(wǎng)站，幫人PS旅行圖片，賺個(gè)雞腿吃
3).首播過(guò)億，火爆全網(wǎng)，我分析了《乘風(fēng)破浪的姐姐》，發(fā)現(xiàn)了這些秘密 
4).80行代碼！用Python做一個(gè)哆來(lái)A夢(mèng)分身 
5).你必須掌握的20個(gè)python代碼，短小精悍，用處無(wú)窮 
6).30個(gè)Python奇淫技巧集 
7).我總結(jié)的80頁(yè)《菜鳥(niǎo)學(xué)Python精選干貨.pdf》,都是干貨 
8).再見(jiàn)Python！我要學(xué)Go了！2500字深度分析！
9).發(fā)現(xiàn)一個(gè)舔狗福利！這個(gè)Python爬蟲(chóng)神器太爽了，自動(dòng)下載妹子圖片

點(diǎn)閱讀原文，看B站我的20個(gè)視頻！