API Routes and Full CRUD Functionality

api/api/crud.py

@@ -1,72 +1,118 @@
+import json
+from typing import Sequence, Optional
+
+from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from . import models, schemas
 
-import json
+
+def get_session(db: Session, session_id: int, user_id: Optional[str] = None):
+    stmt = select(models.Session).where(models.Session.id == session_id)
+    if user_id is not None:
+        stmt = stmt.where(models.Session.user_id == user_id)
+    return db.scalars(stmt).one_or_none()
+    # return db.query(models.Session).filter(models.Session.id == session_id).first()
 
 
-def get_session(db: Session, session_id: int):
-    return db.query(models.Session).filter(models.Session.id == session_id).first()
+def get_sessions(db: Session, user_id: str, location_id: str | None = None) -> Sequence[schemas.Session]:
+    stmt = select(models.Session).where(models.Session.user_id == user_id).where(models.Session.is_active.is_(True))
 
+    if location_id is not None:
+        stmt = stmt.where(models.Session.location_id == location_id)
 
-def get_sessions(db: Session, user_id: str, location_id: str | None = None):
-    filtered_by_user = db.query(models.Session).filter(
-        models.Session.user_id == user_id
-    )
-    filtered_by_location = (
-        filtered_by_user.filter(models.Session.location_id == location_id)
-        if location_id is not None
-        else filtered_by_user
-    )
-    return (
-        filtered_by_location.filter(models.Session.is_active == True)
-        .order_by(models.Session.created_at.desc())
-        .all()
-    )
+    return db.scalars(stmt).all()
 
+    # filtered_by_user = db.query(models.Session).filter(
+    #     models.Session.user_id == user_id
+    # )
+    # filtered_by_location = (
+    #     filtered_by_user.filter(models.Session.location_id == location_id)
+    #     if location_id is not None
+    #     else filtered_by_user
+    # )
+    # return (
+    #     filtered_by_location.filter(models.Session.is_active.is_(True))
+    #     .order_by(models.Session.created_at.desc())
+    #     .all()
+    # )
 
