## Flask

是一个基于Python实现的Web开发微框架（基于MVC设计模式），官方文档：[https://flask.palletsprojects.com/en/3.0.x/](https://flask.palletsprojects.com/en/3.0.x/)

安装Flask：`pip install flask`

查看Flask版本：`pip show flask`

## 插件

### Flask-SQLAlchemy

Flask-SQLAlchemy是对SQLAlchemy的一个封装，使得在Flask中使用SQLAlchemy更加方便。因为 Flask-SQLAlchemy依赖SQLAlchemy，所以只要安装了Flask-SQLAlchemy，SQLAlchemy会自动安装。

安装命令：`pip install flask-sqlalchemy`

Flask想要操作数据库，必须要先安装Python操作MySQL驱动，这里选用pymysql作为驱动程序，安装命令：`pip install pymysql`

### Flask-Migrate

Flask-Migrate是处理Flask应用程序SQLAlchemy数据库迁移的扩展。

安装命令：`pip install flask-migrate`

数据库操作可通过Flask命令行界面进行，ORM模型映射成表三步：

1. flask db init：这步只需执行一次
2. flask db migrate：识别ORM模型的改变，生成迁移脚本
3. flask db upgrade：运行迁移脚本，同步到数据库中

### Flask-Mail

在Flask中可使用Flask-Mail发送邮箱验证码。

安装命令：`pip install flask-mail`

### Flask-WTF

Flask-WTF是简化WTForms操作的第三方库。WTForms表单的两个主要功能是验证用户提交数据的合法性以及渲染模板，而Flask-WTF还包括一些其他的功能：CSRF保护，文件上传等。

安装Flask-WTF默认也会安装WTForms，安装命令：`pip install flask-wtf`

## run参数

    from flask import Flask
    app = Flask(name)
    @app.route("/")
    def index():
    return "Home"
    if name == "main":
    app.run(debug=True)

- debug：是否开启调试模式，开启后（默认关闭），修改保存Python代码后会自动重启，方便调试
- port：指定服务器端口号，默认是5000
- host：主机，默认是127.0.0.1，指定为0.0.0.0代表所有ip

## 返回值

### 返回字符串

    from flask import Flask
    app = Flask(name)
    @app.route("/")
    def index():
    return "<h1>Home</h1>"  # 1. 返回 字符串，支持HTML标签
    if name == "main":
    app.run(debug=True)

### 返回JSON

    from flask import Flask
    import json
    app = Flask(name)
    @app.route("/")
    def index():
    data = {"name": "张三", "age": 18}
    response = app.response_class(
    response=json.dumps(data, ensure_ascii=False),  # ensure_ascii设置为False，避免出现中文乱码
    status=200,
    mimetype="application/json",
    )
    return response
    if name == "main":
    app.run(debug=True)

若返回数据不包含中文：

    from flask import Flask, jsonify
    app = Flask(name)
    @app.route("/")
    def index():
    return jsonify({"name": "jack", "age": 18})
    if name == "main":
    app.run(debug=True)

### 模板渲染

templates\index.html文件：

    <!DOCTYPE html>
    <html lang="zh-CN">
    <head>
    <meta charset="UTF-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
    <title>首页</title>
    <link rel="stylesheet" href="{{ url_for('static',filename='index.css')}}" />
    </head>
    <body>
    <h2>{{name}}</h2>
    </body>
    </html>

static\index.css文件：

    h2 {
    color: red;
    }

app.py文件：

    from flask import Flask, render_template
    app = Flask(name)
    @app.route("/")
    def index():
    return render_template("index.html", name="张三")
    if name == "main":
    app.run(debug=True)

### 自定义Response对象

    from flask import Flask, make_response
    app = Flask(name)
    @app.route("/")
    def index():
    html = "<h1>Home</h1>"
    res = make_response(html, 200)
    return res
    if name == "main":
    app.run(debug=True)

也可以使用Flask的响应对象类Response，它允许更灵活地创建和自定义响应对象，以满足特定需求。对于更复杂的自定义响应，使用Response类会更有优势。

### 重定向

    from flask import Flask, redirect
    app = Flask(name)
    @app.route("/")
    def index():
    return redirect("https://www.baidu.com/")
    # return redirect("/home/")  # 也可以这么用
    if name == "main":
    app.run(debug=True)

url_for反向解析，通过视图函数发过来找路由，url_for("蓝图名称.视图函数名")。

    from flask import Flask, redirect, Blueprint, url_for
    app = Flask(name)
    blue = Blueprint("user", name)
    @blue.route("/response/")
    def get_response():
    return "response ok"
    @blue.route("/")
    def index():
    ret = url_for("user.get_response")
    return redirect(ret)
    if name == "main":
    app.register_blueprint(blueprint=blue)
    app.run(debug=True)

url_for也能传参：url_for("user.get_response", name="张三", age=18)。

## 请求Request

服务器在接收到客户端的请求后，Flask框架会自动创建Request对象，Request对象不可修改。属性有：

- url 完整请求地址
- base_url 去掉GET参数的URL
- host_url 只有主机和端口号的URL
- path 路由中的路径
- method 请求方法
- remote_addr 请求的客户端地址
- args GET请求参数
- form POST请求参数
- files 文件上传
- headers 请求头
- cookies 请求中的cookie

类字典结构对象，获取指定key值：dict.get('key')和dict.getlist('key')。

## 路由参数

### string

    @app.route("/string/<name>/")
    def get_name(name):
    return name

### int

    @app.route("/int/<int:id>/")
    def get_int(id):
    return str(id)

### float

    @app.route("/float/<float:pi>/")
    def get_float(pi):
    return str(pi)

### path

支持/的字符串，例如/path/3/14/，返回3/14。

    @app.route("/path/<path:name>/")
    def get_path(name):
    return str(name)

### uuid

    @app.route("/get-uuid/")
    def get_uuid():
    import uuid
    return str(uuid.uuid4())
    @app.route("/uuid/<uuid:id>/")
    def uuid(id):
    return str(id)

### any

从列出的项目中选择一个。

    @app.route("/any/<any(apple,orange, banana):fruit>/")
    def get_any(fruit):
    return str(fruit)

## 项目拆分

    Project
    |   app.py
    |
    ---App
    |   models.py
    |   views.py
    |   init.py
    |
    +---static
    |       index.css
    |
    +---templates
    |       index.html

## cookie

App\templates\index.html文件：

    <!DOCTYPE html>
    <html lang="zh-CN">
    <head>
    <meta charset="UTF-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
    <title>首页</title>
    <link rel="stylesheet" href="{{ url_for('static',filename='index.css')}}" />
    </head>
    <body>
    <h2>首页</h2>
    <hr />
    {% if username %}
    <p>当前登录用户：{{username}} <a href="/logout/">注销</a></p>
    {% else %}
    <a href="/login/">登录</a>
    {% endif %}
    </body>
    </html>

App\templates\login.html文件：

    <!DOCTYPE html>
    <html lang="zh-CN">
    <head>
    <meta charset="UTF-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
    <title>Document</title>
    </head>
    <body>
    <h2>登录</h2>
    <hr />
    &lt;form action=&quot;&quot; method=&quot;post&quot;&gt;
      &lt;p&gt;用户名：&lt;input type=&quot;text&quot; name=&quot;username&quot; /&gt;&lt;/p&gt;
      &lt;p&gt;密码：&lt;input type=&quot;password&quot; name=&quot;password&quot; /&gt;&lt;/p&gt;
      &lt;p&gt;&lt;button&gt;提交&lt;/button&gt;&lt;/p&gt;
    &lt;/form&gt;
    
    </body>
    </html>

App\views.py文件：

    import datetime
    from flask import Blueprint, request, render_template, redirect
    from .models import *
    创建蓝图
    blue = Blueprint("user", name)
    @blue.route("/")
    def index():
    # 4. 获取cookie
    username = request.cookies.get("user")
    return render_template("index.html", username=username)
    @blue.route("/login/", methods=["GET", "POST"])
    def login():
    if request.method == "GET":
    return render_template("login.html")
    elif request.method == "POST":
    # 1. 获取前端提交过来的数据
    username = request.form.get("username")
    password = request.form.get("password")
        # 2. 模拟登录：用户名和密码验证
        if username == &quot;admin&quot; and password == &quot;123&quot;:
            response = redirect(&quot;/&quot;)
    
            # 3. 设置cookie
            response.set_cookie(&quot;user&quot;, username)  # 浏览器关闭失效
            # response.set_cookie(&quot;user&quot;, username, max_age=3600 * 24)  # 过期时间 秒
            # response.set_cookie(
            #     &quot;user&quot;, username, expires=datetime.datetime(2023, 12, 31)
            # )
    
            return response
        else:
            return &quot;用户名或密码错误！&quot;
    
    @blue.route("/logout/")
    def logout():
    response = redirect("/")
    # 5. 删除cookie
    response.delete_cookie("user")
    return response</code></pre><h2>session</h2><p>App\views.py文件：</p><pre><code class="lang-python">import datetime
    
    from flask import Blueprint, request, render_template, redirect, session
    from .models import *
    创建蓝图
    blue = Blueprint("user", name)
    @blue.route("/")
    def index():
    # 4. 获取session
    username = session.get("user")
    return render_template("index.html", username=username)
    @blue.route("/login/", methods=["GET", "POST"])
    def login():
    if request.method == "GET":
    return render_template("login.html")
    elif request.method == "POST":
    # 1. 获取前端提交过来的数据
    username = request.form.get("username")
    password = request.form.get("password")
        # 2. 模拟登录：用户名和密码验证
        if username == &quot;admin&quot; and password == &quot;123&quot;:
            response = redirect(&quot;/&quot;)
    
            # 3. 设置session
            session[&quot;user&quot;] = username
            session.permanent = True
    
            return response
    
        else:
            return &quot;用户名或密码错误！&quot;
    
    @blue.route("/logout/")
    def logout():
    response = redirect("/")
    # 5. 删除session
    session.pop("user")
    # session.clear()  # 慎用，会删除所有session
    return response</code></pre><p>还需要配置session，App\__init__.py文件：</p><pre><code class="lang-python">import datetime
    
    from flask import Flask
    from .views import blue
    def create_app():
    app = Flask(name)
    # 注册蓝图
    app.register_blueprint(blueprint=blue)
    # session配置
    # print(app.config)  # Flask配置信息
    app.config[&quot;SECRET_KEY&quot;] = &quot;abc123&quot;
    app.config[&quot;PERMANENT_SESSION_LIFETIME&quot;] = datetime.timedelta(days=3)
    
    return app</code></pre>
    
    


Python轻量级Web应用框架Flask学习笔记

并行：多线程同时操作多个资源类（两个或多个事件在同一时刻发生）

并发：多线程交替操作同一资源类（宏观上的并行，微观上的串行）

## concurrent.futures

是Python 3.2引入的新模块，它是在threading和multiprocessing之上的一个通用抽象层，提供了ThreadPoolExecutor和ProcessPoolExecutor两个类，以便使用线程池/进程池并发地执行任务。

## map与submit方法

map方法很简单，map的结果和入参顺序是对应的。

future模式更强大，用as_completed顺序是不定的。

import httpx from parsel import Selector from concurrent.futures import ThreadPoolExecutor, as_completed urls = [f"https://news.cnblogs.com/n/page/{page}" for page in range(1, 10 + 1)] def craw(url): r = httpx.get(url) return r.text def parse(html): selector = Selector(html) title = selector.css(".content h2.news_entry a::text")[0].get() return title with ThreadPoolExecutor() as pool: htmls = pool.map(craw, urls) htmls = list(zip(urls, htmls)) for url, html in htmls: print(url, len(html)) print("craw over") with ThreadPoolExecutor() as pool: futures = {} for url, html in htmls: future = pool.submit(parse, html) futures[future] = url for future in as_completed(futures): url = futures[future] print(url, future.result()) print("parse over")

## Web服务多线程加速

import flask import json import time from concurrent.futures import ThreadPoolExecutor

app = flask.Flask(**name**)
pool = ThreadPoolExecutor()

def read_file():
time.sleep(0.1)
return "file rusult"

def read_db():
time.sleep(0.2)
return "db rusult"

def read_api():
time.sleep(0.1)
return "api rusult"

@app.route("/")
def index():
result_file = pool.submit(read_file)
result_db = pool.submit(read_db)
result_api = pool.submit(read_api)
return json.dumps(
{
"result_file": result_file.result(),
"result_db": result_db.result(),
"result_api": result_api.result(),
}
)

if **name** == "**main**":
app.run()

## Web服务多进程加速

from flask import Flask import json import math from concurrent.futures import ProcessPoolExecutor

app = Flask(**name**)

def is_prime(n):
if n < 2:
return False
if n == 2:
return True
if n % 2 == 0:
return False

    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n, 2):
        if n % i == 0:
            return False
    
    return True
    

