示例

app/api/v1/api.py

@@ -0,0 +1,97 @@
+from typing import List, Optional
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.ext.asyncio import AsyncSession
+from app.core.database import get_db
+from app.crud.event import event
+from app.schemas.event import EventList, EventDetail, EventUpdate, EventQuery, TestEvent
+
+router = APIRouter()
+
+@router.get("/hello")
+async def get_hello():
+    
+    return {"data":"hello"}
+
+# 路径 /events response_model 自动将返回数据序列化为EventList模型的列表格式.
+@router.get("/events", response_model=List[EventList])
+async def get_events(
+    db: AsyncSession = Depends(get_db), # 数据库依赖注入
+    start_time: Optional[str] = None,
+    end_time: Optional[str] = None,
+    etypeName: Optional[str] = None,
+    area: Optional[str] = None,
+    skip: int = Query(0, ge=0),
+    limit: int = Query(100, ge=1, le=1000)
+):
+    """
+    获取事件列表
+    - 支持时间范围查询
+    - 支持事件类型和区域筛选
+    - 支持分页
+    """
+    query = EventQuery(
+        start_time=start_time,
+        end_time=end_time,
+        etypeName=etypeName,
+        area=area,
+        skip=skip,
+        limit=limit
+    )
+    events = await event.get_multi_with_query(db, query=query)
+    
+    # 3. 序列化返回结果
+    return [EventList.model_validate(event) for event in events]
+
+@router.get("/events/{event_id}", response_model=EventDetail)
+async def get_event(
+    event_id: str,
+    db: AsyncSession = Depends(get_db)
+):
+    """
+    获取事件详情
+    """
+    event_obj = await event.get_by_id(db, event_id=event_id)
+    if not event_obj:
+        raise HTTPException(status_code=404, detail="事件不存在")
+    return EventDetail.model_validate(event_obj)
+
+@router.put("/events/{event_id}", response_model=EventDetail)
+async def update_event(
+    event_id: str,
+    event_in: EventUpdate,
+    db: AsyncSession = Depends(get_db)
+):
+    """
+    更新事件信息
+    """
+    event_obj = await event.update_event(
+        db,
+        event_id=event_id,
+        obj_in=event_in
+    )
+    if not event_obj:
+        raise HTTPException(status_code=404, detail="事件不存在")
+    return EventDetail.model_validate(event_obj)
+
+@router.delete("/events/{event_id}", response_model=EventDetail)
+async def delete_event(
+    event_id: str,
+    db: AsyncSession = Depends(get_db)
+):
+    """
+    删除事件
+    """
+    event_obj = await event.delete_event(db, event_id=event_id)
+    if not event_obj:
+        raise HTTPException(status_code=404, detail="事件不存在")
+    return EventDetail.model_validate(event_obj) 
+
+# 响应类型要写对啊
+@router.get("/testEvent", response_model=List[TestEvent])
+async def testEvent(
+    db: AsyncSession = Depends(get_db)
+):
+    event_obj = await event.get_test(db)
+    return event_obj
+
+    # return  [TestEvent.model_validate(event) for event in event_obj]

app/api/v1/testLogin.py

