python-fastapi-mm-zhcs-yj-api/routers/api/rainfall/dzzh.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

from fastapi import APIRouter, Request, Depends, Query, HTTPException, status
from common.security import valid_access_token
from sqlalchemy.orm import Session
from sqlalchemy.sql import func
from common.auth_user import *
from sqlalchemy import  text
from pydantic import BaseModel
from common.BigDataCenterAPI import *
from database import get_db
from typing import List
from models import *
from utils import *
from utils.spatial import *
from utils.rainfall_util import *
import json
import traceback
from jobs.rainfall_conditions_job import get_stcd_data
from datetime import datetime,timedelta


router = APIRouter()


@router.get("/list")
async def get_list(
    area_name: str= Query(None),
    history_time:int = Query(None),
    future_time:int = Query(None),
    db: Session = Depends(get_db),
    page: int = Query(1, gt=0, description='页码'),
    pageSize: int = Query(10, gt=0, description='每页条目数量')
):
    try:
        # 计算 OFFSET 值
        offset = (page - 1) * pageSize
        # 构造基础查询
        base_query = text("SELECT * FROM sharedb.midmap_dzzh")
        if area_name:
            base_query = base_query.bindparams(area_name=area_name).where(text("area = :area_name"))

        # 构造分页查询
        paginated_query = f"{base_query} LIMIT :limit OFFSET :offset"
        paginated_query = text(paginated_query).bindparams(limit=pageSize, offset=offset)

        # 构造统计总数据量的查询
        count_query = select(func.count()).select_from(text("sharedb.midmap_dzzh"))
        if area_name:
            count_query = count_query.where(text("area = :area_name")).bindparams(area_name=area_name)

        # 执行统计查询并获取总数据量
        total = db.execute(count_query).scalar()

        # 执行分页查询并获取结果
        result = db.execute(paginated_query).fetchall()

        # 将结果转换为rain_pits.py字典列表
        result_list = []
        for row in result:
            data = dict(row)
            if history_time:
                real_code = get_real_code(db, data['longitude'], data['latitude'])
                rainfall = get_rainfall(real_code, history_time, db)
                data['rainfall'] = rainfall
            if future_time:
                data['rainfall'] = 0
            result_list.append(data)
        return {
            "code": 200,
            "msg": "操作成功",
            "data": result_list,
            "total": total,
            "page": page,
            "pageSize": pageSize,
            "totalPages": (total + pageSize - 1) // pageSize
        }
    except Exception as e:
        traceback.print_exc()
        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")