[ Python ] Flask 基於 Web開發 大型程序的結構實例解析

  

  做爲一個編程入門新手,Flask是我接觸到的第一個Web框架。想要深刻學習,就從《FlaskWeb開發:基於Python的Web應用開發實戰》這本書入手,本書因爲是翻譯過來的中文版,理解起來不是很順暢。可是對着代碼理解也是能應對的,學到  第七章:大型程序結構  這章節的時候,發現難度有所提高,網上能參考的完整實例沒有,因而根據本身的理解記下來。html

 

程序結構圖:python

 

 

 

 README

(1)本程序是基於Flask微型Web框架開發,使用Jinja2模版引擎
(2)頁面展現了一個文本框和一個按鈕,輸入文本框點擊按鈕提交,文本框爲空沒法提交(輸入文本框的數據爲一個模擬用戶);
(3)當在文本框中輸入新用戶提交,歡迎詞和文本框中輸入老用戶提交不一致;
(4)文本框輸入新用戶提交後,將新用戶保存至SQLite數據庫,並使用異步發送郵件至管理員郵箱;
(5)頁面刷新,瀏覽器不會再次提示:是否提交


項目結構

flasky     # 程序根目錄
├── app     # 核心模塊目錄
│   ├── email.py     # 郵件發送模版
│   ├── __init__.py
│   ├── main     # 藍圖模塊目錄
│   │   ├── errors.py     # 錯誤處理模塊
│   │   ├── forms.py     # 頁面表單模塊
│   │   ├── __init__.py
│   │   └── views.py     # 正常處理模塊
│   ├── models.py     # 對象關係映射模塊
│   ├── static     # 頁面靜態資源目錄
│   │   └── favicon.ico     # 頁面收藏夾圖標
│   └── templates     # 默認存放頁面模版目錄
│       ├── 404.html
│       ├── base.html
│       ├── index.html
│       ├── mail     # 郵件模塊目錄
│       │   ├── new_user.html
│       │   └── new_user.txt
│       └── user.html
├── config.py     # 程序配置文件
├── data-dev.sqlite     # 程序數據庫文件
├── manage.py     # 程序管理啓動文件
├── migrations     # 數據庫遷移目錄
│   ├── alembic.ini
│   ├── env.py
│   ├── README
│   ├── script.py.mako
│   └── versions
├── requirements.txt     # 全部依賴包文件
└── tests     # 測試文件目錄
    ├── __init__.py
    └── test_basics.py
README

 

 程序代碼總彙

"/"web

# -*- coding: utf-8 -*-
# Author: hkey
import os

basedir = os.path.abspath(os.path.dirname(__file__))

class Config(object):   # 全部配置類的父類,通用的配置寫在這裏
    SECRET_KEY = os.environ.get('SECRET_KEY') or 'hard to guess string'
    SQLALCHEMY_COMMIT_ON_TEARDOWN = True
    SQLALCHEMY_TRACK_MODIFICATIONS = True
    FLASKY_MAIL_SUBJECT_PREFIX = '[Flasky]'
    FLASKY_MAIL_SENDER = 'Flasky Admin <xxx@126.com>'
    FLASKY_ADMIN = 'xxx@qq.com'

    @staticmethod
    def init_app(app):  # 靜態方法做爲配置的統一接口,暫時爲空
        pass

class DevelopmentConfig(Config):    # 開發環境配置類
    DEBUG = True
    MAIL_SERVER = 'smtp.126.com'
    MAIL_PORT = 465
    MAIL_USE_SSL = True
    MAIL_USERNAME = 'xxx@126.com'
    MAIL_PASSWORD = 'xxxxxx'
    SQLALCHEMY_DATABASE_URI = \
    'sqlite:///' + os.path.join(basedir, 'data-dev.sqlite')

class TestingConfig(Config):    # 測試環境配置類
    TESTING = True
    SQLALCHEMY_DATABASE_URI = \
        'sqlite:///' + os.path.join(basedir, 'data-test.sqlite')

class ProductionConfig(Config):     # 生產環境配置類
    SQLALCHEMY_DATABASE_URI = \
        'sqlite:///' + os.path.join(basedir, 'data.sqlite')