@@ -0,0 +1,122 @@
+from datetime import datetime, timedelta
+from typing import Optional
+
+from fastapi import Depends, FastAPI, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+from jose import JWTError, jwt
+from passlib.context import CryptContext
+from pydantic import BaseModel
+
+# 配置参数（应从环境变量读取）
+SECRET_KEY = "your-secret-key-keep-it-secret!"
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+# 模拟用户数据库
+fake_users_db = {
+    "johndoe": {
+        "username": "johndoe",
+        "full_name": "John Doe",
+        "email": "johndoe@example.com",
+        # 哈希后的密码（明文是 secret）
+        "hashed_password": "$2b$12$EixZaYVK1fsbY1eZIbOnjesN9NwG1s3Z6FDcjyH103a2.dJgD0L4q",
+        "disabled": False,
+    }
+}
+
+# Token 模型
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+
+class TokenData(BaseModel):
+    username: Optional[str] = None
+
+# 用户模型
+class User(BaseModel):
+    username: str
+    email: Optional[str] = None
+    full_name: Optional[str] = None
+    disabled: Optional[bool] = None
+
+class UserInDB(User):
+    hashed_password: str
+
+# 密码哈希配置
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+# OAuth2 配置
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+app = FastAPI()
+
+# 密码验证工具
+def verify_password(plain_password: str, hashed_password: str):
+    return pwd_context.verify(plain_password, hashed_password)
+
+# 获取用户
+def get_user(db, username: str):
+    if username in db:
+        user_dict = db[username]
+        return UserInDB(**user_dict)
+
+# 用户认证
+def authenticate_user(fake_db, username: str, password: str):
+    user = get_user(fake_db, username)
+    if not user:
+        return False
+    # if not verify_password(password, user.hashed_password):
+    #     return False
+    return user
+
+# 创建访问令牌
+def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+# 获取当前用户（依赖注入）
+async def get_current_user(token: str = Depends(oauth2_scheme)):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        username: str = payload.get("sub")
+        if username is None:
+            raise credentials_exception
+        token_data = TokenData(username=username)
+    except JWTError:
+        raise credentials_exception
+    user = get_user(fake_users_db, username=token_data.username)
+    if user is None:
+        raise credentials_exception
+    return user
+
+# 登录路由
+@app.post("/token", response_model=Token)
+async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
+    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    access_token = create_access_token(
+        data={"sub": user.username}, expires_delta=access_token_expires
+    )
+    return {"access_token": access_token, "token_type": "bearer"}
+
+# 受保护路由
+@app.get("/users/me/", response_model=User)
+async def read_users_me(current_user: User = Depends(get_current_user)):
+    return current_user

app/core/config.py

@@ -0,0 +1,19 @@
+from pydantic_settings import BaseSettings
+
+class Settings(BaseSettings):
+    # 数据库配置
+    DB_HOST: str = "10.0.0.17"
+    DB_PORT: int = 3306
+    DB_USER: str = "root"
+    DB_PASSWORD: str = "root"
+    DB_NAME: str = "kangda"
+    DB_CHARSET: str = "utf8mb4"
+    DB_POOL_SIZE: int = 5
+    DB_MAX_OVERFLOW: int = 10
+    DB_POOL_TIMEOUT: int = 30
+    DB_POOL_RECYCLE: int = 1800
+
+    class Config:
+        env_file = ".env"
+
+settings = Settings() 

app/core/database.py

@@ -0,0 +1,32 @@
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession  
+
+from sqlalchemy.orm import sessionmaker, declarative_base 
+from app.core.config import settings 
+
+engine = create_async_engine(
+    f"mysql+aiomysql://{settings.DB_USER}:{settings.DB_PASSWORD}@{settings.DB_HOST}:{settings.DB_PORT}/{settings.DB_NAME}",
+    pool_size=settings.DB_POOL_SIZE,        # 连接池常驻连接数
+    max_overflow=settings.DB_MAX_OVERFLOW,  # 池最大溢出连接数
+    pool_timeout=settings.DB_POOL_TIMEOUT,  # 获取连接超时时间（秒）
+    pool_recycle=settings.DB_POOL_RECYCLE,  # 连接回收间隔（秒）
+    echo=False                             # 关闭SQL日志输出
+)
+
+# expire_on_commit=False 禁用提交后过期对象 --> 提交后原对象仍然可以使用.
+async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+
+# ORM模型基类
+Base = declarative_base()
+
+
+# 获取数据库会话
+async def get_db():
+    async with async_session() as session:
+        try:
+            yield session 
+            await session.commit() 
+        except Exception:
+            await session.rollback()
+            raise 
+        finally:
+            await session.close()

app/crud/base.py

