基于Flask的Web应用开发 - qidao123.com技术社区-IT企服评测·应用市场

mysqlclient==2.2.0
SQLAlchemy==2.0.23
Flask-SQLAlchemy==3.1.1

复制代码

mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>

复制代码

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__,
template_folder='../templates',
static_folder='../assets',
static_url_path='/assets')
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+mysqldb://root:root@127.0.0.1/myblog_db'
db = SQLAlchemy(app)
from routes import user_routes
from routes import admin_routes

复制代码

from datetime import datetime
from routes import db
from sqlalchemy import Integer, String, BLOB, TIMESTAMP
from sqlalchemy.orm import Mapped, mapped_column
class Article(db.Model):
"""
踩坑，
1、nullable参数写错
2、格式不对齐
"""
__tablename__ = 'articles'
id: Mapped[int] = mapped_column(Integer, primary_key=True)
title: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
__content: Mapped[bytes] = mapped_column(BLOB, name="content", nullable=False)
create_time: Mapped[datetime] = mapped_column(TIMESTAMP, nullable=False)
update_time: Mapped[datetime] = mapped_column(TIMESTAMP, nullable=True)
@property
def content(self):
return self.__content.decode('utf-8')

复制代码

{% extends 'base.html' %}
{% block title %}
博客主页
{% endblock %}
<--! 拼写错误：在 index.html 文件中，你在循环部分写成了 acticles，应该是 articles。这个拼写错误会导致循环内容无法正确显示。 -->
{% block content %}
<table border="1">
<tr>
<th>标题</th>
<th>时间</th>
</tr>
{% for article in articles %}
<tr>
<td><a target="_blank" href="https://www.cnblogs.com/article/{{ article.id }}.html">{{ article.title }}</a></td>
<td>{{ article.create_time }}123</td>
</tr>
{% endfor %}
</table>
{% endblock %}

复制代码

flask-WTF==1.2.1
flask-login==0.6.3

复制代码

from flask_login import UserMixin
from routes import db, login_manager
from sqlalchemy import Integer, String, BLOB, TIMESTAMP
from sqlalchemy.orm import Mapped, mapped_column
@login_manager.user_loader
def load_user(user_id):
return db.session.get(User, user_id)
class User(db.Model, UserMixin):
__tablename__ = 'user'
id: Mapped[int] = mapped_column(Integer, primary_key=True)
username: Mapped[str] = mapped_column(String(128), unique=True, nullable=False)
password: Mapped[str] = mapped_column(String(255), nullable=False)
fullname: Mapped[str] = mapped_column(String(128), nullable=False)
description: Mapped[str] = mapped_column(String(255), nullable=True)
def check_password_correction(self, attempted_password):
return self.password == attempted_password

复制代码

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager
app = Flask(__name__,
template_folder='../templates',
static_folder='../assets',
static_url_path='/assets')
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+mysqldb://root:root@127.0.0.1/myblog_db'
app.config['SECRET_KEY'] = 'ec9439cfc6c796ae2029594d' #初始化配置
db = SQLAlchemy(app)
login_manager = LoginManager(app) #初始化实例
from routes import user_routes
from routes import admin_routes

复制代码

from datetime import datetimefrom flask_login import UserMixin
from routes import db, login_manager
from sqlalchemy import Integer, String, BLOB, TIMESTAMP
from sqlalchemy.orm import Mapped, mapped_column
@login_manager.user_loader
def load_user(user_id):
return db.session.get(User, user_id)
class User(db.Model, UserMixin):
__tablename__ = 'user'
id: Mapped[int] = mapped_column(Integer, primary_key=True)
username: Mapped[str] = mapped_column(String(128), unique=True, nullable=False)
password: Mapped[str] = mapped_column(String(255), nullable=False)
fullname: Mapped[str] = mapped_column(String(128), nullable=False)
description: Mapped[str] = mapped_column(String(255), nullable=True)
def check_password_correction(self, attempted_password):
return self.password == attempted_password

复制代码

from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, SubmitField
from wtforms.validators import DataRequired
class LoginForm(FlaskForm):
username = StringField(label="用户名:", validators=[DataRequired()])
password = PasswordField(label="密码:", validators=[DataRequired()])
submit = SubmitField(label="登陆")

