Improving API to collect Elevator Demand and compile it as CSV file

.gitignore

@@ -0,0 +1,6 @@
+__pycache__/
+*.pyc
+*.db
+.venv
+build
+*.egg-info

nextlevel-elevator/.tool-versions

@@ -0,0 +1 @@
+python 3.10.13

nextlevel-elevator/Dockerfile

@@ -0,0 +1,8 @@
+FROM python:3.10-slim-bookworm
+
+RUN mkdir -p /app
+WORKDIR /app
+COPY pyproject.toml /app/pyproject.toml
+RUN pip install -e ./
+
+COPY ./src /app/src

nextlevel-elevator/docker-compose.yml

@@ -0,0 +1,10 @@
+version: "3.9"
+
+services:
+  api:
+    build: .
+    ports:
+      - "1337:8000"
+    volumes:
+      - .:/app
+    command: uvicorn main:app --host 0.0.0.0 --port 8000

nextlevel-elevator/main.py

@@ -0,0 +1,21 @@
+from fastapi import FastAPI
+from sqlmodel import SQLModel
+
+from src import db
+from src import api
+
+app = FastAPI()
+
+
+def create_db_and_tables():
+    engine = db.Engine()
+    SQLModel.metadata.create_all(engine)
+
+
+@app.on_event("startup")
+def on_startup():
+    # It's not the Ideal but since this project is only for didatical pourpose
+    create_db_and_tables()
+
+
+app.include_router(api.router)

nextlevel-elevator/pyproject.toml

@@ -0,0 +1,20 @@
+[project]
+name = "nextlevel-elevator"
+version = "0.1.0"
+description = "A basic FastAPI project for NextLevel Elevator."
+dependencies = [
+    "fastapi==0.115.14", 
+    "pydantic==2.11.7", 
+    "sqlmodel==0.0.24", 
+    "uvicorn==0.34.3",
+    "httpx==0.28.1"
+]
+
+[project.optional-dependencies]
+test = [
+    "pytest<8.0.0,>=7.4.3"
+]
+
+[build-system]
+requires = ["setuptools", "wheel"]
+build-backend = "setuptools.build_meta" 

nextlevel-elevator/readme.md

@@ -0,0 +1,20 @@
+## Instanlling
+```bash
+pip install -e ./
+pip install .[test]
+```
+
+## Testsing
+
+```bash
+pytest tests/
+```
+
+## Running
+
+```bash
+docker compose up -d
+```
+## API Reference
+
+http://127.0.0.1:1337/docs

nextlevel-elevator/src/api/__init__.py

@@ -0,0 +1,10 @@
+from fastapi import APIRouter
+
+from . import v1
+
+router = APIRouter(
+    prefix="/api",
+    responses={404: {"description": "Not found"}},
+)
+
+router.include_router(v1.router)

nextlevel-elevator/src/api/v1/__init__.py

@@ -0,0 +1,10 @@
+from fastapi import APIRouter
+
+from . import elevator
+
+router = APIRouter(
+    prefix="/v1",
+    responses={404: {"description": "Not found"}},
+)
+
+router.include_router(elevator.router)

nextlevel-elevator/src/api/v1/elevator.py

