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实例以后在项目中可以实现搞笑的数据库操作。


内容版权声明:除非注明,否则皆为本站原创文章。

转载注明出处:http://www.sulao.cn/post/1001

相关推荐