documentation test

requirements.txt

@@ -1,7 +1,7 @@
-python-dotenv~=1.0.0
-pymongo~=4.4.1
-pydantic~=1.10.12
-fastapi~=0.101.1
+python-dotenv
+pymongo
+pydantic
+fastapi
 uvicorn~=0.23.2
 werkzeug
 starlette

src/__main__.py

@@ -18,7 +18,7 @@
 from src.domain.user import User
 # Imported routes
 from src.routes import index, blog, email, login, user, experiences, links, register, contact, projects, newsletter, \
-    subscriber, comments, github, book, technology
+    subscriber, comments, github, book, technology, media
 from src.services import db
 from src.tags_metadata import tags_metadata
 from src.utils.domain_to_txt import write_fields_to_txt
@@ -36,6 +36,7 @@
 
 app.include_router(index.router, prefix='/index', tags=['Index'])
 app.include_router(blog.router, prefix='/blog', tags=['Blog'])
+app.include_router(media.router, prefix='/media', tags=['Media'])
 app.include_router(github.router, prefix='/github', tags=['Github'])
 app.include_router(book.router, prefix='/book', tags=['Book'])
 app.include_router(technology.router, prefix='/technology', tags=['Technology'])

src/routes/book.py

@@ -1,198 +1,227 @@
-"""
-Routes Overview:
-1. GET / - Retrieve all books from the database.
-2. GET /{_id} - Retrieve a book by its ID.
-3. POST / - Add a new book to the database.
-4. PUT /{_id} - Edit an existing book by its ID.
-5. DELETE /{_id} - Delete a book by its ID.
-"""
-
-from fastapi import APIRouter, Depends, HTTPException
-
+import os
+from fastapi import APIRouter, Depends, HTTPException, UploadFile, File
+from fastapi.responses import FileResponse
 from src.domain.book import Book
 from src.services import db
-
 from src.services.security import get_current_user
 
+books_media = 'media/books_media'
+
 router = APIRouter()
 
 """
-THIS ROUTES ARE PUBLIC
+Public Routes:
+1. GET / - Retrieve all books from the database.
+2. GET /{_id} - Retrieve a book by its ID.
 """
 
 
-# Get all the book from database
+# Get all books from the database
 @router.get('/', operation_id='get_all_book_public')
 async def get_all_book_public() -> list[Book]:
     """
-    This route handles the retrieval of all the book from the database
+    Retrieve all books from the database.
 
-    :return: a list of Book objects containing all the book in the database
+    :return: A list of Book objects containing all the books in the database.
     """
-
-    # Retrieve all blogs from the database using the find method
     cursor = db.process.book.find()
-
-    # Create a list of Blog objects by unpacking data from each document retrieved
     book_list = [Book(**document) for document in cursor]
-
-    # Return the list of Blog objects
     return book_list
 
 
-# Get book by its ID
+# Get a book by its ID
 @router.get('/{_id}', operation_id='get_book_by_id_public')
 async def get_book_by_id_public(_id: str):
     """
-    This route handles the retrieval of one book by its ID from the database
+    Retrieve a book by its ID from the database.
 
-    :param _id: The ID of the book to be retrieved
-    :return: If the book is found, returns the book data; otherwise, returns a 404 error
+    :param _id: The ID of the book to be retrieved.
+    :return: The Book object if found; otherwise, raises a 404 error.
     """
-
-    # Attempt to find book in the database based on the provided ID
     cursor = db.process.book.find_one({'_id': _id})
-
-    # If no book is found, return a 404 error with a relevant detail message
     if cursor is None:
         raise HTTPException(status_code=404, detail=f'Book by ID: ({_id}) does not exist')
-    else:
-        # If the book is found, convert the cursor data into a Book object and return it
-        return Book(**cursor)
+    return Book(**cursor)
 
 
 """
-THIS ROUTES ARE PRIVATE
+Private Routes:
+1. GET /admin/ - Retrieve all books from the database (admin).
+2. GET /admin/{_id} - Retrieve a book by its ID (admin).
+3. POST / - Add a new book to the database (admin).
+4. PUT /{_id} - Edit an existing book by its ID (admin).
+5. DELETE /{_id} - Delete a book by its ID (admin).
 """
 
 
