chemical_company.py 1.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243
  1. #!/usr/bin/env python3
  2. # -*- coding: utf-8 -*-
  3. from fastapi import APIRouter, Request, Depends, Query, HTTPException, status
  4. from common.security import valid_access_token
  5. from sqlalchemy.orm import Session
  6. from sqlalchemy.sql import func
  7. from common.auth_user import *
  8. from sqlalchemy import text
  9. from pydantic import BaseModel
  10. from common.BigDataCenterAPI import *
  11. from database import get_db
  12. from typing import List
  13. from models import *
  14. from utils import *
  15. from utils.spatial import *
  16. import json
  17. import traceback
  18. from jobs.rainfall_conditions_job import get_stcd_data
  19. from datetime import datetime,timedelta
  20. router = APIRouter()
  21. @router.get("/list")
  22. async def get_list(
  23. area_name: str,
  24. db: Session = Depends(get_db)
  25. ):
  26. try:
  27. sql = text("""SELECT * from sharedb.chemical_company""")
  28. if area_name :
  29. sql.bindparams(area_name=area_name).where(text("area = :area_name"))
  30. result = db.execute(sql).fetchall()
  31. # 将结果转换为rain_pits.py字典列表
  32. result_list = []
  33. for row in result:
  34. result_list.append(dict(row))
  35. return result_list
  36. except Exception as e:
  37. traceback.print_exc()
  38. raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")