Flask - 邮件发送和数据模型

发表于 更新于 分类于 > > 本文字数： 9.7k 阅读时长 ≈ 16 分钟

flask-mail

  • 说明：是一个邮件发送的拓展库，使用非常简洁

  • 安装：pip install flask-mail

  • 使用：

    1. 先配置：邮件服务器、用户名、密码

      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      from flask_mail import Mail, Message
      import os

      # 必须先写配置，然后再创建Mail对象，否则配置无效
      # 邮件服务器
      app.config['MAIL_SERVER'] = os.getenv('MAIL_SERVER') or 'smtp.163.com'
      # 配置用户名
      app.config['MAIL_USERNAME'] = os.environ.get('MAIL_USERNAME') or '16657158725@163.com'
      # 配置邮箱密码
      app.config['MAIL_PASSWORD'] = os.getenv('MAIL_PASSWORD') or '123456'

    1. 创建 Mail 对象，专门用于发送邮件

      1
      2
      # 创建Mail对象
      mail = Mail(app)

    2. 创建邮件消息对象，设置相关参数

      1
      2
      3
      4
      5
      6
      7
      8
      # 准备邮件内容
      msg = Message(subject='账户激活',
                    recipients=['15168896730@163.com'],
                    sender=app.config['MAIL_USERNAME'])
      # 添加HTML内容，通过浏览器查看邮件
      msg.html = '<h1>邮件发送测试，请点击链接完成账户激活！</h1>'
      # 添加body内容，命令行接收邮件
      msg.body = '邮件发送测试，请点击链接完成账户激活！'

    3. 发送邮件：mail.send(msg)

      1
      2
      # 发送邮件
      mail.send(msg)

  • 发送邮件函数的封装

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    # 封装函数发送邮件
    def send_mail(to, subject, template, **kwargs):
        # 准备邮件内容
        msg = Message(subject=subject,
                      recipients=[to],
                      sender=app.config['MAIL_USERNAME'])
        # 添加HTML内容，通过浏览器查看邮件
        msg.html = render_template(template + '.html', **kwargs)
        # 添加body内容，命令行接收邮件
        msg.body = render_template(template + '.txt', **kwargs)
        # 发送邮件
        mail.send(msg)

  • 异步发送邮件

    1
    2
    3
    4
    def async_send_mail(app, msg):
        # 发送邮件需要程序的上下文
        with app.app_context():
            mail.send(msg)
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    # 封装函数发送邮件
    def send_mail(to, subject, template, **kwargs):
        # 该函数不在manage.py中
        # 从代理对象中获取原始对象
        app = current_app._get_current_object()
        # 准备邮件内容
        msg = Message(subject=subject,
                      recipients=[to],
                      sender=app.config['MAIL_USERNAME'])
        # 添加HTML内容，通过浏览器查看邮件
        msg.html = render_template(template + '.html', **kwargs)
        # 添加body内容，命令行接收邮件
        msg.body = render_template(template + '.txt', **kwargs)
        # 创建线程
        thr = Thread(target=async_send_mail, args=[app, msg])
        # 启动线程
        thr.start()
        return thr

  • 示例：

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    from flask import Flask, render_template, current_app
    from flask_script import Manager
    from flask_mail import Mail, Message
    from threading import Thread
    import os

    app = Flask(__name__)

    # 必须先写配置，然后再创建Mail对象，否则配置无效
    # 邮件服务器
    app.config['MAIL_SERVER'] = os.getenv('MAIL_SERVER') or 'smtp.163.com'
    # 配置用户名
    app.config['MAIL_USERNAME'] = os.environ.get('MAIL_USERNAME') or '16657158725@163.com'
    # 配置邮箱密码
    app.config['MAIL_PASSWORD'] = os.getenv('MAIL_PASSWORD') or '123456'
    mail = Mail(app)

    manager = Manager(app)


    def async_send_mail(app, msg):
        # 发送邮件需要程序的上下文
        with app.app_context():
            mail.send(msg)


    # 封装函数发送邮件
    def send_mail(to, subject, template, **kwargs):
        # 该函数不在manage.py中
        # 从代理对象中获取原始对象
        app = current_app._get_current_object()
        # 准备邮件内容
        msg = Message(subject=subject,
                      recipients=[to],
                      sender=app.config['MAIL_USERNAME'])
        # 添加HTML内容，通过浏览器查看邮件
        msg.html = render_template(template + '.html', **kwargs)
        # 添加body内容，命令行接收邮件
        msg.body = render_template(template + '.txt', **kwargs)
        # 创建线程
        thr = Thread(target=async_send_mail, args=[app, msg])
        # 启动线程
        thr.start()
        return thr


    @app.route('/')
    def mail():
        send_mail('15168896730@163.com', '账户激活', 'activate', name='苏寅')
        return '邮件已发送'


    if __name__ == '__main__':
        manager.run()
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    <!DOCTYPE html>
    <html lang="en">
    <head>
        <meta charset="UTF-8">
        <title>账户激活</title>
    </head>
    <body>
        <h1>{{ name }}您好！</h1>
        <h2>邮件发送测试，请点击链接完成账户激活！</h2>
    </body>
    </html>