config = {  # config字典註冊了不一樣的配置,默認配置爲開發環境,本例使用開發環境
    'development': DevelopmentConfig,
    'testing': TestingConfig,
    'production': ProductionConfig,
    'default': DevelopmentConfig
}
config.py
# -*- coding: utf-8 -*-
# Author: hkey
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)

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.add_command('db', MigrateCommand)

@manager.command
def test():
    import unittest
    tests = unittest.TestLoader().discover('tests')
    unittest.TextTestRunner(verbosity=2).run(tests)

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

 

"/app"sql

# -*- coding: utf-8 -*-
# Author: hkey
from flask import Flask, render_template
from flask_bootstrap import Bootstrap
from flask_sqlalchemy import SQLAlchemy
from flask_mail import Mail
from config import config

# 因爲還沒有初始化所需的程序實例,因此沒有初始化擴展,建立擴展類時沒有向構造函數傳入參數。
bootstrap = Bootstrap()
mail = Mail()
db = SQLAlchemy()

def create_app(config_name):
    '''工廠函數'''
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)   # 經過config.py統一接口
    bootstrap.init_app(app)     # 該init_app是bootstrap實例的方法調用,與上面毫無關係
    mail.init_app(app)          # 同上
    db.init_app(app)            # 同上

    # 附加路由和自定義錯誤頁面,將藍本註冊到工廠函數
    from .main import main as main_blueprint
    app.register_blueprint(main_blueprint)

    return app
app/__init__.py
# -*- coding: utf-8 -*-
# Author: hkey
from threading import Thread
from flask import render_template, current_app
from flask_mail import Message
from . import mail

def send_async_mail(app, msg):
    '''建立郵件發送函數'''
    with app.app_context():
        mail.send(msg)

def send_mail(to, subject, template, **kwargs):
    app = current_app._get_current_object()
    if app.config['FLASKY_ADMIN']:
        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_mail, args=(app, msg))
        thr.start()     # 經過建立子線程實現異步發送郵件
        return thr
app/email.py
# -*- coding: utf-8 -*-
# Author: hkey

# 對象關係映射類

from . import db

class Role(db.Model):
    __tablename__ = 'roles'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(64), unique=True, index=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
app/models.py

 

''/app/main"shell

# -*- coding: utf-8 -*-
# Author: hkey
from  flask import Blueprint
# 定義藍本
main = Blueprint('main', __name__)

from . import views, errors
app/main/__init__.py
# -*- coding: utf-8 -*-
# Author: hkey
from flask import render_template
from . import main

@main.app_errorhandler(404)     # 路由裝飾器由藍本提供,這裏要調用 app_errorhandler 而不是 errorhandler
def page_not_found(e):
    return render_template('404.html'), 404

@main.app_errorhandler(500)
def internal_server_error(e):
    return render_template('500.html'), 500
app/main/errors.py
# -*- coding: utf-8 -*-
# Author: hkey
from flask_wtf import FlaskForm
from wtforms import StringField, SubmitField
from wtforms.validators import Required

class NameForm(FlaskForm):
    '''經過 flask-wtf 定義表單類'''
    name = StringField('What is your name ?', validators=[Required()])  # 文本框
    submit = SubmitField('Submit')  # 按鈕
app/main/forms.py
# -*- coding: utf-8 -*-
# Author: hkey
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():
    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    # 經過session保存 known爲False,經過web渲染須要
            if current_app.config['FLASKY_ADMIN']:  # 若是配置變量有flasky管理員就發送郵件
                # 異步發送郵件
                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'))  # 經過redirect避免用戶刷新重複提交
    return render_template('index.html', form=form, name=session.get('name'),
                           known=session.get('known', False))
app/main/views.py

 

"/app/main/templates" 頁面數據庫

<!DOCTYPE html>
{% extends "bootstrap/base.html" %}
{% block title %}Flasky{% endblock %}
{% block head %}
{{ super() }}
<link rel="shortcut icon" href="{{ url_for('static', filename = 'favicon.ico')}}"
      type="image/x-icon">