复制代码

{% extends 'base.html' %}
{% block title %}
博客主页
{% endblock %}
{% block content %}
<form method="POST" >
{{ form.hidden_tag() }}
<h1 >博客管理员登录</h1>
{{ form.username.label }}
{{ form.username(, placeholder="输入用户名") }}
{{ form.password.label }}
{{ form.password(, placeholder="输入密码") }}
{{ form.submit() }}
</form>
{% endblock %}

复制代码

······
@app.route('/login.html', methods=['GET', 'POST'])
def login_page():
form = LoginForm()
if form.validate_on_submit():
result = UserService().do_login(username=form.username.data, password=form.password.data)
if result:
flash(f'欢迎{form.username.data}回来',category='success')
return redirect(url_for('home_page'))
else:
flash(f'用户名或密码错误!',category='error')
return render_template('login.html', form=form)

复制代码

from sqlalchemy import Select
from models.user import User
from routes import db
from flask_login import login_user
class UserService:
def do_login(self, username: str, password: str)-> bool:
query = Select(User).where(User.username == username)
attempted_user = db.session.scalar(query)
if attempted_user and attempted_user.check_password_correction(
attempted_password=password
):
login_user(attempted_user)
return True
return False

复制代码

{% with messages = get_flashed_messages(with_categories=true) %}
{% if messages %}
{% for category, message in messages %}
{{ message }}
<button type="button" data-bs-dismiss="alert" aria-label="Close"></button>
{% endfor %}
{% endif %}
{% endwith %}

复制代码

@app.route('/logout.html')
def logout_page():
logout_user()
return redirect(url_for('home_page'))

复制代码

·····
{% if current_user.is_authenticated %}
<ul >
<li >
<a target="_blank" href="https://www.cnblogs.com/#">发布新文章</a>
</li>
<li >
<a target="_blank" href="https://www.cnblogs.com/{{ url_for('logout_page') }}">退出</a>
</li>
</ul>
{% else %}
·····

复制代码

from flask_wtf import FlaskForm
from wtforms import StringField, SubmitField, HiddenField, TextAreaField
from wtforms.validators import DataRequired
class ArticleForm(FlaskForm):
title = StringField(label="标题:", validators=[DataRequired()])
content = TextAreaField(label="内容:", validators=[DataRequired()])
submit = SubmitField(label="保持")

复制代码

{% extends 'base.html' %}
{% block title %}
博客主页
{% endblock %}
{% block content %}
<form method="POST" >
{{ form.hidden_tag() }}
<h1 >添加新文章</h1>
{{ form.title.label() }}
{{ form.title(, placeholder="请输入文章标题") }}
{{ form.content.label() }}
{{ form.content(, placeholder="请输入文章内容") }}
</br>
{{ form.submit()}}
</form>
{% endblock %}

复制代码

class ArticleService:
·····
def create_article(self, article: Article):
db.session.add(article)
db.session.commit()
return article

复制代码

from flask import render_template, url_for, redirect,flash
from flask_login import login_required
from forms.article_form import ArticleForm
from models.article import Article
from routes import app
from services.article_service import ArticleService
@app.route('/createarticle.html', methods=['GET','POST'])
@login_required
def create_article_page():
form = ArticleForm()
if form.validate_on_submit():
new_article = Article()
new_article.title = form.title.data
new_article.content = form.content.data
try:
ArticleService().create_article(new_article)
flash(message=f'发布文章完成', category='success')
return redirect(url_for('home_page'))
except Exception as error:
flash(message=f'发布文章失败: {error}', category='danger')
return render_template(template_name_or_list='editarticle.html', form=form)

复制代码