-# Get all the book from database
+# Get all books from the database (admin only)
 @router.get('/admin/', operation_id='get_all_book_private')
 async def get_all_book_private(current_user: str = Depends(get_current_user)) -> list[Book]:
     """
-    This route handles the retrieval of all the book from the database
+    Retrieve all books from the database. Admin access only.
 
-    :return: a list of Book objects containing all the book in the database
+    :param current_user: The currently authenticated user.
+    :return: A list of Book objects containing all the books in the database.
     """
-
-    # Retrieve all blogs from the database using the find method
     cursor = db.process.book.find()
-
-    # Create a list of Blog objects by unpacking data from each document retrieved
     book_list = [Book(**document) for document in cursor]
-
-    # Return the list of Blog objects
     return book_list
 
 
-# Get book by its ID
+# Get a book by its ID (admin only)
 @router.get('/admin/{_id}', operation_id='get_book_by_id_private')
 async def get_book_by_id_private(_id: str, current_user: str = Depends(get_current_user)):
     """
-    This route handles the retrieval of one book by its ID from the database
+    Retrieve a book by its ID from the database. Admin access only.
 
-    :param current_user: Current user that is registered
-    :param _id: The ID of the book to be retrieved
-    :return: If the book is found, returns the book data; otherwise, returns a 404 error
+    :param _id: The ID of the book to be retrieved.
+    :param current_user: The currently authenticated user.
+    :return: The Book object if found; otherwise, raises a 404 error.
     """
-
-    # Attempt to find book in the database based on the provided ID
     cursor = db.process.book.find_one({'_id': _id})
-
-    # If no book is found, return a 404 error with a relevant detail message
     if cursor is None:
         raise HTTPException(status_code=404, detail=f'Book by ID: ({_id}) does not exist')
-    else:
-        # If the book is found, convert the cursor data into a Book object and return it
-        return Book(**cursor)
+    return Book(**cursor)
 
 
-# This route adds a new book
+# Add a new book to the database (admin only)
 @router.post('/', operation_id='add_new_book_private')
-async def add_new_book_private(book: Book,
-                                      current_user: str = Depends(get_current_user)) -> Book | None:
+async def add_new_book_private(book: Book, current_user: str = Depends(get_current_user)) -> Book | None:
     """
-    Handles the addition of a new book to the database.
+    Add a new book to the database. Admin access only.
 
     :param book: The Book object representing the new book to be added.
-    :param current_user: The current user, obtained from the authentication system.
-    :return: If the addition is successful, returns the newly added Book object; otherwise, returns None.
+    :param current_user: The currently authenticated user.
+    :return: The newly added Book object if successful; otherwise, returns None.
     """
-
-    # Convert the Book object to a dictionary
     book_dict = book.dict(by_alias=True)
-
-    # Insert book data into database
     insert_result = db.process.book.insert_one(book_dict)
-
-    # Check if the insertion was acknowledged by the database
     if insert_result.acknowledged:
-
-        # Update the dictionary with the newly assigned _id
         book_dict['_id'] = str(book_dict['_id'])
-
-        # Return the newly added Book object
         return Book(**book_dict)
-    else:
-
-        # If the insertion was not acknowledged, return None
-        return None
+    return None
 
 
-# Edit book by its ID
+# Edit an existing book by its ID (admin only)
 @router.put('/{_id}', operation_id='edit_book_by_id_private')
-async def edit_book_by_id_private(_id: str, book: Book,
-                                         current_user: str = Depends(get_current_user)):
+async def edit_book_by_id_private(_id: str, book: Book, current_user: str = Depends(get_current_user)):
     """
-    Handles the editing of book by its ID in the database.
+    Edit an existing book by its ID in the database. Admin access only.
 
     :param _id: The ID of the book to be edited.
     :param book: The updated Book object with the new data.
-    :param current_user: The current user, obtained from the authentication system.
-    :return: If the book are successfully edited, returns the updated Book object; otherwise, returns None.
+    :param current_user: The currently authenticated user.
+    :return: The updated Book object if successful; otherwise, returns None.
     """
-
-    # Convert the Book object to a dictionary
     book_dict = book.dict(by_alias=True)
-
-    # Delete the '_id' field from the book dictionary to avoid updating the ID
     del book_dict['_id']
-
-    # Update the book in the database using the update_one method
     cursor = db.process.book.update_one({'_id': _id}, {'$set': book_dict})
-
-    # Check if the book were successfully updated
     if cursor.modified_count > 0:
