Python Flask项目步骤
阅读原文时间:2023年07月09日阅读:2

步骤一:构建基础项目框架

创建manage.py文件

from flask import Flask

app = Flask(__name__)

""" 配置信息 """
""" 数据库 """

@app.route("/index")
def index():
return "index"

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

步骤二:配置信息

class Config(object):
"""配置信息"""
DEBUG = True

SECRET\_KEY = "SAJCHIAAO\_ASV+\_C\_S+\*()"

# 数据库  
SQLALCHEMY\_DATABASE\_URL = "mysql+pymysql://root:root@127.0.0.1:3306/novel"  
# 追踪对象的修改并且发送信号  
SQLALCHEMY\_TRACK\_MODIFICATIONS = True

# redis的配置信息  
REDIS\_HOST = "127.0.0.1"  
REDIS\_PORT = 6379  
REDIS\_PWD = "hpredis"

# flask-session配置信息,将数据session保存到redis中  
SESSION\_TYPE = "redis"  
SESSION\_REDIS = redis.StrictRedis(host=REDIS\_HOST,port=REDIS\_PORT,password=REDIS\_PWD)  
SESSION\_USE\_SIGNER = True           # 对cookie里的sessionId混淆处理(隐藏)  
PERMANENT\_SESSION\_LIFETIME = 86400  # session数据的有效期,单位秒

app.config.from_object(Config)

步骤三:创建数据库

from flask_sqlalchemy import SQLAlchemy
import redis

db = SQLAlchemy(app)

创建redis连接对象

redis_story = redis.StrictRedis(
host=Config.REDIS_HOST,
port=Config.REDIS_PORT,
password=Config.REDIS_PWD
)

步骤四:保存session

利用flask-session,将session数据保存到redis中

from flask_session import Session

利用flask-session,将session数据保存到redis中

Session(app)

整体结构 

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_session import Session

import redis

app = Flask(__name__)

class Config(object):
"""配置信息"""
DEBUG = True

SECRET\_KEY = "SAJCHIAAO\_ASV+\_C\_S+\*()"

# 数据库  
SQLALCHEMY\_DATABASE\_URL = "mysql+pymysql://root:root@127.0.0.1:3306/novel"  
# 追踪对象的修改并且发送信号  
SQLALCHEMY\_TRACK\_MODIFICATIONS = True

# redis的配置信息  
REDIS\_HOST = "127.0.0.1"  
REDIS\_PORT = 6379  
REDIS\_PWD = "hpredis"

# flask-session配置信息,将数据session保存到redis中  
SESSION\_TYPE = "redis"  
SESSION\_REDIS = redis.StrictRedis(  
    host=REDIS\_HOST,  
    port=REDIS\_PORT,  
    password=REDIS\_PWD  
)  
SESSION\_USE\_SIGNER = True  # 对cookie里的sessionId混淆处理(隐藏)  
PERMANENT\_SESSION\_LIFETIME = 86400  # session数据的有效期,单位秒

app.config.from_object(Config)

数据库

db = SQLAlchemy(app)

创建redis连接对象

redis_story = redis.StrictRedis(host=Config.REDIS_HOST,port=Config.REDIS_PORT,password=Config.REDIS_PWD)

利用flask-session,将session数据保存到redis中

Session(app)

@app.route("/index")
def index():
return "index"

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

拆分成多个文件

manage.py:启动文件

config.py:配置文件

ihome:项目文件夹

ihome/__init__.py:项目文件

ihome/models.py:数据库文件

ihome/web_html.py:静态文件蓝图

ihome/static:存放静态文件的文件夹,存储静态文件(html,css,js等文件)

ihome/api_1_0:存放api接口的蓝图文件

ihome/utils:存放自定义方法文件

1. manage.py启动文件的相关配置

# coding:utf-8
from ihom import create_app,db
from flask_script import Manager
from flask_migrate import Migrate,MigrateCommand

创建flask的应用对象

app = create_app("develop")

数据库迁移命令

manager = Manager(app)
Migrate(app,db)
manager.add_command("db",MigrateCommand)

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

2.config.py配置文件的相关配置

Flask-SQLAlchemy的配置值

flask-session的配置值

# coding:utf-8
import redis

class Config(object):
"""配置信息"""

SECRET\_KEY = "dscsdv\*VDSVDSUBck98sd7767"

# 数据库  
SQLALCHEMY\_DATABASE\_URI = "mysql+pymysql://root:root@127.0.0.1:3306/ihome"  
SQLALCHEMY\_TRACK\_MODIFICATIONS = True

# redis  
REDIS\_HOST = "127.0.0.1"  
REDIS\_PORT = 6379

