如何用Python+Flask-SQLAlchemy避免PostgreSQL插入相近经纬度记录

阿华AIGC实验室

2026-5-19

嘿，这个需求在地理数据处理场景里挺常见的！我来给你一步步拆解怎么用Flask-SQLAlchemy结合PostgreSQL的地理空间功能实现：

核心思路

我们要同时满足两个去重规则：

避免重复的user.key主键（这个你已经有基础逻辑了）
插入前检查数据库中是否存在与新经纬度距离小于设定阈值的记录，借助PostgreSQL的空间函数高效计算球面距离

步骤1：确保PostgreSQL支持空间函数

首先得确认你的PostgreSQL实例安装了postgis扩展（大部分云服务商的PostgreSQL默认会自带，如果没有的话，手动执行这条SQL开启：CREATE EXTENSION postgis;）。这个扩展提供了我们需要的专业地理计算函数，比自己写Haversine公式靠谱多了。

步骤2：定义Flask-SQLAlchemy模型

保留user_key作为主键，同时存储纬度和经度字段：

from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import func

db = SQLAlchemy()

class Location(db.Model):
    __tablename__ = "locations"
    user_key = db.Column(db.String, primary_key=True, nullable=False)
    latitude = db.Column(db.Float, nullable=False)
    longitude = db.Column(db.Float, nullable=False)

步骤3：实现“相近经纬度检查”逻辑

用PostGIS的ST_DistanceSphere函数计算两个经纬度点之间的球面距离（单位是米），这个函数能直接利用空间索引优化性能。写一个辅助函数来做检查：

def is_location_near_existing(new_lat: float, new_lon: float, threshold_meters: int = 100) -> bool:
    """
    检查新经纬度是否与数据库中已有记录距离小于阈值（默认100米）
    返回True表示存在相近地点，False表示可以插入
    """
    # 重点提醒：PostGIS的ST_MakePoint是先经度、后纬度，别搞反顺序！
    existing_location = Location.query.filter(
        func.ST_DistanceSphere(
            func.ST_SetSRID(func.ST_MakePoint(Location.longitude, Location.latitude), 4326),
            func.ST_SetSRID(func.ST_MakePoint(new_lon, new_lat), 4326)
        ) < threshold_meters
    ).first()
    return existing_location is not None

步骤4：封装完整的插入逻辑

把主键检查和经纬度检查结合起来，放在同一个事务里避免并发场景下的重复插入问题：

def add_new_location(user_key: str, lat: float, lon: float) -> tuple[bool, str]:
    # 先检查主键是否已存在
    if Location.query.get(user_key):
        return False, "该user_key已存在对应的记录"
    
    # 检查经纬度是否与已有地点过近
    if is_location_near_existing(lat, lon):
        return False, "该地点与数据库中已有记录距离过近"
    
    # 所有检查通过，插入新记录
    try:
        new_loc = Location(user_key=user_key, latitude=lat, longitude=lon)
        db.session.add(new_loc)
        db.session.commit()
        return True, "记录插入成功"
    except Exception as e:
        db.session.rollback()
        return False, f"插入失败：{str(e)}"

步骤5：优化查询性能（可选但强烈推荐）

如果你的表数据量较大，每次全表扫描计算距离会很慢。给经纬度创建空间索引能大幅提升查询速度：

你可以直接在PostgreSQL中执行这条SQL：

CREATE INDEX idx_locations_geopoint ON locations 
USING GIST (ST_SetSRID(ST_MakePoint(longitude, latitude), 4326));

或者在Flask-SQLAlchemy的迁移脚本（比如用Alembic）里添加索引定义：

from alembic import op
import sqlalchemy as sa
from sqlalchemy.sql import func

def upgrade():
    op.create_index(
        "idx_locations_geopoint",
        "locations",
        [func.ST_SetSRID(func.ST_MakePoint(sa.column("longitude"), sa.column("latitude")), 4326)],
        postgresql_using="gist"
    )

def downgrade():
    op.drop_index("idx_locations_geopoint", table_name="locations")

注意事项