-def create_session(db: Session, user_id: str, session: schemas.SessionCreate):
-    db_session = models.Session(
+def create_session(db: Session, user_id: str, session: schemas.SessionCreate) -> models.Session:
+    honcho_session = models.Session(
         user_id=user_id,
         location_id=session.location_id,
         session_data=json.dumps(session.session_data),
     )
-    db.add(db_session)
+    db.add(honcho_session)
     db.commit()
-    db.refresh(db_session)
-    return db_session
+    db.refresh(honcho_session)
+    return honcho_session
 
-# TODO add a method to update a session and change the active state
+def update_session(db: Session, user_id: str, session_id: int, metadata: dict) -> bool:
+    # stmt = select(models.Session).where(models.Session.id == session_id).where(models.Session.user_id == user_id)
+    # honcho_session = db.scalars(stmt).one_or_none()
+    honcho_session = get_session(db, session_id=session_id, user_id=user_id)
+    # honcho_session = db.get(models.Session, session_id)
+    if honcho_session is None:
+        raise ValueError("Session not found or does not belong to user")
+    honcho_session.metadata = metadata
+    db.commit()
+    db.refresh(honcho_session)
+    return honcho_session
 
-def create_message(db: Session, message: schemas.MessageCreate, session_id: int):
-    db_message = models.Message(
+def delete_session(db: Session, user_id: str, session_id: int) -> bool:
+    stmt = select(models.Session).where(models.Session.id == session_id).where(models.Session.user_id == user_id)
+    honcho_session = db.scalars(stmt).one_or_none()
+    # honcho_session = db.get(models.Session, session_id)
+    if honcho_session is None:
+        return False
+    honcho_session.is_active = False
+    db.commit()
+    return True
+
+def create_message(db: Session, message: schemas.MessageCreate, user_id: str, session_id: int) -> models.Message:
+    honcho_session = get_session(db, session_id, user_id)
+    if honcho_session is None:
+        raise ValueError("Session not found or does not belong to user")
+
+    honcho_message = models.Message(
         session_id=session_id,
         is_user=message.is_user,
         content=message.content,
     )
-    db.add(db_message)
+    db.add(honcho_message)
     db.commit()
-    db.refresh(db_message)
-    return db_message
+    db.refresh(honcho_message)
+    return honcho_message
 
+def get_messages(db: Session, user_id: str, session_id: int) -> Sequence[schemas.Message]:
+    session = get_session(db, session_id, user_id)
+    if session is None:
+        raise ValueError("Session not found or does not belong to user")
+    stmt = select(models.Message).where(models.Message.session_id == session_id)
+    return db.scalars(stmt).all()
+    # return (
+    #     db.query(models.Message)
+    #     .filter(models.Message.session_id == session_id)
+    #     .all()
+    # )
 
-def get_metacognitions(db: Session, message_id: int):
-    return (
-        db.query(models.Metacognitions)
-        .filter(models.Metacognitions.message_id == message_id)
-        .all()
-    )
+# def get_metacognitions(db: Session, message_id: int):
+#     return (
+#         db.query(models.Metacognitions)
+#         .filter(models.Metacognitions.message_id == message_id)
+#         .all()
+#     )
 
-
-def create_metacognition(
-    db: Session, metacognition: schemas.MetacognitionsCreate, message_id: int
-):
-    db_metacognition = models.Metacognitions(
-        message_id=message_id,
-        metacognition_type=metacognition.metacognition_type,
-        content=metacognition.content,
-    )
-    db.add(db_metacognition)
-    db.commit()
-    db.refresh(db_metacognition)
-    return db_metacognition
+# def create_metacognition(
+#     db: Session, metacognition: schemas.MetacognitionsCreate, message_id: int
+# ):
+#     honcho_metacognition = models.Metacognitions(
+#         message_id=message_id,
+#         metacognition_type=metacognition.metacognition_type,
+#         content=metacognition.content,
+#     )
+#     db.add(honcho_metacognition)
+#     db.commit()
+#     db.refresh(honcho_metacognition)
+#     return honcho_metacognition

api/api/main.py

@@ -1,14 +1,16 @@
 from fastapi import Depends, FastAPI, HTTPException
 from sqlalchemy.orm import Session
+import uvicorn
 
 from . import crud, models, schemas
 from .db import SessionLocal, engine
 
-models.Base.metadata.create_all(bind=engine)
+models.Base.metadata.create_all(bind=engine) # Scaffold Database if not already done
 
 app = FastAPI()
 
 def get_db():
+    """FastAPI Dependency Generator for Database"""
     db = SessionLocal()
     try:
         yield db
@@ -30,39 +32,46 @@ def create_session(
 ):
     return crud.create_session(db, user_id, session)
 
-@app.put("/users/{user_id}/sessions/{session_id}")
+@app.put("/users/{user_id}/sessions/{session_id}", response_model=schemas.Session)
 def update_session(
     user_id: str,
     session_id: int,
-    session: schemas.SessionCreate,
+    session: schemas.SessionUpdate,
     db: Session = Depends(get_db),
     ):
-    # TODO finish implementation
-    pass
+    if session.session_data is None:
+        raise HTTPException(status_code=400, detail="Session data cannot be empty") # TODO TEST if I can set the metadata to be blank with this 
+    try:
+        return crud.update_session(db, user_id=user_id, session_id=session_id, metadata=session.session_data) 
+    except ValueError:
+        raise HTTPException(status_code=404, detail="Session not found")
 
 @app.delete("/users/{user_id}/sessions/{session_id}")
 def delete_session(
     user_id: str,
     session_id: int,
-    session: schemas.SessionCreate,
     db: Session = Depends(get_db),
     ):
-    # TODO finish implementation
-    pass
+    response = crud.delete_session(db, user_id=user_id, session_id=session_id)
+    if response:
+        return {"message": "Session deleted successfully"}
+    else:
+        raise HTTPException(status_code=404, detail="Session not found")
 
 @app.get("/users/{user_id}/sessions/{session_id}", response_model=schemas.Session)
 def get_session(user_id: str, session_id: int, db: Session = Depends(get_db)):
-    db_session = crud.get_session(db, session_id)
-    if db_session is None:
+    honcho_session = crud.get_session(db, session_id=session_id, user_id=user_id)
+    if honcho_session is None:
         raise HTTPException(status_code=404, detail="Session not found")
-    return db_session
+    return honcho_session
 
 ########################################################
 # Message Routes
 ########################################################
 
 @app.post(
-    "/users/{user_id}/sessions/{session_id}/messages/", response_model=schemas.Message
+    "/users/{user_id}/sessions/{session_id}/messages/",
+    response_model=schemas.Message
 )
 def create_message_for_session(
     user_id: str,
@@ -70,34 +79,54 @@ def create_message_for_session(
     message: schemas.MessageCreate,
     db: Session = Depends(get_db),
 ):
-    return crud.create_message(db, message, session_id)
+    try:
+        return crud.create_message(db, message=message, user_id=user_id, session_id=session_id)
+    except ValueError:
+        raise HTTPException(status_code=404, detail="Session not found")
+
+@app.get(
+    "/users/{user_id}/sessions/{session_id}/messages/", 
+    response_model=list[schemas.Message]
+)
+def get_messages_for_session(
+        user_id: str,
+        session_id: int,
+        db: Session = Depends(get_db),
+        ):
+    try: 
+        return crud.get_messages(db, user_id=user_id, session_id=session_id)
+    except ValueError:
+        raise HTTPException(status_code=404, detail="Session not found")
 
 ########################################################
 # Metacognition Routes
 ########################################################
 
-@app.get(
-    "/users/{user_id}/sessions/{session_id}/messages/{message_id}/metacognitions/",
-    response_model=list[schemas.Metacognitions],
-)
-def get_metacognitions_for_message(
-    user_id: str,
-    session_id: int,
-    message_id: int,
-    db: Session = Depends(get_db),
-):
-    return crud.get_metacognitions(db, message_id)
+# @app.get(
+#     "/users/{user_id}/sessions/{session_id}/messages/{message_id}/metacognitions/",
+#     response_model=list[schemas.Metacognitions],
+# )
+# def get_metacognitions_for_message(
+#     user_id: str,
+#     session_id: int,
+#     message_id: int,
+#     db: Session = Depends(get_db),
+# ):
+#     return crud.get_metacognitions(db, message_id)
 
 
-@app.post(
-    "/users/{user_id}/sessions/{session_id}/messages/{message_id}/metacognitions/",
-    response_model=schemas.Metacognitions,
-)
-def create_metacognition_for_message(
-    user_id: str,
-    session_id: int,
-    message_id: int,
-    metacognition: schemas.MetacognitionsCreate,
-    db: Session = Depends(get_db),
-):
-    return crud.create_metacognition(db, metacognition, message_id)
+# @app.post(
+#     "/users/{user_id}/sessions/{session_id}/messages/{message_id}/metacognitions/",
+#     response_model=schemas.Metacognitions,
+# )
+# def create_metacognition_for_message(
+#     user_id: str,
+#     session_id: int,
+#     message_id: int,
+#     metacognition: schemas.MetacognitionsCreate,
+#     db: Session = Depends(get_db),
+# ):
+#     return crud.create_metacognition(db, metacognition, message_id)
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=8000)

api/api/models.py

@@ -21,6 +21,9 @@ class Session(Base):
     created_at: Mapped[datetime.datetime] = mapped_column(default=datetime.datetime.utcnow)
     messages = relationship("Message", back_populates="session")
 
+    def __repr__(self) -> str:
+        return f"Session(id={self.id}, user_id={self.user_id}, location_id={self.location_id}, is_active={self.is_active}, created_at={self.created_at})"
+
 
 class Message(Base):
     __tablename__ = "messages"
@@ -35,6 +38,8 @@ class Message(Base):
 
     session = relationship("Session", back_populates="messages")
     metacognitions = relationship("Metacognitions", back_populates="message")
+    def __repr__(self) -> str:
+        return f"Message(id={self.id}, session_id={self.session_id}, is_user={self.is_user}, content={self.content[10:]})"
 
 
 # TODO: add metacognitive data to messages
@@ -49,3 +54,6 @@ class Metacognitions(Base):
     # content = Column(String)
 
     message = relationship("Message", back_populates="metacognitions")
+
+    def __repr__(self) -> str:
+        return f"Metacognitions(id={self.id}, message_id={self.message_id}, metacognition_type={self.metacognition_type}, content={self.content[10:]})"

api/api/schemas.py

@@ -27,6 +27,10 @@ class SessionCreate(SessionBase):
     session_data: dict | None = None
 
 
+class SessionUpdate(SessionBase):
+    session_data: dict | None = None
+
+
 class Session(SessionBase):
     id: int
     messages: list[Message]