# flask-session配置,将session值存放到redis中  
SESSION\_TYPE = "redis"  
SESSION\_REDIS = redis.StrictRedis(host=REDIS\_HOST,port=REDIS\_PORT,password="hpredis")  
SESSION\_USE\_SIGNER = True           # 对cookie里的sessionId混淆处理(隐藏)  
PERMANENT\_SESSION\_LIFETIME = 86400  # session数据的有效期,单位秒

class DevelopmentConfig(Config):
"""开发模式的配置信息"""
DEBUG = True

class ProductConfig(Config):
"""生产环境的配置信息"""
pass

config_map = {
"develop":DevelopmentConfig,
"product":ProductConfig
}

3.ihome/__init__.py文件的相关配置

# coding:utf-8
import redis
import logging
from flask import Flask
from config import config_map
from flask_session import Session
from flask_sqlalchemy import SQLAlchemy
from flask_wtf import CSRFProtect
from logging.handlers import RotatingFileHandler

from ihom.utils.commons import ReConverter # 引入自定义的方法

数据库

db = SQLAlchemy()

创建redis连接对象

redis_store = None

配置日志信息

设置日志的记录等级

logging.basicConfig(level=logging.INFO)

创建日志记录器,指明日志保存的路径、每个日志文件的最大大小,保存日志文件个数上限

file_log_handler = RotatingFileHandler("logs/log",maxBytes=1024*1024*100,backupCount=10)

创建日志记录的格式

formatter = logging.Formatter("%(levelname)s %(filename)s:%(lineno)d %(message)s")

为刚创建的日志记录器设置日志的记录格式

file_log_handler.setFormatter(formatter)

为全局的日志工具对象(flask app 使用的)添加日志记录器

logging.getLogger().addHandler(file_log_handler)

工厂模式(在正式开发中,有debug(调试)环境,与线上环境,采用工厂模式可随时转换)

def create_app(config_name):
"""
创建flask应用对象
:param config_name: str配置模式的模式名字 ("develop","product") (调试,线上)
:return: app
"""
app = Flask(__name__)

# 根据配置模式的名字获取配置参数的类  
config\_class = config\_map.get(config\_name)  
app.config.from\_object(config\_class)    # 一定在db之前

# 使用app初始化db  
db.init\_app(app)

# 初始化redis工具  
global redis\_store  
redis\_store = redis.StrictRedis(host=config\_class.REDIS\_HOST, port=config\_class.REDIS\_PORT,password="hpredis")

# 利用flask-session ,键session保存到数据库的Redis中  
Session(app)

# 为flask补充csrf防护  
CSRFProtect(app)

# 为flask添加自定义的转换器  
app.url\_map.converters\["re"\] = ReConverter

# 注册蓝图(api接口蓝图)  
from ihom import api\_1\_0  
app.register\_blueprint(api\_1\_0.api,url\_prefix="/api/v1.0")

# 注册提供静态文件的蓝图(前端静态文件的蓝图)  
from ihom import web\_html  
app.register\_blueprint(web\_html.html)

return  app  

 

4.ihome/api_1_0接口蓝图配置

__init__.py的配置

# coding:utf-8

from flask import Blueprint

创建蓝图对象

api = Blueprint("api_1_0",__name__)

导入蓝图视图函数

from . import demo

demo.py的配置

# coding:utf-8

from . import api

import logging

from ihom import db,models
from flask import current_app

@api.route("/index")
def index():
# logging.error() # 记录错误信息
# logging.warn() # 记录警告信息
# logging.info() # 记录信息
# logging.debug() # 调试
current_app.logger.error("error info")
current_app.logger.warn("warn info")
current_app.logger.info("info info")
current_app.logger.debug("debug info")

return "index page"

5.ihome/models.py数据库的基本配置

5.1迁移数据库命令分为三步

第一步:创建迁移仓库

python manage.py db init

第二步:创建迁移脚本

python manage.py db migrate -m 'init table'

'init table'是注释

第三步:更新数据库

python manage.py db upgrade

注意

如果数据库迁移成功但没有数据库没表的时候

1.进入shell

python manage.py shell

2.导入你要迁移的数据表模型类

from apps.cms.models import *

3.创建表

db.create_all()

5.2数据库的配置

from . import db
from datetime import datetime

class BaseModel(object):
"""模型基类, 创建时间与更新时间"""
create_time = db.Column(db.DateTime, default=datetime.now) # 记录创建时间
updata_time = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now) # 记录更新时间

class User(BaseModel, db.Model):
"""用户"""

\_\_tablename\_\_ = "ih\_user\_profile"

