Flask 学习 六 大型程序结构

pip freeze >requirement.txt 自动生成版本号

pip install -r requirement.txt 自动下载对应的库

梳理结构

config.py

#！/usr/bin/env python
# -*- coding:utf-8 -*-
import os
basedir = os.path.abspath(os.path.dirname(__file__))
class Config:
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'hard to guess string'  # 可以用来存储框架，扩展，程序等的配置变量
    SQLALCHEMY_COMMIT_ON_TEARDOWN = True  # 每次请求结束后自动提交数据库的变动
    FLASKY_MAIL_SUBJECT_PREFIX= '[Flasky]'
    FLASKY_MAIL_SENDER= 'Flasky Admin <flasky@example.com>'  # 发件人
    FLASKY_ADMIN = os.environ.get('FLASKY_ADMIN')  # 收件人

    @staticmethod
    def init_app(app):
        pass

class DevelopmentConfig(Config):
    DEBUG=True
    # 邮件配置
    MAIL_SERVER= 'smtp.qq.com'
    MAIL_PORT = 465
    MAIL_USE_TLS = True
    MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
    MAIL_PASSWORD= os.environ.get('MAIL_PASSWORD')
    SQLALCHEMY_DATABASE_URI= os.environ.get('DEV_DATABASE_URL') or 'sqlite:///' + os.path.join(basedir, 'data-dev.sqlite')

class TestingConfig(Config):
    TESTING=True
    SQLALCHEMY_DATABASE_URI = os.environ.get('TEST_DATABASE_URL') or 'sqlite:///' + os.path.join(basedir,'data-test.sqlite')

class ProductionConfig(Config):
    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///' + os.path.join(basedir,'data.sqlite')

config={
    'development':DevelopmentConfig,
    'testing':TestingConfig,
    'production':ProductionConfig,
    'default':DevelopmentConfig
}

app/__init__.py   使用程序工厂函数

#！/usr/bin/env python
# -*- coding:utf-8 -*-
from flask import Flask
from flask_moment import Moment
from flask_bootstrap import Bootstrap
from flask_mail import Mail
from flask_sqlalchemy import SQLAlchemy
from config import config

bootstrap = Bootstrap()
mail =Mail()
moment = Moment()
db = SQLAlchemy()

def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)
    bootstrap.init_app(app)
    mail.init_app(app)
    moment.init_app(app)
    db.init_app(app)
    
    # 附加使用蓝本路由和错误页面
    from . import main as main_blueprint
    app.register_blueprint(main_blueprint)
    return app

蓝本中实现路由和自定义错误页面

创建蓝本 app/main/__init__.py

#！/usr/bin/env python
# -*- coding:utf-8 -*-
from flask import Blueprint
main = Blueprint('main',__name__)
from . import views,errors

注册蓝本 app/__init__.py 

# 附加使用蓝本路由和错误页面
from .main import main as main_blueprint
app.register_blueprint(main_blueprint)

app/main/error.py

#！/usr/bin/env python
# -*- coding:utf-8 -*-
from flask import render_template
from . import main

#主程序的errorhandler
@main.errorhandler(404)
def page_not_find(e):
    return render_template('404.html'), 404

@main.errorhandler(500)
def internal_server_error(e):
    return render_template('500.html'), 500

main/views.py

#！/usr/bin/env python
# -*- coding:utf-8 -*-
from flask import render_template,session,redirect,url_for,current_app
from . import main
from .forms import NameForm
from .. import db
from ..models import User
from ..email import send_mail

# 使用蓝本自定义路由
@main.route('/', methods=['get', 'post'])
def index():
    #name = None
    form = NameForm()
    if form.validate_on_submit():
        user = User.query.filter_by(username=form.name.data).first()
        if user is None:
            user = User(username=form.name.data)
            db.session.add(user)
            session['known']=False
            if current_app.config['FLASKY_ADMIN']:
                send_mail(current_app.config['FLASKY_ADMIN'],'New user','mail/new_user',user=user)
        else:
            session['known'] = True
        session['name']=form.name.data
        form.name.data=''
        return redirect(url_for('.index'))# 蓝本中index函数在main.index下

return render_template('index.html', name=session.get('name'), form=form, known=session.get('known',False))

main/forms.py

