Reafctor app

app/routers/chargepoint_v1.py

@@ -0,0 +1,114 @@
+import random
+import string
+from uuid import UUID
+from fastapi import APIRouter, HTTPException, Security
+from fastapi.params import Depends
+from sqlalchemy.orm import Session
+
+from app.database import get_db
+from app.ocpp_proto import chargepoint_manager
+from app.schemas.chargepoint import ChargePoint, ChargePointCreate, ChargePointUpdate, ChargePointPassword, ChargePointConnectionInfo
+from app.models.chargepoint import ChargePoint as DBChargePoint
+from app.security import get_api_key
+
+router = APIRouter(
+    prefix="/chargepoint",
+    tags=["chargepoint"],
+)
+
+@router.get(path="/", response_model=list[ChargePoint])
+async def get_chargepoints(
+    skip: int = 0,
+    limit: int = 20,
+    api_key: str = Security(get_api_key),
+    db: Session = Depends(get_db)
+):
+    return db.query(DBChargePoint).offset(skip).limit(limit).all()
+
+@router.get(path="/{chargepoint_id}", response_model=ChargePoint)
+async def get_chargepoint(
+    chargepoint_id: UUID,
+    api_key: str = Security(get_api_key),
+    db: Session = Depends(get_db)
+):
+    chargepoint = db.query(DBChargePoint).filter(DBChargePoint.id == chargepoint_id).first()
+    if chargepoint is None:
+        raise HTTPException(status_code=404, detail="Chargepoint not found")
+    return chargepoint
+
+@router.get(path="/{chargepoint_id}/password", response_model=ChargePointPassword)
+async def get_chargepoint_password(
+    chargepoint_id: UUID,
+    api_key: str = Security(get_api_key),
+    db: Session = Depends(get_db)
+):
+    chargepoint = db.query(DBChargePoint).filter(DBChargePoint.id == chargepoint_id).first()
+    if chargepoint is None:
+        raise HTTPException(status_code=404, detail="Chargepoint not found")
+    return ChargePointPassword(password=chargepoint.password)
+
+@router.delete(path="/{chargepoint_id}/password", response_model=ChargePointPassword)
+async def reset_chargepoint_password(
+    chargepoint_id: UUID,
+    api_key: str = Security(get_api_key),
+    db: Session = Depends(get_db)
+):
+    chargepoint = db.query(DBChargePoint).filter(DBChargePoint.id == chargepoint_id).first()
+    if chargepoint is None:
+        raise HTTPException(status_code=404, detail="Chargepoint not found")
+    chargepoint.password = ''.join(random.choice(string.ascii_letters + string.digits) for i in range(24))
+    db.commit()
+    return ChargePointPassword(password=chargepoint.password)
+
+@router.post(path="/", response_model=ChargePoint)
+async def create_chargepoint(
+    chargepoint: ChargePointCreate,
+    api_key: str = Security(get_api_key),
+    db: Session = Depends(get_db)
+):
+    chargepoint_db = DBChargePoint(
+        friendly_name=chargepoint.friendly_name,
+        is_active=chargepoint.is_active,
+        password=''.join(random.choice(string.ascii_letters + string.digits) for i in range(24))
+    )
+    db.add(chargepoint_db)
+    db.commit()
+    db.refresh(chargepoint_db)
+    return chargepoint_db
+
+@router.patch(path="/{chargepoint_id}", response_model=ChargePoint)
+async def update_chargepoint(
+    chargepoint_id: UUID,
+    chargepoint_update: ChargePointUpdate,
+    api_key: str = Security(get_api_key),
+    db: Session = Depends(get_db)
+):
+    chargepoint = db.query(DBChargePoint).filter(DBChargePoint.id == chargepoint_id).first()
+    if chargepoint is None:
+        raise HTTPException(status_code=404, detail="Chargepoint not found")
+    for key, value in chargepoint_update.model_dump(exclude_unset=True).items():
+        setattr(chargepoint, key, value)
+    db.commit()
+    return chargepoint
+
+@router.delete(path="/{chargepoint_id}", response_model=[])
+async def delete_chargepoint(
+    chargepoint_id: UUID,
+    api_key: str = Security(get_api_key),
+    db: Session = Depends(get_db)
+):
+    chargepoint = db.query(DBChargePoint).filter(DBChargePoint.id == chargepoint_id).first()
+    if chargepoint is None:
+        raise HTTPException(status_code=404, detail="Chargepoint not found")
+    db.delete(chargepoint)
+    db.commit()
+    return []
+
+@router.get(path="/{chargepoint_id}/status", response_model=ChargePointConnectionInfo)
+async def get_chargepoint_status(
+    chargepoint_id: UUID,
+    api_key: str = Security(get_api_key)
+):
+    return ChargePointConnectionInfo(
+        connected=chargepoint_manager.is_connected(chargepoint_id)
+    )