feat(data-platform): implement PRP-2 schema and migrations#11
Closed
w7-mgfcode wants to merge 2 commits into
Closed
feat(data-platform): implement PRP-2 schema and migrations#11w7-mgfcode wants to merge 2 commits into
w7-mgfcode wants to merge 2 commits into
Conversation
Create comprehensive Product Requirements Prompt for INITIAL-2 (Data Platform) including: - Mini-warehouse schema with dimension tables (store, product, calendar) - Fact tables (sales_daily, price_history, promotion, inventory_snapshot_daily) - SQLAlchemy 2.0 ORM model patterns with type annotations - Alembic migration guidance and constraint naming conventions - Grain protection via unique constraints for idempotent upserts - Composite indexes for time-series query optimization - Unit and integration test specifications - Example SQL queries for KPIs and exogenous feature joins Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Add mini-warehouse schema for retail demand forecasting: Dimension tables: - store: locations with code, region, city, store_type - product: catalog with sku, category, brand, base_price - calendar: time dimension with day_of_week, is_holiday Fact tables: - sales_daily: grain-protected (date, store_id, product_id) - price_history: validity windows with valid_from/valid_to - promotion: discount mechanics with pct and amount - inventory_snapshot_daily: stockout detection Key features: - SQLAlchemy 2.0 patterns (Mapped[], mapped_column()) - Grain protection via UniqueConstraint for idempotent upserts - Check constraints for data quality (positive qty, valid dates) - Composite indexes for time-range + store/product queries - Pydantic v2 schemas for API validation - 32 unit tests + 11 integration tests Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Reviewer's GuideImplements the PRP-2 retail data platform mini-warehouse: adds SQLAlchemy 2.0 ORM models for 3 dimension and 4 fact tables, a baseline Alembic migration, Pydantic v2 schemas, integration/unit tests for constraints and relationships, shared async DB fixtures, and example schema/query documentation. Entity relationship diagram for PRP2 data platform schemaerDiagram
Store {
int id PK
string code UK
string name
string region
string city
string store_type
}
Product {
int id PK
string sku UK
string name
string category
string brand
decimal base_price
decimal base_cost
}
Calendar {
date date PK
int day_of_week
int month
int quarter
int year
bool is_holiday
string holiday_name
}
SalesDaily {
int id PK
date date FK
int store_id FK
int product_id FK
int quantity
decimal unit_price
decimal total_amount
UNIQUE date_store_product
}
PriceHistory {
int id PK
int product_id FK
int store_id FK
decimal price
date valid_from
date valid_to
}
Promotion {
int id PK
int product_id FK
int store_id FK
string name
decimal discount_pct
decimal discount_amount
date start_date
date end_date
}
InventorySnapshotDaily {
int id PK
date date FK
int store_id FK
int product_id FK
int on_hand_qty
int on_order_qty
bool is_stockout
UNIQUE date_store_product
}
Store ||--o{ SalesDaily : has
Store ||--o{ PriceHistory : has
Store ||--o{ Promotion : has
Store ||--o{ InventorySnapshotDaily : has
Product ||--o{ SalesDaily : has
Product ||--o{ PriceHistory : has
Product ||--o{ Promotion : has
Product ||--o{ InventorySnapshotDaily : has
Calendar ||--o{ SalesDaily : has
Calendar ||--o{ InventorySnapshotDaily : has
Class diagram for new ORM models and Pydantic schemasclassDiagram
class Base
class TimestampMixin {
datetime created_at
datetime updated_at
}
class Store {
+int id
+str code
+str name
+str region
+str city
+str store_type
+list~SalesDaily~ sales
+list~PriceHistory~ price_history
+list~Promotion~ promotions
+list~InventorySnapshotDaily~ inventory_snapshots
}
class Product {
+int id
+str sku
+str name
+str category
+str brand
+Decimal base_price
+Decimal base_cost
+list~SalesDaily~ sales
+list~PriceHistory~ price_history
+list~Promotion~ promotions
+list~InventorySnapshotDaily~ inventory_snapshots
}
class Calendar {
+date date
+int day_of_week
+int month
+int quarter
+int year
+bool is_holiday
+str holiday_name
+list~SalesDaily~ sales
+list~InventorySnapshotDaily~ inventory_snapshots
}
class SalesDaily {
+int id
+date date
+int store_id
+int product_id
+int quantity
+Decimal unit_price
+Decimal total_amount
+Store store
+Product product
+Calendar calendar
}
class PriceHistory {
+int id
+int product_id
+int store_id
+Decimal price
+date valid_from
+date valid_to
+Product product
+Store store
}
class Promotion {
+int id
+int product_id
+int store_id
+str name
+Decimal discount_pct
+Decimal discount_amount
+date start_date
+date end_date
+Product product
+Store store
}
class InventorySnapshotDaily {
+int id
+date date
+int store_id
+int product_id
+int on_hand_qty
+int on_order_qty
+bool is_stockout
+Calendar calendar
+Store store
+Product product
}
TimestampMixin <|-- Store
Base <|-- Store
TimestampMixin <|-- Product
Base <|-- Product
TimestampMixin <|-- Calendar
Base <|-- Calendar
TimestampMixin <|-- SalesDaily
Base <|-- SalesDaily
TimestampMixin <|-- PriceHistory
Base <|-- PriceHistory
TimestampMixin <|-- Promotion
Base <|-- Promotion
TimestampMixin <|-- InventorySnapshotDaily
Base <|-- InventorySnapshotDaily
Store "1" o-- "*" SalesDaily : sales
Product "1" o-- "*" SalesDaily : sales
Calendar "1" o-- "*" SalesDaily : sales
Store "1" o-- "*" PriceHistory : price_history
Product "1" o-- "*" PriceHistory : price_history
Store "1" o-- "*" Promotion : promotions
Product "1" o-- "*" Promotion : promotions
Store "1" o-- "*" InventorySnapshotDaily : inventory_snapshots
Product "1" o-- "*" InventorySnapshotDaily : inventory_snapshots
Calendar "1" o-- "*" InventorySnapshotDaily : inventory_snapshots
class StoreBase {
+str code
+str name
+str region
+str city
+str store_type
}
class StoreCreate
class StoreRead {
+int id
}
class ProductBase {
+str sku
+str name
+str category
+str brand
+Decimal base_price
+Decimal base_cost
}
class ProductCreate
class ProductRead {
+int id
}
class CalendarBase {
+date date
+int day_of_week
+int month
+int quarter
+int year
+bool is_holiday
+str holiday_name
}
class CalendarCreate
class CalendarRead
class SalesDailyBase {
+date date
+int store_id
+int product_id
+int quantity
+Decimal unit_price
+Decimal total_amount
}
class SalesDailyCreate
class SalesDailyRead {
+int id
}
class PriceHistoryBase {
+int product_id
+int store_id
+Decimal price
+date valid_from
+date valid_to
}
class PriceHistoryCreate
class PriceHistoryRead {
+int id
}
class PromotionBase {
+int product_id
+int store_id
+str name
+Decimal discount_pct
+Decimal discount_amount
+date start_date
+date end_date
}
class PromotionCreate
class PromotionRead {
+int id
}
class InventorySnapshotDailyBase {
+date date
+int store_id
+int product_id
+int on_hand_qty
+int on_order_qty
+bool is_stockout
}
class InventorySnapshotDailyCreate
class InventorySnapshotDailyRead {
+int id
}
StoreBase <|-- StoreCreate
StoreBase <|-- StoreRead
ProductBase <|-- ProductCreate
ProductBase <|-- ProductRead
CalendarBase <|-- CalendarCreate
CalendarBase <|-- CalendarRead
SalesDailyBase <|-- SalesDailyCreate
SalesDailyBase <|-- SalesDailyRead
PriceHistoryBase <|-- PriceHistoryCreate
PriceHistoryBase <|-- PriceHistoryRead
PromotionBase <|-- PromotionCreate
PromotionBase <|-- PromotionRead
InventorySnapshotDailyBase <|-- InventorySnapshotDailyCreate
InventorySnapshotDailyBase <|-- InventorySnapshotDailyRead
StoreRead ..> Store : from_attributes
ProductRead ..> Product : from_attributes
CalendarRead ..> Calendar : from_attributes
SalesDailyRead ..> SalesDaily : from_attributes
PriceHistoryRead ..> PriceHistory : from_attributes
PromotionRead ..> Promotion : from_attributes
InventorySnapshotDailyRead ..> InventorySnapshotDaily : from_attributes
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
|
Caution Review failedThe pull request is closed. 📝 WalkthroughWalkthroughIntroduces a comprehensive Data Platform schema with SQLAlchemy ORM models implementing a star schema (3 dimensions: Store, Product, Calendar; 4 fact tables), Alembic migrations, Pydantic validation schemas, comprehensive tests validating constraints and models, and SQL query examples for analytics. Changes
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes Poem
✨ Finishing touches
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
Hey - I've found 3 issues, and left some high level feedback:
- There are two separate
db_sessionfixtures (intests/conftest.pyandapp/features/data_platform/tests/conftest.py) with nearly identical logic; consider consolidating to a single shared fixture to avoid divergence in how test databases are initialized and cleaned up. - The
db_sessionfixtures create tables withBase.metadata.create_all()while your production schema is managed via Alembic migrations; consider switching the test setup to runalembic upgrade headinstead so tests exercise the same schema (constraints/indexes) as deployed environments.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- There are two separate `db_session` fixtures (in `tests/conftest.py` and `app/features/data_platform/tests/conftest.py`) with nearly identical logic; consider consolidating to a single shared fixture to avoid divergence in how test databases are initialized and cleaned up.
- The `db_session` fixtures create tables with `Base.metadata.create_all()` while your production schema is managed via Alembic migrations; consider switching the test setup to run `alembic upgrade head` instead so tests exercise the same schema (constraints/indexes) as deployed environments.
## Individual Comments
### Comment 1
<location> `app/features/data_platform/models.py:211-212` </location>
<code_context>
+ valid_to: Mapped[date | None] = mapped_column(Date)
+
+ # Relationships
+ product: Mapped["Product"] = relationship(back_populates="price_history")
+ store: Mapped["Store | None"] = relationship(back_populates="price_history")
+
+ __table_args__ = (
</code_context>
<issue_to_address>
**issue (bug_risk):** Forward-ref type for `store` uses the `| None` syntax inside the string, which will not resolve correctly.
In SQLAlchemy, string-based annotations should contain only the model name, without typing operators. `Mapped["Store | None"]` will be treated as the literal name `"Store | None"` and won’t resolve to `Store`. Use `Mapped[Store | None]` (with `from __future__ import annotations`) or `Mapped["Store"]` and rely on `nullable=True` on the FK to represent optionality.
</issue_to_address>
### Comment 2
<location> `app/features/data_platform/tests/test_constraints.py:23-32` </location>
<code_context>
+from app.features.data_platform.models import Calendar, Product, SalesDaily, Store
+
+
+@pytest.mark.integration
+class TestSalesDailyConstraints:
+ """Integration tests for SalesDaily constraints."""
</code_context>
<issue_to_address>
**suggestion (testing):** Add integration tests for `PriceHistory` and `Promotion` database constraints.
The current integration tests cover constraints on `SalesDaily`, `InventorySnapshotDaily`, `Store`, `Product`, and `Calendar`, but none for `PriceHistory` or `Promotion`. To validate those schemas as well, please add tests that:
* `PriceHistory`
* Assert `price < 0` raises `IntegrityError` (`ck_price_history_price_positive`).
* Assert `valid_to < valid_from` raises `IntegrityError` (`ck_price_history_valid_dates`).
* Assert a valid row inserts successfully.
* `Promotion`
* Assert `end_date < start_date` raises `IntegrityError` (`ck_promotion_valid_dates`).
* Assert `discount_pct < 0` and `discount_pct > 1` raise `IntegrityError` (`ck_promotion_discount_pct_range`).
* Assert `discount_amount < 0` raises `IntegrityError` (`ck_promotion_discount_amount_positive`).
* Assert a valid row inserts successfully.
You can follow the pattern of `TestSalesDailyConstraints` and reuse `sample_store`, `sample_product`, and `sample_calendar` fixtures where needed.
Suggested implementation:
```python
from app.features.data_platform.models import (
Calendar,
InventorySnapshotDaily,
PriceHistory,
Product,
Promotion,
SalesDaily,
Store,
)
```
```python
@pytest.mark.integration
class TestPriceHistoryConstraints:
"""Integration tests for PriceHistory constraints."""
async def test_price_must_be_non_negative(
self,
db_session: AsyncSession,
sample_store: Store,
sample_product: Product,
sample_calendar: Calendar,
):
"""ck_price_history_price_positive: price < 0 should raise IntegrityError."""
price_history = PriceHistory(
store_id=sample_store.id,
product_id=sample_product.id,
calendar_id=sample_calendar.id,
price=-1,
valid_from=sample_calendar.date,
valid_to=sample_calendar.date,
)
db_session.add(price_history)
with pytest.raises(IntegrityError):
await db_session.flush()
async def test_valid_to_must_be_on_or_after_valid_from(
self,
db_session: AsyncSession,
sample_store: Store,
sample_product: Product,
sample_calendar: Calendar,
):
"""ck_price_history_valid_dates: valid_to < valid_from should raise IntegrityError."""
# Assuming sample_calendar provides a valid date; create an earlier date manually
earlier_date = sample_calendar.date - datetime.timedelta(days=1)
price_history = PriceHistory(
store_id=sample_store.id,
product_id=sample_product.id,
calendar_id=sample_calendar.id,
price=10,
valid_from=sample_calendar.date,
valid_to=earlier_date,
)
db_session.add(price_history)
with pytest.raises(IntegrityError):
await db_session.flush()
async def test_valid_price_history_inserts_successfully(
self,
db_session: AsyncSession,
sample_store: Store,
sample_product: Product,
sample_calendar: Calendar,
):
"""A valid PriceHistory row should insert successfully."""
price_history = PriceHistory(
store_id=sample_store.id,
product_id=sample_product.id,
calendar_id=sample_calendar.id,
price=10,
valid_from=sample_calendar.date,
valid_to=sample_calendar.date,
)
db_session.add(price_history)
await db_session.flush()
@pytest.mark.integration
class TestPromotionConstraints:
"""Integration tests for Promotion constraints."""
async def test_end_date_must_be_on_or_after_start_date(
self,
db_session: AsyncSession,
sample_store: Store,
sample_product: Product,
sample_calendar: Calendar,
):
"""ck_promotion_valid_dates: end_date < start_date should raise IntegrityError."""
later_date = sample_calendar.date + datetime.timedelta(days=1)
promotion = Promotion(
store_id=sample_store.id,
product_id=sample_product.id,
start_date=later_date,
end_date=sample_calendar.date,
discount_pct=0.1,
discount_amount=0,
)
db_session.add(promotion)
with pytest.raises(IntegrityError):
await db_session.flush()
async def test_discount_pct_must_be_between_0_and_1(
self,
db_session: AsyncSession,
sample_store: Store,
sample_product: Product,
sample_calendar: Calendar,
):
"""ck_promotion_discount_pct_range: discount_pct outside [0, 1] should raise IntegrityError."""
base_kwargs = dict(
store_id=sample_store.id,
product_id=sample_product.id,
start_date=sample_calendar.date,
end_date=sample_calendar.date,
discount_amount=0,
)
promotion_negative_pct = Promotion(discount_pct=-0.1, **base_kwargs)
db_session.add(promotion_negative_pct)
with pytest.raises(IntegrityError):
await db_session.flush()
await db_session.rollback()
promotion_over_one_pct = Promotion(discount_pct=1.1, **base_kwargs)
db_session.add(promotion_over_one_pct)
with pytest.raises(IntegrityError):
await db_session.flush()
async def test_discount_amount_must_be_non_negative(
self,
db_session: AsyncSession,
sample_store: Store,
sample_product: Product,
sample_calendar: Calendar,
):
"""ck_promotion_discount_amount_positive: discount_amount < 0 should raise IntegrityError."""
promotion = Promotion(
store_id=sample_store.id,
product_id=sample_product.id,
start_date=sample_calendar.date,
end_date=sample_calendar.date,
discount_pct=0,
discount_amount=-1,
)
db_session.add(promotion)
with pytest.raises(IntegrityError):
await db_session.flush()
async def test_valid_promotion_inserts_successfully(
self,
db_session: AsyncSession,
sample_store: Store,
sample_product: Product,
sample_calendar: Calendar,
):
"""A valid Promotion row should insert successfully."""
promotion = Promotion(
store_id=sample_store.id,
product_id=sample_product.id,
start_date=sample_calendar.date,
end_date=sample_calendar.date,
discount_pct=0.2,
discount_amount=0,
)
db_session.add(promotion)
await db_session.flush()
@pytest.mark.integration
class TestSalesDailyConstraints:
"""Integration tests for SalesDaily constraints."""
```
1. Ensure `datetime` and `IntegrityError` are imported at the top of the file if they are not already:
```python
import datetime
from sqlalchemy.exc import IntegrityError
```
2. If `PriceHistory` or `Promotion` use different field names or relationships (e.g., `store`, `product`, `calendar` relationships instead of `*_id` columns), adjust the constructor arguments to match the actual model definitions.
3. If the `sample_calendar` fixture exposes fields differently (e.g., `sample_calendar.date_id` vs `sample_calendar.date`), update `valid_from`, `valid_to`, `start_date`, and `end_date` accordingly.
</issue_to_address>
### Comment 3
<location> `app/features/data_platform/tests/__init__.py:1` </location>
<code_context>
+"""Tests for data platform feature."""
</code_context>
<issue_to_address>
**suggestion (testing):** Add unit tests for the Pydantic schemas in `schemas.py` to validate API-facing constraints.
I don’t see any tests exercising the Pydantic schemas in `app/features/data_platform/schemas.py`. Since they’re the API-facing validation layer, please add unit tests that cover:
* Field length/required-ness (e.g. `StoreBase.code` max_length, `name` non-empty).
* Numeric bounds (e.g. `base_price`/`base_cost` `ge=0`, `discount_pct` in `[0, 1]`, `on_hand_qty`/`on_order_qty` non-negative).
* Optional vs required IDs (e.g. `store_id` nullable vs positive when set).
* Any date consistency you choose to enforce at the schema level.
A small parametrized `test_schemas.py` that instantiates each `*Create` schema with valid data and a few invalid variants (asserting `ValidationError`) should be sufficient.
Suggested implementation:
```python
"""Tests for data platform feature schemas."""
from datetime import date, timedelta
import pytest
from pydantic import ValidationError
from app.features.data_platform import schemas
# --- Store schemas ---------------------------------------------------------
def test_store_create_valid():
"""StoreCreate accepts valid minimal payload."""
store = schemas.StoreCreate(
code="STORE_001",
name="Main store",
)
assert store.code == "STORE_001"
assert store.name == "Main store"
@pytest.mark.parametrize(
"field, value",
[
("code", ""), # empty code should be rejected by max_length/min_length or custom validators
("code", "x" * 256), # past a typical max_length=255
("name", ""), # name should be non-empty
],
)
def test_store_create_string_constraints(field, value):
"""StoreCreate enforces code max length and name non-emptiness."""
base_payload = {
"code": "STORE_001",
"name": "Main store",
}
base_payload[field] = value
with pytest.raises(ValidationError):
schemas.StoreCreate(**base_payload)
# --- Product / pricing schemas --------------------------------------------
def _valid_product_payload():
"""Helper: minimally valid product payload for ProductCreate-like schema."""
return {
"sku": "SKU-123",
"name": "Test product",
"base_price": 10.0,
"base_cost": 5.0,
"discount_pct": 0.25,
}
def test_product_create_valid():
"""ProductCreate accepts valid numeric ranges."""
product = schemas.ProductCreate(**_valid_product_payload())
assert product.base_price == 10.0
assert product.base_cost == 5.0
assert 0 <= product.discount_pct <= 1
@pytest.mark.parametrize(
"field, value",
[
("base_price", -0.01),
("base_cost", -1),
("discount_pct", -0.01),
("discount_pct", 1.01),
],
)
def test_product_create_numeric_bounds(field, value):
"""ProductCreate enforces non-negative prices and discount_pct in [0, 1]."""
payload = _valid_product_payload()
payload[field] = value
with pytest.raises(ValidationError):
schemas.ProductCreate(**payload)
# --- Inventory / stock schemas --------------------------------------------
def _valid_inventory_payload():
"""Helper: minimally valid inventory payload for InventoryCreate-like schema."""
today = date.today()
return {
"sku": "SKU-123",
"store_id": None,
"on_hand_qty": 0,
"on_order_qty": 0,
"effective_date": today,
"end_date": today + timedelta(days=1),
}
def test_inventory_create_valid_with_null_store_id():
"""InventoryCreate allows nullable store_id."""
inv = schemas.InventoryCreate(**_valid_inventory_payload())
assert inv.store_id is None
assert inv.on_hand_qty == 0
assert inv.on_order_qty == 0
assert inv.effective_date <= inv.end_date
def test_inventory_create_valid_with_positive_store_id():
"""InventoryCreate accepts positive store_id when provided."""
payload = _valid_inventory_payload()
payload["store_id"] = 123
inv = schemas.InventoryCreate(**payload)
assert inv.store_id == 123
@pytest.mark.parametrize("field", ["on_hand_qty", "on_order_qty"])
def test_inventory_create_non_negative_quantities(field):
"""InventoryCreate enforces non-negative inventory quantities."""
payload = _valid_inventory_payload()
payload[field] = -1
with pytest.raises(ValidationError):
schemas.InventoryCreate(**payload)
def test_inventory_create_store_id_positive_when_set():
"""InventoryCreate requires store_id, when set, to be positive."""
payload = _valid_inventory_payload()
payload["store_id"] = 0
with pytest.raises(ValidationError):
schemas.InventoryCreate(**payload)
def test_inventory_create_date_consistency():
"""InventoryCreate enforces effective_date <= end_date."""
today = date.today()
payload = _valid_inventory_payload()
payload["effective_date"] = today
payload["end_date"] = today - timedelta(days=1)
with pytest.raises(ValidationError):
schemas.InventoryCreate(**payload)
```
These tests assume the following schema names and fields in `app/features/data_platform/schemas.py`:
- `StoreCreate` with fields: `code` (str with max_length/min_length), `name` (non-empty str).
- `ProductCreate` with fields: `sku` (str), `name` (str), `base_price` (float, `ge=0`), `base_cost` (float, `ge=0`), `discount_pct` (float between 0 and 1 inclusive).
- `InventoryCreate` with fields: `sku` (str), `store_id` (Optional[int], must be positive when not None), `on_hand_qty` and `on_order_qty` (non-negative numeric), `effective_date` and `end_date` (dates with a validator enforcing `effective_date <= end_date`).
If your actual schema names or fields differ, adjust the imports and field names in these tests accordingly, or add the corresponding `*Create` schemas and Pydantic field constraints so these tests pass.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+211
to
+212
| product: Mapped["Product"] = relationship(back_populates="price_history") | ||
| store: Mapped["Store | None"] = relationship(back_populates="price_history") |
There was a problem hiding this comment.
issue (bug_risk): Forward-ref type for store uses the | None syntax inside the string, which will not resolve correctly.
In SQLAlchemy, string-based annotations should contain only the model name, without typing operators. Mapped["Store | None"] will be treated as the literal name "Store | None" and won’t resolve to Store. Use Mapped[Store | None] (with from __future__ import annotations) or Mapped["Store"] and rely on nullable=True on the FK to represent optionality.
Comment on lines
+23
to
+32
| @pytest.mark.integration | ||
| class TestSalesDailyConstraints: | ||
| """Integration tests for SalesDaily constraints.""" | ||
|
|
||
| async def test_unique_constraint_prevents_duplicates( | ||
| self, | ||
| db_session: AsyncSession, | ||
| sample_store: Store, | ||
| sample_product: Product, | ||
| sample_calendar: Calendar, |
There was a problem hiding this comment.
suggestion (testing): Add integration tests for PriceHistory and Promotion database constraints.
The current integration tests cover constraints on SalesDaily, InventorySnapshotDaily, Store, Product, and Calendar, but none for PriceHistory or Promotion. To validate those schemas as well, please add tests that:
PriceHistory- Assert
price < 0raisesIntegrityError(ck_price_history_price_positive). - Assert
valid_to < valid_fromraisesIntegrityError(ck_price_history_valid_dates). - Assert a valid row inserts successfully.
- Assert
Promotion- Assert
end_date < start_dateraisesIntegrityError(ck_promotion_valid_dates). - Assert
discount_pct < 0anddiscount_pct > 1raiseIntegrityError(ck_promotion_discount_pct_range). - Assert
discount_amount < 0raisesIntegrityError(ck_promotion_discount_amount_positive). - Assert a valid row inserts successfully.
- Assert
You can follow the pattern of TestSalesDailyConstraints and reuse sample_store, sample_product, and sample_calendar fixtures where needed.
Suggested implementation:
from app.features.data_platform.models import (
Calendar,
InventorySnapshotDaily,
PriceHistory,
Product,
Promotion,
SalesDaily,
Store,
)@pytest.mark.integration
class TestPriceHistoryConstraints:
"""Integration tests for PriceHistory constraints."""
async def test_price_must_be_non_negative(
self,
db_session: AsyncSession,
sample_store: Store,
sample_product: Product,
sample_calendar: Calendar,
):
"""ck_price_history_price_positive: price < 0 should raise IntegrityError."""
price_history = PriceHistory(
store_id=sample_store.id,
product_id=sample_product.id,
calendar_id=sample_calendar.id,
price=-1,
valid_from=sample_calendar.date,
valid_to=sample_calendar.date,
)
db_session.add(price_history)
with pytest.raises(IntegrityError):
await db_session.flush()
async def test_valid_to_must_be_on_or_after_valid_from(
self,
db_session: AsyncSession,
sample_store: Store,
sample_product: Product,
sample_calendar: Calendar,
):
"""ck_price_history_valid_dates: valid_to < valid_from should raise IntegrityError."""
# Assuming sample_calendar provides a valid date; create an earlier date manually
earlier_date = sample_calendar.date - datetime.timedelta(days=1)
price_history = PriceHistory(
store_id=sample_store.id,
product_id=sample_product.id,
calendar_id=sample_calendar.id,
price=10,
valid_from=sample_calendar.date,
valid_to=earlier_date,
)
db_session.add(price_history)
with pytest.raises(IntegrityError):
await db_session.flush()
async def test_valid_price_history_inserts_successfully(
self,
db_session: AsyncSession,
sample_store: Store,
sample_product: Product,
sample_calendar: Calendar,
):
"""A valid PriceHistory row should insert successfully."""
price_history = PriceHistory(
store_id=sample_store.id,
product_id=sample_product.id,
calendar_id=sample_calendar.id,
price=10,
valid_from=sample_calendar.date,
valid_to=sample_calendar.date,
)
db_session.add(price_history)
await db_session.flush()
@pytest.mark.integration
class TestPromotionConstraints:
"""Integration tests for Promotion constraints."""
async def test_end_date_must_be_on_or_after_start_date(
self,
db_session: AsyncSession,
sample_store: Store,
sample_product: Product,
sample_calendar: Calendar,
):
"""ck_promotion_valid_dates: end_date < start_date should raise IntegrityError."""
later_date = sample_calendar.date + datetime.timedelta(days=1)
promotion = Promotion(
store_id=sample_store.id,
product_id=sample_product.id,
start_date=later_date,
end_date=sample_calendar.date,
discount_pct=0.1,
discount_amount=0,
)
db_session.add(promotion)
with pytest.raises(IntegrityError):
await db_session.flush()
async def test_discount_pct_must_be_between_0_and_1(
self,
db_session: AsyncSession,
sample_store: Store,
sample_product: Product,
sample_calendar: Calendar,
):
"""ck_promotion_discount_pct_range: discount_pct outside [0, 1] should raise IntegrityError."""
base_kwargs = dict(
store_id=sample_store.id,
product_id=sample_product.id,
start_date=sample_calendar.date,
end_date=sample_calendar.date,
discount_amount=0,
)
promotion_negative_pct = Promotion(discount_pct=-0.1, **base_kwargs)
db_session.add(promotion_negative_pct)
with pytest.raises(IntegrityError):
await db_session.flush()
await db_session.rollback()
promotion_over_one_pct = Promotion(discount_pct=1.1, **base_kwargs)
db_session.add(promotion_over_one_pct)
with pytest.raises(IntegrityError):
await db_session.flush()
async def test_discount_amount_must_be_non_negative(
self,
db_session: AsyncSession,
sample_store: Store,
sample_product: Product,
sample_calendar: Calendar,
):
"""ck_promotion_discount_amount_positive: discount_amount < 0 should raise IntegrityError."""
promotion = Promotion(
store_id=sample_store.id,
product_id=sample_product.id,
start_date=sample_calendar.date,
end_date=sample_calendar.date,
discount_pct=0,
discount_amount=-1,
)
db_session.add(promotion)
with pytest.raises(IntegrityError):
await db_session.flush()
async def test_valid_promotion_inserts_successfully(
self,
db_session: AsyncSession,
sample_store: Store,
sample_product: Product,
sample_calendar: Calendar,
):
"""A valid Promotion row should insert successfully."""
promotion = Promotion(
store_id=sample_store.id,
product_id=sample_product.id,
start_date=sample_calendar.date,
end_date=sample_calendar.date,
discount_pct=0.2,
discount_amount=0,
)
db_session.add(promotion)
await db_session.flush()
@pytest.mark.integration
class TestSalesDailyConstraints:
"""Integration tests for SalesDaily constraints."""- Ensure
datetimeandIntegrityErrorare imported at the top of the file if they are not already:
import datetime
from sqlalchemy.exc import IntegrityError- If
PriceHistoryorPromotionuse different field names or relationships (e.g.,store,product,calendarrelationships instead of*_idcolumns), adjust the constructor arguments to match the actual model definitions. - If the
sample_calendarfixture exposes fields differently (e.g.,sample_calendar.date_idvssample_calendar.date), updatevalid_from,valid_to,start_date, andend_dateaccordingly.
| @@ -0,0 +1 @@ | |||
| """Tests for data platform feature.""" | |||
There was a problem hiding this comment.
suggestion (testing): Add unit tests for the Pydantic schemas in schemas.py to validate API-facing constraints.
I don’t see any tests exercising the Pydantic schemas in app/features/data_platform/schemas.py. Since they’re the API-facing validation layer, please add unit tests that cover:
- Field length/required-ness (e.g.
StoreBase.codemax_length,namenon-empty). - Numeric bounds (e.g.
base_price/base_costge=0,discount_pctin[0, 1],on_hand_qty/on_order_qtynon-negative). - Optional vs required IDs (e.g.
store_idnullable vs positive when set). - Any date consistency you choose to enforce at the schema level.
A small parametrized test_schemas.py that instantiates each *Create schema with valid data and a few invalid variants (asserting ValidationError) should be sufficient.
Suggested implementation:
"""Tests for data platform feature schemas."""
from datetime import date, timedelta
import pytest
from pydantic import ValidationError
from app.features.data_platform import schemas
# --- Store schemas ---------------------------------------------------------
def test_store_create_valid():
"""StoreCreate accepts valid minimal payload."""
store = schemas.StoreCreate(
code="STORE_001",
name="Main store",
)
assert store.code == "STORE_001"
assert store.name == "Main store"
@pytest.mark.parametrize(
"field, value",
[
("code", ""), # empty code should be rejected by max_length/min_length or custom validators
("code", "x" * 256), # past a typical max_length=255
("name", ""), # name should be non-empty
],
)
def test_store_create_string_constraints(field, value):
"""StoreCreate enforces code max length and name non-emptiness."""
base_payload = {
"code": "STORE_001",
"name": "Main store",
}
base_payload[field] = value
with pytest.raises(ValidationError):
schemas.StoreCreate(**base_payload)
# --- Product / pricing schemas --------------------------------------------
def _valid_product_payload():
"""Helper: minimally valid product payload for ProductCreate-like schema."""
return {
"sku": "SKU-123",
"name": "Test product",
"base_price": 10.0,
"base_cost": 5.0,
"discount_pct": 0.25,
}
def test_product_create_valid():
"""ProductCreate accepts valid numeric ranges."""
product = schemas.ProductCreate(**_valid_product_payload())
assert product.base_price == 10.0
assert product.base_cost == 5.0
assert 0 <= product.discount_pct <= 1
@pytest.mark.parametrize(
"field, value",
[
("base_price", -0.01),
("base_cost", -1),
("discount_pct", -0.01),
("discount_pct", 1.01),
],
)
def test_product_create_numeric_bounds(field, value):
"""ProductCreate enforces non-negative prices and discount_pct in [0, 1]."""
payload = _valid_product_payload()
payload[field] = value
with pytest.raises(ValidationError):
schemas.ProductCreate(**payload)
# --- Inventory / stock schemas --------------------------------------------
def _valid_inventory_payload():
"""Helper: minimally valid inventory payload for InventoryCreate-like schema."""
today = date.today()
return {
"sku": "SKU-123",
"store_id": None,
"on_hand_qty": 0,
"on_order_qty": 0,
"effective_date": today,
"end_date": today + timedelta(days=1),
}
def test_inventory_create_valid_with_null_store_id():
"""InventoryCreate allows nullable store_id."""
inv = schemas.InventoryCreate(**_valid_inventory_payload())
assert inv.store_id is None
assert inv.on_hand_qty == 0
assert inv.on_order_qty == 0
assert inv.effective_date <= inv.end_date
def test_inventory_create_valid_with_positive_store_id():
"""InventoryCreate accepts positive store_id when provided."""
payload = _valid_inventory_payload()
payload["store_id"] = 123
inv = schemas.InventoryCreate(**payload)
assert inv.store_id == 123
@pytest.mark.parametrize("field", ["on_hand_qty", "on_order_qty"])
def test_inventory_create_non_negative_quantities(field):
"""InventoryCreate enforces non-negative inventory quantities."""
payload = _valid_inventory_payload()
payload[field] = -1
with pytest.raises(ValidationError):
schemas.InventoryCreate(**payload)
def test_inventory_create_store_id_positive_when_set():
"""InventoryCreate requires store_id, when set, to be positive."""
payload = _valid_inventory_payload()
payload["store_id"] = 0
with pytest.raises(ValidationError):
schemas.InventoryCreate(**payload)
def test_inventory_create_date_consistency():
"""InventoryCreate enforces effective_date <= end_date."""
today = date.today()
payload = _valid_inventory_payload()
payload["effective_date"] = today
payload["end_date"] = today - timedelta(days=1)
with pytest.raises(ValidationError):
schemas.InventoryCreate(**payload)These tests assume the following schema names and fields in app/features/data_platform/schemas.py:
StoreCreatewith fields:code(str with max_length/min_length),name(non-empty str).ProductCreatewith fields:sku(str),name(str),base_price(float,ge=0),base_cost(float,ge=0),discount_pct(float between 0 and 1 inclusive).InventoryCreatewith fields:sku(str),store_id(Optional[int], must be positive when not None),on_hand_qtyandon_order_qty(non-negative numeric),effective_dateandend_date(dates with a validator enforcingeffective_date <= end_date).
If your actual schema names or fields differ, adjust the imports and field names in these tests accordingly, or add the corresponding *Create schemas and Pydantic field constraints so these tests pass.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Implements PRP-2: Data Platform Schema + Migrations for the retail demand forecasting mini-warehouse.
Tables Created
Dimensions:
store,product,calendarFacts:
sales_daily,price_history,promotion,inventory_snapshot_dailyKey Features
UNIQUE(date, store_id, product_id)on sales_dailyMapped[],mapped_column())Test plan
uv run ruff check app/features/data_platform/passesuv run mypy app/features/data_platform/*.pypasses (0 errors)uv run pyright app/features/data_platform/passes (0 errors)uv run pytest app/features/data_platform/tests/test_models.py -v(32 passed)uv run pytest -m integration(11 passed)uv run alembic upgrade headsucceedsuv run alembic downgrade -1 && uv run alembic upgrade head(reversible)🤖 Generated with Claude Code
Summary by Sourcery
Introduce a new data platform vertical slice with a retail forecasting mini-warehouse schema, migrations, validation schemas, tests, and example queries.
New Features:
Enhancements:
Documentation:
Tests:
Summary by CodeRabbit
Release Notes
New Features
Documentation
✏️ Tip: You can customize this high-level summary in your review settings.