1. 程式人生 > >[ Python ] Flask 基於 Web開發 大型程序的結構實例解析

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

精確 object commit static AS AI .sql version bar

  

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

程序結構圖:

技術分享圖片

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

程序代碼總匯

"/"

技術分享圖片
# -*- 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 <[email protected]>
    FLASKY_ADMIN = [email protected]

    @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 = [email protected]
    MAIL_PASSWORD = hk470033365
    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():
    ‘‘‘Run the unit tests.‘‘‘
    import unittest
    tests = unittest.TestLoader().discover(tests)
    unittest.TextTestRunner(verbosity=2).run(tests)

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

"/app"

技術分享圖片
# -*- 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"

技術分享圖片
# -*- 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/main/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/main/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/main/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/main/templates/mail/new_user.html 技術分享圖片
User {{ user.username }} has joined.
app/main/templates/mail/new_user.txt

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

創建需求文件

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

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

創建數據庫

(venv) E:\flasky>python manage.py shell
>>> db.create_all()
>>> admin_role = Role(name=‘Admin‘)
>>> mod_role = Role(name=‘Moderator‘)
>>> user_role = Role(name=‘User‘)
>>> user_john = User(username=‘john‘, role=admin_role)
>>> user_susan = User(username=‘susan‘, role=user_role)
>>> user_david = User(username=‘david‘, role=user_role)
>>> db.session.add(admin_role)
>>> db.session.add(mod_role)
>>> db.session.add(user_role)
>>> db.session.add(user_john)
>>> db.session.add(user_susan)
>>> db.session.add(user_david)
>>> db.session.commit()
>>> 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

技術分享圖片

輸入用戶名並提交:

技術分享圖片

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

技術分享圖片

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