From 08ae1f44c0b7b7260aced898e8122580c8f89bdd Mon Sep 17 00:00:00 2001 From: haotian <2421912570@qq.com> Date: Thu, 5 Jun 2025 18:09:51 +0800 Subject: [PATCH] =?UTF-8?q?=E7=A4=BA=E4=BE=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/v1/api.py | 97 ++++++++++++++++++++++++++++++ app/api/v1/testLogin.py | 122 +++++++++++++++++++++++++++++++++++++ app/core/config.py | 19 ++++++ app/core/database.py | 32 ++++++++++ app/crud/base.py | 88 +++++++++++++++++++++++++++ app/crud/event.py | 130 ++++++++++++++++++++++++++++++++++++++++ app/main.py | 25 ++++++++ app/models/models.py | 122 +++++++++++++++++++++++++++++++++++++ app/schemas/event.py | 116 +++++++++++++++++++++++++++++++++++ run.py | 14 +++++ test_generic.py | 27 +++++++++ test_requests.py | 87 +++++++++++++++++++++++++++ 12 files changed, 879 insertions(+) create mode 100644 app/api/v1/api.py create mode 100644 app/api/v1/testLogin.py create mode 100644 app/core/config.py create mode 100644 app/core/database.py create mode 100644 app/crud/base.py create mode 100644 app/crud/event.py create mode 100644 app/main.py create mode 100644 app/models/models.py create mode 100644 app/schemas/event.py create mode 100644 run.py create mode 100644 test_generic.py create mode 100644 test_requests.py diff --git a/app/api/v1/api.py b/app/api/v1/api.py new file mode 100644 index 0000000..2013901 --- /dev/null +++ b/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] \ No newline at end of file diff --git a/app/api/v1/testLogin.py b/app/api/v1/testLogin.py new file mode 100644 index 0000000..11a6457 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/app/core/config.py b/app/core/config.py new file mode 100644 index 0000000..25fe352 --- /dev/null +++ b/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() \ No newline at end of file diff --git a/app/core/database.py b/app/core/database.py new file mode 100644 index 0000000..9cd0d08 --- /dev/null +++ b/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() \ No newline at end of file diff --git a/app/crud/base.py b/app/crud/base.py new file mode 100644 index 0000000..22103a5 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/app/crud/event.py b/app/crud/event.py new file mode 100644 index 0000000..2e0634d --- /dev/null +++ b/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) \ No newline at end of file diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..c393011 --- /dev/null +++ b/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": "测试成功"} \ No newline at end of file diff --git a/app/models/models.py b/app/models/models.py new file mode 100644 index 0000000..651ae00 --- /dev/null +++ b/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'), + ) \ No newline at end of file diff --git a/app/schemas/event.py b/app/schemas/event.py new file mode 100644 index 0000000..ae13b77 --- /dev/null +++ b/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) + \ No newline at end of file diff --git a/run.py b/run.py new file mode 100644 index 0000000..7358eb5 --- /dev/null +++ b/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 + ) \ No newline at end of file diff --git a/test_generic.py b/test_generic.py new file mode 100644 index 0000000..8eafe5f --- /dev/null +++ b/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))) \ No newline at end of file diff --git a/test_requests.py b/test_requests.py new file mode 100644 index 0000000..e3ffc93 --- /dev/null +++ b/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() \ No newline at end of file