背??景其实一开始我用的是pymysql，但是发现维护起来比较麻烦，还有代码注入的风险，所以索性直接用了ORM框架。ORM全称ObjectRelationalMapper，可以简单理解为数据库表和Python类之间的映射。通过操作Python类，可以间接操作数据库。Python的ORM框架比较有名的有SQLAlchemy和Peewee。这里不做对比，只是简单说明一下个人对SQLAlchemy的一些使用，希望能给朋友们带来帮助。sqlalchemyversion:1.3.15pymysqlversion:0.9.3mysqlversion:5.7初始化一般使用ORM框架，会有一些初始化工作，比如数据库连接，定义基本映射等。以MySQL为例，你只需要传入DSN字符串就可以创建数据库连接。其中echo表示是否输出相应的SQL语句，有助于调试。fromsqlalchemyimportcreate_engineengine=create_engine('mysql+pymysql://$user:$password@$host:$port/$db?charset=utf8mb4',echo=True)对于我个人的设计，在引入ORM框架时，我的项目将参考MVC模式做如下设计。其中，model存放了一些数据库模型，即映射到数据库表的Python类；model_op存储了每个模型对应的操作，即增删改查；当调用者（如main.py）进行数据库操作时，只需要调用model_op层，不用关心model层，从而达到解耦。├──main.py├──model│├──__init__.py│├──base_model.py│├──ddl.sql│└──py_orm_model.py└──model_op├──__init__.py└───py_orm_model_op.py映射声明（模型介绍）比如我们有这样一张测试表。createtablepy_orm(`id`int(11)NOTNULLAUTO_INCREMENTCOMMENT'uniqueid',`name`varchar(255)NOTNULLDEFAULT''COMMENT'name',`attr`JSONNOTNULLCOMMENT'attribute',`ct`timestampNOTNULLDEFAULTCURRENT_TIMESTAMPCOMMENT'创建时间',`ut`timestampNOTNULLDEFAULTCURRENT_TIMESTAMPONupdateCURRENT_TIMESTAMPCOMMENT'更新时间',PRIMARYKEY(`id`))ENGINE=InnoDBCOMMENT'测试表';在ORM框架中，映射的结果是以下Python类。#py_orm_model.pyfrom.base_modelimportBasefromsqlalchemyimportColumn,Integer,String,TIMESTAMP,text,JSONclassPyOrmModel(Base):__tablename__='py_orm'id=Column(Integer,autoincrement=True,primary_key=True,comment='唯一ID')name=Column(String(255),nullable=False,default='',comment='name')attr=Column(JSON,nullable=False,comment='attribute')ct=Column(TIMESTAMP,nullable=False,server_default=text('CURRENT_TIMESTAMP'),comment='创建时间')ut=Column(TIMESTAMP,nullable=False,server_default=text('CURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMP'),comment='更新时间')首先，我们可以看到PyOrmModel继承了Base类，这是sqlalchemy提供的一个基类。它会对我们声明的Python类做一些检查，我会把它放在base_model中。#base_model.py#一般base_model做一些初始化工作fromsqlalchemyimportcreate_enginefromsqlalchemy.ext.declarativeimportdeclarative_base=declarative_base()engine=create_engine("mysql+pymysql://root:123456@127.0.0.1:33306/orm_test?charset=utf8mb4",echo=False)其次，每个Python类都必须包含__tablename__属性，否则找不到对应的表。三、创建数据表有两种方式。第一种当然是在MySQL中手动创建。只要你的Python类定义没有问题，就可以正常运行；第二种是通过orm框架来创建，比如下面这样。#main.py#注意这里的导入路径。Base在创建表的时候，会寻找继承它的子类。如果路径错误，将无法创建成功。fromsqlachlemy_labimportBase,engineif__name__=='__main__':Base.metadata.create_all(engine)创建效果：...2020-04-0410:12:53,974INFOsqlalchemy.engine.base.EngineCREATETABLEpy_orm(idINTEGERNOTNULLAUTO_INCREMENT,nameVARCHAR(255)NOTNULLDEFAULT''COMMENT'name',attrJSONNOTNULLCOMMENT'property',ctTIMESTAMPNOTNULLDEFAULTCURRENT_TIMESTAMP,utTIMESTAMPNOTNULLDEFAULTCURRENT_TIMESTAMP更新CURRENT_TIMESTAMP,PRIMARYKEY(id))四、关于字段属性：1.primary_key和autoincrement比较好理解，就是MySQL的主键和自增属性。2、如果是int类型，则不需要指定长度，如果是varchar类型，则必须指定。3.Nullable对应MySQL中的NULL和NOTNULL4.关于default和server_default：default表示ORM框架层面的默认值，即如果插入时字段没有赋值，我们定义的默认值将使用;server_default表示数据库级别的默认值，即DDL语句中的default关键字。Session的介绍在SQLAlchemy的文档中有提到，对数据库的增删改查都是通过session进行的。>>>fromsqlalchemy.ormimportsessionmaker>>>Session=sessionmaker(bind=engine)>>>session=Session()>>>orm=PyOrmModel(id=1,name='test',attr={})>>>session.add(orm)>>>session.commit()>>>session.close()由上可知，对于每一个操作，我们都需要获取、提交和释放session。这样就太多余了，也太麻烦了，所以我们一般都会进行一层封装。1、使用contextmanager处理session的异常回滚和关闭。这部分与引用的文章几乎一致。#base_model.pyfromcontextlibimportcontextmanagerfromsqlalchemy.ormimportsessionmaker,scoped_sessiondef_get_session():"""Getsession"""returnscoped_session(sessionmaker(bind=engine,expire_on_commit=False))()#这里统一session管理，包括get，提交，回滚和关闭@contextmanagerdefdb_session(commit=True):session=_get_session()try:yieldsessionifcommit:session.commit()exceptExceptionase:session.rollback()raiseefinally:ifsession:session.close()2.在PyOrmModel中添加两个方法，用于model和dict之间的转换。类PyOrmModel(Base):...@staticmethoddeffields():return['id','name','attr']@staticmethoddefto_json(model):fields=PyOrmModel.fields()json_data={}for字段中的字段：json_data[field]=model.__getattribute__(field)returnjson_data@staticmethoddeffrom_json(data:dict):fields=PyOrmModel.fields()model=PyOrmModel()forfieldinfields:iffieldindata:model.__setattr__(field,data[field])返回model3。数据库操作的封装与参考文章不同。我直接调用session，这样调用者就不需要关注model层，降低了耦合度。#py_orm_model_op.pyfromsqlachlemy_lab.modelimportdb_sessionfromsqlachlemy_lab.modelimportPyOrmModelclassPyOrmModelOp:def__init__(self):pass@staticmethoddefsave_data(data:dict):withdb_session()assession:model=PyOrmModel.from_json(data)会话。add(model)#查询操作，不需要commit@staticmethoddefquery_data(pid:int):data_list=[]withdb_session(commit=False)assession:data=session.query(PyOrmModel).filter(PyOrmModel.id==pid)fordindata:data_list.append(PyOrmModel.to_json(d))returndata_list4。调用方法：#main.pyfromsqlachlemy_lab.model_opimportPyOrmModelOpif__name__=='__main__':PyOrmModelOp.save_data({'id':1、'名称':'测试','属性':{}})