flask-sqlalchemy

  • 说明：提供了大多数关系数据模型的支持，提供了 ORM (对象关系映射)

  • 安装：pip install flask-sqlalchemy

  • 连接地址：就是指定操作的数据库

    • MySQL：mysql + pymysql://username:password@host/database?charset-utf8
    • sqlite3：
      • windows：sqlite:///c:path/to/database.db
      • linux：sqlite:////path/to/database.db
    • 选项：SQLALCHEMY_DATABASE_URI

  • 使用流程

    1. 先配置连接地址

      1
      2
      3
      4
      5
      6
      7
      from flask_sqlalchemy import SQLAlchemy

      # 数据库配置
      # 链接地址
      base_dir = os.path.abspath(os.path.dirname(__file__))
      database_uri = 'sqlite:///' + os.path.join(base_dir, 'data.sqlite')
      app.config['SQLALCHEMY_DATABASE_URI'] = database_uri

    2. 创建数据库操作对象

      1
      2
      # 创建对象
      db = SQLAlchemy(app)

    3. 创建数据库：db.create_all()

      1
      2
      3
      4
      5
      @app.route('/create/')
      def create():
          # 创建数据库
          db.create_all()
          return '数据库创建成功'

    4. 删除数据库：db.drop_all()

      1
      2
      3
      4
      5
      @app.route('/drop/')
      def drop():
          # 删除数据表
          db.drop_all()
          return '数据表已删除'

  • 添加终端操作命令

    • 使用装饰器：@manager.command
    • 装饰器函数名就是终端的命令名
    • 使用：python manage.py createall|dropall
      1
      2
      3
      4
      5
      6
      7
      # 添加命令行的删除数据库命令
      @manager.command
      def dropall():
          if prompt_bool('你确定要删库跑路吗？'):
              db.drop_all()
              return '数据表已删除'
          return '删库需谨慎！'