@@ -0,0 +1,88 @@
+from typing import Any, Dict, Generic, List, Optional, Type, TypeVar, Union
+from fastapi.encoders import jsonable_encoder
+from pydantic import BaseModel
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, update, delete
+from app.core.database import Base
+
+ModelType = TypeVar("ModelType", bound=Base)
+CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
+UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
+
+'''
+    Generic[ModelType, CreateSchemaType, UpdateSchemaType] 是 Python 类型提示中泛型编程的关键语法，用于声明一个​​泛型类​​。
+    它的作用是让 CRUDBase 类具备类型参数化的能力，允许在继承或实例化时动态绑定具体类型，从而实现代码复用和类型安全
+    声明 CRUDBase 类需要三个类型参数：ModelType、CreateSchemaType、UpdateSchemaType。
+    这些类型参数会在类的内部方法中使用（如 get、create、update），确保类型一致性。
+
+'''
+class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
+    def __init__(self, model: Type[ModelType]):
+        """
+        CRUD对象与SQLAlchemy模型类一起使用
+        :param model: SQLAlchemy模型类
+        """
+        self.model = model
+
+    # 根据id获取对象
+    async def get(self, db: AsyncSession, id: Any) -> Optional[ModelType]:
+        """
+        通过ID获取对象
+        """
+        query = select(self.model).where(self.model.id == id)
+        result = await db.execute(query)
+        return result.scalar_one_or_none()
+
+    # 分页查询
+    async def get_multi(
+        self, db: AsyncSession, *, skip: int = 0, limit: int = 100
+    ) -> List[ModelType]:
+        """
+        获取多个对象
+        """
+        query = select(self.model).offset(skip).limit(limit)
+        result = await db.execute(query)
+        return result.scalars().all()
+
+    async def create(self, db: AsyncSession, *, obj_in: CreateSchemaType) -> ModelType:
+        """
+        创建对象
+        """
+        obj_in_data = jsonable_encoder(obj_in)
+        db_obj = self.model(**obj_in_data)
+        db.add(db_obj)
+        await db.commit()
+        await db.refresh(db_obj)
+        return db_obj
+
+    async def update(
+        self,
+        db: AsyncSession,
+        *,
+        db_obj: ModelType,
+        obj_in: Union[UpdateSchemaType, Dict[str, Any]]
+    ) -> ModelType:
+        """
+        更新对象
+        """
+        obj_data = jsonable_encoder(db_obj)
+        if isinstance(obj_in, dict):
+            update_data = obj_in
+        else:
+            update_data = obj_in.dict(exclude_unset=True)
+        for field in obj_data:
+            if field in update_data:
+                setattr(db_obj, field, update_data[field])
+        db.add(db_obj)
+        await db.commit()
+        await db.refresh(db_obj)
+        return db_obj
+
+    async def remove(self, db: AsyncSession, *, id: Any) -> ModelType:
+        """
+        删除对象
+        """
+        obj = await self.get(db=db, id=id)
+        await db.delete(obj)
+        await db.commit()
+        return obj 

app/crud/event.py