id = db.Column(db.Integer, primary\_key=True)  # 用户编号  
name = db.Column(db.String(32), unique=True, nullable=False)  # 用户名称  
password\_hash = db.Column(db.String(128), nullable=False)  # 加密密码  
mobile = db.Column(db.String(11), unique=True, nullable=False)  # 手机号  
real\_name = db.Column(db.String(32))  # 真实姓名  
id\_card = db.Column(db.String(20))  # 身份证  
avatar\_url = db.Column(db.String(128))  # 用户头像路径  
houses = db.relationship("House", backref="user")  # 用户发布的房屋  
orders = db.relationship("Order", backref="user")  # 用户的订单

class Area(BaseModel, db.Model):
"""城区"""

\_\_tablename\_\_ = "ih\_area\_info"

id = db.Column(db.Integer, primary\_key=True)  # 区域编号  
name = db.Column(db.String(32), nullable=False)  # 区域名字  
houses = db.relationship("House", backref="area")  # 区域的房屋

房屋设施表,建立房屋与房屋设施的多对多关系

house_facility = db.Table(
"ih_house_facility",
db.Column("house_id", db.Integer, db.ForeignKey("ih_house_info.id"), primary_key=True), # 房屋编号
db.Column("facility_id", db.Integer, db.ForeignKey("ih_facility_info.id"), primary_key=True), # 房屋设施
)

class House(BaseModel, db.Model):
"""房屋信息"""

\_\_tablename\_\_ = "ih\_house\_info"

id = db.Column(db.Integer, primary\_key=True)  # 房屋编号  
user\_id = db.Column(db.Integer, db.ForeignKey("ih\_user\_profile.id"), nullable=False)  # 房屋主人的用户  
area\_id = db.Column(db.Integer, db.ForeignKey("ih\_area\_info.id"), nullable=False)  # 归属地的区域编号  
title = db.Column(db.String(64), nullable=False)  # 标题  
price = db.Column(db.Integer, default=0)  # 单价,单位分  
address = db.Column(db.String(512), default="")  # 地址  
room\_count = db.Column(db.Integer, default=1)  # 房间数目  
acreage = db.Column(db.Integer, default=0)  # 房屋面积  
unit = db.Column(db.String(32), default="")  # 房屋单元,几房几厅  
capacity = db.Column(db.Integer, default=1)  # 房屋容纳的人数  
beds = db.Column(db.String(64), default="")  # 房屋床铺的配置  
deposit = db.Column(db.Integer, default=0)  # 房屋押金  
min\_days = db.Column(db.Integer, default=0)  # 最少入住天数  
max\_days = db.Column(db.Integer, default=0)  # 最多入住天数  
order\_count = db.Column(db.Integer, default=0)  # 预定完成的该房屋的订单数  
index\_image\_url = db.Column(db.String(256), default="")  # 房屋主图的路径

facilities = db.relationship("Facility", secondary=house\_facility)  # 房屋的设施

image = db.relationship("HouseImage")  # 房屋的图片  
orders = db.relationship("Order", backref="house")  # 房屋的订单

class Facility(BaseModel, db.Model):
"""设施信息"""
__tablename__ = "ih_facility_info"

id = db.Column(db.Integer, primary\_key=True)  # 设施编号  
name = db.Column(db.String(32), nullable=False)  # 设施名字

class HouseImage(BaseModel, db.Model):
"""房屋图片"""
__tablename__ = "ih_house_image"

id = db.Column(db.Integer, primary\_key=True)  
house\_id = db.Column(db.Integer, db.ForeignKey("ih\_house\_info.id"), nullable=False)  # 房屋编号  
url = db.Column(db.String(256), nullable=False)  # 图片路径

class Order(BaseModel, db.Model):
"""订单"""
__tablename__ = "ih_order_info"

id = db.Column(db.Integer, primary\_key=True)  # 订单编号  
user\_id = db.Column(db.Integer, db.ForeignKey("ih\_user\_profile.id"), nullable=False)  # 下单的用户编号  
house\_id = db.Column(db.Integer, db.ForeignKey("ih\_house\_info.id"), nullable=False)  # 预订房间编号  
begin\_data = db.Column(db.DateTime, nullable=False)  # 预定的起始时间  
end\_date = db.Column(db.DateTime, nullable=False)  # 预定的结束时间  
days = db.Column(db.Integer, nullable=False)  # 预定的总天数  
house\_price = db.Column(db.Integer, nullable=False)  # 房屋单价  
amount = db.Column(db.Integer, nullable=False)  # 订单的总金额  
status = db.Column(  # 订单状态  
    db.Enum(  
        "WAIT\_ACCEPT",  # 待接单  
        "WAIT\_PAYMENT",  # 待支付  
        "PAID",  # 已支付  
        "WAIT\_COMMENT",  # 待评价  
        "COMPLETE",  # 已完成  
        "CANCELED",  # 已取消  
        "REJECTED"  # 已拒单  
    ),  
    default="WAIT\_ACCEPT", index=True)  
