Skip to content

Commit

Permalink
WIP: implement create status API endpoint
Browse files Browse the repository at this point in the history
  • Loading branch information
jmaupetit committed May 7, 2024
1 parent 5248972 commit 2008ccb
Show file tree
Hide file tree
Showing 2 changed files with 74 additions and 9 deletions.
39 changes: 30 additions & 9 deletions src/api/qualicharge/api/v1/routers/dynamic.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,13 +3,18 @@
import logging
from typing import Annotated, List

from fastapi import APIRouter, Path, status
from fastapi import APIRouter, Depends, HTTPException, Path
from fastapi import status as fa_status
from sqlmodel import Session, select

from qualicharge.db import get_session
from qualicharge.models.dynamic import (
SessionCreate,
StatusCreate,
StatusRead,
)
from qualicharge.schemas import PointDeCharge, Status
from qualicharge.schemas import Session as QCSession

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -57,25 +62,41 @@ async def read_status_history(
raise NotImplementedError


@router.post("/status/", status_code=status.HTTP_201_CREATED, tags=["Status"])
async def create_status(status: StatusCreate) -> None:
@router.post("/status/", status_code=fa_status.HTTP_201_CREATED, tags=["Status"])
async def create_status(
status: StatusCreate,
session: Session = Depends(get_session),
) -> None:
"""Create a status."""
raise NotImplementedError


@router.post("/status/bulk", status_code=status.HTTP_201_CREATED, tags=["Status"])
pdc = session.exec(
select(PointDeCharge).where(
PointDeCharge.id_pdc_itinerance == status.id_pdc_itinerance
)
).one_or_none()
if pdc is None:
raise HTTPException(
status_code=fa_status.HTTP_404_NOT_FOUND,
detail="Attached point of charge does not exist",
)
db_status = Status(**status.model_dump(exclude={"id_pdc_itinerance"}))
db_status.point_de_charge_id = pdc.id
session.add(db_status)
session.commit()


@router.post("/status/bulk", status_code=fa_status.HTTP_201_CREATED, tags=["Status"])
async def create_status_bulk(statuses: List[StatusCreate]) -> None:
"""Create a statuses batch."""
raise NotImplementedError


@router.post("/session/", status_code=status.HTTP_201_CREATED, tags=["Session"])
@router.post("/session/", status_code=fa_status.HTTP_201_CREATED, tags=["Session"])
async def create_session(session: SessionCreate) -> None:
"""Create a session."""
raise NotImplementedError


@router.post("/session/bulk", status_code=status.HTTP_201_CREATED, tags=["Session"])
@router.post("/session/bulk", status_code=fa_status.HTTP_201_CREATED, tags=["Session"])
async def create_session_bulk(sessions: List[SessionCreate]) -> None:
"""Create a sessions batch."""
raise NotImplementedError
44 changes: 44 additions & 0 deletions src/api/tests/api/v1/routers/test_dynamic.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
"""Tests for the QualiCharge API dynamic router."""

import json

from fastapi import status
from sqlmodel import select

from qualicharge.factories.dynamic import StatusCreateFactory
from qualicharge.factories.static import StatiqueFactory
from qualicharge.schemas import PointDeCharge, Status
from qualicharge.schemas.utils import save_statique


def test_create_status_with_invalid_id_pdc_itinerance(db_session, client_auth):
"""Test the /status/ create endoiint."""
id_pdc_itinerance = "ESZUNE1111ER1"
qc_status = StatusCreateFactory.build(id_pdc_itinerance=id_pdc_itinerance)

# Point of charge does not exist yet
response = client_auth.post(
"/dynamique/status/", json=json.loads(qc_status.model_dump_json())
)
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.json() == {"detail": "Attached point of charge does not exist"}

# Create point of charge
save_statique(
db_session, StatiqueFactory.build(id_pdc_itinerance=id_pdc_itinerance)
)
response = client_auth.post(
"/dynamique/status/", json=json.loads(qc_status.model_dump_json())
)
assert response.status_code == status.HTTP_201_CREATED
assert response.json() is None

# Query database to check created status and relations
pdc = db_session.exec(
select(PointDeCharge).where(
PointDeCharge.id_pdc_itinerance == qc_status.id_pdc_itinerance
)
).one()
db_status = db_session.exec(select(Status)).one()
assert db_status.point_de_charge_id == pdc.id
assert db_status in pdc.statuses

0 comments on commit 2008ccb

Please sign in to comment.