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

原查询数据条件分页优化

Browse Source
This commit is contained in:
lhx
2025-11-06 11:00:16 +08:00
parent 0980d48fda
commit 1cbea1bb2a
5 changed files with 100 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
app/services/base.py
View File

@@ -52,7 +52,7 @@ class BaseService(Generic[ModelType]):
return db.query(self.model).filter(field == field_value).all()
return []
def search_by_conditions(self, db: Session, conditions: Dict[str, Any]) -> List[ModelType]:
def search_by_conditions(self, db: Session, conditions: Dict[str, Any], skip: int = 0, limit: Optional[int] = None) -> List[ModelType]:
"""根据多个条件搜索记录"""
query = db.query(self.model)
for field_name, field_value in conditions.items():
@@ -62,4 +62,19 @@ class BaseService(Generic[ModelType]):
query = query.filter(field.like(f"{field_value}"))
else:
query = query.filter(field == field_value)
return query.all()
query = query.offset(skip)
if limit is not None:
query = query.limit(limit)
return query.all()
def search_by_conditions_count(self, db: Session, conditions: Dict[str, Any]) -> int:
"""根据多个条件搜索记录总数"""
query = db.query(self.model)
for field_name, field_value in conditions.items():
if hasattr(self.model, field_name) and field_value is not None:
field = getattr(self.model, field_name)
if isinstance(field_value, str):
query = query.filter(field.like(f"{field_value}"))
else:
query = query.filter(field == field_value)
return query.count()

44
app/services/section_data.py
View File

@@ -22,7 +22,7 @@ class SectionDataService(BaseService[SectionData]):
return sections[0] if sections else None
def get_by_account_id(self, db: Session, account_id: str) -> Optional[SectionData]:
"""根据账号ID获取断面数据"""
accounts = self.get_by_field(db, "account_id", account_id)
accounts = self.get_by_field(db, "account_id", account_id)
return accounts if accounts else None
def get_by_number(self, db: Session, number: str) -> List[SectionData]:
"""根据桥梁墩(台)编号获取断面数据"""
@@ -36,7 +36,9 @@ class SectionDataService(BaseService[SectionData]):
number: Optional[str] = None,
status: Optional[str] = None,
basic_types: Optional[str] = None,
account_id: Optional[str] = None) -> List[SectionData]:
account_id: Optional[str] = None,
skip: int = 0,
limit: Optional[int] = None) -> List[SectionData]:
"""根据多个条件搜索断面数据"""
conditions = {}
if section_id is not None:
@@ -56,7 +58,7 @@ class SectionDataService(BaseService[SectionData]):
if account_id is not None:
conditions['account_id'] = account_id
return self.search_by_conditions(db, conditions)
return self.search_by_conditions(db, conditions, skip, limit)
def search_sections_with_checkpoints(self, db: Session,
id: Optional[int] = None,
@@ -65,9 +67,32 @@ class SectionDataService(BaseService[SectionData]):
work_site: Optional[str] = None,
number: Optional[str] = None,
status: Optional[str] = None,
account_id: Optional[str] = None) -> List[Dict[str, Any]]:
"""查询断面数据并返回带观测点的结果"""
sections = self.search_section_data(db, id, section_id, mileage, work_site, number, status, account_id=account_id)
account_id: Optional[str] = None,
skip: int = 0,
limit: Optional[int] = None) -> Dict[str, Any]:
"""查询断面数据并返回带观测点的结果（支持分页）"""
# 构建查询条件
conditions = {}
if section_id is not None:
conditions["section_id"] = section_id
if work_site is not None:
conditions["work_site"] = work_site
if number is not None:
conditions["number"] = number
if status is not None:
conditions["status"] = status
if id is not None:
conditions['id'] = id
if mileage is not None:
conditions['mileage'] = mileage
if account_id is not None:
conditions['account_id'] = account_id
# 获取总数
total_count = self.search_by_conditions_count(db, conditions)
# 获取分页数据
sections = self.search_by_conditions(db, conditions, skip, limit)
result = []
for section in sections:
@@ -101,7 +126,12 @@ class SectionDataService(BaseService[SectionData]):
}
result.append(section_dict)
return result
return {
"data": result,
"total": total_count,
"skip": skip,
"limit": limit
}
def get_section_with_checkpoints(self, db: Session, section_id: str) -> Dict[str, Any]:
"""获取断面数据及其关联的观测点"""

34
app/services/settlement_data.py
View File

@@ -37,6 +37,7 @@ class SettlementDataService(BaseService[SettlementData]):
nyid: Optional[str] = None,
sjName: Optional[str] = None,
workinfoname: Optional[str] = None,
skip: int = 0,
limit: Optional[int] = None) -> List[SettlementData]:
"""根据多个条件搜索沉降数据，按上传时间倒序排序"""
query = db.query(SettlementData)
@@ -55,7 +56,9 @@ class SettlementDataService(BaseService[SettlementData]):
# 按上传时间倒序排序
query = query.order_by(SettlementData.createdate.desc())
# 如果指定了limit，则限制返回数量
# 添加分页支持
if skip > 0:
query = query.offset(skip)
if limit is not None and limit > 0:
query = query.limit(limit)
@@ -67,9 +70,25 @@ class SettlementDataService(BaseService[SettlementData]):
nyid: Optional[str] = None,
sjName: Optional[str] = None,
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)
skip: int = 0,
limit: Optional[int] = None) -> Dict[str, Any]:
"""查询沉降数据并返回格式化结果，按上传时间倒序排序（支持分页）"""
# 先获取总数（不计分页）
count_query = db.query(SettlementData)
if id is not None:
count_query = count_query.filter(SettlementData.id == id)
if point_id is not None:
count_query = count_query.filter(SettlementData.point_id == point_id)
if nyid is not None:
count_query = count_query.filter(SettlementData.NYID == nyid)
if sjName is not None:
count_query = count_query.filter(SettlementData.sjName == sjName)
if workinfoname is not None:
count_query = count_query.filter(SettlementData.workinfoname == workinfoname)
total_count = count_query.count()
# 获取分页数据
settlement_data = self.search_settlement_data(db, id, point_id, nyid, sjName, workinfoname, skip, limit)
result = []
for settlement in settlement_data:
@@ -97,7 +116,12 @@ class SettlementDataService(BaseService[SettlementData]):
}
result.append(settlement_dict)
return result
return {
"data": result,
"total": total_count,
"skip": skip,
"limit": limit
}
def search_settlement_checkpoint_data_formatted(self, db: Session,
id: Optional[int] = None,