[Python知识库]Python后端Flask学习项目实践---搭建一个问答网站

1.项目效果展示

这里主要以后端为主，前端的代码不做展示，如果需要代码可以评论或者私信

用户注册、登录相关：

用邮箱进行注册，并通过向邮箱发送验证码来进行判断，一个邮箱只能注册一个账号

?

?首页相关：

用户登录后可以进行发布问题和回答，同时也提供搜索功能，在首页展示所有问题

?搜索：

?

?评论:

?

2.项目工程目录

blueprints:

项目蓝图包括问答的逻辑实现和用户的逻辑时间

migrations：

项目数据库迁移

static、templates：

前端相关，css、html文件等

3.项目实现：

3.1数据库

主要是需要设计一个用户表user表，一个邮箱对应验证码EmailCaptcha的表，一个问题question表，一个评论Answer表

利用flask中提供的SQLAlchemy不用我们自己手动写SQL代码，用面向对象的思维解决就好

（1）新建db对象

因为在很多文件中都需要用到db对象，所以用一个专门的文件etx.py储存，并在app.py中进行初始化：

etx.py:

from flask_mail import Mail
from flask_sqlalchemy import SQLAlchemy


db = SQLAlchemy()
mail = Mail()

app.py:

app = Flask(__name__)
# 配置项
app.config.from_object(config)

db.init_app(app)
mail.init_app(app)

migrate = Migrate(app, db)
# 组装蓝图 将book、course、user模块都组装在main.py中
app.register_blueprint(qa_bp)
app.register_blueprint(user_bp)

(2)配置数据库：

# 数据库的配置变量
HOSTNAME = '127.0.0.1'
PORT = '3306'
DATABASE = 'flask'
USERNAME = 'root'
PASSWORD = '*****'
DB_URI = 'mysql+pymysql://{}:{}@{}:{}/{}'.format(USERNAME, PASSWORD, HOSTNAME, PORT, DATABASE)
SQLALCHEMY_DATABASE_URI= DB_URI
# 关闭数据库修改跟踪操作[提高性能]，可以设置为True，这样可以跟踪操作：
SQLALCHEMY_TRACK_MODIFICATIONS=False

# 开启输出底层执行的sql语句
SQLALCHEMY_ECHO = True

（3）设计models：

from exts import db


class EmailCaptchaModel(db.Model):
    __tablename__="email_captcha"
    id=db.Column(db.Integer,primary_key=True,autoincrement=True)
    email=db.Column(db.String(100),nullable=True,unique=True)
    captcha=db.Column(db.String(10),nullable=False)
    create_time=db.Column(db.DateTime)


class UserModel(db.Model):
    __tablename__ = "user"
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    username = db.Column(db.String(200),nullable=False,unique=True)
    email = db.Column(db.String(100),nullable=False,unique=True)
    password = db.Column(db.String(200),nullable=False)
    join_time = db.Column(db.DateTime)


class QuestionModel(db.Model):
    __tablename__ = "question"
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    title = db.Column(db.String(200), nullable=False)
    content = db.Column(db.Text,nullable=False)
    create_time = db.Column(db.DateTime)
    author_id = db.Column(db.Integer,db.ForeignKey("user.id"))
    author = db.relationship("UserModel",backref="questions")


class AnswerModel(db.Model):
    __tablename__ = "answer"
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    content = db.Column(db.Text,nullable=False)
    create_time = db.Column(db.DateTime)
    question_id = db.Column(db.Integer,db.ForeignKey("question.id"))
    author_id = db.Column(db.Integer, db.ForeignKey("user.id"))

    question = db.relationship("QuestionModel",backref=db.backref("answers",order_by=create_time.desc()))
    author = db.relationship("UserModel",backref="answers")

（4）生成数据库

利用flask中提供的数据库迁移功能可以在更新数据库后直接迁移

在命令行进行输入

step1：

flask db init

这条语句执行完后会生成上述文件中的migrate文件夹

step2：

flask db migrate

step3：

flask db upgrade

更新完成！

3.2 发送邮件功能

在注册时点击发送验证码后利用一个固定的邮箱给注册的邮箱发送验证码信息，同时存储到数据库中验证输入的验证码是否与收到的验证码一致。

主要是利用flask_mail进行邮箱发送。

（1）邮箱设置

# 邮箱配置
# 项目中用的是QQ邮箱
MAIL_SERVER = "smtp.qq.com"
MAIL_PORT = 465
MAIL_USE_TLS = False
MAIL_USE_SSL = True
MAIL_DEBUG = True
MAIL_USERNAME = "774747245@qq.com"
MAIL_PASSWORD = "*****"
MAIL_DEFAULT_SENDER = "774747245@qq.com"

（2）得到验证码