数据的 CURD 操作

  • 增加数据

    1. 准备数据：创建数据模型对象
    2. 添加多条数据
      • 一条：db.session.add()
      • 多条：db.session.add_all()
    3. 需要手动提交才会保存数据
    4. 配置自动添加
      • app.config['SQLALCHEMY_COMMIT_ON_TEARDOWM'] = True
    5. 关闭数据修改的追踪
      • app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

    • 示例：

      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      @app.route('/insert/')
      def insert():
          # 创建数据模型对象
          # qiqi = User(username='qiqi', email='qiqi@163.com')
          # 添加到数据库，增加一条数据
          # db.session.add(qiqi)

          # 添加多条数据
          fei = User(username='fei', email='fei@163.com')
          ouya = User(username='ouya', email='ouya@163.com')
          sunqi = User(username='sunqi', email='sunqi@163.com')
          db.session.add_all([fei, ouya, sunqi])

          # 提交操作
          db.session.commit()
          return '数据已插入'

  • 查询数据

    • 根据主键进行查询：User.query.get(主键)
    • 返回：查到返回一个对象，没有找到返回 None
    • 示例：
      1
      2
      3
      4
      5
      6
      7
      8
      # 查询数据
      @app.route('/select/<uid>')
      def select(uid):
          # 根据主键查询
          u = User.query.get(uid)
          if u:
              return u.username
          return '查无此人'

  • 修改数据

    1. 根据主键找到对象
    2. 修改对应的属性
    3. 重新添加到数据库（add），若添加的数据带 ID，则会自动识别为更新操作

    • 示例：

      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      # 修改数据
      @app.route('/update/<uid>')
      def update(uid):
          u = User.query.get(uid)
          if u:
              u.email = 'xxx@163.com'
              # 没有专门的更新操作当添加的对象有ID时会自动识别为更新操作
              db.session.add(u)
              db.session.commit()
              return '数据已更新'
          return '查无此人'

  • 删除数据

    1. 根据主键找到对象
    2. 删除对象：db.session.delete()
    3. 说明：很多时候我们不会做物理删除，只需要加一个用于删除的字段，修改字段就相当于删除，这叫逻辑删除。

    • 示例：

      1
      2
      3
      4
      5
      6
      7
      8
      9
      # 删除数据
      @app.route('/delete/<uid>')
      def delete(uid):
          u = User.query.get(uid)
          if u:
              db.session.delete(u)
              db.session.commit()
              return '数据已删除'
          return '查无此人'

模型设计参考

  • 常见字段类型

    类型名 python 类型 说明
    Integer int 32 位
    SmallInteger int 16 位
    BigInteger int/long 不受限制
    Float float 浮点数
    String str 变长字符串
    Text str 优化后的变长字符串
    Boolean bool 布尔值
    Date datetime.date 日期
    Time datetime.time 时间
    DateTime datetime.datetime 日期时间
    Interval datetime.timedelta 时间间隔

  • 常见字段选项

    选项 说明
    primary_key 是否作为主键索引，默认为 False
    unique 是否作为唯一索引，默认为 False
    index 是否作为普通索引，默认为 False
    nullable 是否可以为空，默认为 True
    default 设置默认值

  • 总结：

    • 插入数据可以不传值的字段，自增的主键，有默认值，可以为空
    • flask-sqlalchemy 要求每一个模型都有一个主键，名称通常为 id