{% extends 'base.html' %}
{% block title %}
博客主页
{% endblock %}
{% block content %}
<--! 拼写错误：在 index.html 文件中，你在循环部分写成了 acticles，应该是 articles。这个拼写错误会导致循环内容无法正确显示。 -->
{% for article in articles %}
<ul >
<li >
<a target="_blank" href="https://www.cnblogs.com/article/{{ article.id }}.html" >{{ article.title }}</a>
</li>
{% if current_user.is_authenticated %}
<li >
<small >
<a target="_blank" href="https://www.cnblogs.com/editartical/{{ article.id }}.html">编辑</a>
</small>
</li>
{% endif %}
</ul>
<p >
<a target="_blank" href="https://www.cnblogs.com/article/{{ article.id }}.html">{{ article.content }}</a>
</p>
<ul >
<small >发布时间：{{ article.create_time }}</small>
</ul>
{% endfor %}
{% endblock %}

复制代码

@app.route('/createarticle.html', methods=['GET','POST'])
@login_required
def create_article_page():
form = ArticleForm()
if form.validate_on_submit():
new_article = Article()
new_article.title = form.title.data
new_article.content = form.content.data
try:
article, error_msg = ArticleService().create_article(new_article)
if error_msg:
flash(message=f'发布文章错误', category='danger')
else:
flash(message=f'发布文章完成', category='success')
return redirect(url_for('home_page'))
except Exception as error:
flash(message=f'发布文章失败: {error}', category='danger')
return render_template(template_name_or_list='editarticle.html', form=form)

复制代码

····
# 发布文章
@app.route('/createarticle.html', methods=['GET','POST'])
@login_required
def create_article_page():
form = ArticleForm()
if form.validate_on_submit():
new_article = Article()
new_article.title = form.title.data
new_article.content = form.content.data
try:
article, error_msg = ArticleService().create_article(new_article)
if error_msg:
flash(message=f'发布文章错误:{error_msg}', category='danger')
else:
flash(message=f'发布文章完成', category='success')
return redirect(url_for('home_page'))
except Exception as error:
flash(message=f'发布文章失败: {error}', category='danger')
return render_template(template_name_or_list='editarticle.html', form=form)
# 更新文章
@app.route('/editarticle/<article_id>.html', methods=['GET','POST'])
@login_required
def edit_article_page(article_id: str):
form = ArticleForm()
if request.method == 'GET':
try:
article = ArticleService().get_article(int(article_id))
if not article:
flash(message=f'修改的文章不存在', category='danger')
return redirect(url_for('home_page'))
else:
form.title.data = article.title
form.content.data = article.content
except Exception as ex:
flash(massage=f'提取文件失败: {ex}', category='danger')
return redirect(url_for('home_page'))
if form.validate_on_submit():
try:
updated_article = Article()
updated_article.id = int(article_id)
updated_article.title = form.title.data
updated_article.content = form.content.data
article, error_msg = ArticleService().update_article(updated_article)
if error_msg:
flash(message=f'更新文章失败', category='danger')
else:
flash(message=f'更新文章成功', category='success')
return redirect(url_for('home_page'))
return redirect(url_for('home_page'))
except Exception as error:
flash(message=f'发布文章失败: {error}', category='danger')
return render_template(template_name_or_list='editarticle.html', form=form)

复制代码

····
# 发布文章对数据库进行比对
def create_article(self, article: Article):
query = Select(Article).where(Article.title == article.title)
# db.session.scalar和 db.session.execute。这里使用execute 有问题，无法判断是否查询到数据所以使用scalar
exit_article = db.session.scalar(query)
if exit_article:
return article, '同标题的文章已存在'
db.session.add(article)
db.session.commit()
return article, None
# 更新文章
def update_article(self, article: Article):
exit_article = db.session.get(Article, article.id)
if not exit_article:
return article, '文章不存在'
# TODO: 检查同标题文章是否存在
qury = Select(Article).where(and_(Article.title == article.title, Article.id != article.id))
same_title_article = db.session.scalar(qury)
if same_title_article :
return article, '更新同标题的文章已存在'
exit_article.title = article.title
exit_article.content = article.content
exit_article.update_time = func.now()

复制代码

····
# 发布文章
@app.route('/createarticle.html', methods=['GET','POST'])
@login_required
def create_article_page():
·····
# 通过传递 is_edit参数判断编辑/更新
return render_template(template_name_or_list='editarticle.html', form=form, is_edit=False)
# 更新文章
@app.route('/editarticle/<article_id>.html', methods=['GET','POST'])
@login_required
def edit_article_page(article_id: str):
····
# 通过传递 is_edit参数判断编辑/更新
return render_template(template_name_or_list='editarticle.html', form=form, is_edit=True)

