@contextmanager:Python实现with结构的好
2021-08-25 本文已影响0人
空口言_1d2e
常见with结构：
# 常规写法
f = open("1.txt")
f.write("hello world")
f.close()
# 优化 采用with结构
with open("1.txt","w") as f:
f.write("hello world")
with 结构的好处，不会因为忘了关闭文件而造成不必要的错误。类似的，当我们在某些时候不希望遗忘一些重要语句的时候，可以自己封装个with结构，比如关闭数据库链接等情况
一般实现方法
with结构一般的实现方法是在定义类的时候重载enter方法和exit方法
class myOpen():
def __init__(self, name, state):
self.f = open(name, state)
def __enter__(self):
return self.f
def __exit__(self, exc_type, exc_val, exc_tb):
self.f.close()
if __name__ == '__main__':
with myOpen("1.txt", "w") as f:
f.write("hello world")
高逼格的
@contextmanager方法
from contextlib import contextmanager
@contextmanager
def myOpen(name, state):
try:
f = open(name, state)
yield f
finally:
f.close()
if __name__ == "__main__":
with myOpen("test.txt", "w") as f:
f.write("hello world!")
源码
def contextmanager(func):
"""@contextmanager decorator.
Typical usage:
@contextmanager
def some_generator(<arguments>):
<setup>
try:
yield <value>
finally:
<cleanup>
This makes this:
with some_generator(<arguments>) as <variable>:
<body>
equivalent to this:
<setup>
try:
<variable> = <value>
<body>
finally:
<cleanup>
"""
@wraps(func)
def helper(*args, **kwds):
return GeneratorContextManager(func(*args, **kwds))
return helper
flask-SQLAlchemy数据库模型插入数据的时候使用session.commit()必须处理异常回滚db.session.rollback()
from flask_sqlalchemy import SQLAlchemy as BaseSQLAlchemy
from contextlib import contextmanager
#自定义一个SQLAlchemy继承flask_sqlalchemy的,方便自定义方法！！！
class SQLAlchemy(BaseSQLAlchemy):
#利用contextmanager管理器,对try/except语句封装，使用的时候必须和with结合！！！
@contextmanager
def auto_commit_db(self):
try:
yield
self.session.commit()
except Exception as e:
# 加入数据库commit提交失败，必须回滚！！！
self.session.rollback()
raise e
db = SQLAlchemy()
利用with上下文管理方法调用！
from kirin_app.db_models.table_user import User
from kirin_app.db_models import db
class RegisterViewModel:
def __init__(self,form_data):
self.email = form_data["email"]
self.nickname = form_data["nickname"]
self.password = form_data["password"]
self.__add_db_data()
# 把数据添加进入数据库
def __add_db_data(self):
with db.auto_commit_db():
user_db = User(email=self.email, nickname=self.nickname, password=self.password)
db.session.add(user_db)
image