数据库的迁移

  • 说明：
    项目开发中，总是避免不了的进行数据模型的更改，若已经有了数据库，此时数据模型与数据库中的表将不再一致；将数据模型的更改应用到对应数据库中的过程叫数据的迁移。前面的先删除后创建有点粗暴（副作用有点大，数据全部丢失），最好的方式是既做到数据库的更新，又不使数据丢失。若自己不会，可以借助第三方扩展库 flask-migrate 来完成。
  • 安装：pip install flask-migrate
  • 配置：
    1. 导入类库：from flask_migrate import Migrate, MigrateCommand
    2. 创建对象，传递 app，db 作为参数：migrate = Migrate(app, db)
    3. 将数据库迁移命令添加到终端：manager.add_command('db', MigrateCommand)
  • 使用：
    1. 初始化数据库迁移的厂库（会创建一个 migrations 的目录，及相关脚本），只需要一次
      • python manage.py db init
    2. 创建迁移脚本（根据数据模型与数据库的差异）
      • python manage.py db migrate
    3. 执行迁移脚本
      • python manage.py db upgrade
    • 说明：
      1. 第一步只需要执行一次，以后第二步和和第三步反复使用。
      2. 不一定每次自动迁移都会成功，迁移后要做检查，出问题需要手动处理。
  • 示例：
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
    16
    17
    18
    19
    20
    21
    22
    23
    24
    25
    26
    27
    28
    29
    30
    31
    32
    33
    34
    35
    36
    37
    38
    39
    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89
    90
    91
    92
    93
    94
    95
    96
    97
    98
    99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124
    125
    126
    127
    128
    129
    130
    131
    132
    133
    from flask import Flask
    from flask_script import Manager, prompt_bool
    from flask_sqlalchemy import SQLAlchemy
    from flask_migrate import Migrate, MigrateCommand
    import os

    app = Flask(__name__)
    manager = Manager(app)

    # 数据库配置
    # 链接地址
    base_dir = os.path.abspath(os.path.dirname(__file__))
    database_uri = 'sqlite:///' + os.path.join(base_dir, 'data.sqlite')
    app.config['SQLALCHEMY_DATABASE_URI'] = database_uri
    # 是否追踪数据的变化，发出警告，如果没有必要可以关闭
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
    # 配置自动提交，否则需要每次手动提交
    app.config['SQLALCHEMY_COMMIT_ON_TEARDOWM'] = True
    # 创建对象
    db = SQLAlchemy(app)

    # 创建数据库迁移对象
    migrate = Migrate(app, db)
    # 将迁移命令添加到命令行
    manager.add_command('db', MigrateCommand)


    # 设计数据模型类
    class User(db.Model):
        # 指定表明，不指定时做如下转换
        # 表名：大驼峰 转换为 小写+下划线
        # 如：UserModel => user_model
        __tablename__ = 'users'
        id = db.Column(db.Integer, primary_key=True)
        username = db.Column(db.String(20), unique=True)
        email = db.Column(db.String(60), unique=True)
        age = db.Column(db.Integer)


    # 增加数据
    @app.route('/insert/')
    def insert():
        # 创建数据模型对象
        # qiqi = User(username='qiqi', email='qiqi@163.com')
        # 添加到数据库，增加一条数据
        # db.session.add(qiqi)

        # 添加多条数据
        fei = User(username='fei', email='fei@163.com')
        ouya = User(username='ouya', email='ouya@163.com')
        sunqi = User(username='sunqi', email='sunqi@163.com')
        db.session.add_all([fei, ouya, sunqi])

        # 提交操作
        db.session.commit()
        return '数据已插入'


    # 查询数据
    @app.route('/select/<uid>')
    def select(uid):
        # 根据主键查询
        u = User.query.get(uid)
        if u:
            return u.username
        return '查无此人'


    # 修改数据
    @app.route('/update/<uid>')
    def update(uid):
        u = User.query.get(uid)
        if u:
            u.email = 'xxx@163.com'
            # 没有专门的更新操作当添加的对象有ID时会自动识别为更新操作
            db.session.add(u)
            db.session.commit()
            return '数据已更新'
        return '查无此人'


    # 删除数据
    @app.route('/delete/<uid>')
    def delete(uid):
        u = User.query.get(uid)
        if u:
            db.session.delete(u)
            db.session.commit()
            return '数据已删除'
        return '查无此人'


    @app.route('/')
    def index():
        return '数据模型测试'


    @app.route('/create/')
    def create():
        # 数据表已经存在，当需要数据迁移时不会再次创建，可以简单粗暴的先删除再创建
        db.drop_all()
        # 创建数据表，第一次会创建数据库
        db.create_all()
        return '数据表已创建'


    @app.route('/drop/')
    def drop():
        # 删除数据表
        db.drop_all()
        return '数据表已删除'


    # 添加命令行的创建数据库命令
    # 通过该装饰器修饰的函数名就是终端命令名
    @manager.command
    def createall():
        db.drop_all()
        db.create_all()
        return '数据表已创建'


    # 添加命令行的删除数据库命令
    @manager.command
    def dropall():
        if prompt_bool('你确定要删库跑路吗？'):
            db.drop_all()
            return '数据表已删除'
        return '删库需谨慎！'


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