复制代码

{% extends 'base.html' %}
{% block title %}
博客主页-
{% if is_edit %}
编辑文章
{% else %}
添加新文章
{% endif %}
{% endblock %}
{% block content %}
<form method="POST" >
{{ form.hidden_tag() }}
<h1 >
{% if is_edit %}
编辑文章
{% else %}
添加新文章
{% endif %}
</h1>
{{ form.title.label() }}
{{ form.title(, placeholder="请输入文章标题") }}
{{ form.content.label() }}
{{ form.content(, placeholder="请输入文章内容") }}
</br>
{{ form.submit()}}
</form>
{% endblock %}

复制代码

·····
{% if current_user.is_authenticated %}
·····
<li >
<small >
<a data-bs-toggle="modal" data-bs-target="#Mdal-DeleteConfirm-{{ article.id }}">删除</a>
</small>
</li>
{% endif %}

复制代码

from flask_wtf import FlaskForm
from wtforms import HiddenField, SubmitField
from wtforms.validators import DataRequired
class DeleteArticleForm(FlaskForm):
article_id = HiddenField(validators=[DataRequired()])
submit = SubmitField(label='删除')

复制代码

<dev >
<h5 id="deleteModalLabel">{{ article.title }}</h5>
<button type="button" data-bs-dismiss="modal" aria-label="Close"></button>
<form method="POST">
{{ delete_article_form.csrf_token }}
{{ delete_article_form.article_id(value=article.id) }}
<h4 >确定要删除"{{ article.title }}"吗？</h4>
<button type="button" data-bs-dismiss="modal">取消</button>
<button type="button" >确定</button>
</form>
</dev>

复制代码

{% for article in articles %}
{% if current_user.is_authenticated %}
{% include 'includes/article_modals.html' %}
{% endif %}

复制代码

·······
def delete_article(self, article_id: int):
article = db.session.get(Article, article_id)
if article:
db.session.delete(article)
db.session.commit()
return article, None
else:
return False, '文章不存在'

复制代码

@app.route('/', methods=['GET', 'POST'])
@app.route('/index.html', methods=['GET', 'POST'])
def home_page():
if current_user.is_authenticated:
delete_article_form = DeleteArticleForm()
if delete_article_form.validate_on_submit():
if delete_article_form.validate_on_submit():
result, error = ArticleService().delete_article(int(delete_article_form.article_id.data))
if result:
flash(message=f'删除文章成功', category='success')
return redirect(url_for('home_page'))
else:
flash(message=f'删除文章成功', category='danger')
articles = ArticleService().get_articles()
if current_user.is_authenticated:
return render_template(template_name_or_list='index.html', articles=articles, delete_article_form=delete_article_form)
return render_template(template_name_or_list='index.html', articles=articles)

复制代码

·····
·····

复制代码

{% extends 'base.html' %}
{% block title %}
博客 -{{ article.title }}
{% endblock %}
{% block content %}
<textarea id="article_content" >{{ article.content }}</textarea>
<h4><p >{{ article.title }}</p></h4>
<p >最后更新： {{ article.update_time }}</p>
<p id="article_viewer"></p>
"
{% endblock %}

复制代码

$(function (){
var converter = new showdown.Converter();
var article_html = converter.makeHtml($('#article_content').val())
$('#article_content').html(article_html)
})

复制代码

{% extends 'base.html' %}
{% block title %}
博客主页-
{% if is_edit %}
编辑文章
{% else %}
添加新文章
{% endif %}
{% endblock %}
{% block content %}
<form method="POST" >
{{ form.hidden_tag() }}
<h1 >
{% if is_edit %}
编辑文章
{% else %}
添加新文章
{% endif %}
</h1>
{{ form.title.label() }}
{{ form.title(, placeholder="请输入文章标题") }}
{{ form.content.label() }}
{{ form.content(, placeholder="请输入文章内容") }}
</br>
{{ form.submit()}}
<a target="_blank" href="https://www.cnblogs.com/#" id="article_preview_btn" >预览</a>
文章内容:
</form>
{% endblock %}

