Flask入门

1.浅谈FLask

“微”的含义

“微”并不代表整个应用只能塞在一个 Python 文件内，当然塞在单一文件内也 没有问题。 “微”也不代表 Flask 功能不强。微框架中的“微”字表示 Flask 的目标是保持核心简单而又可扩展。 Flask 不会替你做出许多决定，比如选用何 种数据库。类似的决定，如使用何种模板引擎，是非常容易改变的。 Flask 可以 变成你任何想要的东西，一切恰到好处，由你做主。

FLask完整版<=Django
#本博客开始从web开发讲解FLask，先将前后端不分离的项目，后将vue +flask,后从运维开发讲解FLask从源码分析flask为什么这么牛逼
#html，网络请看我其他博客

2.初识Flask

1.安装Flask

pip install flask

启动最小的hello, word

from flask import Flask

app = Flask(__name__)

@app.route('/')
def hello_world():
    return "<p>Hello, World!</p>"'

if __name__ == '__main__':
    app.run()
  1. 首先我们导入了 Flask 类。该类的实例将会成为我们的 WSGI 应用。
  2. 接着我们创建一个该类的实例。第一个参数是应用模块或者包的名称。 __name__ 是一个适用于大多数情况的快捷方式。有了这个参数， Flask 才能知道在哪里可以找到模板和静态文件等东西。
  3. 然后我们使用 route() 装饰器来告诉 Flask 触发函数 的 URL 。
  4. 函数返回需要在用户浏览器中显示的信息。默认的内容类型是 HTML ，因此字 符串中的 HTML 会被浏览器渲染。

2.路由

url_for() 函数用于构建指定函数的 URL。它把函数名称作为第一个 参数。它可以接受任意个关键字参数，每个关键字参数对应 URL 中的变量。未知变量 将添加到 URL 中作为查询参数。

from flask import Flask, url_for

app = Flask(__name__)


@app.route('/')
def hello1():
    return 'Hello World!'


@app.route('/zz/<int:post_id>')
def hello2(name='zzz'):
    return f'Hello World!{name}'

@app.route('/test')
def test_url_for():
    # 下面是一些调用示例（请在命令行窗口查看输出的 URL）：
    print(url_for('hello1'))  # 输出：/
    print(url_for('hello2', name='213'))  # /zz/123
    print(url_for('test_url_for', num=2))  # 输出：/test?num=2
    return 'Test page'

if __name__ == '__main__':
    app.run()
string （缺省值） 接受任何不包含斜杠的文本
int 接受正整数
float 接受正浮点数
path 类似 string ，但可以包含斜杠
uuid 接受 UUID 字符串

3.redict重定向

from flask import Flask, redirect

app = Flask(__name__)

@app.route('/hello/')
def hello():
    return redirect("/index")
    #return redirect(url_for('index')) 同上面的结果是一样的

@app.route("/index")
def index():
    return "123"
    
if __name__ == '__main__':
    app.run()

hello尾部有一个斜杠，看起来就如同一个文件 夹。访问一个没有斜杠结尾的 URL 时 Flask 会自动进行重 定向，帮您在尾部加上一个斜杠

index的 URL 没有尾部斜杠，因此其行为表现与一个文件类似。如果访问这 个 URL 时添加了尾部斜杠（就会得到一个 404 “未找到” 错 误。这样可以保持 URL 唯一，并有助于搜索引擎重复索引同一页面。

4.Jinja2

举例：

