当前位置：首页 > news >正文

FastApi地理坐标数据存取实践

news 2025/7/25 7:29:19

说明：

应用Pydantic Model 验证/出入数据， SqlAlchemy Model数据实体，Fastapi提供API机制支持。
数据表的坐标字段采用Mysql的GEOMETRY类型
目前还没成功使用Pydantic的Coordinate类型，待后续改良

要点：

输出的结果是DeviceLocationSimpleOut模型，里面的验证/转换方法需通过，否则不能录入成功

设计：

建表sql

/*==============================================================*/
/* Table: ia_iot_device_location                                */
/*==============================================================*/
create table ia_iot_device_location
(id                   bigint not null auto_increment  comment '',iot_device_id        bigint default 0  comment '',label                varchar(255)  comment '',coordinates          GEOMETRY not null  comment '地理坐标',create_datetime      datetime not null default CURRENT_TIMESTAMP  comment '创建时间',update_datetime      datetime not null default CURRENT_TIMESTAMP  comment '更新时间',delete_datetime      datetime  comment '',is_delete            tinyint not null default 0  comment '',primary key (id)
);/*==============================================================*/
/* Index: Index_1                                               */
/*==============================================================*/
create index Index_1 on ia_iot_device_location
(iot_device_id
);/*==============================================================*/
/* Index: Index_2                                               */
/*==============================================================*/
create SPATIAL index Index_2 on ia_iot_device_location
(coordinates
);

Pydantic Model

(apps\vadmin\iot\schemas\DeviceLocation.py)

#!/usr/bin/python
# -*- coding: utf-8 -*-
# @version        : 1.0
# @Create Time    : 2024/05/22 20:45
# @File           : Device.py
# @IDE            : PyCharm
# @desc           : pydantic 模型，用于数据库序列化操作from pydantic import BaseModel, Field, ConfigDict, ValidationError, validator, field_validator, constr
from core.data_types import DatetimeStr
from datetime import datetime, timezone, timedelta
from apps.vadmin.iot.models.data_types import *
from apps.vadmin.iot.utils import utils
from application import settings
from pydantic_extra_types.coordinate import Coordinate
from geoalchemy2.shape import to_shapeclass DeviceLocation(BaseModel):label: str | None = Field(None, title="标签")iot_device_id: int | None = Field(..., title="None")# coordinates: Coordinate | None = Field(..., title="地理坐标")coordinates: str | None = Field(..., title="地理坐标")class DeviceLocationSimpleIn(DeviceLocation):passclass DeviceLocationSimpleOut(DeviceLocation):model_config = ConfigDict(from_attributes=True)id: int = Field(..., title="编号")create_datetime: DatetimeStr = Field(..., title="创建时间")update_datetime: DatetimeStr = Field(..., title="更新时间")@validator("create_datetime", "update_datetime", pre=True)def convert_utc_to_local(cls, value):return utils.convert_utc_to_local(value)@field_validator("coordinates", mode="before")def turn_coordinates_into_wkt(cls, value):return to_shape(value).wkt

SqlAlchemy Model

(apps\vadmin\iot\models\models.py)

from typing import List, Optional
from datetime import datetime
from sqlalchemy import BigInteger, Column, DateTime, ForeignKey, ForeignKeyConstraint, Index, Integer, String, Table, Text, text
from sqlalchemy.dialects.mysql import TINYINT
from sqlalchemy.orm import Mapped, declarative_base, mapped_column, relationship
from sqlalchemy.orm.base import Mapped
from geoalchemy2 import Geometry, WKBElement
from sqlalchemy.orm import relationship, Mapped, mapped_column
from db.db_base import BaseModel
from .data_types import DeviceType
import uuid
import secretsmetadata = BaseModel.metadataclass DeviceLocation(BaseModel):__tablename__ = 'ia_iot_device_location'__table_args__ = (Index('Index_1', 'iot_device_id'),Index('Index_2', 'coordinates'))id = mapped_column(BigInteger, primary_key=True)coordinates: Mapped[WKBElement] = mapped_column(Geometry(geometry_type='POINT', spatial_index=True), nullable=False, comment='地理坐标')iot_device_id = mapped_column(BigInteger, server_default=text("'0'"))label = mapped_column(String(255, 'utf8mb4_general_ci'))

FastApi 入口

(apps\vadmin\iot\views.py)

###########################################################
#    设备地理位置
###########################################################
@app.get("/device-location", summary="获取设备地理位置列表", tags=["设备地理位置"])
async def get_deviceLocation_list(p: params.DeviceLocation = Depends(), auth: Auth = Depends(AllUserAuth())):datas, count = await crud.DeviceLocationDal(auth.db).get_datas(**p.dict(), v_return_count=True)return SuccessResponse(datas, count=count)@app.post("/device-location", summary="创建设备地理位置", tags=["设备地理位置"])
async def create_deviceLocation(data: schemas.DeviceLocation, auth: Auth = Depends(AllUserAuth())):return SuccessResponse(await crud.DeviceLocationDal(auth.db).create_data(data=data))@app.delete("/device-location", summary="删除设备地理位置", description="硬删除", tags=["设备地理位置"])
async def delete_deviceLocation_list(ids: IdList = Depends(), auth: Auth = Depends(AllUserAuth())):await crud.DeviceLocationDal(auth.db).delete_datas(ids=ids.ids, v_soft=False)return SuccessResponse("删除成功")@app.put("/device-location/{data_id}", summary="更新设备地理位置", tags=["设备地理位置"])
async def put_deviceLocation(data_id: int, data: schemas.DeviceLocation, auth: Auth = Depends(AllUserAuth())):return SuccessResponse(await crud.DeviceLocationDal(auth.db).put_data(data_id, data))@app.get("/device-location/{data_id}", summary="获取设备地理位置信息", tags=["设备地理位置"])
async def get_deviceLocation(data_id: int, db: AsyncSession = Depends(db_getter)):schema = schemas.deviceLocationSimpleOutreturn SuccessResponse(await crud.DeviceLocationDal(db).get_data(data_id, v_schema=schema))