复制代码

$(function (){
$('#article_preview_btn').click(function (){
var converter = new showdown.Converter();
var content_html = converter.makeHtml($('#content').val());
$('#article_preview').html(content_html);
});
});

复制代码

$ pip install bcrypt

复制代码

Flask==3.0.0mysqlclient==2.2.0
SQLAlchemy==2.0.23
Flask-SQLAlchemy==3.1.1flask-WTF==1.2.1
flask-login==0.6.3bcrypt==4.1.1

复制代码

>>> import bcrypt
>>> pd='admin'
>>> hashed = bcrypt.hashpw(pd.encode(), bcrypt.gensalt())
>>> print(hashed.decode('utf-8'))
$2b$12$U3PhlQenadR1WCb63.1Rxu83TrnFxv884YpPOPjYZI0wzbl.oG4Iq

复制代码

·····
def check_password_correction(self, attempted_password):
password_hashed = self.password.encode()
return bcrypt.checkpw(attempted_password.encode(), password_hashed)

复制代码

from flask_wtf import FlaskForm
from flask_wtf.file import FileField, FileRequired
from wtforms import SubmitField
class ImageUploadForm(FlaskForm):
image_file = FileField(label="选择图片", validators=[FileRequired()])
submit = SubmitField(label="上传")

复制代码

{% extends 'base.html' %}
{% block title %}
博客主页
{% endblock %}
{% block content %}
<form method="POST" enctype="multipart/form-data">
{{ form.hidden_tag() }}
<h1 >
上传页面
</h1>
{{ form.image_file.label }}
{{ form.image_file() }}
{{ form.submit() }}
</form>
{% endblock %}

复制代码

from pathlib import Path
class Profile:
__images_path = None
@staticmethod
def get_images_path():
home_path = Path(__file__).parent.parent
images_path = home_path.joinpath("data/images")
if not images_path.exists():
images_path.mkdir(parents=True)
return images_path

复制代码

from pathlib import Path
def get_file_name_parts( filename: str):
pos = filename.rfind('.')
if pos == -1:
return filename, ''
return filename[:pos], filename[pos + 1:]
def get_save_filepaths(file_path: Path, filename: str):
save_file = file_path.joinpath(filename)
if not save_file.exists():
return save_file
name, ext = get_file_name_parts(filename)
for index in range(1, 100):
save_file = file_path.joinpath(f'{name}_{index}.{ext}')
if not save_file.exists():
return save_file
return file_path.joinpath(f'{name}_override.{ext}')

复制代码

·····
@app.route('/images.html', methods=['GET', 'POST'])
@login_required
def images_page():
form = ImageUploadForm()
if form.validate_on_submit():
image_file = form.image_file.data
images_path = Profile.get_images_path()
image_filename = secure_filename(image_file.filename)
image_fullpath = utils.get_save_filepaths(images_path, image_filename)
image_file.save(image_fullpath)
flash(message=f'上传图片成功: {image_fullpath}', category='success')
return render_template(template_name_or_list='images.html', form=form)

复制代码

·····
{% if current_user.is_authenticated %}
<ul >
<li >
<a target="_blank" href="https://www.cnblogs.com/{{ url_for('images_page') }}">图片管理</a>
</li>
<li >
<a target="_blank" href="https://www.cnblogs.com/{{ url_for('create_article_page') }}">发布新文章</a>
</li>
<li >
<a target="_blank" href="https://www.cnblogs.com/{{ url_for('logout_page') }}">退出</a>
</li>
</ul>
{% else %}
······

复制代码

·····
@app.route('/image/<image_filename>')
def download_image(image_filename: str):
image_path = Profile.get_images_path()
image_filepath = image_path.joinpath(image_filename)
if not image_filepath:
return abort(404)
return send_from_directory(directory=image_filepath, path=image_filename)

复制代码

from common.profile import Profile
class ImageService:
def get_image_filename_list(self):
image_paht = Profile.get_images_path()
filename_list = []
if image_paht.exists():
for item in image_paht.iterdir():
if item.is_file():
filename_list.append(item.name)
return filename_list

