其實一開始用的是pymysql，但發現維護比較麻煩，還存在程式碼注入的風險，所以就乾脆直接用ORM框架。
ORM即Object Relational Mapper，可以簡單理解為資料庫表和Python類別之間的映射，透過操作Python類，可以間接操作資料庫。
Python的ORM框架比較有名的是SQLAlchemy和Peewee，這裡不做比較，只是單純講解個人對SQLAlchemy的一些使用，希望能給各位朋友帶來幫助。
from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://$user:$password@$host:$port/$db?charset=utf8mb4', echo=True)├── main.py ├── model │ ├── __init__.py │ ├── base_model.py │ ├── ddl.sql │ └── py_orm_model.py └── model_op ├── __init__.py └── py_orm_model_op.py
create table py_orm ( `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '唯一id', `name` varchar(255) NOT NULL DEFAULT '' COMMENT '名称', `attr` JSON NOT NULL COMMENT '属性', `ct` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', `ut` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON update CURRENT_TIMESTAMP COMMENT '更新时间', PRIMARY KEY(`id`) )ENGINE=InnoDB COMMENT '测试表';
# 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='名称')
attr = Column(JSON, nullable=False, comment='属性')
ct = Column(TIMESTAMP, nullable=False, server_default=text('CURRENT_TIMESTAMP'), comment='创建时间')
ut = Column(TIMESTAMP, nullable=False, server_default=text('CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP'), comment='更新时间')# 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:123456@127.0.0.1:33306/orm_test?charset=utf8mb4", echo=False)# main.py # 注意这里的导入路径，Base创建表时会寻找继承它的子类，如果路径不对，则无法创建成功 from sqlachlemy_lab import Base, engine if __name__ == '__main__': Base.metadata.create_all(engine)
... 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 '名称', 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) )
>>> 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()# 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))() # 在这里对session进行统一管理，包括获取，提交，回滚和关闭 @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()
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# 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) # 查询操作，不需要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
# main.py
from sqlachlemy_lab.model_op import PyOrmModelOp
if __name__ == '__main__':
PyOrmModelOp.save_data({'id': 1, 'name': 'test', 'attr': {}})以上是五分鐘菜鳥學會Python玩轉SQL的神器！的詳細內容。更多資訊請關注PHP中文網其他相關文章！