#！/usr/bin/env python
# -*- coding:utf-8 -*-
from flask_wtf import FlaskForm
from wtforms import StringField, SubmitField
from wtforms.validators import DataRequired

class NameForm(FlaskForm):
    name = StringField('姓名', validators=[DataRequired()])
    submit = SubmitField('提交')

app/email.py

#！/usr/bin/env python
# -*- coding:utf-8 -*-
from threading import Thread
from flask import render_template,current_app
from . import mail
from flask_mail import Message
def send_async_email(app,msg):
    with app.app_context():
        mail.send(msg)

def send_mail(to,subject,template,**kwargs):
    app = current_app._get_current_object()
    msg = Message(app.config['FLASKY_MAIL_SUBJECT_PREFIX'] + subject,sender=app.config['FLASKY_MAIL_SENDER'],recipients=[to])
    msg.body=render_template(template + '.txt',**kwargs)
    msg.html = render_template(template + '.html', **kwargs)
    thr = Thread(target=send_async_email,args=[app,msg])
    thr.start()
    return thr

app/models.py

#！/usr/bin/env python
# -*- coding:utf-8 -*-
from app import db
class Role(db.Model):
    __tablename__='roles'
    id = db.Column(db.Integer,primary_key=True)
    name = db.Column(db.String(64),unique=True)
    users = db.relationship('User',backref='role',lazy='dynamic')

    def __repr__(self):
        return '<Role %r>'% self.name


class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(64), unique=True,index=True)
    role_id=db.Column(db.Integer,db.ForeignKey('roles.id'))

    def __repr__(self):
        return '<User %r>' % self.username

manage.py

#！/usr/bin/env python
# -*- coding:utf-8 -*-
import os
from app import create_app,db
from app.models import User,Role
from flask_script import Manager,Shell
from flask_migrate import Migrate,MigrateCommand

app=create_app(os.getenv('FLASK_CONFIG')or 'default')
manager = Manager(app)
# 数据库迁移
migrate = Migrate(app,db)
manager.add_command('db',MigrateCommand)
# 集成python shell
def make_shell_context():
    return dict(app=app, db=db, User=User, Role=Role)
manager.add_command("shell", Shell(make_context=make_shell_context))

@manager.command
def test():
    '''启动单元测试'''
    import unittest
    tests=unittest.TestLoader().discover('tests')
    unittest.TextTestRunner(verbosity=2).run(tests)

if __name__ == '__main__':
    manager.run()

requirements.txt