comment = db.Column(db.Text)  # 订单的评论信息或者拒单原因

6.ihome/web_html.py静态文件蓝图相关配置

# coding:utf-8

from flask import Blueprint,current_app,make_response
from flask_wtf import csrf

提供静态文件的蓝图

html = Blueprint("web_html",__name__)

127.0.0.1:5000/()

127.0.0.1:5000/(index.html)

127.0.0.1:5000/register.html

127.0.0.1:5000/favicon.ico # 游览器认为的网站标识,游览器自己请求这个资源

@html.route("/")
def get_html(html_file_name):
"""提供HTML文件"""
# 如果html_file_name为"",表示访问的路径是/,请求主页
if not html_file_name:
html_file_name = "index.html"

# 如果资源名不是favicon.ico  
if html\_file\_name != "favicon.ico":  
    html\_file\_name = "html/" + html\_file\_name

# 创建一个csrf\_token值  
csrf\_token = csrf.generate\_csrf()

# flask提供返回静态文件的方法  
resp = make\_response(current\_app.send\_static\_file(html\_file\_name))

# 设置cookie值(本次打开游览器才有效)  
resp.set\_cookie("csrf\_token",csrf\_token)

return resp

ihome/api_1_0 文件夹下 新建文件 verify_code.py 文件,用于验证码接口。

verify_code.py 验证码接口

# coding:utf-8

from . import api
from ihom.utils.captcha.captcha import captcha
from ihom import redis_store,constants
from flask import current_app,jsonify,make_response
from ihom.utils.response_code import RET

GET 127.0.0.1/api/v1.0/image_codes/

@api.route("/image_codes/")
def get_image_code(image_image_id):
"""
获取图片验证码
: params image_code_id:图片验证码编号
:return: 正常:验证码图片 异常:返回json
"""
# 业务逻辑处理
# 生成验证码图片
# 名字,真实文本,图片数据
name,text,image_data = captcha.generate_captcha()

# 将验证码真实值与编号保存到redis中,设置有效期  
# redis: 字符串 列表 哈希  set  
# "key":xxx  
# 使用哈希维护有效期只能整体设置  
# "image\_codes":{"id1":"1","编号2":""} 哈希类型    hset("image\_codes","id1","1")

# 单条维护记录,选用字符串  
# "image\_code\_编号":"真实值"  
# 第一步:设置值  
# redis\_store.set("image\_code\_%s" % image\_image\_id, text)  
# 第二步:设置有效期  
# redis\_store.expire("image\_code\_%s" % image\_image\_id, constants.IMAGE\_CODE\_REDIS\_EXPIRES)  
# 可以一步设置完成  
try:  
    redis\_store.setex("image\_code\_%s" % image\_image\_id, constants.IMAGE\_CODE\_REDIS\_EXPIRES, text)  
except Exception as e:  
    # 记录日志  
    current\_app.logger.error(e)  
    return  jsonify(errno=RET.DBERR,errmsg="保存图片验证码信息失败")

# 返回图片  
resp = make\_response(image\_data)  
# 设置响应头为图片类型  
resp.headers\["Content-Type"\] = "image/jpg"  
return resp

captcha.py 生成验证码图片

Python 生成图片验证码

constants.py存放常量

# coding:utf-8

保存常量的文件

图片验证码的redis有效期

IMAGE_CODE_REDIS_EXPIRES = 180

使用第三方工具:云通讯

注册账号

添加测试号码

下载SDK:把SDK的代码拷贝到libs文件下

SMS.py存放短信函数

新建SMS.py文件,编写发送短信的函数

from ronglian_sms_sdk import SmsSDK
import json

accId = '容联云通讯分配的主账号ID'
accToken = '容联云通讯分配的主账号TOKEN'
appId = '容联云通讯分配的应用ID'

class CCP(object):
"""自己封装的发送短信辅助类"""
# 用来保存对象的类属性
instance = None

def \_\_new\_\_(cls):  
    # 判断CCP类有没有已经创建好的对象  
    # 如果没有,创建一个对象,并保存  
    # 如果有,则将保存将保存的对象直接返回  
    if cls.instance is None:  
        obj = super(CCP,cls).\_\_new\_\_(cls)  
        cls.instance = obj  
        obj.sdk = SmsSDK(accId, accToken, appId)  
    return cls.instance

