Python Flask框架—快速开发API

 2024/11/11 

需要手动安装
pip3 install flask

Flask 是一个用 Python 编写的轻量级 Web 框架，主要用于快速开发 Web 应用程序和 API，它提供了一些基本的功能，可以快速开始，而不需要从头写很多代码。

核心功能
• 路由：Flask 允许你定义 URL 路径（例如 http://你的网址/home），并将其映射到处
理请求的函数上。
• 请求处理：Flask 能接收用户发送的数据（比如表单输入）并做出相应的处理。
• 返回响应：Flask 可以返回网页内容、JSON 数据或其他类型的信息给用户。

Flask：Hello World

#coding=utf-8
from flask import Flask

app = Flask(__name__)
# 根路由，访问时调用home函数
@app.route('/')
def home():
    return 'Hello, world'

if __name__ == '__main__':
    app.run(debug=True)
    
# WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.
#  * Running on http://127.0.0.1:5000
# Press CTRL+C to quit
#  * Restarting with stat
#  * Debugger is active!
#  * Debugger PIN: 432-779-631

自定义路由

和nginx类似，定义了路由（路径）装饰器装饰器后，访问该目录就会返回

定义的函数返回的内容

@app.route('about')
def about():
	return "It's an About page"

# 定义/login，只接受POST请求
@app.route('/login',methods=['POST'])
def login():
	# 通过request获取提交表单中的username信息
	username = request.form['username']
	return f"Welcome,{username}"

案例：服务器状态监控API

自动化监控：定期检查资源使用情况
实时告警：资源超过阈值
数据可视化：生成健康报告

#coding=UTF-8
from flask import Flask,jsonify
import psutil

# 创建flask应用
app=Flask(__name__)

# 定义健康检查api(定义路由
@app.route('/api/health',methods=['GET'])
def health_check():
    cpu_usage=psutil.cpu_percent(interval=1)

    mem=psutil.virtual_memory()
    mem_usage=mem.percent

    disk=psutil.disk_usage('/')
    disk_usage=disk.percent

    status='healthy'
    if cpu_usage > 80 or mem_usage >80 or disk_usage > 80:
        status='unhealthy'
    return jsonify({
        'status':status,
        'cpu_usage': cpu_usage,
        'mem_usage': mem_usage,
        'disk_usage': disk_usage
    })

if __name__ == '__main__':
    app.run(host='0.0.0.0',port=5000)

API获取数据

#coding=utf-8
import requests
import logging
from logging import Formatter

# 配置默认日志记录器
logging.basicConfig(level=logging.INFO,
                    filename='server_health.log',
                    encoding='utf-8',
                    format="%(asctime)s-%(levelname)s-%(name)s-%(message)s",
                    datefmt='%Y-%m-%d-%H:%M:%S')
# 定义时间格式
formatter=Formatter('%(asctime)s-%(levelname)s-%(name)s-%(message)s')
for handler in logging.getLogger().handlers:
    handler.setFormatter(formatter)

# 定义api接口和报警阈值
url='http://127.0.0.1:5000/api/health'
threshold_cpu=80
threshold_mem=80
threshold_disk=80

def check_server_health():
    response=requests.get(url)
    if response.status_code == 200:
        data=response.json()
        cpu_usage=data.get('cpu_usage')
        mem_usage=data.get('mem_usage')
        disk_usage=data.get('disk_usage')
        status=data.get('status')
        # 检查是否超越阈值
        if cpu_usage > threshold_cpu:
            logging.warning(f'告警:CPU使用率达到{cpu_usage}%,超过阈值{threshold_cpu}')
        if mem_usage > threshold_mem:
            logging.warning(f'告警:MEM使用率达到{mem_usage}%,超过阈值{threshold_mem}')
        if disk_usage > threshold_disk:
            logging.warning(f'告警:硬盘使用率达到{disk_usage}%,超过阈值{threshold_disk}')
        # 输出服务器状态
        logging.info(f"服务器状态:{status}")
        logging.info(f"内存使用率:{mem_usage}%")
        logging.info(f"CPU使用率:{cpu_usage}%")
        logging.info(f"硬盘使用率：{disk_usage}%")
    else:
        print("无法获取服务器状态")

if __name__=='__main__':
    check_server_health()

案例：服务状态检查与重启API

#coding=utf-8
from flask import Flask,jsonify
import subprocess

app=Flask(__name__)

@app.route('/api/service/status/<service>',methods=['GET'])
def service_status_check(service):
    sts=subprocess.run(["systemctl","is-active",service],capture_output=True,text=True)
    output=sts.stdout.strip()
    if output == 'active':
        status = 'running'
    elif output == 'inactive':
        status = 'stopped'
    else:
        status = 'unknown'
    return jsonify({
        'service': service,
        'status': status
    })

@app.route('/api/service/restart/<service>',methods=['POST'])
def service_restart(service):
    result=subprocess.run(["systemctl","restart",service],capture_output=True,text=True)
    if result.returncode == 0:
        return jsonify({'message': f"{service} restarted successfully"})
    else:
        return jsonify({'message': f"Failed to restarted {service}"})

if __name__ == '__main__':
    app.run(debug=True,host='0.0.0.0',port=5000)