flask工厂函数集成peewee(ORM)的方法
- 2025-03-06 11:54:14
- 开发
- 147
- shevechco
flask工厂函数主要是用来动态创建应用实例的,我们在一般应用中都需要将peewee数据库初始化的动作放到这个工厂函数里,之前有学习过使用sqlalchemy集成到工厂函数,可以查看这个笔记:http://www.sulao.cn/post/658.html
在工厂函数中需要一个合适的时机来初始化数据库,之前在使用sqlalchemy的时候我们是实例化以后使用init_app函数来挂载到app实例中,peewee没有这个方法,但是有一个数据库代理的Proxy方法,这样就可以延迟初始化到创建app实例的时候,这样通过代理绑定的数据库可以适应不同的场景,下面我们来看下代码:
首先我们在./apps/__init__.py中创建了data_proxy实例
#!/usr/bin/python3 #coding:utf-8 from peewee import Proxy data_proxy = Proxy()
然后我们需要定义表映射的类./apps/modules/Data.py
#!/usr/bin/python3 #coding:utf-8 from apps import data_proxy from peewee import * import datetime class BaseModel(Model): class Meta: database = data_proxy class BPost(BaseModel): id = AutoField() cate_id = SmallIntegerField(6) author_id = IntegerField(11) tag = CharField(max_length=255) alias = CharField(max_length=255) is_top = SmallIntegerField(1) title = CharField(max_length=255) content = TextField() view_num = IntegerField(11) post_time = DateTimeField(formats='%Y-%m-%d %H:%M:%S', default=datetime.datetime.now) update_time = DateTimeField() class BCate(BaseModel): id = AutoField() cate_name = CharField(max_length=50) count = IntegerField(11) order = IntegerField(11) alias = CharField(max_length=50) intro = TextField() create_time = DateTimeField(formats='%Y-%m-%d %H:%M:%S', default=datetime.datetime.now) update_time = DateTimeField() class BConf(BaseModel): id = AutoField() name = CharField(max_length=50) content = TextField() key = CharField(max_length=50) class BUser(BaseModel): id = AutoField() name = CharField(max_length=50) account = CharField(max_length=50) passwd = CharField(max_length=50) level = SmallIntegerField(default=0) status = SmallIntegerField(default=0) email = CharField(max_length=50) remote_ip = CharField(max_length=15) intro = TextField() create_time = DateTimeField(formats='%Y-%m-%d %H:%M:%S', default=datetime.datetime.now) update_time = DateTimeField()
最后在工厂函数中初始化这个数据库代理/apps/application.py
#!/usr/bin/python3 #coding:utf-8 __author__ = 'yang.su' from flask import Flask from peewee import MySQLDatabase from apps import data_proxy #from apps.modules.Data import * from apps.admin import admint from apps.post import article import os import time import logging def create_app(): app = Flask(__name__) app.secret_key = 'yang.su' app.config.from_pyfile("../settings.py") db = MySQLDatabase(app.config["DB_NAME"], user=app.config["DB_USER"], password=app.config["DB_PASS"], host=app.config["DB_HOST"], port=app.config["DB_PORT"]) data_proxy.initialize(db) @app.before_request def bf_request(): data_proxy.connect() @app.after_request def af_request(response): if not data_proxy.is_closed(): data_proxy.close() return response # data_proxy.create_tables([BlogPost, BlogCate, BlogConf, BlogUser]) configure_logger(app) register_blueprint(app) template_filter(app) return app
在工厂函数中使用前置钩子来打开数据库连接,后置方法实现数据库的关闭。
通过上述方法可以高效的集成peewee(ORM),挂载到app实例以后在项目中可以实现搞笑的数据库操作。
内容版权声明:除非注明,否则皆为本站原创文章。