R / Richie全部文章 ↑

Python · 4 分钟阅读

4. Flask 案例:网页查询 IP 归属地

目录


本章用一个最小但完整的小项目把前面三章串起来:

  • 启动 Flask 应用
  • 渲染 Jinja2 模板
  • 处理 GET / POST 表单
  • 调用外部 API(IP 归属地查询)
  • 区分 HTML 页面与 JSON 接口

最终效果:在首页输入 IP,提交后展示该 IP 所在的国家/地区/运营商;
同一份后端逻辑还提供 /api/ip/<ip> JSON 接口。


1. 准备

# 创建并激活虚拟环境
python3 -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate

# 安装依赖
pip install -U flask requests

本教程使用 ip-api.com 的免费接口(无需 Key,每分钟 45 次
限制,仅供学习)。生产环境请购买稳定服务或自建 GeoIP 数据库。


2. 项目结构

ip-lookup/
├── app.py                  # 应用入口
├── services/
│   └── ip_query.py         # IP 查询业务逻辑
├── templates/
│   ├── base.html           # 基础布局
│   └── index.html          # 查询页
└── requirements.txt

3. 业务逻辑:IP 查询

把“访问第三方 API”这件事封装到一个独立函数,方便复用与测试:

# services/ip_query.py
"""IP 归属地查询封装。"""
from __future__ import annotations

from dataclasses import dataclass
from typing import Any

import requests

API_URL = "http://ip-api.com/json/{ip}?lang=zh-CN"
TIMEOUT = 5  # 秒


@dataclass(frozen=True)
class IpInfo:
    ip: str
    country: str
    region: str
    city: str
    isp: str

    @classmethod
    def from_api(cls, ip: str, data: dict[str, Any]) -> "IpInfo":
        return cls(
            ip=ip,
            country=data.get("country", ""),
            region=data.get("regionName", ""),
            city=data.get("city", ""),
            isp=data.get("isp", ""),
        )


def query_ip(ip: str) -> IpInfo:
    """根据 IP 查询归属地。

    Raises:
        ValueError: IP 为空。
        RuntimeError: 接口返回失败。
    """
    ip = (ip or "").strip()
    if not ip:
        raise ValueError("IP 不能为空")

    try:
        resp = requests.get(API_URL.format(ip=ip), timeout=TIMEOUT)
        resp.raise_for_status()
    except requests.RequestException as exc:
        raise RuntimeError(f"调用 IP 接口失败:{exc}") from exc

    data = resp.json()
    if data.get("status") != "success":
        raise RuntimeError(f"未查询到结果:{data.get('message', 'unknown')}")

    return IpInfo.from_api(ip, data)

要点:

  • 使用 dataclass 描述数据,类型清晰。
  • 显式设置 timeout,避免请求挂起拖垮服务。
  • 把不同错误(参数错、第三方失败)归一为自定义异常,方便视图层统一处理。

4. Flask 应用

# app.py
from __future__ import annotations

from flask import Flask, jsonify, render_template, request

from services.ip_query import query_ip, ValueError as _ValueError, RuntimeError as _RuntimeError

app = Flask(__name__)


def _do_query(ip: str) -> dict:
    """执行查询并统一异常处理,返回给模板/JSON 共用的字典。"""
    try:
        info = query_ip(ip)
    except _ValueError as exc:
        return {"ok": False, "error": str(exc)}
    except _RuntimeError as exc:
        return {"ok": False, "error": str(exc)}
    return {
        "ok": True,
        "ip": info.ip,
        "country": info.country,
        "region": info.region,
        "city": info.city,
        "isp": info.isp,
    }


@app.get("/")
def index():
    """GET 渲染查询页,POST 提交查询。"""
    result = None
    if request.method == "POST":
        result = _do_query(request.form.get("ip", ""))
    return render_template("index.html", result=result)


@app.get("/api/ip/<ip>")
def api_ip(ip: str):
    """JSON 接口:GET /api/ip/8.8.8.8"""
    result = _do_query(ip)
    return jsonify(result), (200 if result["ok"] else 400)


if __name__ == "__main__":
    app.run(debug=True, port=8080)

想让同一路径同时支持 GET(显示表单)和 POST(提交查询)?把两个方法都
装饰到同一个视图里:

@app.route("/", methods=["GET", "POST"])
def index(): ...

或者使用 Flask 3.x 的语法糖:

@app.get("/")
@app.post("/")
def index(): ...

5. 模板

5.1 基模板 templates/base.html

