wrote crud routes for transaction

Author: Hardcode3

app/api/endpoints/expenses.py

@@ -0,0 +1,133 @@
+from typing import List
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, status, Query
+from sqlalchemy.orm import Session
+from sqlalchemy.exc import IntegrityError
+
+from app.core.database import get_db
+from app.models.transaction import Transaction
+from app.schemas.transaction import TransactionCreate, TransactionUpdate, TransactionResponse
+
+router = APIRouter()
+
+
+def retrieve_transaction(transaction_id: uuid.UUID, db: Session) -> Transaction:
+    """Gets a single transaction or raise if non existent.
+
+    Raises:
+        HTTPException (404 Not Found) if the transaction does not exist.
+    """
+    transaction = db.query(Transaction).filter(Transaction.id == transaction_id).one_or_none()
+
+    if transaction is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Transaction with ID {transaction_id} not found")
+
+    return transaction
+
+
+@router.post(
+    "/",
+    response_model=TransactionResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+def create_transaction(
+    data: TransactionCreate,
+    db: Session = Depends(get_db)
+) -> TransactionResponse:
+
+    transaction = Transaction(
+        amount=data.amount,
+        method=data.method,
+        date=data.date,
+        description=data.description,
+    )
+    db.add(transaction)
+    try:
+        db.commit()
+        db.refresh(transaction)
+    except IntegrityError as e:
+        db.rollback()
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Database constraint violation") from e
+
+    return transaction.to_response()
+
+
+@router.get(
+    "/{transaction_id}",
+    response_model=TransactionResponse,
+    status_code=status.HTTP_200_OK,
+)
+def get_single_transaction(
+    transaction_id: uuid.UUID,
+    db: Session = Depends(get_db)
+) -> TransactionResponse:
+
+    transaction = retrieve_transaction(transaction_id, db)
+    return transaction.to_response()
+
+
+@router.get(
+    "/",
+    response_model=List[TransactionResponse],
+    status_code=status.HTTP_200_OK,
+)
+def get_all_transactions(
+    db: Session = Depends(get_db),
+    offset: int = Query(0, ge=0, description="Offset must be 0 or greater (defaults to 0)"),
+    limit: int = Query(10, ge=1, le=100, description="Limit must be between 1 and 100 (defaults to 10)"),
+) -> List[TransactionResponse]:
+
+    transactions = db.query(Transaction).offset(offset).limit(limit).all()
+    return [t.to_response() for t in transactions]
+
+
+@router.patch(
+    "/{transaction_id}",
+    response_model=TransactionResponse,
+    status_code=status.HTTP_200_OK,
+)
+def update_transaction(
+    transaction_id: uuid.UUID,
+    data: TransactionUpdate,
+    db: Session = Depends(get_db),
+) -> TransactionResponse:
+
+    transaction = retrieve_transaction(transaction_id, db)
+
+    update_data = data.model_dump(exclude_unset=True)
+
+    field_mapping = {
+        "amount": "amount",
+        "description": "description",
+        "method": "method",
+        "date": "date"
+    }
+
+    for schema_field, db_field in field_mapping.items():
+        if schema_field in update_data:
+            setattr(transaction, db_field, update_data[schema_field])
+
+    db.commit()
+    db.refresh(transaction)
+
+    return transaction.to_response()
+
+
+@router.delete(
+    "/{transaction_id}",
+    status_code=status.HTTP_204_NO_CONTENT
+)
+def delete_transaction(
+        transaction_id: uuid.UUID,
+        db: Session = Depends(get_db),
+) -> None:
+
+    transaction = retrieve_transaction(transaction_id, db)
+
+    try:
+        db.delete(transaction)
+        db.commit()
+    except IntegrityError as e:
+        db.rollback()
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Cannot delete transaction due to related records") from e

app/api/endpoints/transaction.py

@@ -0,0 +1,26 @@
+import logging
+
+from fastapi import HTTPException, status
+from sqlalchemy.exc import IntegrityError
+
+logger = logging.getLogger(__name__)
+
+
+def sqlalchemy_exception_handler(request, exc) -> None:
+    """Global handler for all SQLAlchemy errors."""
+    # pylint: disable = unused-argument
+
+    # log the error and do not include it into user messages
+    # to avoid exposing sensitive database information
+    logger.error("Database error: %s", str(exc))
+
+    if isinstance(exc, IntegrityError):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Database integrity error. Possible duplicate entry or constraint violation."
+        )
+
+    raise HTTPException(
+        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+        detail="A database error occurred. Please try again later."
+    )

app/api/endpoints/users.py

@@ -19,3 +19,10 @@
 engine = create_engine(DATABASE_URL)
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 Base = declarative_base()
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

app/api/exception_handlers.py

@@ -1,14 +1,18 @@
 from fastapi import FastAPI
 from fastapi.responses import RedirectResponse
 
+from sqlalchemy.exc import SQLAlchemyError
+
 from app.core.database import engine, Base
-from app.api.endpoints import expenses
+from app.api.endpoints import transaction
+import app.api.exception_handlers as ExHandler
 
 Base.metadata.create_all(bind=engine)
 
 app = FastAPI(title="Expense Tracker API")
+app.add_exception_handler(SQLAlchemyError, ExHandler.sqlalchemy_exception_handler)
 
-app.include_router(expenses.router, prefix="/expenses", tags=["Expenses"])
+app.include_router(transaction.router, prefix="/transaction", tags=["Transactions"])
 
 @app.get("/", include_in_schema=False)
 async def redirect_to_swagger():

app/core/config.py

@@ -0,0 +1,40 @@
+"""Expense database model. """
+
+from datetime import datetime
+import uuid
+from enum import Enum as PyEnum
+
+from sqlalchemy import Column, String, Float, DateTime, Enum, UUID
+
+from app.core.database import Base
+from app.schemas.transaction import TransactionResponse
+
+
+class TransactionMethod(str, PyEnum):
+    BANK_TRANSFER = "bank_transfer"
+    CREDIT_CARD = "credit_card"
+    DEBIT_CARD = "debit_card"
+    CASH = "cash"
+    PAYPAL = "paypal"
+    OTHER = "other"
+
+
+class Transaction(Base):
+    __tablename__ = "transactions"
+
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, index=True)
+    amount = Column(Float, nullable=False)
+    method = Column(Enum(TransactionMethod), nullable=False)
+    date = Column(DateTime, default=datetime.now, nullable=False)
+    description = Column(String(255), nullable=True)
+
+
+    def to_response(self) -> TransactionResponse:
+        """Convert a Transaction ORM object into a TransactionResponse Pydantic model."""
+        return TransactionResponse(
+            id=self.id,
+            amount=self.amount,
+            method=self.method,
+            date=self.date,
+            description=self.description,
+        )