admin/railway_cloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

时间格式，获取沉降数据接口limit

Browse Source
This commit is contained in:
lhx
2025-10-14 09:14:59 +08:00
parent 667b793aa1
commit ef0d92f86f
6 changed files with 38 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
app/services/settlement_data.py
View File

@@ -28,30 +28,40 @@ class SettlementDataService(BaseService[SettlementData]):
point_id: Optional[str] = None,
nyid: Optional[str] = None,
sjName: Optional[str] = None,
workinfoname: Optional[str] = None) -> List[SettlementData]:
"""根据多个条件搜索沉降数据"""
conditions = {}
if id is not None:
conditions["id"] = id
if point_id is not None:
conditions["point_id"] = point_id
if nyid is not None:
conditions["NYID"] = nyid
if sjName is not None:
conditions["sjName"] = sjName
if workinfoname is not None:
conditions["workinfoname"] = workinfoname
workinfoname: Optional[str] = None,
limit: Optional[int] = None) -> List[SettlementData]:
"""根据多个条件搜索沉降数据，按上传时间倒序排序"""
query = db.query(SettlementData)
return self.search_by_conditions(db, conditions)
if id is not None:
query = query.filter(SettlementData.id == id)
if point_id is not None:
query = query.filter(SettlementData.point_id == point_id)
if nyid is not None:
query = query.filter(SettlementData.NYID == nyid)
if sjName is not None:
query = query.filter(SettlementData.sjName == sjName)
if workinfoname is not None:
query = query.filter(SettlementData.workinfoname == workinfoname)
# 按上传时间倒序排序
query = query.order_by(SettlementData.createdate.desc())
# 如果指定了limit，则限制返回数量
if limit is not None and limit > 0:
query = query.limit(limit)
return query.all()
def search_settlement_data_formatted(self, db: Session,
id: Optional[int] = None,
point_id: Optional[str] = None,
nyid: Optional[str] = None,
sjName: Optional[str] = None,
workinfoname: Optional[str] = None) -> List[Dict[str, Any]]:
"""查询沉降数据并返回格式化结果"""
settlement_data = self.search_settlement_data(db, id, point_id, nyid, sjName, workinfoname)
workinfoname: Optional[str] = None,
limit: Optional[int] = None) -> List[Dict[str, Any]]:
"""查询沉降数据并返回格式化结果，按上传时间倒序排序"""
settlement_data = self.search_settlement_data(db, id, point_id, nyid, sjName, workinfoname, limit)
result = []
for settlement in settlement_data: