feat: Add address management with CRUD operations and integrate into the application

app/app.py

@@ -13,6 +13,7 @@ from app.controllers.reference import router as reference_router
 from app.controllers.bidders import router as bidders_router
 from app.controllers.distributors import router as distributors_router
 from app.controllers.notes import router as notes_router
+from app.controllers.addresses import router as addresses_router
 
 setup_logging(settings.LOG_LEVEL)
 
@@ -58,3 +59,4 @@ app.include_router(employees_router)
 app.include_router(reference_router)
 app.include_router(bidders_router)
 app.include_router(notes_router)
+app.include_router(addresses_router)

app/controllers/addresses.py

@@ -0,0 +1,84 @@
+from fastapi import APIRouter, Depends, status, Query, HTTPException
+from sqlalchemy.orm import Session
+from app.db.session import get_db
+from app.services.address_service import AddressService
+from app.schemas.address import AddressCreate, AddressOut
+from app.schemas.paginated_response import PaginatedResponse
+from typing import Optional
+import logging
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/v1/addresses", tags=["addresses"])
+
+
+@router.get(
+    "/",
+    response_model=PaginatedResponse[AddressOut],
+    summary="List addresses",
+    response_description="Paginated list of addresses"
+)
+def list_addresses(
+    customer_id: Optional[int] = Query(None, description="Optional customer id to filter addresses"),
+    page: int = Query(1, ge=1, description="Page number (1-indexed)"),
+    page_size: int = Query(10, ge=1, le=100, description="Number of items per page"),
+    order_by: Optional[str] = Query("Id", description="Field to order by"),
+    order_dir: Optional[str] = Query("asc", description="Order direction (asc|desc)"),
+    db: Session = Depends(get_db)
+):
+    try:
+        service = AddressService(db)
+        result = service.list(page=page, page_size=page_size, order_by=order_by, order_dir=order_dir, customer_id=customer_id)
+        return result
+    except Exception as e:
+        logger.error(f"Error listing addresses: {e}")
+        return PaginatedResponse[AddressOut](items=[], page=page, page_size=page_size, total=0)
+
+
+@router.get("/{address_id}", response_model=AddressOut, summary="Get address by id")
+def get_address(address_id: int, db: Session = Depends(get_db)):
+    try:
+        service = AddressService(db)
+        return service.get(address_id)
+    except Exception as e:
+        logger.error(f"Error getting address {address_id}: {e}")
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Address {address_id} not found")
+
+
+@router.post("/", response_model=AddressOut, status_code=status.HTTP_201_CREATED)
+def create_address(address_in: AddressCreate, db: Session = Depends(get_db)):
+    try:
+        service = AddressService(db)
+        result = service.create(address_in.dict())
+        return result
+    except ValueError as e:
+        logger.error(f"Validation error creating address: {e}")
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
+    except Exception as e:
+        logger.error(f"Error creating address: {e}")
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to create address")
+
+
+@router.put("/{address_id}", response_model=AddressOut)
+def update_address(address_id: int, address_in: AddressCreate, db: Session = Depends(get_db)):
+    try:
+        service = AddressService(db)
+        result = service.update(address_id, address_in.dict(exclude_unset=True))
+        return result
+    except ValueError as e:
+        logger.error(f"Validation error updating address {address_id}: {e}")
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
+    except Exception as e:
+        logger.error(f"Error updating address {address_id}: {e}")
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Address {address_id} not found")
+
+
+@router.delete("/{address_id}", status_code=status.HTTP_204_NO_CONTENT)
+def delete_address(address_id: int, db: Session = Depends(get_db)):
+    try:
+        service = AddressService(db)
+        service.delete(address_id)
+        return None
+    except Exception as e:
+        logger.error(f"Error deleting address {address_id}: {e}")
+        raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to delete address")

app/db/models/address.py

@@ -0,0 +1,16 @@
+from sqlalchemy import Column, Integer, String, Boolean
+from app.db.base import Base
+
+
+class Address(Base):
+    __tablename__ = "Address"
+
+    Id = Column(Integer, primary_key=True, index=True)
+    Address = Column(String(200), nullable=True)
+    City = Column(String(100), nullable=True)
+    StateId = Column(Integer, nullable=True)
+    PostalCode = Column(String(70), nullable=True)
+    CustomerId = Column(Integer, nullable=True)
+    Enabled = Column(Boolean, nullable=False, default=True)
+    IsCustomAddress = Column(Boolean, nullable=True, default=False)
+    TempAddress = Column(Boolean, nullable=True, default=False)

app/db/repositories/address_repo.py

@@ -0,0 +1,114 @@
+from sqlalchemy.orm import Session
+from sqlalchemy.exc import SQLAlchemyError
+from app.db.models.address import Address
+from app.schemas.address import AddressCreate, AddressOut
+from app.schemas.paginated_response import PaginatedResponse
+from app.core.exceptions import NotFoundException
+from typing import Optional
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class AddressRepository:
+    def __init__(self, db: Session):
+        self.db = db
+
+    def get(self, address_id: int) -> Optional[AddressOut]:
+        try:
+            addr = self.db.query(Address).filter(Address.Id == address_id).first()
+            if addr:
+                return self._map_address(addr.__dict__)
+            return None
+        except SQLAlchemyError as e:
+            logger.error(f"Database error getting address {address_id}: {e}")
+            raise
+
+    def list(self, page: int = 1, page_size: int = 10, order_by: str = "Id", order_direction: str = "ASC", customer_id: int = None) -> PaginatedResponse[AddressOut]:
+        try:
+            query = self.db.query(Address)
+            if customer_id is not None:
+                query = query.filter(Address.CustomerId == customer_id)
+
+            # validate order_by
+            if not hasattr(Address, order_by):
+                order_by = "Id"
+            order_col = getattr(Address, order_by)
+            order_col = order_col.desc() if order_direction.upper() == "DESC" else order_col.asc()
+            query = query.order_by(order_col)
+
+            total = query.count()
+            rows = query.offset((page - 1) * page_size).limit(page_size).all()
+            items = [self._map_address(r.__dict__) for r in rows]
+            return PaginatedResponse[AddressOut](items=items, page=page, page_size=page_size, total=total)
+        except Exception as e:
+            logger.error(f"Error listing addresses: {e}")
+            return PaginatedResponse[AddressOut](items=[], page=page, page_size=page_size, total=0)
+
+    def create(self, address_data: AddressCreate) -> AddressOut:
+        try:
+            new_address = Address(
+                Address=address_data.address,
+                City=address_data.city,
+                StateId=address_data.state_id,
+                PostalCode=address_data.postal_code,
+                CustomerId=address_data.customer_id,
+                Enabled=address_data.enabled if address_data.enabled is not None else True,
+                IsCustomAddress=address_data.is_custom_address or False,
+                TempAddress=address_data.temp_address or False,
+            )
+            self.db.add(new_address)
+            self.db.commit()
+            self.db.refresh(new_address)
+            return self._map_address(new_address.__dict__)
+        except SQLAlchemyError as e:
+            logger.error(f"Database error creating address: {e}")
+            self.db.rollback()
+            raise
+
+    def update(self, address_id: int, address_data: AddressCreate) -> AddressOut:
+        try:
+            addr = self.db.query(Address).filter(Address.Id == address_id).first()
+            if not addr:
+                raise NotFoundException("Address not found for update")
+            addr.Address = address_data.address
+            addr.City = address_data.city
+            addr.StateId = address_data.state_id
+            addr.PostalCode = address_data.postal_code
+            addr.CustomerId = address_data.customer_id
+            addr.Enabled = address_data.enabled if address_data.enabled is not None else addr.Enabled
+            addr.IsCustomAddress = address_data.is_custom_address or addr.IsCustomAddress
+            addr.TempAddress = address_data.temp_address or addr.TempAddress
+            self.db.commit()
+            self.db.refresh(addr)
+            return self._map_address(addr.__dict__)
+        except SQLAlchemyError as e:
+            logger.error(f"Database error updating address {address_id}: {e}")
+            self.db.rollback()
+            raise
+
+    def delete(self, address_id: int) -> bool:
+        try:
+            addr = self.db.query(Address).filter(Address.Id == address_id).first()
+            if not addr:
+                raise NotFoundException("Address not found for delete")
+            self.db.delete(addr)
+            self.db.commit()
+            return True
+        except SQLAlchemyError as e:
+            logger.error(f"Database error deleting address {address_id}: {e}")
+            self.db.rollback()
+            raise
+
+    def _map_address(self, data: dict) -> AddressOut:
+        return AddressOut(
+            id=data.get('Id'),
+            customer_id=data.get('CustomerId'),
+            address=data.get('Address'),
+            city=data.get('City'),
+            state_id=data.get('StateId'),
+            postal_code=data.get('PostalCode'),
+            enabled=data.get('Enabled'),
+            is_custom_address=data.get('IsCustomAddress'),
+            temp_address=data.get('TempAddress')
+        )

app/schemas/address.py

@@ -0,0 +1,28 @@
+from pydantic import BaseModel, Field
+from typing import Optional
+
+
+class AddressCreate(BaseModel):
+    customer_id: Optional[int] = Field(None, description="Customer ID this address belongs to")
+    address: Optional[str] = Field(None, max_length=200, description="Street address")
+    city: Optional[str] = Field(None, max_length=100, description="City")
+    state_id: Optional[int] = Field(None, description="State ID")
+    postal_code: Optional[str] = Field(None, max_length=70, description="Postal/ZIP code")
+    enabled: Optional[bool] = Field(True, description="Enabled flag")
+    is_custom_address: Optional[bool] = Field(False, description="Is a custom address")
+    temp_address: Optional[bool] = Field(False, description="Temporary address flag")
+
+
+class AddressOut(BaseModel):
+    id: int = Field(..., description="Address ID")
+    customer_id: Optional[int] = Field(None, description="Customer ID this address belongs to")
+    address: Optional[str] = Field(None, description="Street address")
+    city: Optional[str] = Field(None, description="City")
+    state_id: Optional[int] = Field(None, description="State ID")
+    postal_code: Optional[str] = Field(None, description="Postal/ZIP code")
+    enabled: Optional[bool] = Field(None, description="Enabled flag")
+    is_custom_address: Optional[bool] = Field(None, description="Is a custom address")
+    temp_address: Optional[bool] = Field(None, description="Temporary address flag")
+
+    class Config:
+        from_attributes = True