alabaster==0.7.9
alembic==0.9.2
anaconda-client==1.6.0
anaconda-navigator==1.5
anaconda-project==0.4.1
astroid==1.4.9
astropy==1.3
attrs==16.3.0
Automat==0.5.0
Babel==2.3.4
backports.shutil-get-terminal-size==1.0.0
beautifulsoup4==4.5.3
bitarray==0.8.1
blaze==0.10.1
blinker==1.4
bokeh==0.12.4
boto==2.45.0
Bottleneck==1.2.0
cffi==1.9.1
chardet==2.3.0
chest==0.2.3
click==6.7
cloudpickle==0.2.2
clyent==1.2.2
colorama==0.3.7
comtypes==1.1.2
conda==4.3.16
configobj==5.0.6
constantly==15.1.0
contextlib2==0.5.4
cryptography==1.7.1
cssselect==1.0.1
cycler==0.10.0
Cython==0.25.2
cytoolz==0.8.2
dask==0.13.0
datashape==0.5.4
decorator==4.0.11
dill==0.2.5
Django==1.11
docutils==0.13.1
dominate==2.3.1
et-xmlfile==1.0.1
fastcache==1.0.2
Flask==0.12
Flask-Bootstrap==3.3.7.1
Flask-Cors==3.0.2
Flask-Mail==0.9.1
Flask-Migrate==2.0.3
Flask-Moment==0.5.1
Flask-Script==2.0.5
Flask-SQLAlchemy==2.2
Flask-WTF==0.14.2
get==0.0.0
gevent==1.2.1
greenlet==0.4.11
h5py==2.6.0
HeapDict==1.0.0
idna==2.2
imagesize==0.7.1
incremental==16.10.1
ipykernel==4.5.2
ipython==5.1.0
ipython-genutils==0.1.0
ipywidgets==5.2.2
isort==4.2.5
itsdangerous==0.24
jdcal==1.3
jedi==0.9.0
jieba==0.38
Jinja2==2.9.4
jsonschema==2.5.1
jupyter==1.0.0
jupyter-client==4.4.0
jupyter-console==5.0.0
jupyter-core==4.2.1
lazy-object-proxy==1.2.2
llvmlite==0.15.0
locket==0.2.0
lxml==3.7.2
Mako==1.0.6
MarkupSafe==0.23
matplotlib==2.0.0
menuinst==1.4.4
mistune==0.7.3
mpmath==0.19
multipledispatch==0.4.9
nbconvert==4.2.0
nbformat==4.2.0
networkx==1.11
nltk==3.2.2
nose==1.3.7
notebook==4.3.1
numba==0.30.1
numexpr==2.6.1
numpy==1.11.3
numpydoc==0.6.0
odo==0.5.0
olefile==0.44
openpyxl==2.4.1
pandas==0.19.2
parsel==1.1.0
partd==0.3.7
path.py==0.0.0
pathlib2==2.2.0
patsy==0.4.1
pep8==1.7.0
pickleshare==0.7.4
Pillow==4.0.0
ply==3.9
post==0.0.0
prompt-toolkit==1.0.9
psutil==5.0.1
public==0.0.0
py==1.4.32
pyasn1==0.1.9
pyasn1-modules==0.0.8
pycosat==0.6.1
pycparser==2.17
pycrypto==2.6.1
pycurl==7.43.0
PyDispatcher==2.0.5
pyflakes==1.5.0
Pygments==2.1.3
pylint==1.6.4
PyMySQL==0.7.11
pyOpenSSL==16.2.0
pyparsing==2.1.4
pytest==3.0.5
python-dateutil==2.6.0
python-editor==1.0.3
pytz==2016.10
pywin32==220
PyYAML==3.12
pyzmq==16.0.2
QtAwesome==0.4.3
qtconsole==4.2.1
QtPy==1.2.1
query-string==0.0.0
queuelib==1.4.2
request==0.0.0
requests==2.12.4
rope-py3k==0.9.4.post1
scikit-image==0.12.3
scikit-learn==0.18.1
scipy==0.18.1
Scrapy==1.3.3
seaborn==0.7.1
service-identity==16.0.0
setupfiles==0.0.0
simplegeneric==0.8.1
singledispatch==3.4.0.3
six==1.10.0
snowballstemmer==1.2.1
sockjs-tornado==1.0.3
sphinx==1.5.1
spyder==3.1.2
SQLAlchemy==1.1.5
statsmodels==0.6.1
sympy==1.0
tables==3.2.2
toolz==0.8.2
tornado==4.4.2
traitlets==4.3.1
Twisted==17.1.0
unicodecsv==0.14.1
visitor==0.1.3
w3lib==1.17.0
wcwidth==0.1.7
Werkzeug==0.11.15
widgetsnbextension==1.2.6
win-unicode-console==0.5
wordcloud==1.3.1
wrapt==1.10.8
WTForms==2.1
xlrd==1.0.0
XlsxWriter==0.9.6
xlwings==0.10.2
xlwt==1.2.0
zope.interface==4.3.3

单元测试

test/test_basic.py

#！/usr/bin/env python
# -*- coding:utf-8 -*-
import unittest
from flask import current_app
from app import create_app,db

class BasicTestCase(unittest.TestCase):
    def setUp(self):
        self.app = create_app('testing')
        self.app_context=self.app.app_context()
        self.app_context.push()
        db.create_all()
    def tearDown(self):
        db.session.remove()
        db.drop_all()
        self.app_context.pop()
    def test_app_exists(self):
        self.assertFalse(current_app is None)
    def test_app_is_testing(self):
        self.assertTrue(current_app.config['TESTING'])

单元测试注册到主程序manager.py

@manager.command
def test():
    '''启动单元测试'''
    import unittest
    tests=unittest.TestLoader().discover('tests')
    unittest.TextTestRunner(verbosity=2).run(tests)

1、创建数据库

python manage.py shell

from app import db

db.create_all()

2、创建数据库迁移

python hello.py db init # 创建 migrations 文件夹，所有迁移脚本都存放其中

3、创建迁移脚本

python manage.py db migrate 

4、更新数据库

python manage.py db upgrade