@@ -0,0 +1,130 @@
+from typing import List, Optional, Dict, Any
+from sqlalchemy import select, and_, or_, join
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+from app.crud.base import CRUDBase
+from app.models.models import Event, Image, Temperature
+from app.schemas.event import EventUpdate, EventQuery, TestEvent
+
+class CRUDEvent(CRUDBase[Event, EventUpdate, EventUpdate]):
+    async def get_by_id(self, db: AsyncSession, *, event_id: str) -> Optional[Event]:
+        """根据ID获取事件"""
+        query = (
+            select(Event)
+            .options(
+                selectinload(Event.images),
+                selectinload(Event.temperatures)
+            )
+            .where(Event.eventId == event_id)
+        )
+        result = await db.execute(query)
+        return result.scalar_one_or_none()
+
+    async def get_multi_with_query(
+        self, 
+        db: AsyncSession, 
+        *, 
+        query: EventQuery
+    ) -> List[Event]:
+        """根据查询条件获取事件列表"""
+        conditions = []
+        
+        if query.start_time:
+            conditions.append(Event.insDate >= query.start_time)
+        if query.end_time:
+            conditions.append(Event.insDate <= query.end_time)
+        if query.etypeName:
+            conditions.append(Event.etypeName == query.etypeName)
+        if query.area:
+            conditions.append(Event.area == query.area)
+            
+        query_stmt = (
+            select(Event)
+            .options(
+                selectinload(Event.images),
+                selectinload(Event.temperatures)
+            )
+        )
+        
+        if conditions:
+            query_stmt = query_stmt.where(and_(*conditions))
+            
+        query_stmt = query_stmt.offset(query.skip).limit(query.limit)
+        
+        result = await db.execute(query_stmt)
+        return result.scalars().all()
+
+    async def update_event(
+        self,
+        db: AsyncSession,
+        *,
+        event_id: str,
+        obj_in: EventUpdate
+    ) -> Optional[Event]:
+        """更新事件信息"""
+        event = await self.get_by_id(db, event_id=event_id)
+        if not event:
+            return None
+            
+        update_data = obj_in.model_dump()
+        for field, value in update_data.items():
+            setattr(event, field, value)
+            
+        # db.add(event)
+        await db.commit()
+        await db.refresh(event)
+        return event
+
+    async def delete_event(
+        self,
+        db: AsyncSession,
+        *,
+        event_id: str
+    ) -> Optional[Event]:
+        """删除事件"""
+        event = await self.get_by_id(db, event_id=event_id)
+        if not event:
+            return None
+            
+        await db.delete(event)
+        await db.commit()
+        return event
+    
+    
+    async def get_test(
+        self,
+        db: AsyncSession,
+   
+        
+    ) -> List[TestEvent]: #响应类型要写对啊
+        
+        '''
+            eventId
+            number
+            name 
+            imageUrl
+            localPath
+            temperature
+            confidence
+            createTime
+        '''
+        
+        query_stmt = (
+            select(Event.eventId, Event.number, Event.name, Image.imageUrl, Image.localPath, Temperature.temperature, Temperature.confidence, Temperature.createTime)
+            .select_from(Event)
+            .outerjoin(Image, Event.eventId == Image.eventId)
+            .outerjoin(Temperature, Image.imageId == Temperature.imageId)
+            # 多个查询条件
+            .where(and_(Event.etypeName=="日常巡检", True))
+        )
+        
+        result = await db.execute(query_stmt)
+        
+        # 获取字典
+        result =  result.mappings().all()
+        
+        # print(result)
+        return [TestEvent(**row) for row in result]
+
+
+event = CRUDEvent(Event) 

app/main.py

@@ -0,0 +1,25 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from app.api.v1.api import router
+
+
+app = FastAPI(
+    title="测试fastapi",
+    description="测试啊",
+    version="1.0.0"
+)
+
+# 配置CORS
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.include_router(router, prefix="/api/v1")
+
+@app.get("/")
+async def root():
+    return {"message": "测试成功"}

app/models/models.py

