| 12345678910111213141516171819202122232425262728293031323334353637383940414243 |
- #!/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 *
- 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,
- db: Session = Depends(get_db)
- ):
- try:
- sql = text("""SELECT * from sharedb.chemical_company""")
- if area_name :
- sql.bindparams(area_name=area_name).where(text("area = :area_name"))
- result = db.execute(sql).fetchall()
- # 将结果转换为rain_pits.py字典列表
- result_list = []
- for row in result:
- result_list.append(dict(row))
- return result_list
- except Exception as e:
- traceback.print_exc()
- raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
|
