背??景其实一开始我用的是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.pymappingdeclaration(Modelintroduction)比如我们有这样一个测试表createtablepy_orm(`id`int(11)NOTNULLAUTO_INCREMENTCOMMENT'uniqueid',`name`varchar(255)NOTNULLDEFAULT''COMMENT'name',`attr`JSONNOTNULLCOMMENT'attribute',`ct`timestampNOTNULLDEFAULTCURRENT_TIMESTAMPCOMMENT'createdtime',`ut`timestampNOTNULLDEFAULTCURRENT_TIMESTAMPONRUPDATE',PRIMARYKEY(`id`))ENGINE=InnoDBCOMMENT'测试表';在ORM框架中，映射的结果是下面的Python类整数，autoincrement=True，primary_key=True，comment='uniqueid')name=Column(String(255),nullable=False,default='',comment='name')attr=Column(JSON,nullable=False,普通t='属性')ct=Column(TIMESTAMP,nullable=False,server_default=text('CURRENT_TIMESTAMP'),comment='创建时间')ut=Column(TIMESTAMP,nullable=False,server_default=text('CURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMP'),comment='updatetime')首先我们可以看到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_TIMESTAMPONUPDATECURRENT_TIMESTAMP,inPRIMARYKEYNOTNULLDEFAULTCURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMP,inPRIMARYKEYMySQL的字段(idkey)很容易理解1.主键和自增属性2.如果是int类型可以不用指定长度，但是如果是是varchar类型，必须指定3.Nullable对应MySQL中的NULL和NOTNULL4.关于default和server_default：default表示ORM框架级别的默认值，即如果字段没有赋值插入时，我们定义的默认值将使用ed；server_default表示数据库级别的默认值，即DDL语句中的default关键字。Session的介绍在SQLAlchemy的文档中有提到，对数据库的增删改查都是通过session进行的。>>>从sqlalchemy.orm导入sessionmaker>>>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))()#这里统一管理，包例如获取，提交，@contextmanagerdefdb_session（commit=true）：session=_get_session（）try：sessionsession.commit（session.commit（）excess.commit（）extectextectaextescionase：session.rollback（session.rollback（）．to_json(模型)：字段=PyOrmModel.fields()json_data={}用于字段中的字段：json_data[field]=model.__getattribute__(field)返回json_data@staticmethoddeffrom_json(数据):fields=PyOrmModel.fields()model=PyOrmModel()forfieldinfields:iffieldindata:model.__setattr__(field,data[field])不同于直接参考文章model3。session被调用，调用者不需要关注model层，降低耦合＃py_orm_model_op.pyfromsqlachlemy_lab.model导入db_sessionfromsqlachlemy_lab.model导入pyormmmodelclclasspyormmodelop：def__init__Init__（self）：pass@staticmethoddef_data：save_data（dec）add(model)#查询操作，不需要提交@staticmethoddefquery_data(pid:int):data_list=[]withdb_session(commit=False)assession:data=session.query(PyOrm)..filter(PyOrm)=数据中的d数据：data_list.append（pyormmodel.to_json（d））返回data_list4。,'名称':'测试','属性':{}})
