华硕固件做网站6,建筑工程网上培训平台,西安百度推广代理商,高端室内设计工作室说明#xff1a;
应用Pydantic Model 验证/出入 数据#xff0c; SqlAlchemy Model数据实体#xff0c;Fastapi提供API机制支持。数据表的坐标字段采用Mysql的GEOMETRY类型目前还没成功使用Pydantic的Coordinate类型#xff0c;待后续改良
要点#xff1a;
输出的结果是…说明
应用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(..., titleNone)# coordinates: Coordinate | None Field(..., title地理坐标)coordinates: str | None Field(..., title地理坐标)class DeviceLocationSimpleIn(DeviceLocation):passclass DeviceLocationSimpleOut(DeviceLocation):model_config ConfigDict(from_attributesTrue)id: int Field(..., title编号)create_datetime: DatetimeStr Field(..., title创建时间)update_datetime: DatetimeStr Field(..., title更新时间)validator(create_datetime, update_datetime, preTrue)def convert_utc_to_local(cls, value):return utils.convert_utc_to_local(value)field_validator(coordinates, modebefore)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_keyTrue)coordinates: Mapped[WKBElement] mapped_column(Geometry(geometry_typePOINT, spatial_indexTrue), nullableFalse, comment地理坐标)iot_device_id mapped_column(BigInteger, server_defaulttext(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_countTrue)return SuccessResponse(datas, countcount)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(datadata))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(idsids.ids, v_softFalse)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_schemaschema))
接口Example 数据库记录 参考
- Working with Spatial Data using FastAPI and GeoAlchemy
- sql server中对geometry类型的常用操作、SRID、GeoTools工具简单使用geometry和geojson格式之间的转换_sqlserver sde geometry 转text-CSDN博客