@@ -0,0 +1,122 @@
+from datetime import datetime
+from sqlalchemy import Column, String, DateTime, Integer, Text, ForeignKey, Index, BigInteger
+from sqlalchemy.orm import relationship
+from app.core.database import Base
+
+
+class Event(Base):
+    __tablename__ = "event"
+
+    eventId = Column(String(50), primary_key=True)
+    tenantInfoId = Column(String(100))
+    reportEventId = Column(String(100))
+    number = Column(String(20))
+    name = Column(String(20))
+    eclassify = Column(String(5))
+    operationType = Column(String(5))
+    etype = Column(String(20))
+    etypeName = Column(String(20))
+    enTypeName = Column(String(30))
+    hkTypeName = Column(String(20))
+    reportStatus = Column(String(5))
+    results = Column(String(5))
+    insDate = Column(DateTime)
+    insDateShow = Column(DateTime)
+    updDate = Column(DateTime)
+    updDateShow = Column(DateTime)
+    fileType = Column(String(5))
+    area = Column(String(20))
+    floor = Column(String(10))
+    map = Column(String(20))
+    staffId = Column(String(40))
+    targetUserId = Column(String(40))
+    position = Column(String(100))
+    actualStaffName = Column(String(20))
+    targetStaffName = Column(String(20))
+    routeName = Column(String(20))
+    phoneAddress = Column(String(500))
+    width = Column(String(10))
+    height = Column(String(10))
+    resolution = Column(String(10))
+    originX = Column(String(20))
+    originY = Column(String(20))
+    imgList = Column(String(20))
+    robotType = Column(String(5))
+    eventFloor = Column(String(10))
+    floorName = Column(String(10))
+    coordId = Column(String(40))
+    coord = Column(String(40))
+    coordName = Column(String(30))
+    positonName = Column(String(20))
+    processingRemark = Column(String(300))
+    carId = Column(String(40))
+    parkingSpaceType = Column(String(10))
+    parkingSpaceNumber = Column(String(40))
+    carNumber = Column(String(40))
+    eno = Column(String(40))
+    instrument = Column(String(40))
+    evideo = Column(String(40))
+    createTime = Column(DateTime, default=datetime.now, comment='本地后台创建时间')
+    updateTime = Column(DateTime, default=datetime.now, onupdate=datetime.now, comment='本地后台更新时间')
+
+    # 关系
+    images = relationship("Image", back_populates="event", cascade="all, delete-orphan")
+    temperatures = relationship("Temperature", back_populates="event", cascade="all, delete-orphan")
+    process_logs = relationship("ProcessLog", back_populates="event", cascade="all, delete-orphan")
+
+
+class Image(Base):
+    __tablename__ = "image"
+
+    imageId = Column(BigInteger, primary_key=True, autoincrement=True, comment='图片ID')
+    eventId = Column(String(50), ForeignKey('event.eventId'), nullable=False, comment='关联事件ID')
+    imageUrl = Column(String(500), nullable=False, comment='图片URL')
+    localPath = Column(String(500), comment='本地存储路径')
+    createTime = Column(DateTime, default=datetime.now, nullable=False, comment='创建时间')
+
+    # 关系
+    event = relationship("Event", back_populates="images" )
+    temperatures = relationship("Temperature", back_populates="image")
+
+    __table_args__ = (
+        Index('idx_image_event_id', 'eventId'),
+    )
+
+
+class Temperature(Base):
+    __tablename__ = "temperature"
+
+    tempId = Column(BigInteger, primary_key=True, autoincrement=True, comment='温度记录ID')
+    eventId = Column(String(50), ForeignKey('event.eventId'), nullable=False, comment='关联事件ID')
+    imageId = Column(BigInteger, ForeignKey('image.imageId'), nullable=False, comment='关联图片ID')
+    temperature = Column(String(20), nullable=False, comment='温度值')
+    # status = Column(String(2), comment='温度是否正常')
+    confidence = Column(String(40), nullable=False, comment='识别置信度')
+    createTime = Column(DateTime, default=datetime.now, nullable=False, comment='创建时间')
+
+    # 关系
+    event = relationship("Event", back_populates="temperatures")
+    image = relationship("Image", back_populates="temperatures")
+
+    __table_args__ = (
+        Index('idx_temp_event_id', 'eventId'),
+        Index('idx_temp_create_time', 'createTime'),
+    )
+
+
+class ProcessLog(Base):
+    __tablename__ = "process_log"
+
+    logId = Column(BigInteger, primary_key=True, autoincrement=True, comment='日志ID')
+    eventId = Column(String(50), ForeignKey('event.eventId'), nullable=False, comment='关联事件ID')
+    processStatus = Column(Integer, nullable=False, comment='处理状态')
+    errorMessage = Column(Text, comment='错误信息')
+    createTime = Column(DateTime, default=datetime.now, nullable=False, comment='创建时间')
+
+    # 关系
+    event = relationship("Event", back_populates="process_logs")
+
+    __table_args__ = (
+        Index('idx_log_event_id', 'eventId'),
+        Index('idx_log_create_time', 'createTime'),
+    ) 

