Orders 02 merge conflict solution

src/config/settings.py

@@ -2,8 +2,11 @@
 from pathlib import Path
 from typing import Any
 
+from dotenv import load_dotenv
 from pydantic_settings import BaseSettings
 
+load_dotenv()
+
 
 class BaseAppSettings(BaseSettings):
     BASE_DIR: Path = Path(__file__).parent.parent

src/database/crud/orders.py

@@ -0,0 +1,154 @@
+from datetime import datetime
+
+from fastapi import HTTPException, status
+from sqlalchemy.exc import SQLAlchemyError
+from sqlalchemy.orm import Session, joinedload
+
+from database import Order, OrderItem, Cart, User
+from schemas import OrderItemResponseSchema, MovieListItemSchema
+
+
+def create_order(user_id: int, db: Session) -> Order:
+    """Creates a new order for the current user."""
+    cart = db.query(Cart).filter(Cart.user_id == user_id).first()
+    if not cart or not cart.items:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST, detail="Cart is empty."
+        )
+
+    try:
+        order = Order(
+            user_id=user_id, total_amount=sum(item.movie.price for item in cart.items)
+        )
+        db.add(order)
+        db.flush()
+
+        for cart_item in cart.items:
+            order_item = OrderItem(
+                order_id=order.id,
+                movie_id=cart_item.movie_id,
+                price_at_order=cart_item.movie.price,
+            )
+            db.add(order_item)
+
+        db.commit()
+        db.refresh(order)
+        return order
+    except SQLAlchemyError as e:
+        db.rollback()
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Failed to create order: {str(e)}"
+        )
+
+
+def update_order_with_stripe_url(order: Order, stripe_url: str, db: Session) -> None:
+    """Updates the order with a Stripe URL."""
+    try:
+        order.stripe_url = stripe_url
+        db.commit()
+    except SQLAlchemyError as e:
+        db.rollback()
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Failed to update order with Stripe URL: {str(e)}"
+        )
+
+
+def get_user_orders(
+    current_user: User,
+    db: Session,
+    user_id: int | None = None,
+    date_from: datetime | None = None,
+    date_to: datetime | None = None,
+    order_status: str | None = None,
+) -> list[OrderItemResponseSchema]:
+    """Retrieves orders for a specific user or all users for admin."""
+    if current_user.group.name != "user":
+        filters = []
+        if user_id:
+            filters.append(Order.user_id == user_id)
+        if date_from:
+            filters.append(Order.created_at >= date_from)
+        if date_to:
+            filters.append(Order.created_at <= date_to)
+        if order_status:
+            filters.append(Order.status == order_status)
+        orders_query = db.query(Order).filter(*filters)
+    else:
+        if user_id or date_from or date_to or order_status:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="You don't have permission.",
+            )
+        orders_query = db.query(Order).filter(Order.user_id == current_user.id)
+
+    orders = orders_query.options(
+        joinedload(Order.order_items).joinedload(OrderItem.movie)
+    )
+    orders = orders.order_by(Order.created_at.desc()).all()
+
+    if not orders:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="No orders found."
+        )
+
+    order_list = [
+        OrderItemResponseSchema(
+            created_at=order.created_at,
+            movies=[
+                MovieListItemSchema(
+                    id=item.movie.id,
+                    name=item.movie.name,
+                    year=item.movie.year,
+                    time=item.movie.time,
+                    description=item.movie.description,
+                )
+                for item in order.order_items
+            ],
+            total_amount=order.total_amount,
+            status=order.status,
+        )
+        for order in orders
+    ]
+
+    return order_list
+
+
+def get_order_by_id(
+        db: Session, order_id: int, current_user_id: int | None = None
+) -> Order:
+    """Retrieve an order by ID and check permissions."""
+    order = db.query(Order).filter(Order.id == order_id).first()
+
+    if not order:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Order not found."
+        )
+
+    if current_user_id and order.user_id != current_user_id:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="You don't have permission to view this order.",
+        )
+
+    return order
+
+
+def format_order_detail(order: Order) -> OrderItemResponseSchema:
+    """Format the order details for the response."""
+    return OrderItemResponseSchema(
+        created_at=order.created_at,
+        movies=[
+            MovieListItemSchema(
+                id=item.movie.id,
+                name=item.movie.name,
+                year=item.movie.year,
+                time=item.movie.time,
+                description=item.movie.description,
+            )
+            for item in order.order_items
+        ],
+        total_amount=order.total_amount,
+        status=order.status,
+    )