app/services/address_service.py

@@ -0,0 +1,37 @@
+from sqlalchemy.orm import Session
+from app.db.repositories.address_repo import AddressRepository
+from app.schemas.address import AddressCreate, AddressOut
+from app.schemas.paginated_response import PaginatedResponse
+from app.core.exceptions import NotFoundException
+from typing import Optional
+
+
+class AddressService:
+    def __init__(self, db: Session):
+        self.repo = AddressRepository(db)
+
+    def get(self, address_id: int) -> AddressOut:
+        addr = self.repo.get(address_id)
+        if not addr:
+            raise NotFoundException("Address not found")
+        return addr
+
+    def list(self, page: int = 1, page_size: int = 10, order_by: str = "Id", order_dir: str = "ASC", customer_id: Optional[int] = None) -> PaginatedResponse[AddressOut]:
+        return self.repo.list(page=page, page_size=page_size, order_by=order_by, order_direction=order_dir, customer_id=customer_id)
+
+    def create(self, data: dict) -> AddressOut:
+        address_data = AddressCreate(**data)
+        return self.repo.create(address_data)
+
+    def update(self, address_id: int, data: dict) -> AddressOut:
+        existing = self.repo.get(address_id)
+        if not existing:
+            raise NotFoundException("Address not found")
+        address_data = AddressCreate(**data)
+        return self.repo.update(address_id, address_data)
+
+    def delete(self, address_id: int):
+        existing = self.repo.get(address_id)
+        if not existing:
+            raise NotFoundException("Address not found")
+        self.repo.delete(address_id)

tests/unit/conftest.py

@@ -0,0 +1,24 @@
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from app.db.base import Base
+
+# Ensure models are imported so metadata is populated for create_all
+import app.db.models.address  # noqa: F401
+
+
+@pytest.fixture
+def db_session():
+    """Create a new database session with an in-memory SQLite DB for tests."""
+    engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
+    testing_session_local = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+    # Create tables for all models
+    Base.metadata.create_all(bind=engine)
+
+    db = testing_session_local()
+    try:
+        yield db
+    finally:
+        db.close()

tests/unit/test_address_repository.py

@@ -0,0 +1,33 @@
+from app.db.repositories.address_repo import AddressRepository
+from app.schemas.address import AddressCreate
+
+
+def test_create_get_delete_address(db_session):
+    repo = AddressRepository(db_session)
+
+    # Create
+    create_in = AddressCreate(
+        customer_id=1,
+        address="123 Test St",
+        city="Testville",
+        state_id=5,
+        postal_code="99999"
+    )
+    created = repo.create(create_in)
+    assert created is not None
+    assert created.id is not None
+    assert created.address == "123 Test St"
+
+    # Get
+    fetched = repo.get(created.id)
+    assert fetched is not None
+    assert fetched.id == created.id
+    assert fetched.city == "Testville"
+
+    # Delete
+    deleted = repo.delete(created.id)
+    assert deleted is True
+
+    # After delete, get should return None
+    after = repo.get(created.id)
+    assert after is None

tests/unit/test_address_service.py

@@ -0,0 +1,40 @@
+import pytest
+
+from app.services.address_service import AddressService
+from app.core.exceptions import NotFoundException
+
+
+def test_service_create_get_update_delete(db_session):
+    service = AddressService(db_session)
+
+    data = {
+        "customer_id": 2,
+        "address": "50 Service Rd",
+        "city": "Servetown",
+        "state_id": 7,
+        "postal_code": "55555"
+    }
+
+    # Create
+    created = service.create(data)
+    assert created is not None
+    assert created.id is not None
+    assert created.customer_id == 2
+
+    # Get
+    fetched = service.get(created.id)
+    assert fetched.id == created.id
+    assert fetched.address == "50 Service Rd"
+
+    # Update
+    update_data = {"address": "51 Service Rd", "city": "Newtown"}
+    updated = service.update(created.id, update_data)
+    assert updated.address == "51 Service Rd"
+    assert updated.city == "Newtown"
+
+    # Delete
+    service.delete(created.id)
+
+    # After delete, get should raise NotFoundException
+    with pytest.raises(NotFoundException):
+        service.get(created.id)