与烧瓶集成了未染色的炼金术
flask-sqlalchemy-unchained的Python项目详细描述
烧瓶sqlalchemy unchen
集成SQLAlchemy Unchained和Flask。这个包是围绕Flask-SQLAlchemy的一个非常薄的包装，就在flask中注册扩展而言，一切都是一样的。
基本用法
# your_app.pyfromflaskimportFlaskfromflask_sqlalchemy_unchainedimportSQLAlchemyUnchainedapp=Flask(__name__)app.config['SQLALCHEMY_DATABASE_URI']='sqlite:///:memory:'db=SQLAlchemyUnchained(app)classUser(db.Model):classMeta:repr=('id','username','email')username=db.Column(db.String(80),unique=True,nullable=False)email=db.Column(db.String(120),unique=True,nullable=False)
现在，让我们创建表并添加一行：
exportFLASK_APP='./your_app.py'exportFLASK_DEBUG='true' flask shell
>>> from your_app import db, User
>>> db.create_all()
>>> user = User(username='fooar', email='foo@bar.com')
>>> db.session.add(user)
>>> db.session.commit()
>>> assert User.query.all() == [user]
实际使用情况
现在让我们看看如何使用应用程序工厂模式。我们的应用程序的目录结构如下：
./your-project
├── app
│ ├── models
│ │ ├── __init__.py
│ │ └── user.py
│ ├── services
│ │ ├── __init__.py
│ │ ├── model_manager.py
│ │ └── user_manager.py
│ ├── __init__.py
│ ├── config.py
│ ├── extensions.py
│ └── factory.py
├── db
│ └── dev.sqlite
├── tests
│ ├── __init__.py
│ └── test_user.py
├── autoapp.py
└── setup.py
我们的应用程序的入口点将是autoapp.py，所以让我们先看看这个：
# app/autoapp.pyimportosfromapp.factoryimportcreate_appapp=create_app(os.getenv('FLASK_ENV','development'))
现在应用程序工厂：
# app/factory.pyfromflaskimportFlaskfrom.configimportDevConfig,ProdConfig,TestConfigfrom.extensionsimportdbCONFIGS={'development':DevConfig,'production':ProdConfig,'test':TestConfig,}defcreate_app(env):config=CONFIGS[env]app=Flask(__name__,template_folder=config.TEMPLATE_FOLDER,static_folder=config.STATIC_FOLDER,static_url_path=config.STATIC_URL_PATH)app.config.from_object(config)db.init_app(app)returnapp
这就引出了config和extensions模块：
# app/config.pyimportosclassBaseConfig:DEBUG=os.getenv('FLASK_DEBUG',False)APP_ROOT=os.path.abspath(os.path.dirname(__file__))PROJECT_ROOT=os.path.abspath(os.path.join(APP_ROOT,os.pardir))SQLALCHEMY_TRACK_MODIFICATIONS=FalseclassDevConfig(BaseConfig):DEBUG=os.getenv('FLASK_DEBUG',True)db_path=os.path.join(BaseConfig.PROJECT_ROOT,'db','dev.sqlite')SQLALCHEMY_DATABASE_URI='sqlite:///'+db_pathclassProdConfig(BaseConfig):SQLALCHEMY_DATABASE_URI= \ '{engine}://{user}:{password}@{host}:{port}/{db_name}'.format(engine='postgresql+psycopg2',user=os.getenv('FLASK_DATABASE_USER','sqlalchemy_demo'),password=os.getenv('FLASK_DATABASE_PASSWORD','sqlalchemy_demo'),host=os.getenv('FLASK_DATABASE_HOST','127.0.0.1'),port=os.getenv('FLASK_DATABASE_PORT',5432),db_name=os.getenv('FLASK_DATABASE_NAME','sqlalchemy_demo'))classTestConfig(BaseConfig):TESTING=TrueDEBUG=TrueSQLALCHEMY_DATABASE_URI='sqlite://'# :memory:
# app/extensions.pyfromflask_sqlalchemy_unchainedimportSQLAlchemyUnchaineddb=SQLAlchemyUnchained()
User模型与之前相同：
# app/models/user.pyfromapp.extensionsimportdbclassUser(db.Model):classMeta:repr=('id','username','email')username=db.Column(db.String(80),unique=True,nullable=False)email=db.Column(db.String(120),unique=True,nullable=False)
因为sqlalchemy使用数据映射器模式，所以最好使用管理器/服务来处理与数据库的交互。一个好的起点可能是这样的：
# app/services/model_manager.pyfromtypingimport*fromapp.extensionsimportdbclassModelManager:model:Type[db.Model]defcreate(self,commit:bool=False,**kwargs)->db.Model:instance=self.model(**kwargs)self.save(instance,commit)returninstancedefupdate(self,instance:db.Model,commit:bool=False,**kwargs)->db.Model:forattr,valueinkwargs.items():setattr(instance,attr,value)self.save(instance,commit)returninstancedefdelete(self,instance:db.Model,commit:bool=False)->None:db.session.delete(instance)ifcommit:self.commit()defsave(self,instance:db.Model,commit:bool=True):db.session.add(instance)ifcommit:self.commit()defcommit(self)->None:db.session.commit()defrollback(self)->None:db.session.rollback()defget(self,id)->db.Model:returndb.session.query(self.model).get(int(id))defget_by(self,**kwargs)->db.Model:returndb.session.query(self.model).filter_by(**kwargs).first()deffind_all(self)->List[db.Model]:returndb.session.query(self.model).all()deffind_by(self,**kwargs)->List[db.Model]:returndb.session.query(self.model).filter_by(**kwargs).all()
然后UserManager类将如下所示：
# app/services/user_manager.pyfrom..modelsimportUserfrom.model_managerimportModelManagerclassUserManager(ModelManager):model=Userdefcreate(self,username,email,**kwargs)->User:returnsuper().create(username=username,email=email,**kwargs)user_manager=UserManager()
这个示例应用程序的完整源代码，包括与Flask-Migrate和Py-YAML-Fixtures的集成，可以在on GitHub中找到。
欢迎加入QQ群-->： 979659372