@app.route("/is-prime/<numbers>")
def api_is_prime(numbers: str):
number_list = [int(x) for x in numbers.split(",")]
results = process_pool.map(is_prime, number_list)
return json.dumps(dict(zip(number_list, results)))

if **name** == "**main**":
process_pool = ProcessPoolExecutor()
app.run()


Python并发编程标准库concurrent.futures模块

## 读者阁

是一个读者杂志免费在线阅读和下载网站，提供《读者》PDF版电子杂志免费在线阅读、下载，而且网站还在不断更新关于读者杂志的信息，直接打开网站即可，无需注册登录，用爱发电，读者杂志通过AList网盘提供分享。

## 读者

《读者》杂志创刊于1981年3月，是读者出版集团主办的一份综合类文摘杂志，月发行量达1000万册，居亚洲第一，世界综合类期刊第三位。

创刊30余年，《读者》始终坚持“博采中外、荟萃精华、启迪思想、开阔眼界”的办刊宗旨和“高雅、清新、隽永”的办刊风格，发掘人性中的真善美，体现人文关怀，在海内外亿万读者中产生了深远的影响，具有很高知名度和美誉度，被誉为“中国人的心灵读本”、“中国期刊第一品牌”。

## 网站地址

读者阁：[https://duzhege.cn/](https://duzhege.cn/)


读者阁-在线阅读故事会、读者等彩色PDF杂志更新快

## RPC协议

Remote Procedure Call Protocol，远程过程调用协议，IPC是进程间通信，RPC是远程进程通信。

## 服务端和客户端编码实现

服务端：

    package main
    import (
    "fmt"
    "net"
    "net/rpc"
    )
    // 定义类对象
    type World struct {
    }
    // 绑定类方法
    func (this *World) SayHello(name string, resp *string) error {
    *resp = name + " 你好！"
    return nil
    }
    func main() {
    // 1. 注册RPC服务，绑定对象方法
    err := rpc.RegisterName("hello", new(World))
    if err != nil {
    fmt.Println("注册RPC服务失败:", err)
    return
    }
    // 2. 创建RPC服务器，监听端口
    listen, err := net.Listen("tcp", "127.0.0.1:8080")
    if err != nil {
    fmt.Println("net.Listen err:", err)
    }
    defer listen.Close()
    fmt.Println("开始监听...")
    // 3. 建立连接
    conn, err := listen.Accept()
    if err != nil {
    fmt.Println("Accept err:", err)
    }
    defer conn.Close()
    fmt.Println("连接成功...")
    // 4. 绑定服务
    rpc.ServeConn(conn)
    }

客户端：

    package main
    import (
    "fmt"
    "net/rpc"
    )
    func main() {
    // 1. 用RPC连接服务器
    conn, err := rpc.Dial("tcp", "127.0.0.1:8080")
    if err != nil {
    fmt.Println("Dial err:", err)
    return
    }
    defer conn.Close()
    // 2. 调用远程函数
    var reply string
    err = conn.Call("hello.SayHello", "张三", &reply)
    if err != nil {
    fmt.Println("Call err:", err)
    return
    }
    fmt.Println("Reply:", reply)
    }

## JSON版RPC

修改服务端代码：`jsonrpc.ServeConn(conn)`

修改客户端代码：`conn, err := jsonrpc.Dial("tcp", "127.0.0.1:8080")`


Go微服务

## IP地址

在IP地址中，全0代表网络地址，全1代表广播地址，这两个特殊用途IP地址，不可作为主机地址使用。

为了科学使用IP地址，减少浪费，将IP地址根据使用数量级划分为：

- A类：1.0.0.1~126.255.255.254，有126个网络，即共有2^7-2个网络号，每个网络最多有2^24-2个主机地址。127开头的地址可以说是A类的保留地址，用于本地软件环回测试
- B类：128.0.0.1~191.255.255.254
- C类：192.0.0.1~223.255.255.254
- D类：224.0.0.0~239.255.255.255
- E类：240.0.0.1~255.255.255.254，科研用
- 常见为ABC类，DE类不常见

       0 1     7  8     15  16     23  24      31
    A: 0 网络地址 主机地址--------------------(24)
    B: 1 0 网络地址---(14)  主机地址----------(16)
    C: 1 1 0 网络地址------------(21)  主机地址(8)
    D: 1 1 1 0 组播、路由器修改---------------(28)
    E: 1 1 1 1 试验用地址---------------------(28)

私有IP地址范围：

- A类：10.0.0.0~10.255.255.255，子网掩码255.0.0.0
- B类：172.16.0.0~172.31.255.255，子网掩码255.255.0.0
- C类：192.168.0.0~192.168.255.255，子网掩码255.255.255.0

同一个网段的IP地址，物理线路接通就可以直接相互通信，不同网段的i地址，即便是物理线路接通，也不可以直接通信，需要路由器才能相互通信（路由器能够帮我们转发给对应网段的主机）。

### 子网掩码

单纯的IP地址（分为网络和主机部分）只是标识了IP地址的种类，是无法辨识一个IP所处的子网的。子网掩码用来区分网络号和主机号，不能单独存在，必须结合IP地址一起使用。

子网掩码在形式上等同于IP地址，其网络部分全为1（二进制），主机部分全为0（二进制），例如子网掩码：255.255.255.0（另一种写法：IP地址/24，即24个1）。

知道子网掩码，就能判断任意两个IP地址是否处在同一个子网络了。方法是将两个IP地址与子网掩码分别进行二进制与运算，然后比较结果是否相同，如果是，表明它们在同一个子网络中，否则就不是。


网络基础知识

这里是 aboss.top 的自留地。写了十几年野路子代码，攒了些摸爬滚打的经验，做成系列课程慢慢讲；也留了一小块地方，记记代码之外的胡思乱想。