feat: add quote routes and service

Author: zobweyt

src/api/v1/__init__.py

@@ -1,15 +1,16 @@
 from fastapi import APIRouter
 
-from src.api.v1 import auth, authors, collections, media, otp, users
+from src.api.v1 import auth, authors, collections, media, otp, quotes, users
 
 v1_router = APIRouter(prefix="/v1")
 
 v1_router.include_router(auth.router)
 v1_router.include_router(authors.router)
 v1_router.include_router(collections.router)
+v1_router.include_router(media.router)
 v1_router.include_router(otp.router)
+v1_router.include_router(quotes.router)
 v1_router.include_router(users.router)
-v1_router.include_router(media.router)
 
 __all__ = [
     "v1_router",

src/api/v1/auth/__init__.py

@@ -1,12 +1,9 @@
 from src.api.v1.auth.deps import OAuth2BearerDepends, OAuth2PasswordRequestFormDepends, OptionalOAuth2BearerDepends
 from src.api.v1.auth.routes import router
-from src.api.v1.auth.schemas import JWT, AccessTokenResponse
 
 __all__ = [
     "router",
     "OAuth2BearerDepends",
     "OptionalOAuth2BearerDepends",
     "OAuth2PasswordRequestFormDepends",
-    "AccessTokenResponse",
-    "JWT",
 ]

src/api/v1/authors/routes.py

@@ -11,7 +11,7 @@
 
 @router.get("/{name}", response_model=AuthorResponse)
 def get_author(name: str, service: AuthorServiceDepends):
-    author = service.get_author(name)
+    author = service.get_author_by_name(name)
 
     if not author:
         raise HTTPException(status.HTTP_404_NOT_FOUND, _("No author found with the name '%s'." % (name,)))

src/api/v1/authors/service.py

@@ -2,16 +2,19 @@
 
 from src.api.params import SearchParams
 from src.api.v1.authors.models import Author
-from src.db.deps import Session
+from src.db.deps import SessionDepends
 
 
 class AuthorService:
-    def __init__(self, session: Session) -> None:
+    def __init__(self, session: SessionDepends) -> None:
         self.session = session
 
-    def get_author(self, name: str) -> Author | None:
+    def get_author_by_name(self, name: str) -> Author | None:
         return self.session.query(Author).filter(Author.name == name).first()
 
+    def get_author_by_id(self, author_id: int) -> Author | None:
+        return self.session.get(Author, author_id)
+
     def get_authors(self, search_params: SearchParams) -> list[Author]:
         query = self.session.query(Author)
 

src/api/v1/collections/routes.py

@@ -4,6 +4,8 @@
 from src.api.v1.collections.deps import CollectionServiceDepends
 from src.api.v1.collections.models import Collection
 from src.api.v1.collections.schemas import CollectionCreateRequest, CollectionResponse, CollectionUpdateRequest
+from src.api.v1.quotes.deps import QuoteServiceDepends
+from src.api.v1.quotes.schemas import QuoteCollectionsResponse, QuoteResponse
 from src.api.v1.users.me.deps import CurrentUser, CurrentUserOrNone
 from src.i18n import gettext as _
 
@@ -70,7 +72,71 @@ def delete_collection(id: int, current_user: CurrentUser, service: CollectionSer
     service.delete_collection(collection)
 
 
-# TODO:
-# @router.get("/{collection_id}/quotes")
-# @router.post("/{collection_id}/quotes")
-# @router.delete("/{collection_id}/quotes/{quote_id}")
+@router.get("/{collection_id}/quotes", response_model=list[QuoteResponse])
+def get_collection_quotes(
+    collection_id: int,
+    search_params: SearchParamsDepends,
+    current_user: CurrentUserOrNone,
+    collection_service: CollectionServiceDepends,
+    quote_service: QuoteServiceDepends,
+):
+    collection = collection_service.get_collection(collection_id)
+
+    if not collection:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, _("No collection found with the ID %s." % (collection_id,)))
+
+    is_private = collection.visibility == Collection.Visibility.PRIVATE
+    has_access = current_user and current_user.id == collection.created_by_user_id
+
+    if is_private and not has_access:
+        raise HTTPException(status.HTTP_403_FORBIDDEN, _("Access denied to this private collection."))
+
+    return quote_service.get_collection_quotes(collection_id, search_params)
+
+
+@router.post("/{collection_id}/quotes", response_model=QuoteCollectionsResponse, status_code=status.HTTP_201_CREATED)
+def add_quote_to_collection(
+    collection_id: int,
+    quote_id: int,
+    current_user: CurrentUser,
+    collection_service: CollectionServiceDepends,
+    quote_service: QuoteServiceDepends,
+):
+    quote = quote_service.get_quote_by_id(quote_id)
+
+    if not quote:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, _("No quote found with the ID %s." % (quote_id,)))
+
+    collection = collection_service.get_collection(collection_id)
+
+    if not collection:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, _("No collection found with the ID %s." % (collection_id,)))
+
+    if current_user.id != collection.created_by_user_id:
+        raise HTTPException(status.HTTP_403_FORBIDDEN, _("Access denied to this private collection."))
+
+    return collection_service.add_quote_to_collection(quote, collection)
+
+
+@router.delete("/{collection_id}/quotes/{quote_id}", status_code=status.HTTP_204_NO_CONTENT)
+def remove_quote_from_collection(
+    collection_id: int,
+    quote_id: int,
+    current_user: CurrentUser,
+    collection_service: CollectionServiceDepends,
+    quote_service: QuoteServiceDepends,
+):
+    quote = quote_service.get_quote_by_id(quote_id)
+
+    if not quote:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, _("No quote found with the ID %s." % (quote_id,)))
+
+    collection = collection_service.get_collection(collection_id)
+
+    if not collection:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, _("No collection found with the ID %s." % (collection_id,)))
+
+    if current_user.id != collection.created_by_user_id:
+        raise HTTPException(status.HTTP_403_FORBIDDEN, _("Access denied to this private collection."))
+
+    collection_service.remove_quote_from_collection(quote, collection)