复制代码

@app.route('/images.html', methods=['GET', 'POST'])
@login_required
def images_page():
form = ImageUploadForm()
if form.validate_on_submit():
image_file = form.image_file.data
images_path = Profile.get_images_path()
image_filename = secure_filename(image_file.filename)
image_fullpath = utils.get_save_filepaths(images_path, image_filename)
image_file.save(image_fullpath)
flash(message=f'上传图片成功: {image_fullpath}', category='success')
image_filenames = ImageService().get_image_filename_list()
return render_template(template_name_or_list='images.html', form=form, image_filenames=image_filenames)

复制代码

<hr/>
{% if image_filenames %}
{% for image_file in image_filenames %}
<b>https://www.cnblogs.com/image/{{ image_file }}</b>
<img src="https://www.cnblogs.com/image/{{ image_file }}" height="200px;"/>
<a target="_blank" href="https://www.cnblogs.com/image/{{ image_file }}" >查看</a>
<a data-bs-toggle="modal" data-bs-target="#Modal-DeleteConfirm-{{ image_file }}">删除</a>
{% endfor %}
{% endif %}

复制代码

import bcrypt
from sqlalchemy import inspect
from routes import app, db
def init_db():
with app.app_context():
inspector = inspect(db.engine)
if not inspector.has_table('users'):
from models.user import User
from models.article import Article
db.create_all()
password_hashed = bcrypt.hashpw('admin'.encode(), bcrypt.gensalt())
user = User(username="root", password=password_hashed.decode('utf-8'), fullname='root', description='')
db.session.add(user)
db.session.commit()
if __name__ == '__main__':
init_db()
app.run(host='0.0.0.0', debug=True, port=8080)

复制代码

import os
from flask import Flask
from flask_login import LoginManager
from flask_sqlalchemy import SQLAlchemy
MYSQL_HOST = os.getenv("MYSQL_HOST", "localhost")
MYSQL_PORT = os.getenv("MYSQL_PORT", "3306")
MYSQL_USER = os.getenv("MYSQL_USER", "root")
MYSQL_PWD = os.getenv("MYSQL_PWD", "test")
MYSQL_DB = os.getenv("MYSQL_DB", "testdb")
app = Flask(__name__,
template_folder='../templates',
static_folder='../assets',
static_url_path='/assets')
app.config['SQLALCHEMY_DATABASE_URI'] = f'mysql+mysqldb://{MYSQL_USER}:{MYSQL_PWD}@{MYSQL_HOST}:{MYSQL_PORT}/{MYSQL_DB}'
app.config['SECRET_KEY'] = 'ec9439cfc6c796ae2029594d'
db = SQLAlchemy(app)
login_manager = LoginManager(app)
from routes import user_routes
from routes import admin_routes

复制代码

FROM ubuntu
COPY . /opt/myblog/
WORKDIR /opt/myblog/
RUN apt-get update
RUN apt-get install -y python3.9 python3-pip
RUN apt-get install -y pkg-config
RUN apt-get install -y libmysqlclient-dev
RUN pip3 install --upgrade pip
RUN pip3 install -r requirements.txt
ENV PYTHONPATH=/opt/myblog/
ENTRYPOINT ["python3", "main.py"]

复制代码

version: '3.8'
services:
myblog_server:
build: .
image: myblog
container_name: myblog_server
ports:
- "80:8080"
links:
- mysql_server
environment:
MYSQL_HOST: mysql_server
MYSQL_DB: myblog_db
MYSQL_USER: root
MYSQL_PWD: nevertellyou
volumes:
- /opt/myblog_data:/opt/myblog/data
depends_on:
mysql_server:
condition: service_healthy
mysql_server:
image: mysql:8.0
container_name: mysql_server
volumes:
- /opt/mysql:/var/lib/mysql
environment:
MYSQL_ROOT_PASSWORD: nevertellyou
MYSQL_DATABASE: myblog_db
healthcheck:
test: ["CMD", "mysqladmin" ,"ping", "-h", "localhost"]
timeout: 20s
retries: 10

复制代码