<!doctype html>
<html lang="zh-CN">
<head>
  <meta charset="utf-8">
  <title>{% block title %}IP 查询{% endblock %}</title>
  <style>
    body { font-family: -apple-system, "Segoe UI", sans-serif;
           max-width: 640px; margin: 40px auto; padding: 0 16px; }
    input[type=text] { padding: 6px 10px; font-size: 14px; width: 240px; }
    button { padding: 6px 14px; font-size: 14px; cursor: pointer; }
    .ok    { color: #1a7f37; }
    .error { color: #cf222e; }
    table  { border-collapse: collapse; margin-top: 12px; }
    th, td { border: 1px solid #ddd; padding: 6px 12px; text-align: left; }
  </style>
</head>
<body>
  <header><h1><a href="{{ url_for('index') }}">IP 归属地查询</a></h1></header>
  <main>{% block content %}{% endblock %}</main>
</body>
</html>

5.2 查询页 templates/index.html

{% extends "base.html" %}

{% block content %}
  <form method="post" action="{{ url_for('index') }}">
    <input type="text" name="ip" placeholder="例如 8.8.8.8" required>
    <button type="submit">查询</button>
  </form>

  {% if result %}
    {% if result.ok %}
      <p class="ok">查询成功 ✅</p>
      <table>
        <tr><th>IP</th><td>{{ result.ip }}</td></tr>
        <tr><th>国家</th><td>{{ result.country }}</td></tr>
        <tr><th>地区</th><td>{{ result.region }}</td></tr>
        <tr><th>城市</th><td>{{ result.city }}</td></tr>
        <tr><th>运营商</th><td>{{ result.isp }}</td></tr>
      </table>
    {% else %}
      <p class="error">查询失败:{{ result.error }}</p>
    {% endif %}
  {% endif %}
{% endblock %}

6. 运行

flask --app app run --debug --port 8080
# 或:python app.py   (app.py 里也写了 app.run)

打开 http://127.0.0.1:8080,输入 8.8.8.8 查询即可。

直接测试 JSON 接口:

curl http://127.0.0.1:8080/api/ip/8.8.8.8
# {"city":"Mountain View","country":"美国","ip":"8.8.8.8","isp":"Google LLC",
#  "ok":true,"region":"加利福尼亚州"}

7. 进阶:更严谨的写法

7.1 用应用工厂 + 蓝图拆分

项目变大后,把上面的代码拆成:

ip-lookup/
├── app/
│   ├── __init__.py          # create_app()
│   ├── blueprints/
│   │   └── ip.py            # 路由
│   ├── services/
│   │   └── ip_query.py
│   └── templates/
├── config.py
└── wsgi.py
# app/blueprints/ip.py
from flask import Blueprint, jsonify, render_template, request

from ..services.ip_query import query_ip

bp = Blueprint("ip", __name__)


@bp.route("/", methods=["GET", "POST"])
def index():
    result = None
    if request.method == "POST":
        result = _do_query(request.form.get("ip", ""))
    return render_template("index.html", result=result)


@bp.get("/api/ip/<ip>")
def api_ip(ip: str):
    result = _do_query(ip)
    return jsonify(result), (200 if result["ok"] else 400)
# app/__init__.py
from flask import Flask

def create_app() -> Flask:
    app = Flask(__name__)
    from .blueprints.ip import bp as ip_bp
    app.register_blueprint(ip_bp)
    return app
# wsgi.py
from app import create_app
app = create_app()
flask --app wsgi run --debug --port 8080

7.2 缓存

第三方接口有频率限制,可以用 cachetools 简单加一层 TTL 缓存:

from cachetools import TTLCache, cached

_cache: TTLCache = TTLCache(maxsize=1024, ttl=600)  # 10 分钟

@cached(_cache)
def query_ip_cached(ip: str) -> IpInfo:
    return query_ip(ip)

7.3 异步查询(性能提升)

第三方是 IO 密集型调用,使用 httpx.AsyncClient 配合 Quart / Flask 3 的
async 视图可以获得明显提升;或在 gunicorn 下用 gevent worker。


8. 常见问题

  • 接口返回 status: fail:可能是 IP 格式错误或触发了频率限制。
  • 请求超时:检查网络,或增大 TIMEOUT。
  • 页面里看到一堆 HTML 实体(&lt;):模板里多套了一层 {{ }},去掉一层即可。
  • 部署到服务器:不要再用 flask run,改用 gunicorn:
    pip install gunicorn
    gunicorn -w 2 -b 0.0.0.0:8080 wsgi:app
    
  • 希望用户访问 / 时自动展示访问者自身的 IP:在后端用
    request.headers.get('X-Forwarded-For', request.remote_addr) 取得 IP,
    把它作为表单的默认 value。

小结

能力 用到的 Flask 特性
启动 flask --app app run --debug
渲染 render_template + extends + 条件/循环
表单 request.form + GET/POST 共用视图
JSON @app.get("/api/...") + jsonify
错误处理 自定义异常 + 视图层统一收敛
进阶 应用工厂 + 蓝图 + 缓存

至此,你已经掌握了 Flask 入门所需的全部基础。接下来可以:

  • 用 Flask-SQLAlchemy 把查询历史持久化到数据库
  • 用 Flask-Login + GitHub OAuth 加上用户系统(见 OAuth 实战)
  • 用 Flask-Migrate 管理数据库 schema 变更