可以在浏览器中进行测试

@bp.route("/captcha", methods=['POST'])
def get_captcha():
    email = request.form.get("email")
    # 从letters集合中随机取出4个生成验证码 letters集合是英文和数字的集合
    letters = string.ascii_letters + string.digits
    captcha = "".join(random.sample(letters, 4))

    if email:
        message = Message(
            subject="邮箱测试",
            recipients=[email],
            body=f"您的注册验证码是：{captcha}"
        )
        mail.send(message)
        # 存放到数据库中:先通过email进行查询，如果存在该email就直接更新captcha就行，如果不存在就添加一个记录
        captcha_model = EmailCaptchaModel.query.filter_by(email=email).first()
        if captcha_model:
            captcha_model.captcha = captcha
            # captcha_model.create_time=datetime.time.now()
            db.session.commit()
        else:
            captcha_model = EmailCaptchaModel(email=email, captcha=captcha)
            db.session.add(captcha_model)
            db.session.commit()
        print("captcha:", captcha)
        return jsonify({"code": 200, "message": "suceess"})
    else:
        return jsonify({"code": 400, "message": "mail为空"})

3.3?用户注册和登录、登出

（1）判断注册和登录填写的表单是否符合要求

可以直接利用flask中的wtforms进行格式限制：

新建一个forms.py：

import wtforms
from wtforms.validators import length,email,EqualTo,InputRequired
from models import EmailCaptchaModel,UserModel


class LoginForm(wtforms.Form):
    email = wtforms.StringField(validators=[email()])
    password = wtforms.StringField(validators=[length(min=6,max=20)])


class RegisterForm(wtforms.Form):
    username = wtforms.StringField(validators=[length(min=3,max=20,message="长度在3和20之间")])
    email = wtforms.StringField(validators=[email()])
    captcha = wtforms.StringField(validators=[length(min=4, max=4)])
    password = wtforms.StringField(validators=[length(min=6,max=20,message="长度在6和20之间")])
    password_confirm = wtforms.StringField(validators=[EqualTo("password")])

    def validate_captcha(self,field):
        captcha = field.data
        email = self.email.data
        captcha_model = EmailCaptchaModel.query.filter_by(email=email).first()
        print(captcha_model.captcha)
        if not captcha_model or captcha_model.captcha.lower() != captcha.lower():
            print("验证码错误")
            raise wtforms.ValidationError("邮箱验证码错误！")

    def validate_email(self,field):
        email = field.data
        user_model = UserModel.query.filter_by(email=email).first()
        if user_model:
            print("邮箱已存在")
            raise wtforms.ValidationError("邮箱已经存在！")


class QuestionForm(wtforms.Form):
    title = wtforms.StringField(validators=[length(min=3, max=200)])
    content = wtforms.StringField(validators=[length(min=5)])


class AnswerForm(wtforms.Form):
    content = wtforms.StringField(validators=[length(min=1)])

（2）注册

注册主要是判断格式是否正确，并判断验证码是否正确，如果正确后，新生成一个user对象插入到user表中。同时在存储时进行密码加密存储

@bp.route('/register', methods=['GET', 'POST'])
def register():
    if request.method == 'GET':
        return render_template("register.html")
    else:
        form = RegisterForm(request.form)
        if form.validate():
            print("验证成功")
            username = form.username.data
            email = form.email.data
            password = form.password.data
            # 密码加密
            hash_password = generate_password_hash(password=password)
            captcha = form.captcha.data
            create_time = datetime.datetime.now()
            # 1.通过email查询user表 如果存在就通知已存在该用户 不存在就新建
            user_model = UserModel.query.filter_by(email=email).first()
            if user_model:
                print("该邮箱已被注册，请重新输入")
                return redirect(url_for("user.register"))
            user = UserModel(username=username, email=email, password=hash_password, join_time=create_time)
            db.session.add(user)
            db.session.commit()
            return redirect(url_for("user.login"))
        else:
            print("注册验证失败")
            return redirect(url_for("user.register"))

(3)登录

先查询是否存在这个用户，如果存在进行密码验证

@bp.route('/login', methods=['GET', 'POST'])
def login():
    """登录：guest1:123456
    0.通过验证
    1.通过邮箱查找出user_model
    2.如果存在就比较密码是否正确 正确：登录成功 不正确：密码错误
    3.不存在直接提示用户不存在并返回到注册页面"""
    if request.method == 'GET':
        return render_template("login.html")
    else:
        form = LoginForm(request.form)
        if form.validate():
            email = form.email.data
            password_input = form.password.data
            user_model = UserModel.query.filter_by(email=email).first()
            if user_model:
                if check_password_hash(user_model.password, password=password_input):
                    print("登录成功")
                    session['user_id'] = user_model.id
                    return redirect("/")
                else:
                    print("密码输入错误")
                    flash("密码输入错误")
                    return redirect(url_for("user.login"))
            else:
                print("该用户不存在，请注册")
                flash("该用户不存在，请注册")
                return redirect(url_for("user.register"))
        else:
            print("请输入正确格式的账号或密码")
            flash("请输入正确格式的账号或密码")
            return redirect(url_for("user.login"))

