Realizacao do merge da branch 9_lista_favoritos 11-historico

US11 - Eu, como usuário, quero ver o histórico de vídeos que assisti recentemente, para encontrar facilmente vídeos que vi anteriormente.

src/controller/recordController.py

@@ -0,0 +1,25 @@
+from fastapi import APIRouter, HTTPException, Depends, Query
+from sqlalchemy.orm import Session
+from domain import recordSchema
+from database import get_db
+from repository import recordRepository
+from starlette.responses import JSONResponse
+
+Record = APIRouter(
+ prefix="/record"
+)
+
+@Record.post("/")
+def add_to_record(record: recordSchema.RecordCreate, db: Session = Depends(get_db)):
+   record_user = recordSchema.RecordGet(user_id =record.user_id)
+   videos = recordRepository.get_record(db = db, record = record_user)
+   if videos:
+      return recordRepository.create_record(db=db, record=record, is_create= False)
+
+   return recordRepository.create_record(db=db, record=record, is_create= True)
+
+@Record.get("/get_record")
+def check_record(user_id: str =Query(...) , db: Session = Depends(get_db)):
+   record = recordSchema.RecordGet(user_id =user_id)
+   videos = recordRepository.get_record(db=db, record=record)
+   return {"videos": videos}

src/domain/recordSchema.py

@@ -0,0 +1,12 @@
+from pydantic import BaseModel
+from typing import Dict
+
+class RecordBase(BaseModel):
+    user_id: str
+    videos: Dict[str, str]
+
+class RecordCreate(RecordBase):
+    pass
+
+class RecordGet(BaseModel):
+    user_id: str

src/main.py

@@ -8,7 +8,7 @@
 
 load_dotenv()
 
-from controller import commentController, scheduleController, savedVideosController
+from controller import commentController, scheduleController, savedVideosController, recordController
 from controller.savedVideosController import WatchLater
 
 
@@ -38,6 +38,7 @@
 #app.include_router(prefix="/api", router=commentController.comment)
 app.include_router(prefix="/api", router=scheduleController.schedule)
 app.include_router(prefix="/api", router=savedVideosController.favorite)
+app.include_router(prefix="/api", router=recordController.Record)
 
 @app.get("/")
 async def root():

src/model/recordModel.py

@@ -0,0 +1,10 @@
+import uuid
+from sqlalchemy import Column, String, JSON
+from database import Base
+
+class Record(Base):
+   __tablename__ = 'record'
+   id = Column(String, primary_key=True, index=True, default=lambda: str(uuid.uuid4()))
+   user_id = Column(String, index=True, nullable=False)
+   videos = Column(JSON, nullable=False)
+

src/repository/recordRepository.py

@@ -0,0 +1,40 @@
+from sqlalchemy.orm import Session
+
+from domain import recordSchema
+from model import recordModel
+from fastapi import HTTPException 
+
+def create_record(db: Session, record: recordSchema.RecordCreate, is_create):
+   if is_create == True:
+    db_record = recordModel.Record(
+        user_id=record.user_id.strip(),
+        videos=record.videos
+    )
+    db.add(db_record)
+   else:
+    db_record = db.query(recordModel.Record).filter(recordModel.Record.user_id == record.user_id).first()
+
+    id_video = list(record.videos.keys())[0]
+    timestamp_video = list(record.videos.values())[0]
+
+    video_append = db_record.videos.copy()
+    video_append[id_video] = timestamp_video
+
+    db_record.videos = video_append
+
+   db.commit()
+   db.refresh(db_record)
+   print(f"Created/Updated record: user_id={db_record.user_id}")
+   return db_record
+
+def get_record(db: Session, record: recordSchema.RecordGet):
+   user_id = record.user_id.strip()
+   record_entry = db.query(recordModel.Record).filter(
+       recordModel.Record.user_id == user_id
+   ).first()
+   print(f"Query Result: {record_entry}")
+   if record_entry:
+       print(f"Check Record Founds")
+       return record_entry.videos
+   print(f"Check Record Not Found")
+   return False

tests/test_record.py

@@ -0,0 +1,44 @@
+import pytest, sys, os
+
+sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'src')))
+
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from src.database import Base, get_db
+from src.main import app
+
+# Crie um banco de dados de teste em memória
+SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
+engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
+TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+# Dependência para usar o banco de dados de teste
+def override_get_db():
+    try:
+        db = TestingSessionLocal()
+        yield db
+    finally:
+        db.close()
+
+app.dependency_overrides[get_db] = override_get_db
+
+client = TestClient(app)
+
+@pytest.fixture(scope="module")
+def setup_database():
+    Base.metadata.create_all(bind=engine)
+    yield
+    Base.metadata.drop_all(bind=engine)
+
+def test_add_to_record(setup_database):
+    response = client.post("/api/record/", json={"user_id": "user123", "videos": {"video_id": "timestamp"}})
+    assert response.status_code == 200
+    assert response.json()["user_id"] == "user123"
+    assert response.json()["videos"] == {"video_id": "timestamp"}
+
+def test_check_record(setup_database):
+    response = client.get("/api/record/get_record/", params={"user_id": "user123"})
+    assert response.status_code == 200
+    assert "videos" in response.json()
+    assert response.json()["videos"] == {"video_id": "timestamp"}