115 lines
4 KiB
Python
115 lines
4 KiB
Python
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="/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)
|
|
)
|