src/database/models/__init__.py

@@ -0,0 +1,4 @@
+from .accounts import User
+from .orders import Order, OrderItem
+from .movies import Movie
+from .shopping_cart import Cart, CartItem

src/database/models/orders.py

@@ -2,7 +2,8 @@
 from decimal import Decimal
 from enum import Enum
 
-from sqlalchemy import DECIMAL, DateTime, ForeignKey, Integer, String
+from sqlalchemy import DECIMAL, DateTime, ForeignKey, Integer, func, String
+from sqlalchemy import Enum as SQLAlchemyEnum
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from database.models.accounts import User
@@ -27,12 +28,13 @@ class Order(Base):
         Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False
     )
     created_at: Mapped[datetime] = mapped_column(
-        DateTime, nullable=False, default=datetime.now
+        DateTime(timezone=True), server_default=func.now(), nullable=False
     )
     status: Mapped[OrderStatusEnum] = mapped_column(
-        String(50), nullable=False, default=OrderStatusEnum.PENDING.value
+        SQLAlchemyEnum(OrderStatusEnum), nullable=False, default="PENDING"
     )
     total_amount: Mapped[Decimal | None] = mapped_column(DECIMAL(10, 2))
+    stripe_url: Mapped[str | None] = mapped_column(String)
 
     order_items: Mapped[list["OrderItem"]] = relationship(
         "OrderItem", back_populates="order"

src/main.py

@@ -1,7 +1,9 @@
 from fastapi import FastAPI
 from fastapi_pagination import add_pagination
 
-from routes import accounts_router, movies_router, payments_router, profiles_router
+from routes import (
+    accounts_router, movies_router, payments_router, profiles_router, orders_router
+)
 from routes.shopping_cart import router as shopping_carts_router
 
 app = FastAPI(
@@ -22,6 +24,9 @@
 app.include_router(
     payments_router, prefix=f"{api_version_prefix}/payments", tags=["payments"]
 )
+app.include_router(
+    orders_router, prefix=f"{api_version_prefix}", tags=["orders"]
+)
 
 add_pagination(app)
 app.include_router(

src/routes/__init__.py

@@ -2,3 +2,4 @@
 from routes.payments import router as payments_router
 from routes.profiles import router as profiles_router
 from routes.movies import router as movies_router
+from routes.orders import router as orders_router

src/routes/orders.py

@@ -0,0 +1,98 @@
+from datetime import datetime
+
+from fastapi import APIRouter, Request, Depends, status, Query
+from sqlalchemy.orm import Session
+
+from database import get_db
+from database.crud.orders import (
+    create_order,
+    update_order_with_stripe_url,
+    get_user_orders,
+    format_order_detail,
+    get_order_by_id,
+)
+from database.models.accounts import User
+from database.models.orders import OrderStatusEnum
+from routes.profiles import get_current_user
+from schemas.orders import OrderItemResponseSchema, MessageResponseSchema
+from services import create_checkout_session
+
+router = APIRouter()
+
+
+"""
+1. Place an Order
+Endpoint: POST /orders/
+Description: Allows users to place an order for movies in their cart.
+"""
+@router.post("/orders/")
+def place_order(
+    request: Request,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+) -> MessageResponseSchema:
+    order = create_order(user_id=current_user.id, db=db)
+
+    stripe_url = create_checkout_session(
+        request=request, order=order, user_id=current_user.id, db=db
+    )
+    update_order_with_stripe_url(order, stripe_url, db=db)
+
+    return MessageResponseSchema(
+        message=f"Order placed successfully, your order_id: {order.id}"
+    )
+
+
+"""
+2. View User Orders (depending on whether it is an admin or a user...)
+Endpoint: GET /orders/
+Description: Retrieves a list of all orders placed by a specific user.
+"""
+@router.get(
+    "/orders/",
+    response_model=list[OrderItemResponseSchema],
+    status_code=status.HTTP_200_OK,
+)
+def get_user_orders_route(
+    user_id: int | None = Query(default=None, description="Filter orders by user ID"),
+    date_from: datetime | None = Query(
+        default=None, description="Filter orders from this date"
+    ),
+    date_to: datetime | None = Query(
+        default=None, description="Filter orders until this date"
+    ),
+    order_status: OrderStatusEnum | None = Query(
+        default=None, description="Filter orders by status"
+    ),
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+) -> list[OrderItemResponseSchema]:
+    orders: list[OrderItemResponseSchema] = get_user_orders(
+        current_user=current_user,
+        db=db,
+        user_id=user_id,
+        date_from=date_from,
+        date_to=date_to,
+        order_status=order_status,
+    )
+    return orders
+
+
+"""
+3. Detail view of order
+Endpoint: GET /orders/{order_id}/
+Description: Allows user to view details about order.
+"""
+@router.get(
+    "/orders/{order_id}/",
+    response_model=OrderItemResponseSchema,
+    status_code=status.HTTP_200_OK,
+)
+def get_order_detail(
+    order_id: int,
+    current_user: User = Depends(get_current_user),
+    db: Session = Depends(get_db),
+) -> OrderItemResponseSchema:
+    order = get_order_by_id(db, order_id, current_user.id)
+
+    return format_order_detail(order)

src/routes/profiles.py

@@ -13,15 +13,37 @@
     UserGroupEnum,
     UserProfile,
 )
-from exceptions import BaseSecurityError, S3FileUploadError
+from exceptions import BaseSecurityError, S3FileUploadError, TokenExpiredError
 from schemas.profiles import ProfileCreateSchema, ProfileResponseSchema
 from security.http import get_token
 from security.interfaces import JWTAuthManagerInterface
+from security.token_manager import JWTAuthManager
 from storages import S3StorageInterface
 
 router = APIRouter()
 
 
+def get_current_user(
+    token: str = Depends(get_token),
+    db: Session = Depends(get_db),
+    jwt_manager: JWTAuthManager = Depends(get_jwt_auth_manager)
+) -> User:
+    try:
+        token_data = jwt_manager.decode_access_token(token)
+        user = db.query(User).filter_by(id=token_data.get("user_id")).first()
+        if not user or not user.is_active:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="User not found or not active."
+            )
+        return user
+    except TokenExpiredError:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Token has expired."
+        )
+
+
 @router.post(
     "/users/{user_id}/profile/",
     response_model=ProfileResponseSchema,

src/schemas/__init__.py

@@ -3,3 +3,10 @@
     PaymentHistoryResponse,
     PaymentSchema,
 )
+from schemas.orders import (
+    MessageResponseSchema,
+    OrderItemResponseSchema,
+)
+from schemas.movies import (
+    MovieListItemSchema,
+)

src/schemas/orders.py

@@ -0,0 +1,19 @@
+from datetime import datetime
+
+from pydantic import BaseModel
+
+from database.models.orders import OrderStatusEnum
+from schemas.movies import MovieListItemSchema
+
+
+class MessageResponseSchema(BaseModel):
+    message: str
+
+
+class OrderItemResponseSchema(BaseModel):
+    created_at: datetime
+    movies: list[MovieListItemSchema]
+    total_amount: float | None
+    status: OrderStatusEnum
+
+    model_config = {"from_attributes": True}