-        # Retrieve the updated book from the database
         updated_document = db.process.book.find_one({'_id': _id})
-
-        # Check if the updated book exist
         if updated_document:
             updated_document['_id'] = str(updated_document['_id'])
             return Book(**updated_document)
-
-    # Return None if the book were not updated
     return None
 
 
-# Delete book by ID
+# Delete a book by its ID (admin only)
 @router.delete("/{_id}", operation_id='delete_book_by_id_private')
 async def delete_book_by_id_private(_id: str, current_user: str = Depends(get_current_user)):
     """
-    Handles the deletion of book by its ID from the database.
+    Delete a book by its ID from the database. Admin access only.
 
     :param _id: The ID of the book to be deleted.
-    :param current_user: The current user, obtained from the authentication system.
-    :return: If the book are successfully deleted, returns a success message; otherwise, raises a 404 error.
+    :param current_user: The currently authenticated user.
+    :return: A success message if the book is deleted; otherwise, raises a 404 error.
     """
-
-    # Attempt to delete the book from the database using the delete_one method
     delete_results = db.process.book.delete_one({'_id': _id})
-
-    # Check if the book were successfully deleted
     if delete_results.deleted_count > 0:
-        # Return a success message if the book were found and deleted
-        return {'message': 'Experience deleted successfully'}
-    else:
-        # If the book were not found, raise a 404 error
-        raise HTTPException(status_code=404, detail=f'Book with ID: ({_id}) not found!')
+        return {'message': 'Book deleted successfully'}
+    raise HTTPException(status_code=404, detail=f'Book with ID: ({_id}) not found!')
+
+
+"""
+Media Routes:
+1. POST / - Upload a media file.
+2. GET /{filename} - Retrieve a media file by filename.
+3. GET /images/ - List all media files.
+4. DELETE /{filename} - Delete a media file by filename.
+"""
+
+
+# Upload a media file
+@router.post("/")
+async def upload_media_file(file: UploadFile = File(...)):
+    """
+    Upload a media file to the server.
+
+    :param file: The file to be uploaded.
+    :return: A success message indicating the file was uploaded.
+    """
+    try:
+        upload_directory = books_media
+        os.makedirs(upload_directory, exist_ok=True)
+        contents = await file.read()
+        file_name = file.filename if file.filename else 'uploaded_file'
+        file_path = os.path.join(upload_directory, file_name)
+        with open(file_path, 'wb') as f:
+            f.write(contents)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f'There was an error uploading the file: {str(e)}')
+    finally:
+        await file.close()
+    return {"message": f"Successfully uploaded {file_name} to {upload_directory}"}
+
+
+# Retrieve a media file by filename
+@router.get('/{filename}')
+async def get_book_image(filename: str):
+    """
+    Retrieve a media file by filename.
+
+    :param filename: The name of the file to be retrieved.
+    :return: The file if found; otherwise, raises a 404 error.
+    """
+    upload_directory = books_media
+    file_path = os.path.join(upload_directory, filename)
+    if not os.path.exists(file_path):
+        raise HTTPException(status_code=404, detail='Image not found!')
+    try:
+        return FileResponse(file_path)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f'Error serving file: {str(e)}')
+
+
+# List all media files
+@router.get('/images/')
+async def list_book_images():
+    """
+    List all media files in the upload directory.
+
+    :return: A list of filenames in the upload directory.
+    """
+    upload_directory = books_media
+    if not os.path.exists(upload_directory):
+        raise HTTPException(status_code=404, detail='Upload directory not found!')
+    image_names = [f for f in os.listdir(upload_directory) if os.path.isfile(os.path.join(upload_directory, f))]
+    return {"images": image_names}
+
+
+# Delete a media file by filename
+@router.delete("/{filename}")
+async def delete_book_image(filename: str):
+    """
+    Delete a media file from the upload directory.
+
+    :param filename: The name of the file to be deleted.
+    :return: A success message if the file is deleted; otherwise, raises a 404 error.
+    """
+    upload_directory = books_media
+    file_path = os.path.join(upload_directory, filename)
+    if not os.path.exists(file_path):
+        raise HTTPException(status_code=404, detail="Image not found")
+    try:
+        os.remove(file_path)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"There was an error deleting the file: {str(e)}")
+    return {"message": f"Successfully deleted {filename} from {upload_directory}"}