app/schemas/event.py

@@ -0,0 +1,116 @@
+from datetime import datetime
+from typing import List, Optional
+from pydantic import BaseModel, ConfigDict
+
+class ImageBase(BaseModel):
+    imageUrl: str
+    localPath: Optional[str] = None
+    
+    model_config = ConfigDict(from_attributes=True)
+
+class TemperatureBase(BaseModel):
+    temperature: str
+    confidence: str
+    createTime: datetime
+    
+    model_config = ConfigDict(from_attributes=True)
+
+class EventBase(BaseModel):
+    eventId: str
+    tenantInfoId: Optional[str] = None
+    reportEventId: Optional[str] = None
+    number: Optional[str] = None
+    name: Optional[str] = None
+    etypeName: Optional[str] = None
+    insDate: Optional[datetime] = None
+    
+    model_config = ConfigDict(from_attributes=True)
+
+class EventList(EventBase):
+    images: List[ImageBase] = []
+    temperatures: List[TemperatureBase] = []
+    
+    model_config = ConfigDict(from_attributes=True)
+
+class EventDetail(EventBase):
+    eclassify: Optional[str] = None
+    operationType: Optional[str] = None
+    etype: Optional[str] = None
+    enTypeName: Optional[str] = None
+    hkTypeName: Optional[str] = None
+    reportStatus: Optional[str] = None
+    results: Optional[str] = None
+    insDateShow: Optional[datetime] = None
+    updDate: Optional[datetime] = None
+    updDateShow: Optional[datetime] = None
+    fileType: Optional[str] = None
+    area: Optional[str] = None
+    floor: Optional[str] = None
+    map: Optional[str] = None
+    staffId: Optional[str] = None
+    targetUserId: Optional[str] = None
+    position: Optional[str] = None
+    actualStaffName: Optional[str] = None
+    targetStaffName: Optional[str] = None
+    routeName: Optional[str] = None
+    phoneAddress: Optional[str] = None
+    width: Optional[str] = None
+    height: Optional[str] = None
+    resolution: Optional[str] = None
+    originX: Optional[str] = None
+    originY: Optional[str] = None
+    robotType: Optional[str] = None
+    eventFloor: Optional[str] = None
+    floorName: Optional[str] = None
+    coordId: Optional[str] = None
+    coord: Optional[str] = None
+    coordName: Optional[str] = None
+    positonName: Optional[str] = None
+    processingRemark: Optional[str] = None
+    carId: Optional[str] = None
+    parkingSpaceType: Optional[str] = None
+    parkingSpaceNumber: Optional[str] = None
+    carNumber: Optional[str] = None
+    eno: Optional[str] = None
+    instrument: Optional[str] = None
+    evideo: Optional[str] = None
+    createTime: Optional[datetime] = None
+    updateTime: Optional[datetime] = None
+    images: List[ImageBase] = []
+    temperatures: List[TemperatureBase] = []
+    
+    model_config = ConfigDict(from_attributes=True)
+
+class EventUpdate(BaseModel):
+    number: Optional[str] = None
+    name: Optional[str] = None
+    etypeName: Optional[str] = None
+    area: Optional[str] = None
+    position: Optional[str] = None
+    processingRemark: Optional[str] = None
+    
+    model_config = ConfigDict(from_attributes=True)
+
+class EventQuery(BaseModel):
+    start_time: Optional[datetime] = None
+    end_time: Optional[datetime] = None
+    etypeName: Optional[str] = None
+    area: Optional[str] = None
+    skip: int = 0
+    limit: int = 100
+    
+    model_config = ConfigDict(from_attributes=True) 
+    
+    
+class TestEvent(BaseModel):
+    eventId: str 
+    number: Optional[str] = None
+    name : Optional[str] = None
+    imageUrl : Optional[str] = None
+    localPath: Optional[str] = None
+    temperature: Optional[str] = None
+    confidence : Optional[str] = None
+    createTime: Optional[datetime] = None
+    
+    model_config = ConfigDict(from_attributes=True)
+    