@@ -0,0 +1,177 @@
+import io
+import csv
+from src.models import Elevator, ElevatorDemand, ElevatorDemandHistory
+from fastapi import APIRouter,  HTTPException
+from fastapi.responses import StreamingResponse
+
+from datetime import date, datetime
+from pydantic import BaseModel
+
+from sqlalchemy.exc import IntegrityError
+from sqlmodel import select
+
+from typing import List
+
+
+from src.db import Session
+
+
+router = APIRouter(
+    prefix="/elevator"
+)
+
+
+class DemandParameters(BaseModel):
+    level: int
+
+
+class SteteParameters(BaseModel):
+    level: int
+
+
+class DatasetParameters(BaseModel):
+    format: str = "csv"
+
+class ElevatorParameters(BaseModel):
+    min_level: int
+    max_level: int
+
+
+@router.post("/", status_code=201)
+def create_elevator(params: ElevatorParameters, session: Session) -> Elevator:
+    elevator = Elevator(
+        min_level=params.min_level,
+        max_level=params.max_level
+    )
+    session.add(elevator)
+    try:
+        session.flush()
+    except IntegrityError:
+        session.rollback()
+        raise HTTPException(status_code=400)
+    finally:
+        session.commit()
+        session.refresh(elevator)
+        return elevator
+
+
+@router.put("/{elevator_id}", status_code=202)
+async def call(elevator_id: int, params: DemandParameters, session: Session):
+    """
+        Provide the API to call the given elevator stores a ElevatorDemand,
+        and the demand is unique for the given level, if there is a Intergrity
+        Error it will returns 409 Conflict
+    """
+    elevator = session.get(Elevator, elevator_id)
+    if not elevator:
+        session.rollback()
+        raise HTTPException(status_code=404, detail="Not found")
+
+    if not (elevator.min_level <= params.level <= elevator.max_level):
+        session.rollback()
+        raise HTTPException(status_code=400, detail="Level overflow")
+
+    now = datetime.now()
+    demand = ElevatorDemand(
+        elevator_id=elevator_id,
+        timestamp=now.timestamp(),
+        level=params.level
+    )
+    session.add(demand)
+    try:
+        session.flush()
+    except IntegrityError:
+        session.rollback()
+        raise HTTPException(status_code=409, detail="Conflict, demaind already has been made")
+    finally:
+        session.commit()
+
+    return "Accepted"
+
+
+def create_history(demand: ElevatorDemand):
+    dt = datetime.fromtimestamp(demand.timestamp)
+    history = ElevatorDemandHistory(
+        elevator_id=demand.elevator_id,
+        week_day=dt.weekday(),
+        hour=dt.hour,
+        minute=dt.minute,
+        second=dt.second,
+        level=demand.level
+    )
+    return history
+
+
+@router.post("/{elevator_id}/state")
+async def set_state(elevator_id: int, params: SteteParameters, session: Session):
+    """
+        Provide the sufficient API to set the state, the main business logic 
+        is whenever a elevetor reach the the level, check if there is an 
+        open demand to that level, since demand has unique for level the 
+        very first demand will be stored to that level it will be cleared when
+        the elevator reach that level.
+
+        For that purpose we don't need to store the state itself, just reacting
+    """
+
+    elevator = session.get(Elevator, elevator_id)
+    if not elevator:
+        raise HTTPException(status_code=404, detail="Not found")
+
+    demand_stmt = select(ElevatorDemand)\
+                   .where(ElevatorDemand.elevator_id == elevator.id)\
+                   .where(ElevatorDemand.level == params.level)
+
+    demand  = session.exec(demand_stmt).first()
+    if not demand:
+        session.rollback()
+        return "Noop"
+
+    history = create_history(demand)
+    session.add(history)
+    session.flush()
+    session.delete(demand)
+    session.commit()
+    return "Accepted"
+
+
+def format_dataset_csv(history: List[ElevatorDemandHistory]):
+    output = io.StringIO()
+    writer = csv.writer(output)
+
+    writer.writerow([
+        "elevator_id",
+        "week_day",
+        "hour",
+        "minute",
+        "second",
+        "level"
+    ])
+    for h in history:
+        row = [
+            h.elevator_id,
+            h.week_day,
+            h.hour,
+            h.minute,
+            h.second,
+            h.level,
+        ]
+        writer.writerow(row)
+
+    output.seek(0)  # Rewind to the beginning of the stream
+
+    return StreamingResponse(
+        io.BytesIO(output.getvalue().encode('utf-8')),
+        media_type="text/csv",
+        headers={"Content-Disposition": "attachment; filename=dataset.csv"}
+    )
+
+@router.get("/dataset.{format}")
+async def get_dataset(format: str, session: Session):
+    history_stmt = select(ElevatorDemandHistory)
+
+    history = session.exec(history_stmt)
+    if format == "csv":
+        return format_dataset_csv(history)
+
+    raise HTTPException(status_code=400, detail="Format Not suppoted")

nextlevel-elevator/src/db.py

@@ -0,0 +1,27 @@
+import sqlmodel
+from fastapi import Depends
+
+from typing import Annotated
+from src import db
+
+def create_engine():
+    sqlite_file_name = "database.db"
+    sqlite_url = f"sqlite:///{sqlite_file_name}"
+    return sqlmodel.create_engine(sqlite_url)
+
+
+class Engine:
+    __instance__ = None
+    def __new__(cls):
+        if cls.__instance__ is None:
+            cls.__instance__ = create_engine()
+        return cls.__instance__
+
+
+def get_session():
+    engine = Engine()
+    with sqlmodel.Session(engine) as session:
+        yield session
+
+
+Session = Annotated[sqlmodel.Session, Depends(db.get_session)]

nextlevel-elevator/src/models.py

@@ -0,0 +1,53 @@
+import enum
+from typing import Annotated, Union
+
+from sqlmodel import Field, SQLModel, UniqueConstraint
+
+MAX_LEVEL = 10
+MIN_LEVEL = 1
+
+
+class WeekDay(enum.IntEnum):
+    MONDAY = 0
+    TUESDAY = 1
+    WEDNESDAY = 2
+    THURSDAY = 3
+    FRIDAY = 4
+    SATURDAY = 5
+    SUNDAY = 6
+
+
+class Elevator(SQLModel, table=True):
+    id: int = Field(primary_key=True)
+    min_level: int = Field(),
+    max_level: int = Field()
+
+
+class ElevatorDemand(SQLModel, table=True):
+    __table_args__ = (
+        UniqueConstraint(
+            "elevator_id",
+            "level",
+            name="uniq_elevator_id_timestamp_level"
+        ),
+    )
+    id: int = Field(primary_key=True)
+    elevator_id: int = Field(foreign_key="elevator.id")
+    timestamp: int = Field()
+    level: int = Field()
+
+
+class ElevatorDemandHistory(SQLModel, table=True):
+    """
+        Storing the demand that was completelly attended by the Elevator
+        and splitting the timestamp into week_day, hour, minute and second 
+        for the given demand, so it will be more easy to group demand by any 
+        time heuristics with second precision.
+    """
+    id: int = Field(primary_key=True)
+    elevator_id: int = Field(foreign_key="elevator.id")
+    week_day: int = Field()
+    hour: int = Field()
+    minute: int = Field()
+    second: int = Field()
+    level: int = Field()

nextlevel-elevator/tests/__init__.py

@@ -0,0 +1,3 @@
+
+
+