使用SQLAlchemy进行ORM操作
2022/1/29 19:08:47
本文主要是介绍使用SQLAlchemy进行ORM操作,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!
一、参考文档
http://www.pythondoc.com/flask-sqlalchemy/quickstart.html (官方文档)
https://www.jianshu.com/p/f454a1aa760c
二、ORM解释
Object (code) - Relational (Database) - Mapping
数据库的表(table) --> 类(class)
记录(record,行数据)--> 对象(object)
字段(field)--> 对象的属性(attribute)
三、封装工作
1.Database → Class
使用以下指令生成models文件
sqlacodegen mysql+pymysql://用户名:密码@主机:端口号/数据库名 > 数据库对象存放路径
2.数据库连接
import pymysql from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker class DBSession(object): def __init__(self, db_name=database): engine = create_engine('mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8'.format(user, passwd, host, port, db_name), echo=False) Session = sessionmaker() Session.configure(bind=engine) self.session = Session(autocommit=True) # 可以这样用 db = DBSession() db.session.query(AppNotTest).filter(AppNotTest.appId == app_id).update({AppNotTest.status: 1, AppNotTest.updatedBy: 'tester'}, synchronize_session=False) db.session.flush() # 数据提交 db.session.close() # 数据库连接关闭
3.装饰器组装
def db_fixture(db_name='ibupltplatformdb'): class DBFixture(): def __init__(self): self.db_name = db_name def __db_conn(self): self.db = DBSession(self.db_name) self.db_session = self.db.session return self.db_session def __db_finalizer(self): self.db_session.flush() self.db_session.close() def __call__(self, func): def wrapper(*args, **kwargs): session = self.__db_conn() try: res = func(session=session, *args, **kwargs) return res except Exception as ex: logging.error('[DB ERROR] {}'.format(ex)) session.rollback() finally: self.__db_finalizer() return wrapper return DBFixture()
四、使用
1.查
@db_fixture(db_name='platformdb') def query_app_not_test_by_id(session, app_id): """ 1. 查:根据app_id, 等同于SELECT * FROM app_not_test WHERE app_id= ..; :param session: 由装饰器传,调用该方法时无需传此参数 :param app_id: 调用该方法时传此参数 :return: """ record = session.query(AppNotTest).filter(AppNotTest.appId == app_id).scalar() #scalar():返回结果小于等于1 return record @db_fixture() def query_app_not_test_by_app_list(session, app_list): # 2. 批量查询 records = session.query(AppNotTest).filter(AppNotTest.appId.in_(app_list)).all() return records @db_fixture() def query_app_not_test_with_group(session, app_id): # 3. 联表查询 record = session.query(AppNotTest.appId.label('app'), Appgroup.groupname.label('group')).\ filter(AppNotTest.appId == app_id, AppNotTest.appId == Appgroup.appid).all() return record @db_fixture() def query_distinct_group_name(session): # 4. group by使用 records = session.query(Appgroup.groupname, func.count(func.distinct(Appgroup.groupname))).group_by(Appgroup.groupname).all() return records @db_fixture() def query_app_not_test_by_pagination(session, limit, offset): # 5. 分页查询 records = session.query(AppNotTest).order_by(AppNotTest.datachange_lasttime.desc()).offset(offset).limit(limit).all() return records @db_fixture() def query_by_sql(session, app_id): # 6. sql查询,结果是元组而非object sql = 'SELECT * FROM app_not_test WHERE appId = :app_id' records = session.execute(text(sql), {'app_id': app_id}).fetchall() return records
2. 增
@db_fixture() def add_app_not_test(session, app_id, status): record = app_not_test_assembler(app_id, status) session.add(record) def app_not_test_assembler(app_id, status): app_not_test = AppNotTest() app_not_test.appId = app_id app_not_test.status = status app_not_test.createdBy = 'tester' app_not_test.updatedBy = 'tester' return app_not_test
3.改
@db_fixture() def update_app_not_test_status(session, app_id, status): session.query(AppNotTest).filter(AppNotTest.appId == app_id).update({AppNotTest.status: status}, synchronize_session=False)
4.删
@db_fixture() def delete_app_not_test_by_app_id(session, app_id): session.query(AppNotTest).filter(AppNotTest.appId == app_id).delete(synchronize_session='fetch')
五、其它
1、在使用 create_engine创建引擎时,如果默认不指定连接池设置的话,一般情况下,SQLAlchemy会使用一个 QueuePool绑定在新创建的引擎上。并附上合适的连接池参数。
2、在这种情况下,当你使用了session后就算显式地调用session.close(),也不能把连接关闭。连接会由QueuePool连接池进行管理并复用。
3、如果想禁用SQLAlchemy提供的数据库连接池,只需要在调用create_engine是指定连接池为NullPool,SQLAlchemy就会在执行session.close()后立刻断开数据库连接。当然,如果session对象被析构但是没有被调用session.close(),则数据库连接不会被断开,直到程序终止。
from sqlalchemy.pool import NullPool #连接数据库 def getEngine(): engine = create_engine('mysql://root:123456@127.0.0.1/baa?charset=utf8mb4', poolclass=NullPool, echo=False) #print(engine) return engine def getSession(): engine = getEngine() BaseMode.metadata.create_all(engine)## 数据库生成表 # Session = sessionmaker(bind=engine) # session = Session() DBSession = sessionmaker(bind=engine) session = DBSession() return session
这篇关于使用SQLAlchemy进行ORM操作的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!
- 2024-11-01UniApp 中组件的生命周期是多少-icode9专业技术文章分享
- 2024-11-01如何使用Svg Sprite Icon简化网页图标管理
- 2024-10-31Excel数据导出课程:新手从入门到精通的实用教程
- 2024-10-31Excel数据导入课程:新手入门指南
- 2024-10-31RBAC的权限课程:新手入门教程
- 2024-10-31Svg Sprite Icon课程:新手入门必备指南
- 2024-10-31怎么配置 L2TP 允许多用户连接-icode9专业技术文章分享
- 2024-10-31怎么在FreeBSD上 安装 OpenResty-icode9专业技术文章分享
- 2024-10-31运行 modprobe l2tp_ppp 时收到“module not found”消息提醒是什么-icode9专业技术文章分享
- 2024-10-31FreeBSD的下载命令有哪些-icode9专业技术文章分享