Implement customer listing with pagination and ordering via stored procedure; update models, services, and tests

app/controllers/customers.py

@@ -1,16 +1,29 @@
-from fastapi import APIRouter, Depends, status
+from fastapi import APIRouter, Depends, status, Query
 from sqlalchemy.orm import Session
 from app.db.session import get_db
 from app.services.customer_service import CustomerService
+from app.services.customer_list_service import CustomerListService
 from app.schemas.customer import CustomerCreate, CustomerOut
-from typing import List
+from app.schemas.paginated_response import PaginatedResponse
+from typing import List, Optional
 
 router = APIRouter(prefix="/api/v1/customers", tags=["customers"])
 
-@router.get("/", response_model=List[CustomerOut])
-def list_customers(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
-    service = CustomerService(db)
-    return service.list(skip, limit)
+@router.get(
+    "/",
+    response_model=PaginatedResponse[CustomerOut],
+    summary="List customers with pagination and ordering",
+    response_description="Paginated list of customers"
+)
+def list_customers(
+    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("CustomerID", description="Field to order by (CustomerID, CompanyName, Address, City, etc.)"),
+    order_dir: Optional[str] = Query("desc", description="Order direction (asc|desc)"),
+    db: Session = Depends(get_db)
+):
+    service = CustomerListService(db)
+    return service.list_customers(page=page, page_size=page_size, order_by=order_by, order_dir=order_dir)
 
 @router.get("/{customer_id}", response_model=CustomerOut)
 def get_customer(customer_id: int, db: Session = Depends(get_db)):

app/core/config.py

@@ -11,7 +11,7 @@ class Settings(BaseSettings):
     SQLSERVER_USER: str
     SQLSERVER_PASSWORD: str
     SQLSERVER_HOST: str
-    SQLSERVER_PORT: int = 1433
+    SQLSERVER_PORT: int = 31433
     SQLSERVER_DB: str
     SQLSERVER_DRIVER: str = "ODBC Driver 18 for SQL Server"
     CORS_ORIGINS: Optional[str] = "*"

app/db/models/customer.py

@@ -1,12 +1,24 @@
-from sqlalchemy import Column, Integer, String, DateTime
+from sqlalchemy import Column, Integer, String, DateTime, Boolean, Text
 from app.db.base import Base
-from datetime import datetime
 
 class Customer(Base):
-    __tablename__ = "customers"
-    id = Column(Integer, primary_key=True, index=True)
-    name = Column(String(255), nullable=False)
-    email = Column(String(255), unique=True, index=True, nullable=True)
-    phone = Column(String(50), nullable=True)
-    address = Column(String(255), nullable=True)
-    created_at = Column(DateTime, default=datetime.utcnow)
+    __tablename__ = "Customers"
+    
+    CustomerID = Column(Integer, primary_key=True, index=True)
+    CompanyName = Column(String(200), nullable=True)
+    Address = Column(String(200), nullable=True)
+    City = Column(String(100), nullable=True)
+    PostalCode = Column(String(70), nullable=True)
+    WebAddress = Column(String(50), nullable=True)
+    Referral = Column(String(3), nullable=True)
+    CompanyTypeID = Column(Integer, nullable=True)
+    StateID = Column(Integer, nullable=True)
+    CountryID = Column(Integer, nullable=True)
+    LeadGeneratedFromID = Column(Integer, nullable=True)
+    SpecificSource = Column(String(30), nullable=True)
+    PriorityID = Column(Integer, nullable=True)
+    FollowupDate = Column(DateTime, nullable=True)
+    Purchase = Column(String(10), nullable=True)
+    VendorID = Column(String(10), nullable=True)
+    Enabled = Column(Boolean, nullable=False, default=True)
+    RentalType = Column(String(50), nullable=False, default='AB')

app/db/repositories/customer_sp_repo.py

@@ -0,0 +1,153 @@
+from typing import Tuple, List, Dict
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+class CustomerRepository:
+    """
+    Repository for customer DB access via stored procedure.
+    """
+    def __init__(self, db: Session):
+        self.db = db
+        self.engine = db.get_bind()
+
+    def list_customers_via_sp(
+        self,
+        order_by: str,
+        order_dir: str,
+        page: int,
+        page_size: int,
+    ) -> Tuple[List[Dict], int]:
+        """
+        Calls dbo.spAbCustomersGetList and returns (rows, total_records)
+        Falls back to direct SQL query if stored procedure doesn't return data.
+        Raises Exception on DB error.
+        """
+        try:
+            with self.engine.connect() as conn:
+                # Try the stored procedure first
+                try:
+                    data_query = text("""
+                        DECLARE @TotalRecords INT;
+                        EXEC dbo.spAbCustomersGetList
+                            @OrderBy = :order_by,
+                            @OrderDirection = :order_dir,
+                            @Page = :page,
+                            @PageSize = :page_size,
+                            @TotalRecords = @TotalRecords OUTPUT;
+                    """)
+                    
+                    result = conn.execute(data_query, {
+                        "order_by": order_by,
+                        "order_dir": order_dir,
+                        "page": page,
+                        "page_size": page_size,
+                    })
+                    
+                    # Get the customer rows
+                    rows = []
+                    if result.returns_rows:
+                        customer_rows = result.fetchall()
+                        if customer_rows:
+                            columns = result.keys()
+                            raw_rows = [dict(zip(columns, row)) for row in customer_rows]
+                            # Transform database column names to schema field names
+                            for raw_row in raw_rows:
+                                transformed_row = {
+                                    'id': raw_row.get('CustomerID'),
+                                    'name': raw_row.get('CompanyName'),
+                                    'address': raw_row.get('Address'),
+                                    'city': raw_row.get('City'),
+                                    'postal_code': raw_row.get('PostalCode'),
+                                    'web_address': raw_row.get('WebAddress'),
+                                    'referral': raw_row.get('Referral'),
+                                    'company_type_id': raw_row.get('CompanyTypeID'),
+                                    'state_id': raw_row.get('StateID'),
+                                    'country_id': raw_row.get('CountryID'),
+                                    'lead_generated_from_id': raw_row.get('LeadGeneratedFromID'),
+                                    'specific_source': raw_row.get('SpecificSource'),
+                                    'priority_id': raw_row.get('PriorityID'),
+                                    'followup_date': raw_row.get('FollowupDate'),
+                                    'purchase': raw_row.get('Purchase'),
+                                    'vendor_id': raw_row.get('VendorID'),
+                                    'enabled': raw_row.get('Enabled'),
+                                    'rental_type': raw_row.get('RentalType'),
+                                }
+                                rows.append(transformed_row)
+                    
+                    # If stored procedure worked and returned data, use it
+                    if rows:
+                        # Get total count
+                        count_query = text("SELECT COUNT(*) as total FROM dbo.Customers")
+                        count_result = conn.execute(count_query)
+                        total = count_result.fetchone()[0]
+                        return rows, int(total)
+                
+                except Exception as sp_error:
+                    print(f"Stored procedure failed: {sp_error}")
+                
+                # Fallback: Use direct SQL query
+                print("Falling back to direct SQL query")
+                
+                # Calculate offset
+                offset = (page - 1) * page_size
+                
+                # Build the order clause - ensure valid column name and direction
+                order_clause = f"{order_by} {order_dir.upper()}"
+                
+                # Direct query with pagination
+                query = text(f"""
+                    SELECT CustomerID, CompanyName, Address, City, PostalCode, 
+                           WebAddress, Referral, CompanyTypeID, StateID, CountryID,
+                           LeadGeneratedFromID, SpecificSource, PriorityID, 
+                           FollowupDate, Purchase, VendorID, Enabled, RentalType
+                    FROM dbo.Customers 
+                    ORDER BY {order_clause}
+                    OFFSET :offset ROWS 
+                    FETCH NEXT :page_size ROWS ONLY
+                """)
+                
+                result = conn.execute(query, {
+                    "offset": offset,
+                    "page_size": page_size
+                })
+                
+                rows = []
+                if result.returns_rows:
+                    customer_rows = result.fetchall()
+                    if customer_rows:
+                        columns = result.keys()
+                        raw_rows = [dict(zip(columns, row)) for row in customer_rows]
+                        # Transform database column names to schema field names
+                        rows = []
+                        for raw_row in raw_rows:
+                            transformed_row = {
+                                'id': raw_row.get('CustomerID'),
+                                'name': raw_row.get('CompanyName'),
+                                'address': raw_row.get('Address'),
+                                'city': raw_row.get('City'),
+                                'postal_code': raw_row.get('PostalCode'),
+                                'web_address': raw_row.get('WebAddress'),
+                                'referral': raw_row.get('Referral'),
+                                'company_type_id': raw_row.get('CompanyTypeID'),
+                                'state_id': raw_row.get('StateID'),
+                                'country_id': raw_row.get('CountryID'),
+                                'lead_generated_from_id': raw_row.get('LeadGeneratedFromID'),
+                                'specific_source': raw_row.get('SpecificSource'),
+                                'priority_id': raw_row.get('PriorityID'),
+                                'followup_date': raw_row.get('FollowupDate'),
+                                'purchase': raw_row.get('Purchase'),
+                                'vendor_id': raw_row.get('VendorID'),
+                                'enabled': raw_row.get('Enabled'),
+                                'rental_type': raw_row.get('RentalType'),
+                            }
+                            rows.append(transformed_row)
+                
+                # Get total count
+                count_query = text("SELECT COUNT(*) as total FROM dbo.Customers")
+                count_result = conn.execute(count_query)
+                total = count_result.fetchone()[0]
+                
+                return rows, int(total)
+                
+        except Exception as e:
+            raise Exception(f"DB error in list_customers_via_sp: {e}")

app/db/session.py

@@ -1,13 +1,35 @@
+# app/db/session.py
+import os
+import urllib.parse
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 from app.core.config import settings
 
-DATABASE_URL = (
-    f"mssql+pyodbc://{settings.SQLSERVER_USER}:{settings.SQLSERVER_PASSWORD}" 
-    f"@{settings.SQLSERVER_HOST}:{settings.SQLSERVER_PORT}/{settings.SQLSERVER_DB}?driver={settings.SQLSERVER_DRIVER.replace(' ', '+')}"
+USER = settings.SQLSERVER_USER
+PWD = settings.SQLSERVER_PASSWORD
+HOST = settings.SQLSERVER_HOST
+PORT = settings.SQLSERVER_PORT or 31433   # <- default to 31433 if not set
+DB = settings.SQLSERVER_DB
+
+odbc_str = (
+    "DRIVER={ODBC Driver 18 for SQL Server};"
+    f"SERVER=tcp:{HOST},{PORT};"
+    f"DATABASE={DB};"
+    f"UID={USER};PWD={PWD};"
+    "Encrypt=yes;"
+    "TrustServerCertificate=yes;"   # diagnostic: OK for dev. remove/flip in prod.
+    "LoginTimeout=15;"
+)
+params = urllib.parse.quote_plus(odbc_str)
+DATABASE_URL = f"mssql+pyodbc:///?odbc_connect={params}"
+
+engine = create_engine(
+    DATABASE_URL,
+    pool_pre_ping=True,
+    echo=False,
+    connect_args={"timeout": 30},
 )
 
-engine = create_engine(DATABASE_URL, pool_pre_ping=True)
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
 
 def get_db():
@@ -15,4 +37,4 @@ def get_db():
     try:
         yield db
     finally:
-        db.close()
+        db.close()

app/docker/docker-compose.yml

@@ -6,7 +6,7 @@ services:
       SA_PASSWORD: "YourStrong!Passw0rd"
       ACCEPT_EULA: "Y"
     ports:
-      - "1433:1433"
+      - "31433:31433"
     volumes:
       - mssql_data:/var/opt/mssql
   app:
@@ -17,7 +17,7 @@ services:
       SQLSERVER_USER: sa
       SQLSERVER_PASSWORD: YourStrong!Passw0rd
       SQLSERVER_HOST: mssql
-      SQLSERVER_PORT: 1433
+      SQLSERVER_PORT: 31433
       SQLSERVER_DB: aquabarrier
       SECRET_KEY: supersecretkey
     ports:

app/prompts/mvc_fastapi_sp_module.txt

@@ -0,0 +1,113 @@
+"""
+Prompt: Build an MVC-style FastAPI endpoint for listing entities via a SQL Server stored procedure.
+
+Requirements:
+- Use Pydantic for request/response models.
+- SQL Server stored procedure should support ordering, pagination, and return total records as an output parameter.
+- Follow layered architecture: controller/router → service → repository (DB access).
+- Use dependency injection for DB Session/engine.
+- Return paginated response as JSON: { "items": [...], "page": int, "page_size": int, "total": int }
+- Response model: PaginatedResponse[EntityOut] (generic or explicit schema).
+- Controller should accept query params: page (int, 1-indexed), page_size (int), order_by (str), order_dir (str: asc|desc).
+- Repository should call the stored procedure and fetch both rows and total count in a single DB call.
+- Service should validate/normalize query args, call repository, and compose DTOs/pagination object.
+- Controller should document query params and response.
+- Add error handling and docstrings throughout layers.
+- Provide unit/integration tests for endpoint and layers.
+
+CRITICAL: Database Schema Analysis Required
+Before implementing, you MUST:
+
+1. **Get Actual Table Structure**: 
+   - Request the CREATE TABLE script or INFORMATION_SCHEMA query results
+   - Identify exact column names (case-sensitive for SQL Server)
+   - Note data types and constraints
+   - Example: CustomerID vs customer_id vs id
+
+2. **Verify Stored Procedure Signature**:
+   - Get the stored procedure definition
+   - Confirm parameter names and types (@OrderBy, @OrderDirection, @Page, @PageSize, @TotalRecords OUTPUT)
+   - Test that it actually returns data rows
+
+3. **Column Name Mapping Strategy**:
+   - Create mapping between database columns and Pydantic schema fields
+   - Handle naming convention differences (PascalCase DB vs snake_case Python)
+   - Example mapping:
+     ```python
+     COLUMN_MAPPING = {
+         "id": "CustomerID",           # Schema field -> DB column
+         "name": "CompanyName", 
+         "created_at": "CustomerID",   # Fallback if column doesn't exist
+         "email": "EmailAddress"       # Map to actual DB column
+     }
+     ```
+
+4. **Data Transformation Layer**:
+   - Transform raw database results to match Pydantic schema
+   - Handle missing fields gracefully (set to None if not in DB)
+   - Example:
+     ```python
+     transformed_row = {
+         'id': raw_row.get('CustomerID'),
+         'name': raw_row.get('CompanyName'),
+         'email': raw_row.get('EmailAddress'),  # If exists
+         'phone': None,  # If not available in DB
+     }
+     ```
+
+5. **Fallback Strategy**:
+   - Implement direct SQL query fallback if stored procedure fails
+   - Use OFFSET/FETCH for pagination in SQL Server
+   - Separate total count query if needed
+
+6. **Common Pitfalls to Avoid**:
+   - Don't assume SQLAlchemy model columns match database columns
+   - Don't pass application field names directly to stored procedures
+   - Handle cases where stored procedures don't return rows properly
+   - Account for SQL Server naming conventions (PascalCase)
+   - Validate that order_by columns actually exist in the database
+
+Repository Implementation Pattern:
+```python
+def list_entities_via_sp(self, order_by: str, order_dir: str, page: int, page_size: int):
+    try:
+        # Try stored procedure first
+        result = conn.execute(stored_procedure_query, params)
+        
+        if result.returns_rows and result.fetchall():
+            # Transform database columns to schema fields
+            return transform_rows(result), get_total_count()
+        else:
+            # Fallback to direct SQL with proper column names
+            return direct_sql_query_with_pagination()
+    except Exception as e:
+        raise RepositoryException(f"Database error: {e}")
+```
+
+Service Layer Column Validation:
+```python
+class EntityListService:
+    ALLOWED_ORDER_BY = {"id", "name", "created_at", "updated_at"}
+    COLUMN_MAPPING = {
+        "id": "EntityID",
+        "name": "EntityName",
+        "created_at": "EntityID",  # Fallback
+    }
+    
+    def list_entities(self, order_by: str = None, ...):
+        order_by = order_by if order_by in self.ALLOWED_ORDER_BY else "id"
+        db_column = self.COLUMN_MAPPING.get(order_by, order_by)
+        return self.repo.list_entities_via_sp(db_column, ...)
+```
+
+Example API contract:
+GET /api/v1/entities?page=1&page_size=10&order_by=created_at&order_dir=desc
+Response: { "items": [EntityOut], "page": 1, "page_size": 10, "total": 123 }
+
+Testing Requirements:
+- Test with actual database table structure
+- Verify stored procedure calls work with real column names
+- Test fallback scenarios
+- Validate data transformation
+- Test edge cases (empty results, invalid columns)
+"""

app/schemas/customer.py

@@ -1,18 +1,45 @@
 from pydantic import BaseModel
 from typing import Optional
+from datetime import datetime
 
 class CustomerCreate(BaseModel):
     name: str
-    email: Optional[str] = None
-    phone: Optional[str] = None
     address: Optional[str] = None
+    city: Optional[str] = None
+    postal_code: Optional[str] = None
+    web_address: Optional[str] = None
+    referral: Optional[str] = None
+    company_type_id: Optional[int] = None
+    state_id: Optional[int] = None
+    country_id: Optional[int] = None
+    lead_generated_from_id: Optional[int] = None
+    specific_source: Optional[str] = None
+    priority_id: Optional[int] = None
+    followup_date: Optional[datetime] = None
+    purchase: Optional[str] = None
+    vendor_id: Optional[str] = None
+    enabled: Optional[bool] = True
+    rental_type: Optional[str] = "AB"
 
 class CustomerOut(BaseModel):
     id: int
-    name: str
-    email: Optional[str] = None
-    phone: Optional[str] = None
+    name: Optional[str] = None
     address: Optional[str] = None
+    city: Optional[str] = None
+    postal_code: Optional[str] = None
+    web_address: Optional[str] = None
+    referral: Optional[str] = None
+    company_type_id: Optional[int] = None
+    state_id: Optional[int] = None
+    country_id: Optional[int] = None
+    lead_generated_from_id: Optional[int] = None
+    specific_source: Optional[str] = None
+    priority_id: Optional[int] = None
+    followup_date: Optional[datetime] = None
+    purchase: Optional[str] = None
+    vendor_id: Optional[str] = None
+    enabled: Optional[bool] = None
+    rental_type: Optional[str] = None
 
     class Config:
         orm_mode = True

app/schemas/paginated_response.py

@@ -0,0 +1,13 @@
+from typing import Generic, TypeVar, List
+from pydantic import BaseModel
+
+T = TypeVar("T")
+
+class PaginatedResponse(BaseModel, Generic[T]):
+    items: List[T]
+    page: int
+    page_size: int
+    total: int
+
+    class Config:
+        orm_mode = True

app/services/customer_list_service.py

@@ -0,0 +1,74 @@
+from typing import List, Tuple
+from app.db.repositories.customer_sp_repo import CustomerRepository
+from app.schemas.customer import CustomerOut
+from app.schemas.paginated_response import PaginatedResponse
+
+class CustomerListService:
+    """
+    Service for listing customers with pagination and ordering via stored procedure.
+    Validates query params and calls repository.
+    """
+    ALLOWED_ORDER_BY = {
+        "id", "customer_id", "company_name", "name", "address", "city", 
+        "postal_code", "web_address", "referral", "company_type_id", 
+        "state_id", "country_id", "lead_generated_from_id", "specific_source", 
+        "priority_id", "followup_date", "purchase", "vendor_id", "enabled", 
+        "rental_type", "created_at"
+    }
+    ALLOWED_ORDER_DIR = {"asc", "desc"}
+    DEFAULT_ORDER_BY = "id"  # Changed from created_at to id since created_at doesn't exist
+    DEFAULT_ORDER_DIR = "desc"
+    
+    # Mapping from application field names to actual database column names
+    COLUMN_MAPPING = {
+        "id": "CustomerID",
+        "customer_id": "CustomerID",
+        "company_name": "CompanyName",
+        "name": "CompanyName",  # Alias for company_name
+        "address": "Address",
+        "city": "City",
+        "postal_code": "PostalCode",
+        "web_address": "WebAddress",
+        "referral": "Referral",
+        "company_type_id": "CompanyTypeID",
+        "state_id": "StateID",
+        "country_id": "CountryID",
+        "lead_generated_from_id": "LeadGeneratedFromID",
+        "specific_source": "SpecificSource",
+        "priority_id": "PriorityID",
+        "followup_date": "FollowupDate",
+        "purchase": "Purchase",
+        "vendor_id": "VendorID",
+        "enabled": "Enabled",
+        "rental_type": "RentalType",
+        "created_at": "CustomerID",  # Fallback to CustomerID since created_at doesn't exist in DB
+        "email": "CustomerID"  # Fallback to CustomerID since email doesn't exist in DB
+    }
+    DEFAULT_PAGE = 1
+    DEFAULT_PAGE_SIZE = 10
+    MAX_PAGE_SIZE = 100
+
+    def __init__(self, db):
+        self.repo = CustomerRepository(db)
+
+    def list_customers(self, page: int = 1, page_size: int = 10, order_by: str = None, order_dir: str = None) -> PaginatedResponse:
+        """
+        Returns paginated customer list and metadata.
+        Raises ValueError for invalid params or DB errors.
+        """
+        try:
+            page = max(page, 1)
+            page_size = min(max(page_size, 1), self.MAX_PAGE_SIZE)
+            order_by = order_by if order_by in self.ALLOWED_ORDER_BY else self.DEFAULT_ORDER_BY
+            order_dir = order_dir if order_dir in self.ALLOWED_ORDER_DIR else self.DEFAULT_ORDER_DIR
+
+            # Map the application field name to the database column name
+            db_order_by = self.COLUMN_MAPPING.get(order_by, order_by)
+
+            rows, total = self.repo.list_customers_via_sp(db_order_by, order_dir, page, page_size)
+            items = [CustomerOut(**row) for row in rows]
+            return PaginatedResponse[
+                CustomerOut
+            ](items=items, page=page, page_size=page_size, total=total)
+        except Exception as e:
+            raise ValueError(f"Failed to list customers: {e}")

app/tests/unit/test_customers_list.py

@@ -0,0 +1,45 @@
+import pytest
+from fastapi.testclient import TestClient
+from app.main import app
+
+client = TestClient(app)
+
+def test_list_customers_default():
+    resp = client.get("/api/v1/customers")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert "items" in data
+    assert "page" in data
+    assert "page_size" in data
+    assert "total" in data
+    assert isinstance(data["items"], list)
+    assert data["page"] == 1
+    assert data["page_size"] == 10
+
+def test_list_customers_pagination():
+    resp = client.get("/api/v1/customers?page=2&page_size=5")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["page"] == 2
+    assert data["page_size"] == 5
+
+def test_list_customers_ordering():
+    resp = client.get("/api/v1/customers?order_by=name&order_dir=asc")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["page"] == 1
+    assert data["page_size"] == 10
+
+def test_list_customers_invalid_order_by():
+    resp = client.get("/api/v1/customers?order_by=invalid_field")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["page"] == 1
+    assert data["page_size"] == 10
+
+def test_list_customers_invalid_order_dir():
+    resp = client.get("/api/v1/customers?order_dir=up")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert data["page"] == 1
+    assert data["page_size"] == 10