simple-ocpp-cs/app/routers/chargepoint_v1.py

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 ocpp.v201.call import ResetPayload

from app.database import get_db
from app.ocpp_proto import chargepoint_manager
from app.schemas.chargepoint import (
    ChargePoint,
    ChargePointCreate,
    ChargePointUpdate,
    ChargePointPassword,
    ChargePointConnectionInfo,
    ChargePointResetRequest,
    ChargePointResetResponse
)
from app.models.chargepoint import ChargePoint as DbChargePoint
from app.security import get_api_key

router = APIRouter(
    prefix="/chargepoints",
    tags=["Chargepoint (v1)"],
)

@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.get(DbChargePoint, chargepoint_id)
    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.get(DbChargePoint, chargepoint_id)
    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.get(DbChargePoint, chargepoint_id)
    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="", status_code=201, 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.get(DbChargePoint, chargepoint_id)
    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=None)
async def delete_chargepoint(
    chargepoint_id: UUID,
    api_key: str = Security(get_api_key),
    db: Session = Depends(get_db)
):
    chargepoint = db.get(DbChargePoint, chargepoint_id)
    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)
    )

@router.post(path="/{chargepoint_id}/reset", response_model=ChargePointResetResponse)
async def reset_chargepoint(
    chargepoint_id: UUID,
    reset_request: ChargePointResetRequest,
    api_key: str = Security(get_api_key)
):
    if chargepoint_manager.is_connected(chargepoint_id) == False:
        raise HTTPException(status_code=423, detail="Chargepoint not connected")
    response = await chargepoint_manager.call(
        chargepoint_id,
        payload=ResetPayload(type=reset_request.type, evse_id=reset_request.evse_id)
    )
    return ChargePointResetResponse(status=response.status)