<h1>{{ username }}的个人主页</h1> 
{% if bio %} 
	<p>{{ bio }}</p> {# 这里的缩进只是为了可读性，不是必须的 #} 
{% else %} 
	<p>自我介绍为空。</p> 
{% endif %} {# 大部分 Jinja 语句都需要声明关闭 #}

Jinja2 的语法和 Python 大致相同，你在后面会陆续接触到一些常见的用法。在模

板里，你需要添加特定的定界符将 Jinja2 语句和变量标记出来，下面是三种常用的

定界符：

{{ ... }} 用来标记变量。 

{% ... %} 用来标记语句，比如 if 语句，for 语句等。 

{# ... #} 用来写注释。

编写个人主页

模拟数据库数据data，创建后端程序

from flask import Flask, render_template

app = Flask(__name__)

name = "zbb"
data = [
    {'name': "zbb"},
    {'name': "zbb"},
    {'name': "zxy"},
    {'name': "zxy"},
]


@app.route('/')
def hello():
    return render_template('index.html', name=name, data=data)


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

templates/index.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>{{ name }}的主页</title>
</head>
<body>
    <h2>{{ name }}的主页</h2>
    <p>{{ data|length }} Titles</p>
    <ul>
        {% for z in data %} {# 迭代 data 变量 #}
        <li>{{ z.name }}-{{ z.age }} </li>{# 迭代 data 变量 #}
        {% endfor %} {# 使用 endfor 标签结束 for 语句 #}
    </ul>
</body>
</html>

更多模板用法请看我的这篇博文

https://www.cnblogs.com/zdqc/p/11638144.html

前端知识请看

https://www.cnblogs.com/zdqc/category/1428916.html

5.static

存放静态文件的目录

我们在我们的网页中加入图片

    <link rel="icon" href="{{ url_for('static', filename='123.png') }}">

6.数据库

Flask和Django 数据库上的操作的区别就是ORM

Django内置ORM，ORM是什么？

1. MVC或者MVC框架中包括一个重要的部分，就是ORM，它实现了数据模型与数据库的解耦，即数据模型的设计不需要依赖于特定的数据库，通过简单的配置就可以轻松更换数据库，这极大的减轻了开发人员的工作量，不需要面对因数据库变更而导致的无效劳动
2. ORM是“**对象-关系-映射**”的简称。
3. 类对象--->sql--->pymysql--->mysql服务端--->磁盘，orm其实就是将类对象的语法翻译成sql语句的一个引擎

ORM特点

符合python思想，一切接对象，只会ORM就可以操作基本上所有的关系型数据库，如mysql， Postgres oracle等，书写及其简单，但是要代码转换sql，性能较低，也很难写出高性能的sql语句

想学的ORM的同学可以看看我的这篇文章

https://www.cnblogs.com/zdqc/p/11656606.html
#flask也可以使用ORM
# pip install flask-sqlalchemy

这里使用原生sql，封装pymysql

pip install  pymysql

pmysql.py

import pymysql


class MysqlC:

    def __init__(self, host, user, password, database):
        self.con = None
        self.arguments = {
            "host": host,
            "user": user,
            "password": password,
            "database": database,
            "charset": 'utf8'
        }

    def post(self, sql):
        with self:
            try:
                data = self.cursor.execute(sql)
                self.con.commit()

                return data
            except Exception as e:
                self.con.rollback()
                return e

    def get(self, sql, one=None):
        with self:
            try:
                self.cursor.execute(sql)
                if one:
                    return self.cursor.fetchone()
                else:
                    return self.cursor.fetchall()
            except Exception as e:
                return e

    def __enter__(self):
        if self.con is None:
            try:
                self.con = pymysql.connect(**self.arguments)
                self.cursor = self.con.cursor(pymysql.cursors.DictCursor)
            except Exception:
               raise "数据库连接失败！"

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.cursor.close()
        self.con.close()
        self.con = None


if __name__ == '__main__':
    mysql = MysqlC('123', 'root', 'z21', 'white_ip')
    w = mysql.get("select * from  user")
    print(w)

app.py

from flask import Flask, render_template
from mysql.pmysql import MysqlC

app = Flask(__name__)
pmysql = MysqlC("121.3", "root", "我好想你啊", "zzb")
name = "zbb"


@app.route('/')
def hello():
    data = pmysql.get("select name,age from t")
    # [{'name': 'zbb', 'age': 83}, {'name': 'zbb', 'age': 83}, {'name': 'zbb', 'age': 83}]
    print(data)
    return render_template('index.html', name=name, data=data)


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

index.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <link rel="icon" href="{{ url_for('static', filename='123.png') }}">
    <title>{{ name }}的主页</title>
</head>
<body>
    <h2>{{ name }}的主页</h2>
    <p>{{ data|length }} Titles</p>
    <ul>
        {% for z in data %} {# 迭代 data 变量 #}
        <li>{{ z.name }}-{{ z.age }} </li>{# 迭代 data 变量 #}
        {% endfor %} {# 使用 endfor 标签结束 for 语句 #}
    </ul>
</body>
</html>

目录结构

7.配置文件

from flask import Flask,Response,render_template

app = Flask(__name__,template_folder="templates",static_folder="static")
app.config.from_object("config.settings.DevelopmentConfig") 
@app.route('/login')
def login():
    return "ok"


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

config下的settings.py

class Config:
    DEBUG = False
    TESTING = False
    DATABASE_URI = 'sqlite://:memory:'
 
 
class ProductionConfig(Config):
    DATABASE_URI = 'mysql://user@localhost/foo'
 
 
class DevelopmentConfig(Config):
    DEBUG = True

调用

app.config.get("参数")

默认配置

{
    'DEBUG':                                False,  是否开启Debug模式
    'TESTING':                              False,                          是否开启测试模式
    'PROPAGATE_EXCEPTIONS':                 None,                         
    'PRESERVE_CONTEXT_ON_EXCEPTION':        None,
    'SECRET_KEY':                           None,
    'PERMANENT_SESSION_LIFETIME':           timedelta(days=31),
    'USE_X_SENDFILE':                       False,
    'LOGGER_NAME':                          None,
    'LOGGER_HANDLER_POLICY':               'always',
    'SERVER_NAME':                          None,
    'APPLICATION_ROOT':                     None,
    'SESSION_COOKIE_NAME':                  'session',
    'SESSION_COOKIE_DOMAIN':                None,
    'SESSION_COOKIE_PATH':                  None,
    'SESSION_COOKIE_HTTPONLY':              True,
    'SESSION_COOKIE_SECURE':                False,
    'SESSION_REFRESH_EACH_REQUEST':         True,
    'MAX_CONTENT_LENGTH':                   None,
    'SEND_FILE_MAX_AGE_DEFAULT':            timedelta(hours=12),
    'TRAP_BAD_REQUEST_ERRORS':              False,
    'TRAP_HTTP_EXCEPTIONS':                 False,
    'EXPLAIN_TEMPLATE_LOADING':             False,
    'PREFERRED_URL_SCHEME':                 'http',
    'JSON_AS_ASCII':                        True,
    'JSON_SORT_KEYS':                       True,
    'JSONIFY_PRETTYPRINT_REGULAR':          True,
    'JSONIFY_MIMETYPE':                     'application/json',
    'TEMPLATES_AUTO_RELOAD':                None,
}

8.数据库连接池

DBUtils是Python的一个用于实现数据库连接池的模块。

创建一批连接到连接池，供所有线程共享使用。
PS：由于pymysql、MySQLdb等threadsafety值为1，所以该模式连接池中的线程会被所有线程共享

如果没有连接池，使用pymysql来连接数据库时，单线程应用完全没有问题，但如果涉及到多线程应用那么就需要加锁，一旦加锁那么连接势必就会排队等待，当请求比较多时，性能就会降低了。

加锁正常

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import pymysql
import threading
from threading import RLock

LOCK = RLock()
CONN = pymysql.connect(host='127.0.0.1',
                       port=3306,
                       user='root',
                       password='123',
                       database='pooldb',
                       charset='utf8')


def task(arg):
    with LOCK:
        cursor = CONN.cursor()
        cursor.execute('select * from tb1')
        result = cursor.fetchall()
        cursor.close()

        print(result)


for i in range(10):
    t = threading.Thread(target=task, args=(i,))
    t.start()

无锁报错

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql
import threading
CONN = pymysql.connect(host='127.0.0.1',
                       port=3306,
                       user='root',
                       password='123',
                       database='pooldb',
                       charset='utf8')


def task(arg):
    cursor = CONN.cursor()
    cursor.execute('select * from tb1')
    result = cursor.fetchall()
    cursor.close()

    print(result)


for i in range(10):
    t = threading.Thread(target=task, args=(i,))
    t.start()

用法

import time
import pymysql
import threading
from DBUtils.PooledDB import PooledDB, SharedDBConnection
POOL = PooledDB(
    creator=pymysql,  # 使用链接数据库的模块
    maxconnections=6,  # 连接池允许的最大连接数，0和None表示不限制连接数
    mincached=2,  # 初始化时，链接池中至少创建的空闲的链接，0表示不创建
    maxcached=5,  # 链接池中最多闲置的链接，0和None不限制
    maxshared=3,  # 链接池中最多共享的链接数量，0和None表示全部共享。PS: 无用，因为pymysql和MySQLdb等模块的 threadsafety都为1，所有值无论设置为多少，_maxcached永远为0，所以永远是所有链接都共享。
    blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
    maxusage=None,  # 一个链接最多被重复使用的次数，None表示无限制
    setsession=[],  # 开始会话前执行的命令列表。如：["set datestyle to ...", "set time zone ..."]
    ping=0,
    # ping MySQL服务端，检查是否服务可用。# 如：0 = None = never, 1 = default = whenever it is requested, 2 = when a cursor is created, 4 = when a query is executed, 7 = always
    host='127.0.0.1',
    port=3306,
    user='root',
    password='123',
    database='pooldb',
    charset='utf8'
)


def func():
    # 检测当前正在运行连接数的是否小于最大链接数，如果不小于则：等待或报raise TooManyConnections异常
    # 否则
    # 则优先去初始化时创建的链接中获取链接 SteadyDBConnection。
    # 然后将SteadyDBConnection对象封装到PooledDedicatedDBConnection中并返回。
    # 如果最开始创建的链接没有链接，则去创建一个SteadyDBConnection对象，再封装到PooledDedicatedDBConnection中并返回。
    # 一旦关闭链接后，连接就返回到连接池让后续线程继续使用。
    conn = POOL.connection()

    # print(th, '链接被拿走了', conn1._con)
    # print(th, '池子里目前有', pool._idle_cache, '
')

    cursor = conn.cursor()
    cursor.execute('select * from tb1')
    result = cursor.fetchall()
    conn.close()


func()

实例：

import pymysql
from dbutils.pooled_db import PooledDB


class MysqlC:

    def __init__(self, host, user, password, db):
        self.pool = PooledDB(
            creator=pymysql,  # 使用链接数据库的模块
            maxconnections=200,  # 连接池允许的最大连接数，0和None表示不限制连接数
            mincached=10,  # 初始化时，链接池中至少创建的链接，0表示不创建
            blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
            ping=0,
            host=host,
            port=3306,
            user=user,
            password=password,
            database=db,
            charset='utf8'
        )


    def post(self, sql):
        with self:
            try:
                data = self.cursor.execute(sql)
                self.con.commit()

                return data
            except Exception as e:
                self.con.rollback()
                return e

    def get(self, sql, one=None):
        with self:
            try:
                self.cursor.execute(sql)
                if one:
                    return self.cursor.fetchone()
                else:
                    return self.cursor.fetchall()
            except Exception as e:
                return e

    def __enter__(self):
        self.con = self.pool.connection()
        self.cursor = self.con.cursor(pymysql.cursors.DictCursor)

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.cursor.close()
        self.con.close()


if __name__ == '__main__':
    mysql = MysqlC('12193', 'root', 'zb51', 'white_ip')

    w = mysql.get("select * from  user")
    print(w)

放在配置文件中

app.py

from flask import Flask, Response, render_template, current_app
from mysql.mysql_con import MysqlC

app = Flask(__name__, template_folder="templates", static_folder="static")
app.config.from_object("config.settings.DevelopmentConfig")
mysql = MysqlC(**app.config.get("PYMYSQL"))


@app.route('/')
def login():
    w = mysql.get("select * from  user")
    print(w)
    return "ok"


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

setting.py

import pymysql
from dbutils.pooled_db import PooledDB


class Config:
    PYMYSQL = {
        "host": "1193",
        "user": "root",
        "password": "z1",
        "db": "white_ip",

    }


class ProductionConfig(Config):
    DATABASE_URI = 'mysql://user@localhost/foo'


class DevelopmentConfig(Config):
    HOST = "123"

7.Form表单

创建一个login.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>

<form action="" method="post" enctype="multipart/form-data">
    用户名：<input type="text" name="user">
    密码：<input type="password" name="pwd">
    <input type="submit" value="提交">
</form>

<script></script>
</body>
</html>

resquest可以拿到请求中的数据

from flask import Flask, render_template, request
import os

app = Flask(__name__)
app.debug = True

@app.route("/login", methods=["POST", "GET"])
def login():
    if request.method == "GET":
        # 获取URL中的参数，例如传入参数：http://127.0.0.1:5000/login?id=1
        print(request.args.get("id"))
        # 获取URL中的参数 转换成 字典
        print(request.args.to_dict())
        # 获取请求原始信息
        print(request.environ)
        # 路由地址 /login
        print(request.path)
        # 获取访问路径
        print(request.url)  # http://127.0.0.1:5000/login?id=1
        # 获取URL头,不包含参数 /login
        print(request.base_url)  # http://127.0.0.1:5000/login

        return render_template("login.html")

    if request.method == "POST":
        # 请求头中的数据
        print(request.headers)

        print(request.json)  # 请求头中 Content-type:application/json 数据序列化 request.json
        print(request.data)  # 请求头中 Content-type 不包含 Form or data

        # Formdata 和 Args查询参数 中的数据
        print(request.values)

        username = request.form.get("user")
        password = request.form.get("pwd")
        print(username, password)

        return "200 OK"

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

8.文件上传

用 Flask 处理文件上传很容易，只要确保不要忘记在您的 HTML 表单中设置 enctype="multipart/form-data" 属性就可以了。否则浏览器将不会传送您的文件。

from flask import request

@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
    if request.method == 'POST':
        f = request.files['the_file']
        f.save('/var/www/uploads/uploaded_file.txt')
    ...

如果想要知道文件上传之前其在客户端系统中的名称，可以使用 filename 属性。但是请牢记这个值是 可以伪造的，永远不要信任这个值。如果想要把客户端的文件名作为服务器上的文件名， 可以通过 Werkzeug 提供的 secure_filename() 函数:

from werkzeug.utils import secure_filename

@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
    if request.method == 'POST':
        file = request.files['the_file']
        file.save(f"/var/www/uploads/{secure_filename(file.filename)}")
    ...

9.Session

flask会将你的SessionID存放在客户端的Cookie中

基本用法，具体用法请看练习题

使用在app.py中引入

from flask import session
.......
app = Flask(__name__)
app.secret_key = "i am session"   #这里只需要随机指定一个字符串即可
.......


session["username"] = USER["username"] #设置session

验证session

session.get("username"):    #获取session
session.pop('username') #删除session

11.练习

将今天所学全部用上，写一个用户注册，登录验证模块

加上密码验证

创建login.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <link rel="icon" href="{{ url_for('static', filename='123.png') }}">
</head>
<body>
{{ mes }}
<form action="" method="post" enctype="multipart/form-data">
    用户名：<input type="text" name="username">
    密码：<input type="password" name="pwd">
    <input type="submit" value="提交">
</form>
</body>
</html>

reg.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <link rel="icon" href="{{ url_for('static', filename='123.png') }}">
</head>
<body>
{{ mes }}
<form action="" method="post" enctype="multipart/form-data">
    用户名：<input type="text" name="username">
    密码：<input type="password" name="pwd">
    <input type="submit" value="注册">
</form>
</body>
</html>

app.py

from flask import Flask, redirect, url_for, session, request, render_template
from mysql.pmysql import MysqlC
from werkzeug.security import generate_password_hash, check_password_hash  #hash加密密码和解密密码

app = Flask(__name__)
app.secret_key = "qwerty"  # 设置密钥使用session
app.config.from_object("conf.settings.DevelopmentConfig")  # 获取配置文件中的mysql链接
mysql_con = app.config.get("MYSQL")
pmysql = MysqlC(**mysql_con)  # 调取封装好的pymysq


@app.route('/', methods=["get", "post"])
def login():
    # 获取用户提交的数据
    if request.method == "GET":
        return render_template("login.html")

    if request.method == "POST":
        username = request.form.get("username")
        pwd = request.form.get("pwd")
        if not username or not pwd:
            return render_template("login.html", mes="请输入用户名密码")
        password = pmysql.get(f"select pwd from user where username='{username}'", one=True)
        if password:
            if check_password_hash(password["pwd"], pwd):
                # 将用户名加到session中用于判断用户是否登录了
                session["username"] = username
                return redirect('/index')
            else:
                return render_template("login.html", mes="密码输入错误")
        else:
            return render_template("login.html", mes="用户名输入错误")


@app.route('/reg', methods=["get", "post"])
def reg():
    if request.method == "GET":
        return render_template("reg.html")

    if request.method == "POST":
        username = request.form.get("username")
        pwd = request.form.get("pwd")
        if not username or not pwd:
            return render_template("reg.html", mes="请输入用户名密码")
        #进行密码加密
        password = generate_password_hash(pwd)
        try:
            pmysql.post(f"insert user(username,pwd) values('{username}','{password}')")
            return redirect(url_for('login'))
        except Exception as e:
            return render_template("reg.html", mes="注册失败")


@app.route("/index")
def index():
    # 如果用户登录成功
    if session.get("username"):
        return "用户登陆成功！"
    # 未登录跳转到登陆页面
    else:
        # 未登录跳转到登陆页面
        return redirect('/')

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

封装好的pmysql.sql

import pymysql


class MysqlC:

    def __init__(self, host, user, password, database):
        self.con = None
        self.arguments = {
            "host": host,
            "user": user,
            "password": password,
            "database": database,
            "charset": 'utf8'
        }

    def post(self, sql):
        with self:
            self.cursor = self.con.cursor(pymysql.cursors.DictCursor)
            try:
                data = self.cursor.execute(sql)
                self.con.commit()

                return data
            except Exception as e:
                self.con.rollback()
                return e

    def get(self, sql, one=None):
        with self:
            self.cursor = self.con.cursor(pymysql.cursors.DictCursor)
            try:
                self.cursor.execute(sql)
                if one:
                    return self.cursor.fetchone()
                else:
                    return self.cursor.fetchall()
            except Exception as e:
                return e

    def __enter__(self):
        if self.con is None:
            try:
                self.con = pymysql.connect(**self.arguments)
            except Exception:
                return "数据库连接失败！"

            return self.con

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.cursor.close()
        self.con.close()
        self.con = None

settings.py

class DevelopmentConfig:
    MYSQL = {
        "host": "121.423",
        "user": "root",
        "password": "zbb521",
        "database": "zzb"
    }

目录结构

热爱技术,享受生活,感谢推荐!
【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/zdqc/p/15480341.html