run.py

@@ -0,0 +1,14 @@
+import uvicorn 
+
+if __name__ == "__main__":
+    # uvicorn.run(
+    #     "app.main:app",
+    #     host="10.0.0.202",
+    #     port=12342
+    # )
+    
+    uvicorn.run(
+        "app.api.v1.testLogin:app",
+        host="10.0.0.202",
+        port=12342
+    )

test_generic.py

@@ -0,0 +1,27 @@
+from typing import Generic, TypeVar, Dict
+from typing import Optional
+
+K = TypeVar("K")  # 键类型
+V = TypeVar("V")  # 值类型
+
+class GenericCache(Generic[K, V]):
+    def __init__(self):
+        self._store: Dict[K, V] = {}
+
+    def set(self, key: K, value: V) -> None:
+        self._store[key] = value
+
+    def get(self, key: K) -> Optional[V]:
+        return self._store.get(key)
+
+# 使用示例
+cache = GenericCache[str, int]()  # 键为 str，值为 int
+cache.set("count", 100)
+value: Optional[int] = cache.get("count")  # 返回 100
+
+cache.set(112, "123")
+
+
+for key in cache._store.keys():
+    
+    print(type(cache.get(key)))

test_requests.py

@@ -0,0 +1,87 @@
+import requests
+from requests import Response
+from typing import List, TypeVar, Dict, Any, Optional
+import logging
+
+# 设置日志
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+T = TypeVar('T')  # 泛型类型声明
+
+REQUEST_URL = "http://aijinan.biaofun.com.cn/app/query/shandong"  # 替换为实际API地址
+
+def main():
+    cities = ["济南市", "青岛市", "临沂市", "菏泽市", "威海市", "东营市"]
+    
+    # 修正：每次请求创建新的参数对象，避免累积
+    for city in cities:
+        params = {"city": city}
+        
+        # 发起RPC请求
+        response_list = rpc_get(REQUEST_URL, params, False)
+        
+        # 处理响应
+        if not response_list:
+            logger.warning(f"Empty response for city: {city}")
+            continue
+            
+        # 设置城市名称并处理后续逻辑 (根据实际Response类调整)
+        response_list[0]['name'] = city  # 这里假定response_list是字典列表
+        after(response_list)
+        
+def rpc_get(url: str, params: Dict[str, Any], right_or_wrong: bool) -> Optional[List[T]]:
+    """
+    RPC请求封装
+    :param url: 请求URL
+    :param params: 请求参数
+    :param right_or_wrong: 未使用的标志（保留原Java参数）
+    :return: 响应对象列表
+    """
+    logger.info(f"Request URL: {url}")
+    
+    try:
+        # 发起POST请求
+        response: Response = requests.post(url, data=params)
+        
+        # 检查响应状态
+        if response.status_code != 200:
+            logger.error(f"Request failed, Status Code: {response.status_code}, URL: {url}")
+            return None
+            
+        response_text = response.text
+        logger.debug(f"Response: {response_text}")
+        
+        # 解析响应（需要根据实际API响应格式实现）
+        return parse_response(response_text)
+        
+    except Exception as e:
+        logger.exception(f"RPC request failed: {str(e)}")
+        return None
+
+def parse_response(response_text: str) -> List[T]:
+    """
+    解析API响应 (需要根据实际API响应格式实现)
+    示例实现 - 替换为实际解析逻辑
+    """
+    # 这里需要根据实际API返回格式实现解析
+    # 示例：解析为JSON对象
+    import json
+    try:
+        data = json.loads(response_text)
+        # 根据实际数据结构返回列表
+        return [data] if isinstance(data, dict) else data
+    except json.JSONDecodeError:
+        logger.error("Invalid JSON response")
+        return []
+
+def after(response_list: List[Any]):
+    """
+    后续处理逻辑 (根据实际需求实现)
+    """
+    # 示例实现
+    print(f"Processing response for: {response_list[0].get('name')}")
+    # 实际业务逻辑...
+
+if __name__ == "__main__":
+    main()