（4）登出

删除session即可

@bp.route('/logout')
def logout():
    session.clear()  # 清除session就可
    return redirect(url_for("user.login"))

3.4?访问页面权限管理

用户在未登录时不能进行发布问答的功能，这里可以利用一个装饰器进行实现

"""装饰器"""
from flask import g, redirect, url_for
from functools import wraps

"""如果没有登录就不能访问，跳转到登录页面 如果登录了就正常逻辑处理"""


def login_required(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        if hasattr(g, "user"):
            return func(*args, **kwargs)
        else:
            print("未登录不能进行访问")
            return redirect(url_for("user.login"))

    return wrapper

在需要进行访问限制的方法上面加上

@bp.route('/public_question', methods=['GET', 'POST'])
@login_required
def public_question():

就可以实现访问权限的设置啦！！！

3.5 问答查看、搜索

问答功能：实际上就是生成一个QuestionModel，将符合要求的Question存储到数据库中

（1）QuestionModel：

class QuestionModel(db.Model):
    __tablename__ = "question"
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    title = db.Column(db.String(200), nullable=False)
    content = db.Column(db.Text,nullable=False)
    create_time = db.Column(db.DateTime)
    author_id = db.Column(db.Integer,db.ForeignKey("user.id"))
    author = db.relationship("UserModel",backref="questions")

（2）表单验证：

class QuestionForm(wtforms.Form):
    title = wtforms.StringField(validators=[length(min=3, max=200)])
    content = wtforms.StringField(validators=[length(min=5)])

（3）问答功能

@bp.route('/public_question', methods=['GET', 'POST'])
@login_required
def public_question():
    """发布问答"""
    if request.method == 'GET':
        return render_template("public_question.html")
    else:
        form = QuestionForm(request.form)
        if form.validate():
            title = form.title.data
            content = form.content.data
            create_time = datetime.datetime.now()
            question = QuestionModel(title=title, content=content, author=g.user, create_time=create_time)
            db.session.add(question)
            db.session.commit()
            print("发布问答成功")
            return redirect(url_for("qa.index"))
        else:
            flash("格式不正确")
            return redirect(url_for("qa.public_question"))

（4）问题细节查看

@bp.route("/question/<int:question_id>")
def question_detail(question_id):
    question = QuestionModel.query.get(question_id)
    return render_template("detail.html", question=question)

（5）问题搜索

在标题和内容中都可以进行查询搜索，只有符合其一就输出

@bp.route('/search')
def search():
    q = request.args.get("q")
    questions = QuestionModel.query.filter(
        or_(QuestionModel.title.contains(q), QuestionModel.content.contains(q))).order_by(db.text("create_time"))
    print("搜索结果", questions == None)
    if questions:
        return render_template("index.html", questions=questions)
    else:
        print("搜索结果为空")
        return "搜索结果为空"

3.6 评论功能

在发布问题后可以进行评论，与问题一致的流程

（1）生成AnswerModel：

class AnswerModel(db.Model):
    __tablename__ = "answer"
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    content = db.Column(db.Text,nullable=False)
    create_time = db.Column(db.DateTime)
    question_id = db.Column(db.Integer,db.ForeignKey("question.id"))
    author_id = db.Column(db.Integer, db.ForeignKey("user.id"))

    question = db.relationship("QuestionModel",backref=db.backref("answers",order_by=create_time.desc()))
    author = db.relationship("UserModel",backref="answers")

（2）生成表单验证：

class AnswerForm(wtforms.Form):
    content = wtforms.StringField(validators=[length(min=1)])

（3）进行评论

@bp.route("/answer/<int:question_id>", methods=['POST'])
@login_required
def answer(question_id):
    form = AnswerForm(request.form)
    if form.validate():
        content = form.content.data
        create_time = datetime.datetime.now()
        answer_model = AnswerModel(content=content, author=g.user, question_id=question_id, create_time=create_time)
        db.session.add(answer_model)
        db.session.commit()
        return redirect(url_for("qa.question_detail", question_id=question_id))
    else:
        flash("表单验证失败！")
        return redirect(url_for("qa.question_detail", question_id=question_id))

4.总结

就是简单的一个小项目练练手，新手的可以看这个直接入门，但是项目也需要进行完善啦，比如注册也可以设置头像之类的，这样在首页就可以显示自己的头像之类的。