from bson import ObjectId import bson from fastapi import APIRouter, HTTPException, status from fastapi.params import Depends import pymongo from app.dto import PinDTO, PinShareDTO from app.models import HTTPError from app.models.user import User from .utils import get_current_user, objectid_misformatted import app.config as config # Best workaround found for _id typed as ObjectId (creating Exception bcause JSON doesn't support custom types countrary to BSON, used by Mongo) # also allows to create DTOs at the time, but not at it's best (project structure is chaotic FTM :s) import app.serializers as serializers # Import all serializers (detailed in __init__.py) # Database setup client = pymongo.MongoClient(config.MONGODB_URL, username=config.MONGODB_USERNAME, password=config.MONGODB_PASSWORD) db = client[config.MONGODB_DATABASE] pins_collection = db["pins"] friends_collection = db["friends"] pin_permissions_collection = db["pin_permissions"] pins_router = APIRouter( prefix="/pin", tags=["Pins"] ) @pins_router.get( path="/{id}", responses={401: {"model": HTTPError}, 404: {"model": HTTPError}, 422: {"model": HTTPError}} ) async def get_pin(id: str, current_user: User = Depends(get_current_user)): try: pin = pins_collection.find_one({"_id": ObjectId(id)}) if pin is None: raise HTTPException(status_code=404, detail="Pin not found") # Vérifier si l'utilisateur a la permission de voir le pin if pin["user_id"] != current_user.uid: permission = pin_permissions_collection.find_one({ "pin_id": ObjectId(id), "user_id": current_user.uid }) if not permission: raise HTTPException(status_code=403, detail="You don't have permission to view this pin") return serializers.pin_serialize(pin) except bson.errors.InvalidId: objectid_misformatted() @pins_router.patch( path="/{id}", responses={401: {"model": HTTPError}, 404: {"model": HTTPError}, 422: {"model": HTTPError}, 403: {"model": HTTPError}} ) async def update_pin(id: str, pin: PinDTO, current_user: User = Depends(get_current_user)): try: # Vérifier si le pin existe existing_pin = pins_collection.find_one({"_id": ObjectId(id)}) if existing_pin is None: raise HTTPException(status_code=404, detail="Pin not found") # Vérifier si l'utilisateur a la permission de modifier le pin if existing_pin["user_id"] != current_user.uid: permission = pin_permissions_collection.find_one({ "pin_id": ObjectId(id), "user_id": current_user.uid, "can_edit": True }) if not permission: raise HTTPException(status_code=403, detail="You don't have permission to edit this pin") # Mettre à jour le pin result = pins_collection.update_one({"_id": ObjectId(id)}, {"$set": pin.model_dump()}) return {"message": "Pin updated"} except bson.errors.InvalidId: objectid_misformatted() @pins_router.post( path="/add", responses={401: {"model": HTTPError}} ) async def add_pin(pin: PinDTO, current_user: User = Depends(get_current_user)): pin.user_id = current_user.uid pin_id = pins_collection.insert_one(pin.model_dump()).inserted_id return {"id": str(pin_id)} @pins_router.get( path="s", responses={401: {"model": HTTPError}} ) async def list_pins(current_user: User = Depends(get_current_user)): # Récupérer les IDs des pins partagés avec l'utilisateur shared_pins = pin_permissions_collection.find({"user_id": current_user.uid}) shared_pin_ids = [permission["pin_id"] for permission in shared_pins] # Récupérer tous les pins de l'utilisateur et les pins partagés avec lui pins = pins_collection.find({ "$or": [ {"user_id": current_user.uid}, # Pins de l'utilisateur {"_id": {"$in": shared_pin_ids}} # Pins partagés avec l'utilisateur ] }) return serializers.pins_serialize(pins.to_list()) @pins_router.post( path="/{id}/share", responses={401: {"model": HTTPError}, 404: {"model": HTTPError}, 422: {"model": HTTPError}, 403: {"model": HTTPError}} ) async def share_pin(id: str, share_data: PinShareDTO, current_user: User = Depends(get_current_user)): try: # Vérifier si le pin existe et appartient à l'utilisateur courant pin = pins_collection.find_one({"_id": ObjectId(id)}) if pin is None: raise HTTPException(status_code=404, detail="Pin not found") if pin["user_id"] != current_user.uid: raise HTTPException(status_code=403, detail="You can only share your own pins") # Vérifier si l'utilisateur est ami avec la personne avec qui il veut partager friend = friends_collection.find_one({ "$or": [ {"user_id": current_user.uid, "friend_user_id": share_data.friend_id, "status": "accepted"}, {"user_id": share_data.friend_id, "friend_user_id": current_user.uid, "status": "accepted"} ] }) if friend is None: raise HTTPException( status_code=403, detail="You can only share pins with your friends" ) # Vérifier si la permission existe déjà existing_permission = pin_permissions_collection.find_one({ "pin_id": ObjectId(id), "user_id": share_data.friend_id }) if existing_permission: raise HTTPException( status_code=409, detail="Pin is already shared with this user" ) # Créer la permission permission = { "pin_id": ObjectId(id), "user_id": share_data.friend_id, "can_edit": True, "can_delete": False } pin_permissions_collection.insert_one(permission) return {"message": "Pin shared successfully"} except bson.errors.InvalidId: objectid_misformatted() @pins_router.delete( path="/{id}", responses={401: {"model": HTTPError}, 404: {"model": HTTPError}, 422: {"model": HTTPError}, 403: {"model": HTTPError}} ) async def delete_pin(id: str, current_user: User = Depends(get_current_user)): try: pin = pins_collection.find_one({"_id": ObjectId(id)}) if pin is None: raise HTTPException(status_code=404, detail="Pin not found") # Vérifier si l'utilisateur est le propriétaire du pin if pin["user_id"] != current_user.uid: raise HTTPException(status_code=403, detail="Only the owner can delete the pin") # Supprimer le pin et toutes ses permissions pins_collection.delete_one({"_id": ObjectId(id)}) pin_permissions_collection.delete_many({"pin_id": ObjectId(id)}) return {"message": "Pin deleted successfully"} except bson.errors.InvalidId: objectid_misformatted()