天气预报数据通常由第三方API提供，每次调用会返回当前温度、湿度、风速、天气状况等信息。在个人项目或小型应用中，我们需要把这些数据保存下来用于历史分析、趋势展示或简单的缓存。如果为这个场景单独部署MySQL或PostgreSQL，会引入不必要的依赖。SQLite作为一个嵌入式关系型数据库，不需要独立服务进程，直接以文件形式存储数据，特别适合这种轻量级数据收集任务。下面结合一个实际项目，说明如何设计天气数据的存储结构并用Python完成读写操作。
为什么选择SQLite存储天气数据
SQLite最大的优势在于零配置和单文件存储。开发阶段不需要安装数据库服务器，数据文件可以直接拷贝备份，迁移到其他机器也很简单。对于天气数据这种写入频率不高（通常每小时或每几分钟调用一次API）但查询场景明确的数据集，SQLite的性能完全够用。即使数据量增长到几十万条，通过合理的索引设计，查询速度依然可以保持在毫秒级别。
另一个重要原因是事务支持。天气API返回的数据往往包含多个字段，写入时需要保证原子性，避免出现只写了一半的情况。SQLite支持ACID事务，可以在一个事务中插入多条关联记录。同时，它的并发模型对于单写入者多读取者的场景非常友好，而天气数据收集通常只有一个写入进程，多个读取请求可以同时进行，不会产生锁冲突。
当然，SQLite并不是万能的。如果项目需要多个服务同时写入，或者数据量达到数百GB，应该考虑使用客户端/服务器架构的数据库。但对于个人天气站、家庭自动化系统或原型验证项目，SQLite是性价比极高的选择。
设计天气预报数据的表结构
天气数据的最小记录单位通常是一次API响应，包含城市标识、观测时间、温度、湿度、气压、风速、天气描述等。为了支持多城市查询，应该把城市信息单独拆分成一张表，天气记录表中用外键关联城市ID。这样当城市名称发生变化或需要存储城市经纬度等静态信息时，不会在天气表中产生大量冗余。
天气记录表的核心字段包括：记录ID（自增主键）、城市ID、观测时间（建议存储为ISO 8601格式的文本或Unix时间戳整数）、温度（REAL类型，支持小数）、湿度（INTEGER，百分比）、气压（REAL，单位hPa）、风速（REAL）、风向（INTEGER，角度）、天气状况代码（TEXT）。需要注意观测时间字段一定要建立索引，因为大多数查询都是按时间范围过滤。如果还需要按城市和时间联合查询，可以建立城市ID加观测时间的复合索引。
建表语句可以这样写：
CREATE TABLE cities (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
country TEXT,
latitude REAL,
longitude REAL
);
CREATE TABLE weather_records (
id INTEGER PRIMARY KEY AUTOINCREMENT,
city_id INTEGER NOT NULL,
observed_at TEXT NOT NULL,
temperature REAL,
humidity INTEGER,
pressure REAL,
wind_speed REAL,
wind_direction INTEGER,
weather_code TEXT,
FOREIGN KEY (city_id) REFERENCES cities(id)
);
CREATE INDEX idx_weather_city_time ON weather_records(city_id, observed_at);
使用TEXT存储时间的好处是可读性好，但比较时需要字符串排序，ISO 8601格式（如2025-01-01T12:00:00）可以保证字典序和时间顺序一致。如果更关注存储空间和比较性能，可以使用INTEGER存Unix时间戳。这里根据项目需求选择即可。
用Python实现天气数据写入和查询
Python内置的sqlite3模块可以很方便地操作SQLite数据库。首先需要建立数据库连接并初始化表结构，然后封装一个函数来插入天气API返回的数据。下面是一个完整示例，模拟从天气API获取数据并写入数据库，同时包含查询某城市最近24小时温度的功能。
import sqlite3
import json
from datetime import datetime, timedelta
DB_PATH = "weather.db"
def init_db():
conn = sqlite3.connect(DB_PATH)
cur = conn.cursor()
cur.executescript("""
CREATE TABLE IF NOT EXISTS cities (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE,
country TEXT,
latitude REAL,
longitude REAL
);
CREATE TABLE IF NOT EXISTS weather_records (
id INTEGER PRIMARY KEY AUTOINCREMENT,
city_id INTEGER NOT NULL,
observed_at TEXT NOT NULL,
temperature REAL,
humidity INTEGER,
pressure REAL,
wind_speed REAL,
wind_direction INTEGER,
weather_code TEXT,
FOREIGN KEY (city_id) REFERENCES cities(id)
);
CREATE INDEX IF NOT EXISTS idx_weather_city_time
ON weather_records(city_id, observed_at);
""")
conn.commit()
conn.close()
def save_weather_data(city_name, api_response):
conn = sqlite3.connect(DB_PATH)
cur = conn.cursor()
# 确保城市存在
cur.execute("SELECT id FROM cities WHERE name = ?", (city_name,))
row = cur.fetchone()
if row:
city_id = row[0]
else:
cur.execute("INSERT INTO cities (name) VALUES (?)", (city_name,))
city_id = cur.lastrowid
# 解析API响应，示例结构
weather = api_response.get("current", {})
observed_at = datetime.utcnow().isoformat() + "Z"
cur.execute("""
INSERT INTO weather_records
(city_id, observed_at, temperature, humidity, pressure, wind_speed, wind_direction, weather_code)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (
city_id,
observed_at,
weather.get("temp_c"),
weather.get("humidity"),
weather.get("pressure_mb"),
weather.get("wind_kph"),
weather.get("wind_degree"),
weather.get("condition", {}).get("code")
))
conn.commit()
conn.close()
def query_recent_temperatures(city_name, hours=24):
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
cur = conn.cursor()
since = (datetime.utcnow() - timedelta(hours=hours)).isoformat() + "Z"
cur.execute("""
SELECT wr.observed_at, wr.temperature
FROM weather_records wr
JOIN cities c ON wr.city_id = c.id
WHERE c.name = ? AND wr.observed_at >= ?
ORDER BY wr.observed_at ASC
""", (city_name, since))
rows = cur.fetchall()
conn.close()
return [(r["observed_at"], r["temperature"]) for r in rows]
上面的代码中，save_weather_data函数每次调用都会建立新的数据库连接，这在频繁调用时会产生一些开销。实际项目中可以使用连接池或保持一个长连接，但为了演示清晰，每次都关闭连接。另外，代码里使用了参数化查询来防止SQL注入，这是必须遵守的安全实践。
查询函数返回最近指定小时数内的温度时间序列，可以用于绘制折线图。注意SQL语句中使用了>=，在Python字符串中可以正常书写，但在HTML代码块内已经转义为>=，保证显示正确。
性能优化与常见问题
当天气数据积累到数万条后，写入和查询性能可能会下降。有几个简单的优化手段：第一，使用批量插入。如果一次性获取多个城市的天气数据，可以把多个INSERT语句放在一个事务中提交，大幅减少磁盘I/O。第二，开启WAL模式（Write-Ahead Logging）。默认的SQLite日志模式是DELETE，每次写入都会锁定整个数据库文件。WAL模式允许读写并发，写入性能更好，特别适合读多写少的场景。执行PRAGMA journal_mode=WAL;即可开启。
第三，定期清理旧数据。如果只关心最近几个月的天气，可以设置定时任务删除过期记录，控制数据库文件大小。第四，为常用的过滤条件创建索引。除了城市和时间，如果经常按天气状况查询，可以为weather_code字段建立索引。但要注意索引不是越多越好，每个索引都会增加写入成本。
还有一个常见问题是数据库文件损坏。虽然SQLite非常稳定，但意外断电或程序崩溃仍可能导致数据损坏。备份策略很重要，可以使用SQLite的在线备份API，或者定期复制数据库文件。对于重要数据，建议开启PRAGMA synchronous=FULL;保证数据一致性，但会稍微降低写入速度，可以根据实际需求权衡。
最后，当数据量达到百万级别，查询可能变慢，这时可以考虑按月份分表存储天气记录，或者将历史数据归档到单独的数据库文件。SQLite单个数据库文件大小在大多数文件系统上支持到TB级别，但性能会随着数据量增长而下降，合理的分区策略能显著改善体验。