src/api/v1/collections/service.py

@@ -3,11 +3,12 @@
 from src.api.params import SearchParams
 from src.api.v1.collections.models import Collection
 from src.api.v1.collections.schemas import CollectionCreateRequest, CollectionUpdateRequest
-from src.db.deps import Session
+from src.api.v1.quotes.models import Quote
+from src.db.deps import SessionDepends
 
 
 class CollectionService:
-    def __init__(self, session: Session) -> None:
+    def __init__(self, session: SessionDepends) -> None:
         self.session = session
 
     def get_collection(self, id: int) -> Collection | None:
@@ -54,3 +55,19 @@ def update_collection(self, collection: Collection, args: CollectionUpdateReques
     def delete_collection(self, collection: Collection) -> None:
         self.session.delete(collection)
         self.session.commit()
+
+    def add_quote_to_collection(self, quote: Quote, collection: Collection) -> Quote:
+        collection.quotes.append(quote)
+
+        self.session.add(collection)
+        self.session.commit()
+
+        self.session.refresh(quote)
+
+        return quote
+
+    def remove_quote_from_collection(self, quote: Quote, collection: Collection) -> None:
+        collection.quotes.remove(quote)
+
+        self.session.add(collection)
+        self.session.commit()

src/api/v1/quotes/__init__.py

@@ -1,3 +1,4 @@
 from src.api.v1.quotes.models import Quote
+from src.api.v1.quotes.routes import router
 
-__all__ = ["Quote"]
+__all__ = ["Quote", "router"]

src/api/v1/quotes/deps.py

@@ -0,0 +1,7 @@
+from typing import Annotated
+
+from fastapi import Depends
+
+from src.api.v1.quotes.service import QuoteService
+
+QuoteServiceDepends = Annotated[QuoteService, Depends(QuoteService)]

src/api/v1/quotes/enums.py

@@ -0,0 +1,7 @@
+from enum import StrEnum, auto
+
+
+class UserQuotesType(StrEnum):
+    ALL = auto()
+    SAVED = auto()
+    CREATED = auto()

src/api/v1/quotes/query.py

@@ -0,0 +1,52 @@
+from typing import Self
+
+from sqlalchemy import or_
+from sqlalchemy.orm import Query
+
+from src.api.params import SearchParams
+from src.api.v1.authors.models import Author
+from src.api.v1.collections.models import Collection
+from src.api.v1.quotes.enums import UserQuotesType
+from src.api.v1.quotes.models import Quote
+
+
+class QuoteQuery(Query[Quote]):
+    def filter_by_search_params(self, search_params: SearchParams) -> Self:
+        if search_params.q:
+            self = self.filter(
+                or_(
+                    Quote.content.ilike(f"%{search_params.q}%"),
+                    Quote.author.has(Author.name.ilike(f"%{search_params.q}%")),
+                )
+            )
+
+        return self.offset(search_params.offset).limit(search_params.limit)
+
+    def filter_by_collection_id(self, collection_id: int) -> Self:
+        return self.filter(Quote.collections.any(Collection.id == collection_id))
+
+    def filter_by_user_quotes_type(self, user_id: int, type: UserQuotesType) -> Self:
+        match type:
+            case UserQuotesType.ALL:
+                return self.filter_by_user_id_all(user_id)
+            case UserQuotesType.SAVED:
+                return self.filter_by_user_id_saved(user_id)
+            case UserQuotesType.CREATED:
+                return self.filter_by_user_id_created(user_id)
+
+    def filter_by_user_id_all(self, user_id: int) -> Self:
+        return self.outerjoin(Quote.collections).filter(
+            or_(
+                Quote.collections.any((Collection.created_by_user_id == user_id)),
+                Quote.created_by_user_id == user_id,
+            )
+        )
+
+    def filter_by_user_id_saved(self, user_id: int) -> Self:
+        return self.join(Quote.collections).filter(
+            Quote.collections.any((Collection.created_by_user_id == user_id)),
+            Quote.created_by_user_id != user_id,
+        )
+
+    def filter_by_user_id_created(self, user_id: int) -> Self:
+        return self.filter(Quote.created_by_user_id == user_id)

src/api/v1/quotes/routes.py

@@ -0,0 +1,66 @@
+from fastapi import APIRouter, HTTPException, status
+
+from src.api.deps import SearchParamsDepends
+from src.api.v1.authors import AuthorServiceDepends
+from src.api.v1.quotes.deps import QuoteServiceDepends
+from src.api.v1.quotes.schemas import QuoteCollectionsResponse, QuoteCreateRequest, QuoteResponse, QuoteUpdateRequest
+from src.api.v1.users.me.deps import CurrentUser
+from src.i18n import gettext as _
+
+router = APIRouter(prefix="/quotes", tags=["Quotes"])
+
+
+@router.get("/{quote_id}", response_model=QuoteCollectionsResponse)
+def get_quote(quote_id: int, service: QuoteServiceDepends):
+    quote = service.get_quote_by_id(quote_id)
+
+    if not quote:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, _("No quote found with the ID %s." % (quote_id,)))
+
+    return quote
+
+
+@router.get("/", response_model=list[QuoteResponse])
+def get_quotes(search_params: SearchParamsDepends, service: QuoteServiceDepends):
+    quotes = service.get_quotes(search_params)
+
+    if not quotes:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, _("No quotes found matching the provided search parameters."))
+
+    return quotes
+
+
+@router.post("/", response_model=QuoteResponse, status_code=status.HTTP_201_CREATED)
+def create_quote(
+    args: QuoteCreateRequest,
+    current_user: CurrentUser,
+    quote_service: QuoteServiceDepends,
+    author_service: AuthorServiceDepends,
+):
+    if args.author_id and not author_service.get_author_by_id(args.author_id):
+        raise HTTPException(status.HTTP_404_NOT_FOUND, _("No author found with the ID '%s'." % (args.author_id,)))
+    return quote_service.create_quote(args, created_by_user_id=current_user.id)
+
+
+@router.patch("/{quote_id}", response_model=QuoteResponse)
+def update_quote(quote_id: int, args: QuoteUpdateRequest, current_user: CurrentUser, service: QuoteServiceDepends):
+    quote = service.get_quote_by_id(quote_id)
+
+    if not quote:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, _("No quote found with the ID %s." % (quote_id,)))
+    if quote.created_by_user_id != current_user.id:
+        raise HTTPException(status.HTTP_403_FORBIDDEN, _("You do not have permission to modify this quote."))
+
+    return service.update_quote(quote, args)
+
+
+@router.delete("/{quote_id}", status_code=status.HTTP_204_NO_CONTENT)
+def delete_quote(quote_id: int, current_user: CurrentUser, service: QuoteServiceDepends):
+    quote = service.get_quote_by_id(quote_id)
+
+    if not quote:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, _("No quote found with the ID %s." % (quote_id,)))
+    if quote.created_by_user_id != current_user.id:
+        raise HTTPException(status.HTTP_403_FORBIDDEN, _("You do not have permission to modify this quote."))
+
+    service.delete_quote(quote)

src/api/v1/quotes/schemas.py

@@ -0,0 +1,26 @@
+from pydantic import BaseModel
+
+from src.api.v1.authors.schemas import AuthorResponse
+from src.api.v1.collections.schemas import CollectionResponse
+from src.api.v1.schemas import AuditResponse
+
+
+class QuoteResponse(AuditResponse):
+    id: int
+    content: str
+    created_by_user_id: int | None
+    author: AuthorResponse | None
+
+
+class QuoteCollectionsResponse(QuoteResponse):
+    collections: list[CollectionResponse]
+
+
+class QuoteCreateRequest(BaseModel):
+    content: str
+    author_id: int | None
+
+
+class QuoteUpdateRequest(BaseModel):
+    content: str | None = None
+    author_id: int | None = None