<link rel="icon" href="{{ url_for('static', filename = 'favicon.ico')}}"
      type="image/x-icon">
{% endblock %}
{% block navbar %}
<div class="navbar navbar-inverse" role="navigation">
    <div class="container">
        <div class="navbar-header">
            <button type="button" class="navbar-toggle"
            data-toggle="collapse" data-target=".navbar-collapse">
                <span class="sr-only">Toggle navigation</span>
                <span class="icon-bar"></span>
                <span class="icon-bar"></span>
                <span class="icon-bar"></span>
            </button>
            <a class="navbar-brand" href="/">Flasky</a>
        </div>
        <div class="navbar-collapse collapse">
            <ul class="nav navbar-nav">
                <li><a href="/">Home</a></li>
            </ul>
        </div>
    </div>
</div>
{% endblock %}
{% block content %}
<div class="container">
    {% for message in get_flashed_messages() %}
    <div class="alert alert-warning">
        <button type="button" class="close" data-dismiss="alert">&times;</button>
        {{ message }}
    </div>
    {% endfor %}
    {% block page_content %}{% endblock %}
</div>
{% endblock %}
app/templates/base.html
<!DOCTYPE html>
{% extends "base.html" %}
{% import "bootstrap/wtf.html" as wtf %}
{% block title %}Flasky{% endblock %}
{% block page_content %}
<div class="page-header">
    <h1>Hello, {% if name %}{{ name }}{% else %}Stranger{% endif %}!</h1>
    {% if not known %}
    <p>Pleased to meet you!</p>
    {% else %}
    <p>Happy to see you again!</p>
    {% endif %}
</div>
{{ wtf.quick_form(form) }}
{% endblock %}
app/templates/index.html
<!DOCTYPE html>
{% extends "base.html" %}
{% block title %}Flasky - Page Not Found{% endblock %}
{% block page_content %}
<div class="page-header">
    <h1>Not Found!</h1>
</div>
{% endblock %}
app/templates/404.html

 

"/app/main/templates/mail" 郵件模版編程

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
    User <b>{{ user.username }}</b> has joined.
</head>
<body>

</body>
</html>
app/templates/mail/new_user.html
User {{ user.username }} has joined.
app/templates/mail/new_user.txt

 

"/app/main/static/favicon.ico" 靜態 icon 圖片文件flask

 

建立需求文件

程序中必須包含一個 requirements.txt 文件,用於記錄全部依賴包及其精確的版本號。若是要在另外一臺電腦上從新生成虛擬環境,這個文件的重要性就體現出來了,例如部署程序時使用的電腦。bootstrap

(venv) E:\flasky>pip3 freeze > requirements.txt

 

建立數據庫

 

(venv) E:\flasky>python manage.py shell
>>> db.create_all()
>>> exit()

 

生成數據庫遷移文件

 

(venv) E:\flasky>python manage.py db init
Creating directory E:\flasky\migrations ... done
Creating directory E:\flasky\migrations\versions ... done
Generating E:\flasky\migrations\alembic.ini ... done
Generating E:\flasky\migrations\env.py ... done
Generating E:\flasky\migrations\README ... done
Generating E:\flasky\migrations\script.py.mako ... done
Please edit configuration/connection/logging settings in 'E:\\flasky\\migrations\\alembic.ini' before proceeding.

(venv) E:\flasky>python manage.py db migrate -m "initial migration"
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.env] No changes in schema detected.

(venv) E:\flasky>python manage.py db upgrade
INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.

 

 運行測試

 

(venv) E:\flasky>python manage.py test
test_app_exists (test_basics.BasicsTestCase)
確保程序實例存在 ... ok
test_app_is_testing (test_basics.BasicsTestCase)
確保程序在測試中運行 ... ok

----------------------------------------------------------------------
Ran 2 tests in 2.232s

OK

 

啓動程序

 

(venv) E:\flasky>python manage.py runserver
 * Serving Flask app "app" (lazy loading)
 * Environment: production
   WARNING: Do not use the development server in a production environment.
   Use a production WSGI server instead.
 * Debug mode: on
 * Restarting with stat
 * Debugger is active!
 * Debugger PIN: 138-639-525
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)

 

瀏覽器輸入 http://127.0.0.1:5000瀏覽器

 

 輸入用戶名並提交:

 

 

 

 程序會異步發送郵件,程序控制臺會打印發送日誌。已收到郵件:

 

相關文章
相關標籤/搜索