def send\_template\_sms(self,tid,mobile,datas):  
    resp = self.sdk.sendMessage(tid, mobile, datas)  
    resp = json.loads(resp)  
    status\_code = resp.get("statusCode")  
    if status\_code == "000000":  
        # 表示发送短信成功  
        return 0  
    else:  
        # 发送失败  
        return -1

"""
tid = '容联云通讯创建的模板'
mobile = '手机号1,手机号2'
datas = ('变量1', '变量2')
"""

if __name__ == '__main__':
ccp = CCP()
ret = ccp.send_template_sms(1,"18311111111",('123','5'))
print(ret)

verify_code.py 短信验证码接口

# coding:utf-8

from . import api
from ihom.utils.captcha.captcha import captcha
from ihom import redis_store,constants,db
from flask import current_app,jsonify,make_response,request
from ihom.utils.response_code import RET
from ihom.models import User
from ihom.libs.yuntongxun.sms import CCP
import random

图片验证码

GET 127.0.0.1/api/v1.0/image_codes/

@api.route("/image_codes/")
def get_image_code(image_image_id):
….

短信验证码

GET 127.0.0.1/api/v1.0/sms_codes/?image_code=xxx&image_code_id=xxx

@api.route("/sms_codes/")
def get_sms_code(mobile):
"""获取短信验证码"""
# 获取参数
image_code = request.args.get("image_code")
image_code_id = request.args.get("image_code_id")

# 校验参数  
if not all(\[image\_code,image\_code\_id\]):  
    # 表示参数不完整  
    return jsonify(error=RET.PARAMERR,errmsg="参数不完整")

# 业务逻辑处理  
# 从redis中取出真实的图片验证码  
try:  
    real\_image\_code = redis\_store.get("image\_code\_%s"%image\_code\_id)  
except Exception as e:  
    current\_app.logger.error(e)  
    return  jsonify(errno=RET.DBERR,errmsg="数据库异常")

# 判断图片验证码是否过期  
if real\_image\_code is None:  
    # 表示图片验证码过期  
    return jsonify(errno=RET.NODATA,errmsg="图片验证码失效")

# 删除redis图片验证码,防止用户使用同一个验证码验证多次  
try:  
    redis\_store.delete("image\_code\_%s"%image\_code\_id)  
except Exception as e:  
    current\_app.logger.error(e)

# 与用户填写的值进行对比  
if real\_image\_code.lower() != image\_code.lower():  
    # 用户填写的验证码错误  
    return jsonify(errno=RET.DATAERR,errmsg="验证码错误")

# 判断手机号的操作有没有之前的记录,如有有,则用户操作频繁,不做处理  
try:  
    send\_flag = redis\_store.get("send\_sms\_code\_%s"%mobile)  
except Exception as e:  
    current\_app.logger.error(e)  
else:  
    if send\_flag is not None:  
        # 在60s内之前发送过操作记录  
        return jsonify(errno=RET.REQERR,errmsg="请求过于频繁,请60秒后重试")

# 判断手机号是否存在  
try:  
    user = User.query.filter\_by(mobile=mobile).first()  
except Exception as e:  
    current\_app.logger.error(e)  
else:  
    if user is not None:  
        # 表示手机号已存在  
        return jsonify(errno=RET.DATAEXIST,errmsg="手机号已存在")

# 如果手机号不存在,则生成短信验证码  
sms\_code = "%06d" % random.randint(0,999999)

# 保存真实的短信验证码  
try:  
    redis\_store.setex("sms\_code\_%s"%mobile,constants.SMS\_CODE\_REDIS\_EXPIRES,sms\_code)  
    # 保存发送这个手机号的记录,防止用户60s内再次发送获取验证码  
    redis\_store.setex("send\_sms\_code\_%s"%mobile,constants.SEND\_SMS\_CODE\_INTERVAL,1)  
except Exception as e:  
    current\_app.logger.error(e)  
    return jsonify(errno=RET.DBERR,errmsg="保存短信验证码异常")

# 发送短信  
try:  
    ccp = CCP()  
    result = ccp.send\_template\_sms(1,mobile,\[sms\_code,int(constants.SMS\_CODE\_REDIS\_EXPIRES/60)\])  
except Exception as e:  
    current\_app.logger.error(e)  
    return jsonify(errno=RET.THIRDERR,errmsg="短信发送异常")

# 返回值  
if result == 0:  
    # 发送成功  
    return jsonify(errno=RET.OK,errmsg="短信发送成功")  
else:  
    return jsonify(errno=RET.THIRDERR,errmsg="短信发送失败")